From fe6e1ae7209741f5b596c5f7e3e551fd3505f452 Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Mon, 5 Oct 2026 18:02:24 +0300 Subject: [PATCH] [backfill] write a fetched repo only while its queue entry is still there a backfill read its repo state before getRepo and committed records, that state and tracked = true after it, without checking it still owned the repo. an untrack, a deletion or a requeue that landed during the fetch was overwritten, and since the key was already gone nothing ever looked at the repo again. the full and sparse persists now lock the repo and check their pending key in the transaction that writes, and skip the write if it's gone. untrack takes the repo locks like track and resync do, since every other writer that takes a repo off the queue already held them. the deletion test only passed because erasing the metadata made the commit fail, it now asserts the task finished cleanly. the test pds wakes run_changing with notify_waiters, so a getRepo from an earlier run can't leave a permit that lets the change in early. --- src/backfill/mod.rs | 15 +++++++ src/backfill/sparse.rs | 73 ++++++++++++++++++++++++------- src/backfill/worker/process.rs | 36 +++++++++++++--- src/backfill/worker/task.rs | 78 ++++++++++++++++++++++++++++++---- src/control/repos/indexer.rs | 41 ++++++++++++++++++ src/control/repos/mod.rs | 7 +++ src/db/txn.rs | 13 ++++++ 7 files changed, 232 insertions(+), 31 deletions(-) diff --git a/src/backfill/mod.rs b/src/backfill/mod.rs index 12a02ce..9511dda 100644 --- a/src/backfill/mod.rs +++ b/src/backfill/mod.rs @@ -7,3 +7,18 @@ pub mod sparse; pub mod worker; pub use worker::BackfillWorker; + +use crate::types::RepoState; + +/// what writing a fetched repo came to +pub(crate) enum Persisted { + Stored { + records: usize, + state: Box>, + }, + /// excluded, or a signal filter found nothing it wants + Discarded, + /// the queue entry the backfill ran for went away while it fetched, so + /// an untrack, a deletion or a newer requeue gets the last word + Superseded, +} diff --git a/src/backfill/sparse.rs b/src/backfill/sparse.rs index bcd39a2..1a60aba 100644 --- a/src/backfill/sparse.rs +++ b/src/backfill/sparse.rs @@ -1,3 +1,4 @@ +use crate::backfill::Persisted; use crate::backfill::client::ThrottledHttpClient; use crate::backfill::error::BackfillError; use crate::car::{Blocks, CarBlock, CommitBlocks, CommitCar}; @@ -44,6 +45,8 @@ pub(crate) struct SparseBackfillSuccess { pub(crate) enum SparseBackfillResult { Imported(SparseBackfillSuccess), Discarded, + /// see [`Persisted::Superseded`] + Superseded, Skipped, } @@ -69,10 +72,12 @@ fn sparse_probe(filter: &FilterConfig) -> Option<(SmolStr, &'static str)> { }) } +#[allow(clippy::too_many_arguments)] pub(crate) async fn process_did_sparse( app_state: &Arc, http: &ThrottledHttpClient, did: &Did, + pending_key: &[u8], pds: &url::Url, state: RepoState<'static>, verify_signatures: bool, @@ -288,10 +293,12 @@ pub(crate) async fn process_did_sparse( }) .collect::, _>>()?; - let result = persist_sparse_backfill(app_state, did, state, root_commit, records).await?; - - let Some((records, state)) = result else { - return Ok(SparseBackfillResult::Discarded); + let result = + persist_sparse_backfill(app_state, did, pending_key, state, root_commit, records).await?; + let (records, state) = match result { + Persisted::Stored { records, state } => (records, *state), + Persisted::Discarded => return Ok(SparseBackfillResult::Discarded), + Persisted::Superseded => return Ok(SparseBackfillResult::Superseded), }; Ok(SparseBackfillResult::Imported(SparseBackfillSuccess { @@ -542,12 +549,14 @@ async fn fetch_block_chunk( async fn persist_sparse_backfill( app_state: &Arc, did: &Did, + pending_key: &[u8], mut state: RepoState<'static>, root_commit: Commit, records: Vec<(SmolStr, CarBlock)>, -) -> Result)>, BackfillError> { +) -> Result { let app_state = app_state.clone(); let did = did.clone(); + let pending_key = pending_key.to_vec(); let db = app_state.db.clone(); db.run(move |db| { let filter = app_state.filter.load(); @@ -555,11 +564,11 @@ async fn persist_sparse_backfill( let mut collection_counts: HashMap = HashMap::new(); let mut txn = DbTxn::new(db); - // permanent mode locks before the existing-head snapshot so a live - // write or operator redaction cannot invalidate the diff. ephemeral - // mode has no heads and therefore no shared record state to lock. - if !app_state.ephemeral { - txn.hold_repo_write_lock(&did); + // locks before the existing-head snapshot so a live write or operator + // redaction can't invalidate the diff. ephemeral mode has no heads but + // still writes repo state, which an untrack or deletion must keep + if !txn.lock_repo_and_is_pending(&did, &pending_key)? { + return Ok(Persisted::Superseded); } let prefix = keys::record_prefix_did(&did); @@ -588,7 +597,7 @@ async fn persist_sparse_backfill( let mut signal_seen = filter.mode == FilterMode::Full || filter.signals.is_empty(); let mut record_txn = txn.backfill_records(&app_state, &root_commit.rev, &did)?; if record_txn.is_excluded() { - return Ok::<_, miette::Report>(None); + return Ok(Persisted::Discarded); } for (key, block) in records { @@ -634,7 +643,7 @@ async fn persist_sparse_backfill( if !signal_seen { trace!(signals = ?filter.signals, "no signal-matching sparse records found, discarding repo"); - return Ok::<_, miette::Report>(None); + return Ok(Persisted::Discarded); } state.root = Some(root_commit); @@ -666,7 +675,10 @@ async fn persist_sparse_backfill( txn.commit()?; - Ok::<_, miette::Report>(Some((count, state))) + Ok::<_, miette::Report>(Persisted::Stored { + records: count, + state: Box::new(state), + }) }) .await .map_err(BackfillError::from) @@ -950,13 +962,15 @@ mod tests { Option, ) { let address = spawn_car_server(routes).await; - backfill_at(address, verify_cids, BackfillStrategy::SparseFilter).await + backfill_at(address, verify_cids, BackfillStrategy::SparseFilter, true).await } + /// `queued: false` runs it for a queue entry that's already gone async fn backfill_at( address: std::net::SocketAddr, verify_cids: bool, strategy: BackfillStrategy, + queued: bool, ) -> ( Result, Option, @@ -976,6 +990,13 @@ mod tests { keys::repo_metadata_key(&did), db::ser_repo_meta(&crate::types::RepoMetadata::backfilling(1)).unwrap(), ); + let pending_key = keys::pending_key(1); + if queued { + state + .db + .indexer + .stage_pending_insert(&mut batch, &did, &pending_key); + } batch.commit().unwrap(); let (pds, http) = public_client(address, &state.throttler); @@ -983,6 +1004,7 @@ mod tests { &state, &http, &did, + &pending_key, &pds, RepoState::backfilling(), false, @@ -1126,7 +1148,28 @@ mod tests { Result, Option, ) { - backfill_at(serve(app).await, true, strategy).await + backfill_at(serve(app).await, true, strategy, true).await + } + + #[tokio::test] + async fn a_backfill_whose_queue_entry_went_away_writes_nothing() { + let repo = test_repo(PROOF_DID, &tangled_repo("honest")); + let probe = json_error(StatusCode::NOT_FOUND, "RecordNotFound"); + let app = no_proof_pds(&repo, probe, None, Some(repo.blocks.clone())); + + let (result, stored) = backfill_at( + serve(app).await, + true, + BackfillStrategy::SparseFilter, + false, + ) + .await; + + assert!( + matches!(result, Ok(SparseBackfillResult::Superseded)), + "{result:?}" + ); + assert_eq!(stored, None); } #[tokio::test] diff --git a/src/backfill/worker/process.rs b/src/backfill/worker/process.rs index b1ee8c1..e2b7d0f 100644 --- a/src/backfill/worker/process.rs +++ b/src/backfill/worker/process.rs @@ -14,6 +14,7 @@ use jacquard_common::IntoStatic; use jacquard_common::types::cid::{Cid as AtCid, IpldCid}; use jacquard_common::types::did::Did; +use crate::backfill::Persisted; use crate::backfill::admission::BackfillAdmission; use crate::backfill::client::{ThrottledHttpClient, collect_body_bounded}; use crate::backfill::error::BackfillError; @@ -151,6 +152,7 @@ pub(super) async fn process_did( app_state, http, did, + &pending_key, &pds, state.clone(), verify_signatures, @@ -176,6 +178,12 @@ pub(super) async fn process_did( remove_discarded_repo(app_state, did, &pending_key).await?; return Ok(None); } + Ok(SparseBackfillResult::Superseded) => { + debug!( + "queue entry went away during the sparse fetch, dropping what it fetched" + ); + return Ok(None); + } Ok(SparseBackfillResult::Skipped) => { debug!("sparse backfill skipped, falling back to full getRepo"); } @@ -370,14 +378,18 @@ pub(super) async fn process_did( let app_state = app_state.clone(); let did = did.clone(); let rev = root_commit.rev; + let pending_key = pending_key.clone(); tokio::task::spawn_blocking(move || { let mut count = 0; let mut collection_counts: HashMap = HashMap::new(); let mut txn = DbTxn::new(&app_state.db); + if !txn.lock_repo_and_is_pending(&did, &pending_key)? { + return Ok::<_, miette::Report>(Persisted::Superseded); + } let mut record_txn = txn.backfill_records(&app_state, &rev, &did)?; if record_txn.is_excluded() { - return Ok::<_, miette::Report>(None); + return Ok(Persisted::Discarded); } // the MST and non-record blocks were released before entering this blocking // persistence phase; records holds only leaf keys and their blocks. @@ -451,7 +463,7 @@ pub(super) async fn process_did( if !signal_seen { trace!(signals = ?filter.signals, "no signal-matching records found, discarding repo"); - return Ok::<_, miette::Report>(None); + return Ok(Persisted::Discarded); } state.root = Some(root_commit); state.touch(); @@ -483,16 +495,26 @@ pub(super) async fn process_did( txn.commit()?; - Ok::<_, miette::Report>(Some((count, state))) + Ok(Persisted::Stored { + records: count, + state: Box::new(state), + }) }) .await .into_diagnostic()?? }; - let Some((count, state)) = result else { - // signal mode: no signal-matching records found, clean up the optimistically-added repo - remove_discarded_repo(app_state, did, &pending_key).await?; - return Ok(None); + let (count, state) = match result { + Persisted::Stored { records, state } => (records, *state), + Persisted::Discarded => { + // signal mode: no signal-matching records found, clean up the optimistically-added repo + remove_discarded_repo(app_state, did, &pending_key).await?; + return Ok(None); + } + Persisted::Superseded => { + debug!("queue entry went away during the fetch, dropping what it fetched"); + return Ok(None); + } }; trace!(ops = count, elapsed = %start.elapsed().as_secs_f32(), "did ops"); diff --git a/src/backfill/worker/task.rs b/src/backfill/worker/task.rs index fe75324..dc65234 100644 --- a/src/backfill/worker/task.rs +++ b/src/backfill/worker/task.rs @@ -456,7 +456,7 @@ mod tests { let hold = hold.clone(); async move { requests.fetch_add(1, Ordering::SeqCst); - arrived.notify_one(); + arrived.notify_waiters(); drop(hold.lock().await); let (status, body) = response.read().await.clone(); (status, body) @@ -772,12 +772,18 @@ mod tests { async fn run_changing( &self, pending_key: [u8; 8], - change: impl FnOnce() -> miette::Result<()>, + change: impl Future>, ) -> miette::Result> { let held = self.pds.hold.lock().await; + // registered before the task starts, because `notify_waiters` only wakes + // waiters that already exist and keeps nothing for a later one, so a + // getRepo from an earlier run can't let this change in early + let arrived = self.pds.arrived.notified(); + tokio::pin!(arrived); + arrived.as_mut().enable(); let (result, changed) = tokio::join!(self.run(pending_key), async { - self.pds.arrived.notified().await; - let changed = change(); + arrived.await; + let changed = change.await; drop(held); changed }); @@ -1096,7 +1102,9 @@ mod tests { 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)) + .run_changing(keys::pending_key(7), async { + fixture.reindex_pending(7, 11) + }) .await? .unwrap_err(); @@ -1127,7 +1135,7 @@ mod tests { let body = post("a"); fixture.seed(&seeded, 7, &[("a", body.as_slice())])?; let err = fixture - .run_changing(keys::pending_key(7), || { + .run_changing(keys::pending_key(7), async { fixture.delete_authoritatively(&seeded) }) .await? @@ -1167,13 +1175,16 @@ mod tests { .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), || { + let result = fixture + .run_changing(keys::pending_key(7), async { fixture.delete_authoritatively(&seeded) }) .await?; + assert!( + matches!(result, Ok(TaskDisposition::Finished)), + "{result:?}" + ); let after = fixture.repo()?; assert_eq!(after.status, RepoStatus::Deleted); assert!(!after.active, "the tombstone must stay inactive"); @@ -1185,6 +1196,55 @@ mod tests { Ok(()) } + #[tokio::test] + async fn an_untrack_during_the_fetch_is_not_undone() -> 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; + + let repos = crate::control::ReposControl::new(fixture.state.clone()); + let result = fixture + .run_changing(keys::pending_key(7), async { + let untracked = repos.untrack([fixture.did.clone()]).await?; + assert_eq!(untracked, std::slice::from_ref(&fixture.did)); + Ok(()) + }) + .await?; + + assert!( + matches!(result, Ok(TaskDisposition::Finished)), + "{result:?}" + ); + let metadata = fixture + .state + .db + .repo_metadata + .get(keys::repo_metadata_key(&fixture.did)) + .into_diagnostic()? + .expect("repository metadata row"); + assert!(!db::deser_repo_meta(&metadata)?.tracked); + assert_eq!( + fixture.record_rkeys()?, + ["a"], + "the fetched car must not land" + ); + assert_eq!( + fixture.repo()?.root.map(|root| root.data), + seeded.root.as_ref().map(|root| root.data) + ); + assert!(fixture.pending_keys()?.is_empty()); + assert!(fixture.resync()?.is_none()); + Ok(()) + } + #[tokio::test] async fn a_later_valid_car_replaces_the_error_after_requeue() -> miette::Result<()> { let fixture = Fixture::new().await?; diff --git a/src/control/repos/indexer.rs b/src/control/repos/indexer.rs index 7dbe9d6..8692697 100644 --- a/src/control/repos/indexer.rs +++ b/src/control/repos/indexer.rs @@ -306,6 +306,10 @@ impl ReposControl { .db .run(move |db| { let mut txn = crate::db::Txn::new(db); + // a backfill checks its queue entry under these before it commits + txn.hold_record_lock_indexes_sorted( + dids.iter().map(crate::db::record_lock_index_for_did), + ); let mut untracked: Vec = Vec::new(); for did in dids { @@ -1211,4 +1215,41 @@ mod tests { assert!(page(Some(&third))?.is_none()); Ok(()) } + + #[tokio::test] + async fn untrack_waits_for_a_backfill_that_saw_its_queue_entry() -> miette::Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let config = crate::config::Config { + database_path: tmp.path().to_path_buf(), + ..Default::default() + }; + let state = Arc::new(AppState::new(&config)?); + let repos = ReposControl(state.clone()); + let did = Did::new_static("did:web:one.example").into_diagnostic()?; + repos.track([did.clone()]).await?; + let metadata = state + .db + .repo_metadata + .get(keys::repo_metadata_key(&did)) + .into_diagnostic()? + .expect("tracked repo has metadata"); + let pending_key = keys::pending_key(crate::db::deser_repo_meta(&metadata)?.index_id); + + let mut backfill = crate::db::Txn::new(&state.db); + assert!(backfill.lock_repo_and_is_pending(&did, &pending_key)?); + let mut untrack = tokio::spawn({ + let repos = repos.clone(); + let did = did.clone(); + async move { repos.untrack([did]).await } + }); + assert!( + tokio::time::timeout(std::time::Duration::from_millis(200), &mut untrack) + .await + .is_err(), + "untrack went ahead while a backfill held the repo" + ); + drop(backfill); + assert_eq!(untrack.await.into_diagnostic()??, [did]); + Ok(()) + } } diff --git a/src/control/repos/mod.rs b/src/control/repos/mod.rs index f6fea06..ff66e68 100644 --- a/src/control/repos/mod.rs +++ b/src/control/repos/mod.rs @@ -180,6 +180,13 @@ fn read_queues(state: &AppState, info: &mut RepoInfo) -> Result<()> { #[derive(Clone)] pub struct ReposControl(pub(super) Arc); +#[cfg(all(test, feature = "indexer"))] +impl ReposControl { + pub(crate) fn new(state: Arc) -> Self { + Self(state) + } +} + impl ReposControl { pub(crate) fn iter_states( &self, diff --git a/src/db/txn.rs b/src/db/txn.rs index a77fdd4..cad7e86 100644 --- a/src/db/txn.rs +++ b/src/db/txn.rs @@ -143,6 +143,19 @@ impl<'db> Txn<'db> { .into_diagnostic() } + /// lock this repository, then say whether `pending_key` is still queued. + /// every writer that takes a repo off the queue holds this lock, so a key + /// seen here stays queued until this transaction commits + #[cfg(feature = "indexer")] + pub(crate) fn lock_repo_and_is_pending( + &mut self, + did: &Did, + pending_key: &[u8], + ) -> Result { + self.hold_repo_write_lock(did); + self.db.indexer.is_pending(pending_key).into_diagnostic() + } + /// acquire a transaction's complete record-lock set in one canonical order. /// callers that can touch several DIDs must do this before initializing /// lifecycle counts; dynamic encounter-order locking can deadlock against -- 2.51.2