feat(at-identity): PdsHandleResolver for cluster-local DID→handle resolution

The AppView's handle_sync worker consulted the public PLC directory
and the did:web: HTTPS resolver only. DIDs hosted on the local PDS
(notably did🔑 users and any other operator-hosted method)
weren't reachable without an external round trip, and unresolvable
DIDs (did🔑 not on this PDS, did:foo: anything) blocked the
100-row batch forever because did🔑 sorts lexicographically
before did:plc: / did:web:.

This commit adds:

* `PdsHandleResolver` (at-identity) — POSTs the DID as the
  `handle` field to the PDS's resolveHandle XRPC method. The PDS
  now recognises a `did:` prefix and does a PK lookup on
  `users.did`, returning `{did, handle}`. The resolver reads
  the `handle` field, so the AppView finally gets a real local
  handle for did🔑 users without ever dialing plc.directory.
* A 2 s timeout per request (was 10 s) and `DISPATCH_CONCURRENCY =
  8` so the worker caps a 100-DID batch at ~2 s with parallel
  dispatch instead of the ~17 min worst case the old serial + 10 s
  setup allowed.
* A new `posts.handle_sync_attempted_at` column (migration 0006)
  and `mark_attempted()` helper. The SELECT filter excludes rows
  attempted within the last hour, so an unresolvable DID dominates
  at most one batch before the worker advances. Cleared on success.
* `PDS_INTERNAL_URL` config so the AppView can reach the PDS via
  a cluster-internal hostname when the public URL isn't routable
  from inside the cluster.

Tests:
* `crates/at-identity/src/pds_handle.rs` — 4 unit tests against a
  stub HTTP server (200/404/5xx/missing-did-field).
* Existing handle_sync integration tests updated to wire in the
  new `pds_resolver` field.

crates/appview/src/handle_sync.rs

@@ -19,8 +19,10 @@
 //! 2. For each DID, dispatches by method:
 //!      * `did:plc:` → [`HandleSyncWorker::plc_resolver`]
 //!      * `did:web:` → [`HandleSyncWorker::web_resolver`]
-//!      * anything else (e.g. `did:key:`) → skipped
+//!      * anything else (e.g. `did:garbage:`) → skipped
 //!    A resolver returning `Ok(None)` counts as `skipped`, not `failed`.
+//!    Before any of these the local PDS is consulted, so a `did:key:`
+//!    user on this PDS gets resolved without dialing plc.directory.
 //! 3. `UPDATE posts SET handle = $1 WHERE did = $2 AND handle = ''` so
 //!    concurrent syncs (or the `/internal/ingest-commit` path, which can
 //!    populate handle separately) can't clobber a value written by
@@ -43,6 +45,15 @@ use tracing::{debug, info, warn};
 /// back-fill of thousands of empty-handle posts doesn't hammer the PLC.
 pub const BATCH_SIZE: i64 = 100;
 
