From 86dfa4d50fe588961fa58677f3dcce61b8b3b48a Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Mon, 5 Oct 2026 15:58:08 +0300 Subject: [PATCH] [backfill] skip a pending key whose task already finished the worker hands out keys from a snapshot of the queue, so a key whose task finished after the snapshot comes back once that task leaves in_flight. the second run wrote its stale repo state over the synced one and, with its key already gone, never sent BackfillFinished, so the repo stayed desynchronized for good. a task now checks its key is still queued before it starts. the tests that move the queue under a task now do it while the task waits on getRepo, the window their guards are for, and a deletion that lands during a successful backfill is covered too. --- src/backfill/worker/task.rs | 121 ++++++++++++++++++++++++++++++++++-- 1 file changed, 115 insertions(+), 6 deletions(-) diff --git a/src/backfill/worker/task.rs b/src/backfill/worker/task.rs index de60ff2..e00a908 100644 --- a/src/backfill/worker/task.rs +++ b/src/backfill/worker/task.rs @@ -55,6 +55,16 @@ async fn remove_pending_if_excluded( .await } +async fn is_pending(state: &AppState, pending_key: Slice) -> Result { + state + .db + .run(move |db| { + let value = db.indexer.pending.get(pending_key).into_diagnostic()?; + Ok(value.is_some()) + }) + .await +} + async fn remove_pending(state: &AppState, pending_key: Slice) -> Result<()> { state .db @@ -85,6 +95,12 @@ pub(super) async fn did_task( if remove_pending_if_excluded(state, did, pending_key.clone()).await? { return Ok(TaskDisposition::Finished); } + // the worker hands out keys from a snapshot of the queue, so a key whose task finished after + // the snapshot comes back once that task leaves in_flight. it removed the key first, so this + // read sees it gone, and running again would write its stale repo state over the synced one + if !is_pending(state, pending_key.clone()).await? { + return Ok(TaskDisposition::Finished); + } match process_did( state, @@ -409,6 +425,8 @@ mod tests { address: std::net::SocketAddr, response: Arc>, requests: Arc, + arrived: Arc, + hold: Arc>, } impl PdsStub { @@ -418,16 +436,24 @@ mod tests { Bytes::from_static(NOT_FOUND_BODY), ))); let requests = Arc::new(AtomicUsize::new(0)); + let arrived = Arc::new(tokio::sync::Notify::new()); + let hold = Arc::new(tokio::sync::Mutex::new(())); let app = Router::new().route( "/xrpc/com.atproto.sync.getRepo", get({ let response = response.clone(); let requests = requests.clone(); + let arrived = arrived.clone(); + let hold = hold.clone(); move || { let response = response.clone(); let requests = requests.clone(); + let arrived = arrived.clone(); + let hold = hold.clone(); async move { requests.fetch_add(1, Ordering::SeqCst); + arrived.notify_one(); + drop(hold.lock().await); let (status, body) = response.read().await.clone(); (status, body) } @@ -445,6 +471,8 @@ mod tests { address, response, requests, + arrived, + hold, } } @@ -718,6 +746,24 @@ mod tests { .await } + /// runs the task for `pending_key` and applies `change` while the task waits on getRepo, + /// the way another writer can move the queue under a task that's already in flight + async fn run_changing( + &self, + pending_key: [u8; 8], + change: impl FnOnce() -> miette::Result<()>, + ) -> miette::Result> { + let held = self.pds.hold.lock().await; + let (result, changed) = tokio::join!(self.run(pending_key), async { + self.pds.arrived.notified().await; + let changed = change(); + drop(held); + changed + }); + changed?; + Ok(result) + } + fn repo(&self) -> miette::Result> { let bytes = self .state @@ -894,14 +940,41 @@ mod tests { } #[tokio::test] - async fn repo_not_found_cannot_overwrite_a_replaced_pending_entry() -> miette::Result<()> { + async fn a_key_whose_task_already_finished_is_not_run_again() -> miette::Result<()> { let fixture = Fixture::new().await?; let seeded = rooted("3jzfcijpj2z2a")?; let body = post("a"); fixture.seed(&seeded, 7, &[("a", body.as_slice())])?; - fixture.reindex_pending(7, 11)?; + // the task that owned key 7 already finished and removed it + let mut batch = fixture.state.db.inner.batch(); + batch.remove(&fixture.state.db.indexer.pending, keys::pending_key(7)); + batch.commit().into_diagnostic()?; - let err = fixture.run(keys::pending_key(7)).await.unwrap_err(); + assert_eq!( + fixture.run(keys::pending_key(7)).await?, + TaskDisposition::Finished + ); + + assert_eq!(fixture.pds.requests(), 0); + let after = fixture.repo()?; + assert_eq!(after.status, seeded.status); + assert_eq!( + after.root.as_ref().map(|root| root.data), + seeded.root.as_ref().map(|root| root.data) + ); + Ok(()) + } + + #[tokio::test] + async fn repo_not_found_cannot_overwrite_a_replaced_pending_entry() -> miette::Result<()> { + let fixture = Fixture::new().await?; + let seeded = rooted("3jzfcijpj2z2a")?; + let body = post("a"); + fixture.seed(&seeded, 7, &[("a", body.as_slice())])?; + let err = fixture + .run_changing(keys::pending_key(7), || fixture.reindex_pending(7, 11)) + .await? + .unwrap_err(); assert!( matches!(err, BackfillError::RepoNotFound), @@ -929,9 +1002,12 @@ mod tests { let seeded = rooted("3jzfcijpj2z2a")?; let body = post("a"); fixture.seed(&seeded, 7, &[("a", body.as_slice())])?; - fixture.delete_authoritatively(&seeded)?; - - let err = fixture.run(keys::pending_key(7)).await.unwrap_err(); + let err = fixture + .run_changing(keys::pending_key(7), || { + fixture.delete_authoritatively(&seeded) + }) + .await? + .unwrap_err(); assert!( matches!(err, BackfillError::RepoNotFound), @@ -952,6 +1028,39 @@ mod tests { Ok(()) } + #[tokio::test] + async fn authoritative_deletion_is_not_resurrected_by_a_late_successful_backfill() + -> miette::Result<()> { + let fixture = Fixture::new().await?; + let seeded = rooted("3jzfcijpj2z2a")?; + let body = post("a"); + fixture.seed(&seeded, 7, &[("a", body.as_slice())])?; + let newer = post("b"); + let car = build_car( + "3jzfcijpj2z2b", + &[("a", body.as_slice()), ("b", newer.as_slice())], + ) + .await?; + fixture.pds.serve_car(car.bytes).await; + + // the task may fail or finish, but it can't write over the deletion + let _ = fixture + .run_changing(keys::pending_key(7), || { + fixture.delete_authoritatively(&seeded) + }) + .await?; + + let after = fixture.repo()?; + assert_eq!(after.status, RepoStatus::Deleted); + assert!(!after.active, "the tombstone must stay inactive"); + assert!( + fixture.record_rkeys()?.is_empty(), + "erased records must stay erased" + ); + assert!(fixture.pending_keys()?.is_empty()); + Ok(()) + } + #[tokio::test] async fn a_later_valid_car_replaces_the_error_after_requeue() -> miette::Result<()> { let fixture = Fixture::new().await?; -- 2.51.2