From 3aa5d5c0e390590aacc373765bfffb6be86114c0 Mon Sep 17 00:00:00 2001 From: tomdebone Date: Sat, 18 Jul 2026 17:56:11 +0200 Subject: [PATCH] =?UTF-8?q?feat(at-identity):=20PdsHandleResolver=20for=20?= =?UTF-8?q?cluster-local=20DID=E2=86=92handle=20resolution?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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:key: users and any other operator-hosted method) weren't reachable without an external round trip, and unresolvable DIDs (did:key: not on this PDS, did:foo: anything) blocked the 100-row batch forever because did:key: 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:key: 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. --- Cargo.lock | 1 + crates/appview/Cargo.toml | 1 + crates/appview/src/handle_sync.rs | 106 ++++++++-- crates/appview/src/main.rs | 19 ++ .../appview/tests/handle_sync_integration.rs | 9 +- crates/at-identity/src/lib.rs | 2 + crates/at-identity/src/pds_handle.rs | 184 ++++++++++++++++++ crates/at-shared/src/config.rs | 9 + crates/pds-server/src/routes/identity.rs | 42 +++- crates/pds-server/src/routes/types.rs | 7 + .../appview/0006_handle_sync_attempt.sql | 25 +++ 11 files changed, 378 insertions(+), 27 deletions(-) create mode 100644 crates/at-identity/src/pds_handle.rs create mode 100644 migrations/appview/0006_handle_sync_attempt.sql diff --git a/Cargo.lock b/Cargo.lock index 2695fe1..cc55782 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -51,6 +51,7 @@ dependencies = [ "axum", "base64", "chrono", + "futures", "reqwest", "rustls", "serde", diff --git a/crates/appview/Cargo.toml b/crates/appview/Cargo.toml index 0d7b897..85e8050 100644 --- a/crates/appview/Cargo.toml +++ b/crates/appview/Cargo.toml @@ -37,6 +37,7 @@ at-identity = { workspace = true } uuid = { workspace = true } rustls = { version = "0.23", default-features = false, features = ["ring", "std", "logging", "tls12"] } base64 = { workspace = true } +futures = { workspace = true } [dev-dependencies] tokio = { workspace = true } diff --git a/crates/appview/src/handle_sync.rs b/crates/appview/src/handle_sync.rs index 45e62c4..229b111 100644 --- a/crates/appview/src/handle_sync.rs +++ b/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, pub plc_resolver: Arc, pub web_resolver: Arc, 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> { + // 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 { 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>)> = 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) -> 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, diff --git a/crates/appview/src/main.rs b/crates/appview/src/main.rs index 328d41c..1902ee6 100644 --- a/crates/appview/src/main.rs +++ b/crates/appview/src/main.rs @@ -86,8 +86,27 @@ async fn main() -> Result<()> { cfg.plc_directory_url.clone(), )); let web: Arc = Arc::new(at_identity::WebResolver::new()); + // PDS-local handle resolver. The handle_sync worker consults it + // first, before the public PLC/web resolvers, so `did:key:` + // users (and any other DID the operator hosts on this PDS) get their + // local handle without a round trip to plc.directory. A 404 from + // the PDS falls through to the public resolvers. + // + // Use the cluster-internal URL when configured (e.g. `http://pds:3000` + // inside docker compose) — `pds_public_url` may not be reachable + // from inside the cluster when TLS / DNS is set up for outside + // clients only. + let pds_base_url = cfg + .pds_internal_url + .clone() + .unwrap_or_else(|| cfg.pds_public_url.clone()); + let pds_resolver: Arc = Arc::new( + at_identity::pds_handle::PdsHandleResolver::new(pds_base_url), + ); + let handle_sync = handle_sync::HandleSyncWorker { db: db.clone(), + pds_resolver, plc_resolver: plc, web_resolver: web, interval_secs: cfg.appview_handle_sync_interval_secs, diff --git a/crates/appview/tests/handle_sync_integration.rs b/crates/appview/tests/handle_sync_integration.rs index 19b1444..39a7877 100644 --- a/crates/appview/tests/handle_sync_integration.rs +++ b/crates/appview/tests/handle_sync_integration.rs @@ -67,13 +67,14 @@ impl DidHandleResolver for StubResolver { } } -/// Build a worker whose PLC and web resolvers are both the same stub. -/// The integration tests in this file don't care which method the -/// DID uses — the stub answers for any prefix. +/// Build a worker whose PDS, PLC and web resolvers are all the same +/// stub. The integration tests in this file don't care which method +/// the DID uses — the stub answers for any prefix. fn worker_with(db: sqlx::PgPool, stub: Arc) -> HandleSyncWorker { let r: Arc = stub; HandleSyncWorker { db, + pds_resolver: Arc::clone(&r), plc_resolver: Arc::clone(&r), web_resolver: Arc::clone(&r), interval_secs: 999, @@ -389,6 +390,7 @@ async fn sync_resolves_did_web_via_web_resolver() { let worker = HandleSyncWorker { db: db.clone(), + pds_resolver: Arc::clone(&plc_arc), plc_resolver: plc_arc, web_resolver: web_arc, interval_secs: 999, @@ -442,6 +444,7 @@ async fn sync_resolves_did_plc_via_plc_resolver() { let worker = HandleSyncWorker { db: db.clone(), + pds_resolver: Arc::clone(&plc_arc), plc_resolver: plc_arc, web_resolver: web_arc, interval_secs: 999, diff --git a/crates/at-identity/src/lib.rs b/crates/at-identity/src/lib.rs index 3605725..e2211a2 100644 --- a/crates/at-identity/src/lib.rs +++ b/crates/at-identity/src/lib.rs @@ -1,7 +1,9 @@ pub mod handle; +pub mod pds_handle; pub mod plc; pub mod web; pub use handle::{resolve_handle, DidHandleResolver, HandleResolver}; +pub use pds_handle::PdsHandleResolver; pub use plc::{submit_op, PlcClient}; pub use web::WebResolver; \ No newline at end of file diff --git a/crates/at-identity/src/pds_handle.rs b/crates/at-identity/src/pds_handle.rs new file mode 100644 index 0000000..39d1427 --- /dev/null +++ b/crates/at-identity/src/pds_handle.rs @@ -0,0 +1,184 @@ +//! PDS-first handle resolver. +//! +//! The local PDS is the authoritative source for `did:key:` users (and +//! for any DID the operator hosts on this PDS). The `AppView`'s +//! `handle_sync` worker consults this resolver *before* falling through +//! to the public PLC directory / `did:web:` HTTPS resolver, so local +//! users get their handle without ever dialing out to plc.directory. +//! +//! ### Wire shape +//! +//! PDS endpoint: `POST /xrpc/com.atproto.identity.resolveHandle` +//! with body `{ "handle": "" }`. The PDS's handler is +//! polymorphic on the `handle` field: if it starts with `did:` +//! the PDS looks the row up by `did` (PK), otherwise by `handle`. +//! On success the PDS returns `{ "did": "...", "handle": "..." }` +//! — we read the `handle` field, which is what the worker +//! actually needs to fill `posts.handle`. A 404 means the DID +//! isn't hosted here, and the worker falls through to PLC/Web. +//! +//! Network: `reqwest::Client` with a 10 s timeout. The same client +//! is reused across requests — handle with `Arc` +//! in the worker. + +use crate::handle::DidHandleResolver; +use anyhow::Result; +use async_trait::async_trait; +use serde_json::json; +use std::time::Duration; + +pub struct PdsHandleResolver { + pub base_url: String, + pub client: reqwest::Client, +} + +impl PdsHandleResolver { + pub fn new(base_url: impl Into) -> Self { + // 2 s is plenty for a colocated PDS (typical round-trip + // < 100 ms in dev) but bounds the per-DID cost when the + // PDS is unreachable — at 100 DIDs/pass that caps a + // single pass at ~2 s with parallel dispatch, vs the + // ~17 min worst-case the old 10 s timeout allowed with + // serial dispatch. + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(2)) + .build() + .expect("reqwest client build should never fail"); + Self { + base_url: base_url.into(), + client, + } + } +} + +#[async_trait] +impl DidHandleResolver for PdsHandleResolver { + async fn resolve_handle(&self, did: &str) -> Result> { + // POST /xrpc/com.atproto.identity.resolveHandle with + // { "handle": "" } in the body. The PDS's handler + // recognises a `did:` prefix and does a PK lookup on + // `users.did`, returning `{ did, handle }` on match. + let url = format!( + "{}/xrpc/com.atproto.identity.resolveHandle", + self.base_url + ); + let r = self + .client + .post(&url) + .json(&json!({ "handle": did })) + .send() + .await?; + if r.status().as_u16() == 404 { + return Ok(None); + } + if !r.status().is_success() { + // Non-2xx, non-404: a real error — propagate it so the + // worker logs `failed` instead of silently treating the + // DID as `skipped`. + anyhow::bail!( + "pds handle resolver returned {}", + r.status() + ); + } + let v: serde_json::Value = r.json().await?; + // The PDS returns `{ "did": "...", "handle": "..." }` on + // success. We read `handle` (what we actually want for + // `posts.handle`) and ignore the echoed `did`. + Ok(v.get("handle").and_then(|x| x.as_str()).map(String::from)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + use std::net::SocketAddr; + use tokio::net::TcpListener; + + /// Spawn a one-shot HTTP listener that responds to the + /// resolveHandle XRPC call with the configured body + status. + /// Returns the bound address (so the resolver under test can hit + /// `http://127.0.0.1:`). + async fn spawn_stub( + status: u16, + body: serde_json::Value, + ) -> SocketAddr { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let addr = listener.local_addr().unwrap(); + tokio::spawn(async move { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + loop { + let (mut sock, _) = match listener.accept().await { + Ok(p) => p, + Err(_) => break, + }; + let mut buf = vec![0u8; 8192]; + let n = sock.read(&mut buf).await.unwrap_or(0); + if n == 0 { + continue; + } + let body_s = body.to_string(); + let resp = format!( + "HTTP/1.1 {status} {}\r\n\ + Content-Type: application/json\r\n\ + Content-Length: {}\r\n\ + Connection: close\r\n\r\n{body_s}", + status_text(status), + body_s.len(), + ); + let _ = sock.write_all(resp.as_bytes()).await; + } + }); + addr + } + + fn status_text(s: u16) -> &'static str { + match s { + 200 => "OK", + 404 => "Not Found", + 500 => "Internal Server Error", + _ => "Status", + } + } + + #[tokio::test] + async fn returns_handle_on_match() { + let addr = spawn_stub( + 200, + json!({ "did": "did:plc:abc", "handle": "alice.bsky" }), + ) + .await; + let r = PdsHandleResolver::new(format!("http://{addr}")); + let h = r.resolve_handle("did:plc:abc").await.unwrap(); + assert_eq!(h.as_deref(), Some("alice.bsky")); + } + + #[tokio::test] + async fn returns_none_on_404() { + let addr = spawn_stub(404, json!({ "error": "NotFound" })).await; + let r = PdsHandleResolver::new(format!("http://{addr}")); + let h = r.resolve_handle("did:plc:unknown").await.unwrap(); + assert!(h.is_none()); + } + + #[tokio::test] + async fn returns_err_on_5xx() { + let addr = spawn_stub(500, json!({ "error": "oops" })).await; + let r = PdsHandleResolver::new(format!("http://{addr}")); + let result = r.resolve_handle("did:plc:abc").await; + assert!( + result.is_err(), + "5xx must propagate as Err so the worker logs `failed`, not `skipped`" + ); + } + + #[tokio::test] + async fn returns_handle_even_when_did_field_missing() { + // Some PDS implementations might return `{ "handle": "x" }` + // without echoing the DID. We must still extract the handle. + let addr = spawn_stub(200, json!({ "handle": "bob.bsky" })).await; + let r = PdsHandleResolver::new(format!("http://{addr}")); + let h = r.resolve_handle("did:plc:bob").await.unwrap(); + assert_eq!(h.as_deref(), Some("bob.bsky")); + } +} diff --git a/crates/at-shared/src/config.rs b/crates/at-shared/src/config.rs index f3cdc11..33975f3 100644 --- a/crates/at-shared/src/config.rs +++ b/crates/at-shared/src/config.rs @@ -28,6 +28,14 @@ pub struct AppConfig { pub s3_bucket_pds: String, pub s3_bucket_appview: String, pub plc_directory_url: String, + /// Cluster-internal URL the AppView uses to reach the PDS (e.g. + /// `http://pds-server:3000`). Falls back to `pds_public_url` when + /// unset. Splitting this from `pds_public_url` lets a single + /// deployment point the AppView at the in-cluster PDS hostname + /// (which may not be reachable from outside) while clients + /// still see the public URL. + #[serde(default)] + pub pds_internal_url: Option, /// Optional shared secret for `POST /internal/ingest-commit`. If unset, /// the endpoint accepts anonymous requests (dev mode). If set, callers /// must send `X-Ingest-Secret: `. @@ -68,6 +76,7 @@ impl AppConfig { s3_bucket_pds: env("S3_BUCKET_PDS")?, s3_bucket_appview: env("S3_BUCKET_APPVIEW")?, plc_directory_url: env("PLC_DIRECTORY_URL")?, + pds_internal_url: std::env::var("PDS_INTERNAL_URL").ok(), appview_ingest_secret: std::env::var("APPVIEW_INGEST_SECRET").ok(), appview_handle_sync_interval_secs: std::env::var("APPVIEW_HANDLE_SYNC_INTERVAL_SECS") .ok() diff --git a/crates/pds-server/src/routes/identity.rs b/crates/pds-server/src/routes/identity.rs index 91918d7..e0206ee 100644 --- a/crates/pds-server/src/routes/identity.rs +++ b/crates/pds-server/src/routes/identity.rs @@ -9,17 +9,42 @@ pub async fn resolve_handle( State(state): State, Json(req): Json, ) -> Result, (StatusCode, Json)> { + // Polymorphic input: `handle` may be a bare handle OR a DID + // (`did:plc:…`, `did:web:…`, `did:key:…`). When it's a DID we + // look the row up by `did` — the AppView's `PdsHandleResolver` + // uses this path to fill `posts.handle` for local-PDS users + // (including `did:key:`) without a second round-trip to plc.directory. + if req.handle.starts_with("did:") { + let row: Option<(String,)> = + sqlx::query_as("SELECT handle FROM users WHERE did = $1") + .bind(&req.handle) + .fetch_optional(&state.db) + .await + .map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, e))?; + if let Some((handle,)) = row { + return Ok(Json(ResolveHandleResp { + did: req.handle.clone(), + handle: Some(handle), + })); + } + // DID not hosted here — fall through to the handle lookup + // (returns 404 below). + } if let Some(zone) = state.cfg.pds_handle_dns_zone.strip_prefix(".") { if let Some(stripped) = req.handle.strip_suffix(zone) { let user = stripped.trim_end_matches('.'); let full = format!("{}{}", user, state.cfg.pds_handle_dns_zone); - let row: Option<(String,)> = sqlx::query_as("SELECT did FROM users WHERE handle = $1") - .bind(&full) - .fetch_optional(&state.db) - .await - .map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, e))?; + let row: Option<(String,)> = + sqlx::query_as("SELECT did FROM users WHERE handle = $1") + .bind(&full) + .fetch_optional(&state.db) + .await + .map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, e))?; if let Some((did,)) = row { - return Ok(Json(ResolveHandleResp { did })); + return Ok(Json(ResolveHandleResp { + did, + handle: Some(full), + })); } } } @@ -29,7 +54,10 @@ pub async fn resolve_handle( .await .map_err(|e| err(StatusCode::INTERNAL_SERVER_ERROR, e))?; match row { - Some((did,)) => Ok(Json(ResolveHandleResp { did })), + Some((did,)) => Ok(Json(ResolveHandleResp { + did, + handle: Some(req.handle.clone()), + })), None => { warn!(handle = %req.handle, "handle not found"); Err(err( diff --git a/crates/pds-server/src/routes/types.rs b/crates/pds-server/src/routes/types.rs index b0b3fc8..33ff2ca 100644 --- a/crates/pds-server/src/routes/types.rs +++ b/crates/pds-server/src/routes/types.rs @@ -62,6 +62,13 @@ pub struct ResolveHandleReq { #[derive(Debug, Serialize)] pub struct ResolveHandleResp { pub did: String, + /// Always populated when the lookup succeeds. For + /// handle → DID calls this is just the input echo; for + /// DID → handle calls this is the resolved local handle + /// (used by the AppView's `PdsHandleResolver` to fill + /// `posts.handle` without a second round-trip). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub handle: Option, } #[derive(Debug, Deserialize)] diff --git a/migrations/appview/0006_handle_sync_attempt.sql b/migrations/appview/0006_handle_sync_attempt.sql new file mode 100644 index 0000000..86f8c46 --- /dev/null +++ b/migrations/appview/0006_handle_sync_attempt.sql @@ -0,0 +1,25 @@ +-- AppView database schema 0006: handle-sync attempt tracking. +-- +-- Why +-- The `handle_sync` worker SELECTs DIDs whose `posts.handle` is empty +-- and tries to resolve them via the local PDS → PLC directory → +-- `did:web:` resolver. Some DIDs are *unresolvable* (e.g. a `did:key:` +-- user not hosted on the local PDS, or any `did:foo:` method that +-- neither PLC nor Web understands). Without tracking these, every +-- pass re-selects them and they dominate the 100-row batch — and +-- since `did:key:` sorts lexicographically before `did:plc:` / +-- `did:web:`, the worker would process the same 100 unresolvable +-- `did:key:` rows forever and never reach any resolvable DID. +-- +-- With this column the worker marks each empty-handle row with the +-- time of its last attempt. The SELECT filter excludes rows +-- attempted within the last hour, so an unresolvable DID gets at +-- most one attempt per hour and stops blocking forward progress. +-- Rows whose `handle` later gets filled (by another code path) are +-- naturally no longer in the candidate set. +-- +-- The column is per-post (not per-DID) because the candidate set is +-- already per-post and the update is cheap (the empty-handle slice +-- is small in steady state). + +ALTER TABLE posts ADD COLUMN IF NOT EXISTS handle_sync_attempted_at TIMESTAMPTZ;