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