From 465a88e4e5b1d7c5473d34c9f34361e7ac34b309 Mon Sep 17 00:00:00 2001 From: tomdebone Date: Wed, 9 Sep 2026 21:36:47 +0200 Subject: [PATCH] test(appview): handle-sync-Tests vom globalen DB-Zustand entkoppeln MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fünf handle_sync-Tests schlugen fehl, sobald der Indexer parallel lief oder die Datenbank schon andere handle-lose Posts enthielt: run_once() scannt global, geordnet nach did und gedeckelt auf BATCH_SIZE, also landete der frisch geseedete Test-DID schlicht nicht im Batch — die Assertions über report.resolved sagten dann etwas über fremde Zeilen aus. run_once() ist jetzt select_candidates() + resolve_batch(dids); das Verhalten in Produktion ist unverändert. Die Dispatch-Tests treiben resolve_batch mit ihrem eigenen DID, der Batch-Limit-Test prüft den Deckel dort, wo er sitzt (im SELECT), statt ihn aus einem globalen Report abzuleiten. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_013HC9HLrUU1LNwkzp8nkDLX --- crates/appview/src/handle_sync.rs | 68 +++++++++++++++++++++++++------ 1 file changed, 55 insertions(+), 13 deletions(-) diff --git a/crates/appview/src/handle_sync.rs b/crates/appview/src/handle_sync.rs index 229b111..d7976bb 100644 --- a/crates/appview/src/handle_sync.rs +++ b/crates/appview/src/handle_sync.rs @@ -147,7 +147,14 @@ impl HandleSyncWorker { /// `handle` is filled, so a DID that becomes resolvable later /// (e.g. the user joins the local PDS) gets re-attempted. pub async fn run_once(&self) -> Result { - let dids: Vec<(String,)> = sqlx::query_as( + let dids = self.select_candidates().await?; + self.resolve_batch(dids).await + } + + /// The SELECT half of [`Self::run_once`]: up to [`BATCH_SIZE`] + /// distinct DIDs still waiting for a handle. + async fn select_candidates(&self) -> Result> { + let rows: Vec<(String,)> = sqlx::query_as( r#"SELECT DISTINCT did FROM posts WHERE handle = '' @@ -159,7 +166,18 @@ impl HandleSyncWorker { .bind(BATCH_SIZE) .fetch_all(&self.db) .await?; + Ok(rows.into_iter().map(|(did,)| did).collect()) + } + /// The resolve-and-write half of [`Self::run_once`], split out so + /// tests can drive it with a DID set of their own. + /// + /// `run_once`'s own scan is global and capped at [`BATCH_SIZE`], + /// so against a shared database with a live indexer a test's + /// freshly seeded DID may simply not make the batch — which made + /// the dispatch tests fail for reasons that had nothing to do with + /// dispatch. Passing the DIDs in removes that coupling. + async fn resolve_batch(&self, dids: Vec) -> Result { let mut report = SyncReport::default(); if dids.is_empty() { return Ok(report); @@ -174,7 +192,7 @@ impl HandleSyncWorker { // the parallel-write benefit at this batch size. use futures::stream::{self, StreamExt}; let dispatch_results: Vec<(String, anyhow::Result>)> = stream::iter(dids) - .map(|(did,)| async move { + .map(|did| async move { let r = self.dispatch(&did).await; (did, r) }) @@ -376,7 +394,11 @@ mod tests { .into_arc(); let worker = worker_with(db, resolver); - let report = worker.run_once().await.unwrap(); + // Drive the resolve half with our own DID instead of + // `run_once()`: the global scan is capped at BATCH_SIZE and + // this database is shared with a live indexer, so a freshly + // seeded DID is not guaranteed to make the batch. + let report = worker.resolve_batch(vec![did.clone()]).await.unwrap(); assert!(report.resolved >= 2, "expected ≥2 resolved, got {report:?}"); assert_eq!(report.failed, 0); assert_eq!(report.skipped, 0); @@ -411,10 +433,14 @@ mod tests { .into_arc(); let worker = worker_with(db, resolver); - let report = worker.run_once().await.unwrap(); - assert_eq!(report.resolved, 0); - assert_eq!(report.failed, 0); - assert_eq!(report.skipped, 0); // DID was already filtered out by the SELECT + // The SELECT is what filters an already-handled DID out, so + // this one goes through the full `run_once()` — but scoped to + // the assertion that OUR did is untouched, not to global counts. + let candidates = worker.select_candidates().await.unwrap(); + assert!( + !candidates.contains(&did), + "a DID with a handle must not be selected" + ); let h = get_handle(&worker.db, &did).await; assert_eq!(h.as_deref(), Some("pre-existing.handle")); @@ -456,8 +482,22 @@ mod tests { web_resolver: Arc::clone(&resolver), interval_secs: 999, }; - let report = worker.run_once().await.unwrap(); - // Exactly BATCH_SIZE rows updated (one post per DID). + // The cap lives in the SELECT, so that's what we assert on. + // Asserting `report.resolved == BATCH_SIZE` after a full + // `run_once()` only holds on a database where nothing else is + // waiting for a handle — against the shared dev DB with a live + // Jetstream indexer, other DIDs legitimately fill the batch. + let candidates = worker.select_candidates().await.unwrap(); + assert!( + candidates.len() as i64 <= BATCH_SIZE, + "select must never exceed BATCH_SIZE, got {}", + candidates.len() + ); + + // And the resolve half must handle a full batch of our own + // DIDs: one row per DID, all of them updated. + let batch: Vec = all_dids.iter().take(BATCH_SIZE as usize).cloned().collect(); + let report = worker.resolve_batch(batch).await.unwrap(); assert_eq!( report.resolved as i64, BATCH_SIZE, @@ -465,7 +505,7 @@ mod tests { ); assert_eq!(report.failed, 0); - // The remaining 5 DIDs must still have empty handles. + // The 5 DIDs we left out must still have empty handles. let remaining: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM posts WHERE handle = '' AND did LIKE $1") .bind(format!("{prefix}_%")) @@ -528,7 +568,9 @@ mod tests { .await .unwrap(); - let report = worker.run_once().await.unwrap(); + // Scoped to our own DID: `run_once()`'s global batch would + // report whatever else the indexer left pending. + let report = worker.resolve_batch(vec![did.clone()]).await.unwrap(); assert_eq!(report.resolved, 0, "must not touch already-handled rows"); let h = get_handle(&worker.db, &did).await; assert_eq!(h.as_deref(), Some("from-ingest")); @@ -571,7 +613,7 @@ mod tests { web_resolver: web, interval_secs: 999, }; - let report = worker.run_once().await.unwrap(); + let report = worker.resolve_batch(vec![did.clone()]).await.unwrap(); assert_eq!( report.resolved, 1, "did:web must resolve through the web resolver, got {report:?}" @@ -630,7 +672,7 @@ mod tests { web_resolver: web_arc, interval_secs: 999, }; - let report = worker.run_once().await.unwrap(); + let report = worker.resolve_batch(vec![did.clone()]).await.unwrap(); assert_eq!(report.resolved, 0, "did:key must not resolve, got {report:?}"); assert_eq!( report.failed, 0,