test(appview): handle-sync-Tests messen wieder, was ihr Name sagt

Mit gesetztem DATABASE_URL_APPVIEW liefen diese Tests zum ersten Mal
überhaupt (ohne die Variable überspringen sie sich still) — und fielen
um. Zwei Ursachen:

1. Sechs Integrationstests hingen wie zuvor die Unit-Tests am globalen
   run_once()-Batch. select_candidates/resolve_batch sind dafür jetzt
   pub, damit auch die Integrationstests ihren eigenen DID durchreichen
   können statt zu hoffen, dass er es in den Batch schafft.
2. Die Dispatch-Tests für did:web und did:plc verdrahteten den fremden
   Stub als pds_resolver — also eine lokale PDS, die behauptet, eine
   fremde DID zu kennen. Die Moduldoku sagt ausdrücklich, dass die PDS
   vor der Methodenverzweigung befragt wird, damit ein did:key-Nutzer
   der eigenen PDS ohne Umweg über plc.directory auflöst. Die Fixtures
   haben also gegen die dokumentierte Regel getestet statt gegen die
   Verzweigung, um die es ihnen ging. Jetzt kennt die PDS-Stub die DID
   nicht, wie es der Realität entspricht.

Neu: pds_resolves_did_key_before_method_dispatch pinnt die PDS-zuerst-
Regel selbst — dasselbe DID-Verfahren, umgekehrtes Ergebnis, und der
Unterschied ist allein, ob die PDS den Nutzer hostet.

sync_skips_already_resolved prüft weiter über select_candidates: dass
ein DID mit Handle gar nicht erst bei einem Resolver landet, ist der
Punkt des Tests.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013HC9HLrUU1LNwkzp8nkDLX

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<String> = 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<dyn DidHandleResolver> = plc.into_arc();
     let web_arc: Arc<dyn DidHandleResolver> = 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<dyn DidHandleResolver> =
+        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<dyn DidHandleResolver> = plc.into_arc();
     let web_arc: Arc<dyn DidHandleResolver> = 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<dyn DidHandleResolver> =
+        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:?}"