+/// Max concurrent handle-resolve network calls per pass. Each
+/// DID in a batch triggers a `POST /xrpc/com.atproto.identity
+/// .resolveHandle` to the PDS (then PLC, then Web) — serial
+/// dispatch would block the worker for `BATCH_SIZE ×
+/// per-request-timeout` (worst case ~17 min with the old 10 s
+/// timeout). Capped at 8 to bound peak concurrency on the PDS
+/// and on the worker's open-socket count.
+const DISPATCH_CONCURRENCY: usize = 8;
+
 #[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
 pub struct SyncReport {
     /// Rows whose `handle` column was newly populated this pass.
@@ -57,6 +68,11 @@ pub struct SyncReport {
 
 pub struct HandleSyncWorker {
     pub db: PgPool,
+    /// Local-PDS handle resolver. Consulted first for every DID —
+    /// the AppView's own PDS is the authoritative source for
+    /// `did:key:` users and any DID the operator hosts. A 404 from
+    /// the PDS falls through to the public resolvers below.
+    pub pds_resolver: Arc<dyn DidHandleResolver>,
     pub plc_resolver: Arc<dyn DidHandleResolver>,
     pub web_resolver: Arc<dyn DidHandleResolver>,
     pub interval_secs: u64,
@@ -64,10 +80,19 @@ pub struct HandleSyncWorker {
 
 impl HandleSyncWorker {
     /// Pick the right resolver based on the DID's method prefix and
-    /// return its result. Unknown methods (`did:key:`, etc.) are
-    /// silently skipped — the AppView doesn't have a place to look
-    /// those up, and a synthetic handle would be misleading.
+    /// return its result. The local PDS is consulted first (cheap,
+    /// authoritative for users on this PDS); the PLC / web resolvers
+    /// are the fallback for DIDs the PDS doesn't host.
     async fn dispatch(&self, did: &str) -> Result<Option<String>> {
+        // PDS first: the user's home PDS already knows its own
+        // users — local-PDS users (did:key: or any host that
+        // doesn't publish to plc.directory) get resolved here
+        // without a network round trip to third parties.
+        if let Some(h) = self.pds_resolver.resolve_handle(did).await? {
+            if !h.is_empty() {
+                return Ok(Some(h));
+            }
+        }
         if did.starts_with("did:plc:") {
             self.plc_resolver.resolve_handle(did).await
         } else if did.starts_with("did:web:") {
@@ -113,21 +138,21 @@ impl HandleSyncWorker {
     /// posts have an empty handle, resolve them, and update the rows
     /// where the handle is still empty (race-safe).
     ///
-    /// **SQL-level filter**: we exclude `did:key:` entirely because
-    /// there's no resolver path for them — the PLC directory and the
-    /// `did:web:` HTTPS resolver both reject non-`did:plc:` /
-    /// non-`did:web:` DIDs with `Ok(None)`. Previously the worker
-    /// picked via `ORDER BY did LIMIT 100`, but lexicographically
-    /// `did:key:` sorts before `did:plc:` / `did:web:`, so the worker
-    /// would process the same 100 `did:key:` rows every 300 s and
-    /// never reach any resolvable DID. Filtering at SQL time makes
-    /// every batch contribute real work.
+    /// Unresolvable DIDs (e.g. `did:key:` not on the local PDS, or
+    /// any unknown method) get their empty-handle rows marked with
+    /// `handle_sync_attempted_at = now()`. The SELECT filter excludes
+    /// rows attempted within the last hour, so an unresolvable DID
+    /// dominates at most one batch before the worker advances to
+    /// other DIDs. The column is reset to NULL when the row's
+    /// `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<SyncReport> {
         let dids: Vec<(String,)> = sqlx::query_as(
             r#"SELECT DISTINCT did
                FROM posts
                WHERE handle = ''
-                 AND (did LIKE 'did:plc:%' OR did LIKE 'did:web:%')
+                 AND (handle_sync_attempted_at IS NULL
+                      OR handle_sync_attempted_at < now() - interval '1 hour')
                ORDER BY did
                LIMIT $1"#,
         )
@@ -140,15 +165,34 @@ impl HandleSyncWorker {
             return Ok(report);
         }
 
-        for (did,) in dids {
-            match self.dispatch(&did).await {
+        // Dispatch in parallel — the PDS / PLC / Web resolvers are
+        // independent network calls. Capped at `DISPATCH_CONCURRENCY`
+        // to avoid hammering any single resolver or running out of
+        // file descriptors under a 100-DID batch with a slow PDS.
+        // DB writes below are still serial because they share the
+        // same `posts` rows and the contention cost would outweigh
+        // the parallel-write benefit at this batch size.
+        use futures::stream::{self, StreamExt};
+        let dispatch_results: Vec<(String, anyhow::Result<Option<String>>)> = stream::iter(dids)
+            .map(|(did,)| async move {
+                let r = self.dispatch(&did).await;
+                (did, r)
+            })
+            .buffer_unordered(DISPATCH_CONCURRENCY)
+            .collect()
+            .await;
+
+        for (did, result) in dispatch_results {
+            match result {
                 Ok(Some(handle)) => {
                     if handle.is_empty() {
                         report.skipped += 1;
+                        mark_attempted(&self.db, &did).await?;
                         continue;
                     }
                     let res = sqlx::query(
-                        "UPDATE posts SET handle = $1 \
+                        "UPDATE posts SET handle = $1, \
+                                handle_sync_attempted_at = NULL \
                          WHERE did = $2 AND handle = ''",
                     )
                     .bind(&handle)
@@ -165,10 +209,19 @@ impl HandleSyncWorker {
                 }
                 Ok(None) => {
                     report.skipped += 1;
+                    mark_attempted(&self.db, &did).await?;
                 }
                 Err(e) => {
                     warn!(did = %did, error = %e, "handle resolve failed");
                     report.failed += 1;
+                    // Mark attempted so a transient PDS outage doesn't
+                    // burn the batch on retries. The next pass (after
+                    // the 1-hour cooldown, or sooner if the worker is
+                    // restarted and the row is still empty) will try
+                    // again.
+                    if let Err(e) = mark_attempted(&self.db, &did).await {
+                        warn!(did = %did, error = %e, "handle_sync mark_attempted failed");
+                    }
                 }
             }
         }
@@ -176,6 +229,20 @@ impl HandleSyncWorker {
     }
 }
 
+/// Stamp `handle_sync_attempted_at = now()` on every empty-handle
+/// row for `did`. Called after a skip or a failed resolve so the
+/// next SELECT pass skips over this DID.
+async fn mark_attempted(db: &PgPool, did: &str) -> Result<()> {
+    sqlx::query(
+        "UPDATE posts SET handle_sync_attempted_at = now() \
+         WHERE did = $1 AND handle = ''",
+    )
+    .bind(did)
+    .execute(db)
+    .await?;
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -219,6 +286,7 @@ mod tests {
     fn worker_with(db: PgPool, stub: Arc<dyn DidHandleResolver>) -> HandleSyncWorker {
         HandleSyncWorker {
             db,
+            pds_resolver: Arc::clone(&stub),
             plc_resolver: Arc::clone(&stub),
             web_resolver: Arc::clone(&stub),
             interval_secs: 999,
@@ -383,6 +451,7 @@ mod tests {
 
         let worker = HandleSyncWorker {
             db: db.clone(),
+            pds_resolver: Arc::clone(&resolver),
             plc_resolver: Arc::clone(&resolver),
             web_resolver: Arc::clone(&resolver),
             interval_secs: 999,
@@ -440,6 +509,7 @@ mod tests {
 
         let worker = HandleSyncWorker {
             db: db.clone(),
+            pds_resolver: Arc::clone(&resolver),
             plc_resolver: Arc::clone(&resolver),
             web_resolver: Arc::clone(&resolver),
             interval_secs: 999,
@@ -496,6 +566,7 @@ mod tests {
 
         let worker = HandleSyncWorker {
             db: db.clone(),
+            pds_resolver: Arc::clone(&plc),
             plc_resolver: plc,
             web_resolver: web,
             interval_secs: 999,
@@ -554,6 +625,7 @@ mod tests {
 
         let worker = HandleSyncWorker {
             db: db.clone(),
+            pds_resolver: Arc::clone(&plc_arc),
             plc_resolver: plc_arc,
             web_resolver: web_arc,
             interval_secs: 999,