From 6abe1e8f1d5b2ff49f63cf22f29e576b48d994a4 Mon Sep 17 00:00:00 2001
From: dawn <90008@klbr.net>
Date: Tue, 6 Oct 2026 15:49:07 +0300
Subject: [PATCH] [crawler] test the relay-first pass marker and cursor display
 reads

relay_first::read_u64 and ListReposProducer::next_completion read the
relay-first pass counters as big endian u64s, and cursor_display decodes
the msgpack crawler cursor, all through codec functions passed as values
that the representation audit couldn't see. the new tests store a known
marker or cursor and check what each reader gets back, including the
error or fallback for a malformed value. flipping the reads to
from_le_bytes, or the cursor decode to json, fails them.
---
 src/crawler/list_repos/producer.rs | 69 +++++++++++++++++++++++++++++-
 src/crawler/relay_first.rs         | 18 ++++++++
 2 files changed, 85 insertions(+), 2 deletions(-)

diff --git a/src/crawler/list_repos/producer.rs b/src/crawler/list_repos/producer.rs
index 43c34fa..570dc58 100644
--- a/src/crawler/list_repos/producer.rs
+++ b/src/crawler/list_repos/producer.rs
@@ -345,8 +345,8 @@ fn pass_cursor_updates(
 #[cfg(test)]
 mod tests {
     use super::{
-        ListReposProducer, next_completion_number, pass_cursor_updates, should_enqueue_unknown,
-        should_persist_inactive_unknown,
+        ListReposProducer, cursor_display, next_completion_number, pass_cursor_updates,
+        should_enqueue_unknown, should_persist_inactive_unknown,
     };
     use crate::config::Config;
     use crate::crawler::{CrawlerStats, CrawlerWorker, InFlight, SignalChecker};
@@ -406,6 +406,71 @@ mod tests {
         assert!(next_completion_number(Some(u64::MAX), None).is_err());
     }
 
+    #[tokio::test]
+    async fn relay_first_marker_continues_from_the_stored_big_endian_pass() -> Result<()> {
+        let tmp = tempdir().into_diagnostic()?;
+        let state = Arc::new(AppState::new(&Config {
+            database_path: tmp.path().to_path_buf(),
+            ..Default::default()
+        })?);
+        let pass_key = keys::relay_first_pass_key("http://relay.example/");
+        state
+            .db
+            .cursors
+            .insert(&pass_key, [0, 0, 0, 0, 0, 0, 1, 2])
+            .into_diagnostic()?;
+        let (tx, _rx) = tokio::sync::mpsc::channel(1);
+        let mut producer = ListReposProducer {
+            url: Url::parse("http://relay.example/").into_diagnostic()?,
+            checker: SignalChecker {
+                http: crate::net::PublicHttpClient::test(reqwest::Client::new()),
+                state: state.clone(),
+                throttler: state.throttler.clone(),
+            },
+            in_flight: InFlight::new(),
+            tx,
+            enabled: state.crawler_enabled.subscribe(),
+            stats: CrawlerStats::new(state.clone(), Vec::new(), state.throttler.clone()),
+            completion_key: Some(pass_key.clone()),
+            last_completion: None,
+        };
+        assert_eq!(
+            producer.next_completion(true).await?,
+            Some((pass_key.clone(), 259))
+        );
+
+        state
+            .db
+            .cursors
+            .insert(&pass_key, [0; 7])
+            .into_diagnostic()?;
+        assert!(producer.next_completion(true).await.is_err());
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn crawler_cursor_display_decodes_the_msgpack_cursor() -> Result<()> {
+        let tmp = tempdir().into_diagnostic()?;
+        let state = AppState::new(&Config {
+            database_path: tmp.path().to_path_buf(),
+            ..Default::default()
+        })?;
+        let relay = Url::parse("https://relay.example/").into_diagnostic()?;
+        assert_eq!(cursor_display(&state, &relay).await, "none");
+
+        let key = keys::crawler_cursor_key(relay.as_str());
+        state
+            .db
+            .cursors
+            .insert(&key, b"\xa3abc")
+            .into_diagnostic()?;
+        assert_eq!(cursor_display(&state, &relay).await, "abc");
+
+        state.db.cursors.insert(&key, b"abc").into_diagnostic()?;
+        assert_eq!(cursor_display(&state, &relay).await, "none");
+        Ok(())
+    }
+
     #[tokio::test]
     async fn relay_first_pass_persists_relay_inactive_unknown_as_authoritative() -> Result<()> {
         const DID: &str = "did:web:relay-inactive.example";
diff --git a/src/crawler/relay_first.rs b/src/crawler/relay_first.rs
index 3516d16..7c7c7c2 100644
--- a/src/crawler/relay_first.rs
+++ b/src/crawler/relay_first.rs
@@ -392,6 +392,24 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn pass_markers_are_read_big_endian() -> Result<()> {
+        let tmp = tempdir().into_diagnostic()?;
+        let db = crate::db::Db::open(&Config {
+            database_path: tmp.path().to_path_buf(),
+            ..Default::default()
+        })?;
+        let key = keys::relay_first_pass_key("https://relay.example/");
+        assert_eq!(read_u64(&db.cursors, &key)?, None);
+        db.cursors
+            .insert(&key, [0, 0, 0, 0, 0, 0, 1, 2])
+            .into_diagnostic()?;
+        assert_eq!(read_u64(&db.cursors, &key)?, Some(258));
+        db.cursors.insert(&key, [0; 7]).into_diagnostic()?;
+        assert!(read_u64(&db.cursors, &key).is_err());
+        Ok(())
+    }
+
     #[tokio::test]
     async fn pds_pages_resume_and_commit_only_after_terminal_page() -> Result<()> {
         let requests = Arc::new(AtomicUsize::new(0));
-- 
2.51.2