test(appview): handle-sync-Tests vom globalen DB-Zustand entkoppeln
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013HC9HLrUU1LNwkzp8nkDLX
This commit is contained in:
tomdebone
co-authored by
Claude Opus 5
Claude Opus 5
parent
c4ca218d97
commit
465a88e4e5
@@ -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<SyncReport> {
|
||||
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<Vec<String>> {
|
||||
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<String>) -> Result<SyncReport> {
|
||||
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<Option<String>>)> = 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<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,
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user