diff --git a/crates/appview/src/handle_sync.rs b/crates/appview/src/handle_sync.rs index d7976bb..9ac47f1 100644 --- a/crates/appview/src/handle_sync.rs +++ b/crates/appview/src/handle_sync.rs @@ -153,7 +153,10 @@ impl HandleSyncWorker { /// The SELECT half of [`Self::run_once`]: up to [`BATCH_SIZE`] /// distinct DIDs still waiting for a handle. - async fn select_candidates(&self) -> Result> { + /// + /// Public so integration tests can assert on the batch cap without + /// depending on what else the live indexer left pending. + pub async fn select_candidates(&self) -> Result> { let rows: Vec<(String,)> = sqlx::query_as( r#"SELECT DISTINCT did FROM posts @@ -177,7 +180,7 @@ impl HandleSyncWorker { /// 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 { + pub async fn resolve_batch(&self, dids: Vec) -> Result { let mut report = SyncReport::default(); if dids.is_empty() { return Ok(report); @@ -576,6 +579,66 @@ mod tests { assert_eq!(h.as_deref(), Some("from-ingest")); } + /// The documented PDS-first rule: the local PDS is asked before the + /// method dispatch, so a `did:key:` user hosted here resolves + /// without ever dialing plc.directory. This is the flip side of + /// `unknown_methods_are_skipped` — same DID method, opposite + /// outcome, and the difference is solely whether the PDS hosts it. + #[tokio::test] + async fn pds_resolves_did_key_before_method_dispatch() { + let Some(db) = try_test_db().await else { + eprintln!("appview DB unavailable; skipping"); + return; + }; + let did = format!("did:key:z{}", uuid::Uuid::new_v4().simple()); + let _ = sqlx::query("DELETE FROM posts WHERE did = $1") + .bind(&did) + .execute(&db) + .await + .unwrap(); + seed_post(&db, &did, "rk", "").await.unwrap(); + + // The PDS hosts this user; the outbound resolvers know nothing + // and must never be consulted. + let pds = StubResolver::new(HashMap::from([( + did.clone(), + Some("local-user.maarcadetweet.local".into()), + )])) + .into_arc(); + let plc_log: Arc>> = Arc::new(Mutex::new(Vec::new())); + let plc = TrackingResolver::new( + HashMap::from([(did.clone(), Some("must-not-be-used".into()))]), + Arc::clone(&plc_log), + ); + let plc_arc: Arc = Arc::new(plc); + + let worker = HandleSyncWorker { + db: db.clone(), + pds_resolver: pds, + plc_resolver: Arc::clone(&plc_arc), + web_resolver: plc_arc, + interval_secs: 999, + }; + let report = worker.resolve_batch(vec![did.clone()]).await.unwrap(); + assert_eq!( + report.resolved, 1, + "a did:key hosted by the local PDS must resolve, got {report:?}" + ); + assert_eq!( + get_handle(&worker.db, &did).await.as_deref(), + Some("local-user.maarcadetweet.local") + ); + assert!( + plc_log.lock().unwrap().is_empty(), + "the PDS answered, so no outbound resolver may be consulted" + ); + + let _ = sqlx::query("DELETE FROM posts WHERE did = $1") + .bind(&did) + .execute(&db) + .await; + } + /// Dispatch test: a `did:web:` DID must be routed to the /// `web_resolver` (not the PLC one). Without this routing, every /// `did:web:` post would stay `@…` forever. @@ -606,9 +669,18 @@ mod tests { )])) .into_arc(); + // The local PDS is consulted before the method dispatch (see the + // module docs), and it does NOT host this DID — a foreign + // `did:web:` is exactly the case where it answers "don't know". + // Wiring one of the other stubs in here instead would make the + // PDS claim a DID it doesn't have, and the test would be + // asserting against the documented PDS-first rule rather than + // against the method dispatch it's named for. + let pds = StubResolver::new(HashMap::new()).into_arc(); + let worker = HandleSyncWorker { db: db.clone(), - pds_resolver: Arc::clone(&plc), + pds_resolver: pds, plc_resolver: plc, web_resolver: web, interval_secs: 999, @@ -665,9 +737,16 @@ mod tests { let plc_arc: Arc = Arc::new(plc); let web_arc: Arc = Arc::new(web); + // A DID the local PDS does not host — otherwise the PDS-first + // rule would (correctly) resolve it and this test would be + // measuring the wrong thing. The "local PDS *does* host it" + // case is covered by `pds_resolves_did_key_before_method_dispatch`. + let pds_arc: Arc = + Arc::new(StubResolver::new(HashMap::new())); + let worker = HandleSyncWorker { db: db.clone(), - pds_resolver: Arc::clone(&plc_arc), + pds_resolver: pds_arc, plc_resolver: plc_arc, web_resolver: web_arc, interval_secs: 999, diff --git a/crates/appview/tests/handle_sync_integration.rs b/crates/appview/tests/handle_sync_integration.rs index 39a7877..9578f50 100644 --- a/crates/appview/tests/handle_sync_integration.rs +++ b/crates/appview/tests/handle_sync_integration.rs @@ -175,8 +175,13 @@ async fn sync_resolves_known_did() { seed_post(&db, &did, "rkb", "", "second").await.unwrap(); assert_eq!(count_empty_handle_for(&db, &did).await.unwrap(), 2); + // Drive the resolve half with our own DID. `run_once()` scans + // globally, ordered by DID and capped at BATCH_SIZE, so on a + // database that a live indexer keeps topping up, a freshly seeded + // DID isn't guaranteed to make the batch — the assertions below + // would then be measuring someone else's rows. let worker = worker_with(db.clone(), stub.clone()); - let report: SyncReport = worker.run_once().await.unwrap(); + let report: SyncReport = worker.resolve_batch(vec![did.clone()]).await.unwrap(); assert_eq!(report.resolved, 2, "{report:?}"); assert_eq!(report.failed, 0); assert_eq!(report.skipped, 0); @@ -224,9 +229,16 @@ async fn sync_skips_already_resolved() { .into_arc(); let worker = worker_with(db.clone(), stub.clone()); - let report = worker.run_once().await.unwrap(); - assert_eq!(report.resolved, 0, "{report:?}"); - assert_eq!(report.failed, 0); + // This test is about the SELECT: a DID whose rows already carry a + // handle must never reach a resolver in the first place. So assert + // on `select_candidates()` rather than forcing the DID through + // `resolve_batch` — that would consult the resolver by definition + // and defeat the `query_count == 0` check below. + let candidates = worker.select_candidates().await.unwrap(); + assert!( + !candidates.contains(&did), + "a DID that already has a handle must not be selected" + ); // Both rows must still carry the pre-existing handle. let (cnt,): (i64,) = sqlx::query_as( @@ -279,7 +291,16 @@ async fn sync_respects_limit() { let stub = StubResolver::new(mapping).into_arc(); let worker = worker_with(db.clone(), stub.clone()); - let report = worker.run_once().await.unwrap(); + // The cap lives in the SELECT, so assert it there; the report of a + // full `run_once()` depends on what else is pending database-wide. + let candidates = worker.select_candidates().await.unwrap(); + assert!( + candidates.len() as i64 <= BATCH_SIZE, + "select must never exceed BATCH_SIZE, got {}", + candidates.len() + ); + 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, @@ -339,7 +360,7 @@ async fn sync_skips_unresolvable_dids() { let stub = StubResolver::new(HashMap::new()).into_arc(); let worker = worker_with(db.clone(), stub.clone()); - let report = worker.run_once().await.unwrap(); + let report = worker.resolve_batch(vec![did.clone()]).await.unwrap(); assert_eq!(report.resolved, 0); assert_eq!(report.failed, 0); assert_eq!(report.skipped, 1, "{report:?}"); @@ -388,14 +409,21 @@ async fn sync_resolves_did_web_via_web_resolver() { let plc_arc: Arc = plc.into_arc(); let web_arc: Arc = web.into_arc(); + // The local PDS is consulted before the method dispatch and does + // not host a foreign `did:web:` — wiring one of the other stubs in + // here would make it claim a DID it doesn't have, and the test + // would assert against the PDS-first rule instead of the dispatch. + let pds_arc: Arc = + StubResolver::new(HashMap::new()).into_arc(); + let worker = HandleSyncWorker { db: db.clone(), - pds_resolver: Arc::clone(&plc_arc), + pds_resolver: pds_arc, plc_resolver: plc_arc, 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, 1, "did:web must resolve through the web resolver, got {report:?}" @@ -442,14 +470,19 @@ async fn sync_resolves_did_plc_via_plc_resolver() { let plc_arc: Arc = plc.into_arc(); let web_arc: Arc = web.into_arc(); + // Same reasoning as the did:web test: the PDS doesn't host this + // DID, so the method dispatch is what's under test. + let pds_arc: Arc = + StubResolver::new(HashMap::new()).into_arc(); + let worker = HandleSyncWorker { db: db.clone(), - pds_resolver: Arc::clone(&plc_arc), + pds_resolver: pds_arc, plc_resolver: plc_arc, 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, 1, "did:plc must resolve through the PLC resolver, got {report:?}"