From d77b97d2d2b64f06720628fd2132bcf8cbf0c801 Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Mon, 5 Oct 2026 16:16:19 +0300 Subject: [PATCH] [db] reach the backfill queue only through IndexerDb accessors every place that queued or dropped a backfill wrote the `pending` keyspace directly. it's `pub(super)` now, like `records`: writes go through `stage_pending_insert` and `stage_pending_remove`, reads through `is_pending`, `pending_iter` and `pending_range`. both writes take the did already, since a did-ordered copy of the queue has to change in the same places. no behavior change. --- src/api/xrpc/resolve_mini_doc.rs | 15 +++---- src/backfill/manager.rs | 13 +++--- src/backfill/worker.rs | 2 +- src/backfill/worker/process.rs | 6 ++- src/backfill/worker/task.rs | 66 ++++++++++++++++-------------- src/control/repos/indexer.rs | 60 ++++++++++++++------------- src/control/repos/mod.rs | 4 +- src/crawler/list_repos/producer.rs | 2 +- src/crawler/worker.rs | 61 +++++++++++++-------------- src/db/indexer.rs | 4 +- src/db/keyspaces.rs | 44 +++++++++++++++++++- src/db/lifecycle_counts.rs | 44 ++++++++------------ src/ingest/indexer/shard.rs | 35 +++++++--------- src/ops.rs | 12 +++--- 14 files changed, 193 insertions(+), 175 deletions(-) diff --git a/src/api/xrpc/resolve_mini_doc.rs b/src/api/xrpc/resolve_mini_doc.rs index febd92b..8e7893e 100644 --- a/src/api/xrpc/resolve_mini_doc.rs +++ b/src/api/xrpc/resolve_mini_doc.rs @@ -425,13 +425,11 @@ mod tests { } fn seed_pending(&self) -> miette::Result<()> { - self.hydrant - .state - .db - .indexer - .pending - .insert(keys::pending_key(7), keys::repo_key(&self.did)) - .into_diagnostic() + let db = &self.hydrant.state.db; + let mut batch = db.inner.batch(); + db.indexer + .stage_pending_insert(&mut batch, &self.did, &keys::pending_key(7)); + batch.commit().into_diagnostic() } fn durable_state(&self) -> miette::Result> { @@ -463,8 +461,7 @@ mod tests { .state .db .indexer - .pending - .iter() + .pending_iter() .map(|entry| { let (key, value) = entry.into_inner().into_diagnostic()?; Ok((key.to_vec(), value.to_vec())) diff --git a/src/backfill/manager.rs b/src/backfill/manager.rs index 2b9c6eb..1fc82f2 100644 --- a/src/backfill/manager.rs +++ b/src/backfill/manager.rs @@ -40,12 +40,9 @@ fn queue_resync_if( metadata.index_id = rand::random(); txn.batch.remove(&state.db.indexer.resync, &did_key); - txn.batch.remove(&state.db.indexer.pending, old_pending); - txn.batch.insert( - &state.db.indexer.pending, - keys::pending_key(metadata.index_id), - &did_key, - ); + let indexer = &state.db.indexer; + indexer.stage_pending_remove(&mut txn.batch, did, &old_pending); + indexer.stage_pending_insert(&mut txn.batch, did, &keys::pending_key(metadata.index_id)); txn.batch.insert( &state.db.repo_metadata, metadata_key, @@ -246,7 +243,7 @@ mod tests { .contains_key(keys::repo_key(&did())) .unwrap() ); - assert!(state.db.indexer.pending.is_empty().unwrap()); + assert_eq!(state.db.indexer.pending_iter().count(), 0); } #[test] @@ -268,6 +265,6 @@ mod tests { .contains_key(keys::repo_key(&did())) .unwrap() ); - assert_eq!(state.db.indexer.pending.len().unwrap(), 1); + assert_eq!(state.db.indexer.pending_iter().count(), 1); } } diff --git a/src/backfill/worker.rs b/src/backfill/worker.rs index b316e47..47d4576 100644 --- a/src/backfill/worker.rs +++ b/src/backfill/worker.rs @@ -94,7 +94,7 @@ impl BackfillWorker { self.enabled.wait_enabled("backfill").await; let mut spawned = 0; - for guard in self.state.db.indexer.pending.iter() { + for guard in self.state.db.indexer.pending_iter() { let (key, value) = match guard.into_inner() { Ok(kv) => kv, Err(e) => { diff --git a/src/backfill/worker/process.rs b/src/backfill/worker/process.rs index 58c2468..b8c8d0c 100644 --- a/src/backfill/worker/process.rs +++ b/src/backfill/worker/process.rs @@ -246,7 +246,8 @@ pub(super) async fn process_did( pending_key.as_ref(), GaugeState::Resync(None), )?; - txn.batch.remove(&db.indexer.pending, pending_key.clone()); + db.indexer + .stage_pending_remove(&mut txn.batch, &did, &pending_key); if applied { crate::db::Db::update_repo_state( &mut txn.batch, @@ -525,7 +526,8 @@ async fn remove_discarded_repo( } let applied = txn.transition_pending_key(&did, pending_key.as_ref(), GaugeState::Synced)?; - txn.batch.remove(&db.indexer.pending, pending_key); + db.indexer + .stage_pending_remove(&mut txn.batch, &did, &pending_key); if applied { txn.batch.remove(&db.repos, &did_key); txn.batch.remove(&db.repo_metadata, &metadata_key); diff --git a/src/backfill/worker/task.rs b/src/backfill/worker/task.rs index 5951c5c..64e5880 100644 --- a/src/backfill/worker/task.rs +++ b/src/backfill/worker/task.rs @@ -32,7 +32,8 @@ fn stage_excluded_pending_cleanup( return Ok(false); } txn.transition_pending_key(did, pending_key, GaugeState::Synced)?; - txn.batch.remove(&db.indexer.pending, pending_key); + db.indexer + .stage_pending_remove(&mut txn.batch, did, pending_key); Ok(true) } @@ -58,19 +59,18 @@ async fn remove_pending_if_excluded( 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()) - }) + .run(move |db| db.indexer.is_pending(&pending_key).into_diagnostic()) .await } -async fn remove_pending(state: &AppState, pending_key: Slice) -> Result<()> { +async fn remove_pending(state: &AppState, did: &Did, pending_key: Slice) -> Result<()> { + let did = did.clone(); state .db .run(move |db| { let mut batch = db.inner.batch(); - batch.remove(&db.indexer.pending, pending_key); + db.indexer + .stage_pending_remove(&mut batch, &did, &pending_key); batch.commit().into_diagnostic() }) .await @@ -132,7 +132,8 @@ pub(super) async fn did_task( pending_key.as_ref(), GaugeState::Synced, )?; - txn.batch.remove(&db.indexer.pending, pending_key.clone()); + db.indexer + .stage_pending_remove(&mut txn.batch, &did, &pending_key); if applied { txn.batch.remove(&db.indexer.resync, &did_key); } @@ -174,7 +175,7 @@ pub(super) async fn did_task( Ok(TaskDisposition::Finished) } Ok(None) => { - remove_pending(state, pending_key).await?; + remove_pending(state, did, pending_key).await?; Ok(TaskDisposition::Finished) } Err(BackfillError::PdsBusy) => Ok(TaskDisposition::Deferred), @@ -184,7 +185,7 @@ pub(super) async fn did_task( // retrying the same invalid endpoint. A later identity event can // enqueue fresh work with new evidence. warn!(%reason, "unsafe PDS endpoint, parking backfill until identity evidence changes"); - remove_pending(state, pending_key).await?; + remove_pending(state, did, pending_key).await?; Ok(TaskDisposition::Finished) } Err(BackfillError::Deleted) => { @@ -202,7 +203,8 @@ pub(super) async fn did_task( return Ok::<_, miette::Report>(()); } txn.transition_pending_key(&did, pending_key.as_ref(), GaugeState::Synced)?; - txn.batch.remove(&db.indexer.pending, pending_key); + db.indexer + .stage_pending_remove(&mut txn.batch, &did, &pending_key); txn.commit()?; Ok::<_, miette::Report>(()) } @@ -264,7 +266,7 @@ pub(super) async fn did_task( let mut retry_count = match existing_state { Some(ResyncState::Error { retry_count, .. }) => retry_count, Some(ResyncState::Gone { .. }) => { - remove_pending(state, pending_key).await?; + remove_pending(state, did, pending_key).await?; return Ok(TaskDisposition::Finished); } None => 0, @@ -318,7 +320,8 @@ pub(super) async fn did_task( pending_key.as_ref(), GaugeState::Resync(Some(error_kind)), )?; - txn.batch.remove(&db.indexer.pending, pending_key.clone()); + db.indexer + .stage_pending_remove(&mut txn.batch, &did, &pending_key); if applied { txn.batch .insert(&db.indexer.resync, &did_key, serialized_resync_state); @@ -383,7 +386,10 @@ mod tests { crate::db::filter::exclude_key(did.as_str())?, [], ); - batch.insert(&state.db.indexer.pending, pending_key, keys::repo_key(&did)); + state + .db + .indexer + .stage_pending_insert(&mut batch, &did, &pending_key); batch.commit().into_diagnostic()?; assert!( @@ -391,13 +397,11 @@ mod tests { .await? ); assert!( - state + !state .db .indexer - .pending - .get(pending_key) + .is_pending(&pending_key) .into_diagnostic()? - .is_none() ); Ok(()) } @@ -673,10 +677,10 @@ mod tests { keys::repo_metadata_key(&self.did), db::ser_repo_meta(&RepoMetadata::backfilling(index_id))?, ); - batch.insert( - &self.state.db.indexer.pending, - keys::pending_key(index_id), - keys::repo_key(&self.did), + self.state.db.indexer.stage_pending_insert( + &mut batch, + &self.did, + &keys::pending_key(index_id), ); for &(rkey, body) in records { self.state.db.indexer.stage_record( @@ -691,12 +695,9 @@ mod tests { /// mirrors the retry worker: a completed queue entry gives way to a fresh pending key. fn reindex_pending(&self, from: u64, to: u64) -> miette::Result<()> { let mut batch = self.state.db.inner.batch(); - batch.remove(&self.state.db.indexer.pending, keys::pending_key(from)); - batch.insert( - &self.state.db.indexer.pending, - keys::pending_key(to), - keys::repo_key(&self.did), - ); + let indexer = &self.state.db.indexer; + indexer.stage_pending_remove(&mut batch, &self.did, &keys::pending_key(from)); + indexer.stage_pending_insert(&mut batch, &self.did, &keys::pending_key(to)); batch.insert( &self.state.db.repo_metadata, keys::repo_metadata_key(&self.did), @@ -802,8 +803,7 @@ mod tests { self.state .db .indexer - .pending - .iter() + .pending_iter() .map(|guard| Ok(guard.into_inner().into_diagnostic()?.0.to_vec())) .collect() } @@ -1058,7 +1058,11 @@ mod tests { fixture.seed(&seeded, 7, &[("a", body.as_slice())])?; // 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)); + fixture.state.db.indexer.stage_pending_remove( + &mut batch, + &fixture.did, + &keys::pending_key(7), + ); batch.commit().into_diagnostic()?; assert_eq!( diff --git a/src/control/repos/indexer.rs b/src/control/repos/indexer.rs index 35a1756..a1474f4 100644 --- a/src/control/repos/indexer.rs +++ b/src/control/repos/indexer.rs @@ -21,8 +21,7 @@ impl ReposControl { self.0 .db .indexer - .pending - .range((start_bound, std::ops::Bound::Unbounded)) + .pending_range((start_bound, std::ops::Bound::Unbounded)) .map(move |g| { let (id_raw, did_key) = g.into_inner().into_diagnostic()?; let id = u64::from_be_bytes( @@ -124,20 +123,19 @@ impl ReposControl { // skip if already in pending queue let is_pending = db .indexer - .pending - .get(keys::pending_key(metadata.index_id)) - .into_diagnostic()? - .is_some(); + .is_pending(&keys::pending_key(metadata.index_id)) + .into_diagnostic()?; if !is_pending { metadata.tracked = true; // insert into pending with new index_id let old_pending = keys::pending_key(metadata.index_id); - txn.batch.remove(&db.indexer.pending, old_pending); + db.indexer + .stage_pending_remove(&mut txn.batch, did, &old_pending); metadata.index_id = rand::Rng::next_u64(&mut rand::rng()); - txn.batch.insert( - &db.indexer.pending, - keys::pending_key(metadata.index_id), - &did_key, + db.indexer.stage_pending_insert( + &mut txn.batch, + did, + &keys::pending_key(metadata.index_id), ); txn.batch.remove(&db.indexer.resync, &did_key); txn.batch.insert( @@ -246,10 +244,10 @@ impl ReposControl { &metadata_key, crate::db::ser_repo_meta(&metadata)?, ); - txn.batch.insert( - &db.indexer.pending, - keys::pending_key(metadata.index_id), - &did_key, + db.indexer.stage_pending_insert( + &mut txn.batch, + &did, + &keys::pending_key(metadata.index_id), ); txn.counts.add_repos(1); txn.transition_lifecycle(&did, GaugeState::Pending)?; @@ -305,8 +303,11 @@ impl ReposControl { &metadata_key, crate::db::ser_repo_meta(&metadata)?, ); - txn.batch - .remove(&db.indexer.pending, keys::pending_key(metadata.index_id)); + db.indexer.stage_pending_remove( + &mut txn.batch, + &did, + &keys::pending_key(metadata.index_id), + ); txn.batch.remove(&db.indexer.resync, &did_key); txn.transition_lifecycle(&did, GaugeState::Synced)?; untracked.push(did); @@ -354,8 +355,11 @@ impl ReposControl { let metadata_key = keys::repo_metadata_key(&did); if let Some(bytes) = db.repo_metadata.get(&metadata_key).into_diagnostic()? { let mut metadata = crate::db::deser_repo_meta(&bytes)?; - txn.batch - .remove(&db.indexer.pending, keys::pending_key(metadata.index_id)); + db.indexer.stage_pending_remove( + &mut txn.batch, + &did, + &keys::pending_key(metadata.index_id), + ); metadata.tracked = false; txn.batch.insert( &db.repo_metadata, @@ -775,11 +779,10 @@ mod tests { keys::repo_metadata_key(&did), crate::db::ser_repo_meta(&metadata)?, ); - batch.insert( - &state.db.indexer.pending, - keys::pending_key(7), - keys::repo_key(&did), - ); + state + .db + .indexer + .stage_pending_insert(&mut batch, &did, &keys::pending_key(7)); batch.insert(&state.db.crawler, keys::crawler_retry_key(&did), []); state .db @@ -820,10 +823,11 @@ mod tests { let db = &state.db; assert!(gone(db.repos.raw(), keys::repo_key(&did))); assert!(gone(db.repo_metadata.raw(), keys::repo_metadata_key(&did))); - assert!(gone( - db.indexer.pending.raw(), - keys::pending_key(7).to_vec() - )); + assert!( + !db.indexer + .is_pending(&keys::pending_key(7)) + .into_diagnostic()? + ); assert!(gone(db.crawler.raw(), keys::crawler_retry_key(&did))); assert!(gone( db.counts.raw(), diff --git a/src/control/repos/mod.rs b/src/control/repos/mod.rs index 28ff2be..32543ec 100644 --- a/src/control/repos/mod.rs +++ b/src/control/repos/mod.rs @@ -420,10 +420,8 @@ fn read_mini_doc_snapshot(db: &Db, did: &Did) -> Result> metadata .map(|metadata| { db.indexer - .pending - .get(keys::pending_key(metadata.index_id)) + .is_pending(&keys::pending_key(metadata.index_id)) .into_diagnostic() - .map(|value| value.is_some()) }) .transpose()? .unwrap_or(false) diff --git a/src/crawler/list_repos/producer.rs b/src/crawler/list_repos/producer.rs index dd4032a..7418df2 100644 --- a/src/crawler/list_repos/producer.rs +++ b/src/crawler/list_repos/producer.rs @@ -481,7 +481,7 @@ mod tests { let repo = crate::db::deser_repo_state(&bytes)?; assert!(!repo.active); assert_eq!(repo.status, crate::types::RepoStatus::Takendown); - assert_eq!(state.db.indexer.pending.iter().count(), 0); + assert_eq!(state.db.indexer.pending_iter().count(), 0); producer.abort(); worker.abort(); diff --git a/src/crawler/worker.rs b/src/crawler/worker.rs index d1584eb..c223675 100644 --- a/src/crawler/worker.rs +++ b/src/crawler/worker.rs @@ -230,10 +230,10 @@ impl CrawlerWorker { crate::db::ser_repo_meta(&metadata)?, ); #[cfg(feature = "indexer")] - txn.batch.insert( - &db.indexer.pending, - keys::pending_key(metadata.index_id), - &did_key, + db.indexer.stage_pending_insert( + &mut txn.batch, + &guard, + &keys::pending_key(metadata.index_id), ); txn.batch .remove(&db.crawler, keys::crawler_retry_key(&guard)); @@ -382,7 +382,7 @@ fn reconcile_listing( txn.transition_lifecycle(&listing.did, GaugeState::Synced)?; state.status = RepoStatus::Synced; } else { - enqueue_listing_backfill(txn, &listing.did, &did_key)?; + enqueue_listing_backfill(txn, &listing.did)?; txn.transition_lifecycle(&listing.did, GaugeState::Pending)?; state.status = RepoStatus::Desynchronized; } @@ -421,8 +421,11 @@ fn reconcile_listing( .into_diagnostic()? { let metadata = crate::db::deser_repo_meta(&metadata_bytes)?; - txn.batch - .remove(&db.indexer.pending, keys::pending_key(metadata.index_id)); + db.indexer.stage_pending_remove( + &mut txn.batch, + &listing.did, + &keys::pending_key(metadata.index_id), + ); } txn.batch.insert( &db.indexer.resync, @@ -455,15 +458,12 @@ fn listing_is_pending(db: &crate::db::Db, did: &Did) -> Result { return Ok(false); }; let metadata = crate::db::deser_repo_meta(&metadata_bytes)?; - Ok(db - .indexer - .pending - .get(keys::pending_key(metadata.index_id)) - .into_diagnostic()? - .is_some()) + db.indexer + .is_pending(&keys::pending_key(metadata.index_id)) + .into_diagnostic() } -fn enqueue_listing_backfill(txn: &mut crate::db::Txn<'_>, did: &Did, did_key: &[u8]) -> Result<()> { +fn enqueue_listing_backfill(txn: &mut crate::db::Txn<'_>, did: &Did) -> Result<()> { let db = txn.db; let metadata_key = keys::repo_metadata_key(did); let existing = db @@ -473,8 +473,8 @@ fn enqueue_listing_backfill(txn: &mut crate::db::Txn<'_>, did: &Did, did_key: &[ .map(|bytes| crate::db::deser_repo_meta(&bytes)) .transpose()?; if let Some(metadata) = &existing { - txn.batch - .remove(&db.indexer.pending, keys::pending_key(metadata.index_id)); + db.indexer + .stage_pending_remove(&mut txn.batch, did, &keys::pending_key(metadata.index_id)); } let mut metadata = existing.unwrap_or_else(|| RepoMetadata::backfilling(rand::random())); metadata.index_id = rand::random(); @@ -483,11 +483,8 @@ fn enqueue_listing_backfill(txn: &mut crate::db::Txn<'_>, did: &Did, did_key: &[ metadata_key, crate::db::ser_repo_meta(&metadata)?, ); - txn.batch.insert( - &db.indexer.pending, - keys::pending_key(metadata.index_id), - did_key, - ); + db.indexer + .stage_pending_insert(&mut txn.batch, did, &keys::pending_key(metadata.index_id)); Ok(()) } @@ -600,10 +597,10 @@ mod tests { ) .into_diagnostic()?; if pending { + let mut batch = db.inner.batch(); db.indexer - .pending - .insert(keys::pending_key(index_id), did_key) - .into_diagnostic()?; + .stage_pending_insert(&mut batch, &did, &keys::pending_key(index_id)); + batch.commit().into_diagnostic()?; } Ok(()) } @@ -638,11 +635,9 @@ mod tests { assert_eq!(apply(&db, &listing)?, ReconcileOutcome::AccountChanged); assert!( - db.indexer - .pending - .get(keys::pending_key(7)) + !db.indexer + .is_pending(&keys::pending_key(7)) .into_diagnostic()? - .is_none() ); let state = read_state(&db)?; assert!(!state.active); @@ -666,7 +661,7 @@ mod tests { let state = read_state(&db)?; assert!(!state.active); assert_eq!(state.status, RepoStatus::Takendown); - assert_eq!(db.indexer.pending.iter().count(), 0); + assert_eq!(db.indexer.pending_iter().count(), 0); assert!( db.indexer .resync @@ -690,7 +685,7 @@ mod tests { }; assert_eq!(apply(&db, &listing)?, ReconcileOutcome::Unchanged); - assert_eq!(db.indexer.pending.iter().count(), 0); + assert_eq!(db.indexer.pending_iter().count(), 0); Ok(()) } @@ -710,7 +705,7 @@ mod tests { }; assert_eq!(apply(&db, &listing)?, ReconcileOutcome::StateOnly); - assert_eq!(db.indexer.pending.iter().count(), 0); + assert_eq!(db.indexer.pending_iter().count(), 0); Ok(()) } @@ -727,7 +722,7 @@ mod tests { }; assert_eq!(apply(&db, &listing)?, ReconcileOutcome::StateOnly); - assert_eq!(db.indexer.pending.iter().count(), 1); + assert_eq!(db.indexer.pending_iter().count(), 1); let state = read_state(&db)?; assert!(state.active); assert_eq!(state.status, RepoStatus::Desynchronized); @@ -865,7 +860,7 @@ mod tests { .await?; assert!(events.try_recv().is_err()); - assert_eq!(state.db.indexer.pending.iter().count(), 1); + assert_eq!(state.db.indexer.pending_iter().count(), 1); Ok(()) } } diff --git a/src/db/indexer.rs b/src/db/indexer.rs index 2b31c7a..6ecc47a 100644 --- a/src/db/indexer.rs +++ b/src/db/indexer.rs @@ -377,8 +377,8 @@ pub(crate) fn erase_repo(db: &Db, did: &Did) -> Result { let metadata_key = keys::repo_metadata_key(did); if let Some(bytes) = db.repo_metadata.get(&metadata_key).into_diagnostic()? { let metadata = crate::db::deser_repo_meta(&bytes)?; - txn.batch - .remove(&db.indexer.pending, keys::pending_key(metadata.index_id)); + db.indexer + .stage_pending_remove(&mut txn.batch, did, &keys::pending_key(metadata.index_id)); txn.batch.remove(&db.repo_metadata, metadata_key); } let repo_key = keys::repo_key(did); diff --git a/src/db/keyspaces.rs b/src/db/keyspaces.rs index d87d97a..0316878 100644 --- a/src/db/keyspaces.rs +++ b/src/db/keyspaces.rs @@ -11,6 +11,8 @@ use std::cell::RefCell; use std::sync::Arc; use crate::config::Config; +#[cfg(feature = "indexer")] +use jacquard_common::types::did::Did; #[cfg(any( feature = "indexer", @@ -60,8 +62,9 @@ pub struct IndexerDb { pub(super) history: Ks, /// operator-redacted versions: `{record key} 00 {cid}` -> empty pub(super) redactions: Ks, - /// backfill queue of `{ID}` -> empty - pub pending: Ks, + /// backfill queue of `{ID}` -> `{DID}`, written only by `stage_pending_insert` + /// and `stage_pending_remove` + pub(super) pending: Ks, /// per-repo resync/retry state pub resync: Ks, /// live events buffered during backfill @@ -133,6 +136,43 @@ impl IndexerDb { { batch.insert(&self.history, key, value); } + + /// queues a backfill of `did` under `pending_key` + pub(crate) fn stage_pending_insert( + &self, + batch: &mut fjall::OwnedWriteBatch, + did: &Did, + pending_key: &[u8], + ) { + batch.insert(&self.pending, pending_key, super::keys::repo_key(did)); + } + + /// drops `did`'s queue entry under `pending_key` + pub(crate) fn stage_pending_remove( + &self, + batch: &mut fjall::OwnedWriteBatch, + _did: &Did, + pending_key: &[u8], + ) { + batch.remove(&self.pending, pending_key); + } + + pub(crate) fn is_pending(&self, pending_key: &[u8]) -> fjall::Result { + self.pending.contains_key(pending_key) + } + + /// the queue in `pending_key` order, each entry's value is its repo key + pub(crate) fn pending_iter(&self) -> fjall::Iter { + self.pending.iter() + } + + pub(crate) fn pending_range(&self, range: R) -> fjall::Iter + where + K: AsRef<[u8]>, + R: std::ops::RangeBounds, + { + self.pending.range(range) + } } #[cfg(feature = "indexer_stream")] diff --git a/src/db/lifecycle_counts.rs b/src/db/lifecycle_counts.rs index 5ffe287..9012669 100644 --- a/src/db/lifecycle_counts.rs +++ b/src/db/lifecycle_counts.rs @@ -124,13 +124,11 @@ impl<'a> LifecycleCountBatch<'a> { .map(|index_id| { self.db .indexer - .pending - .get(keys::pending_key(index_id)) + .is_pending(&keys::pending_key(index_id)) .into_diagnostic() }) .transpose()? - .flatten() - .is_some(); + .unwrap_or(false); let gauge = if is_pending { GaugeState::Pending @@ -154,13 +152,7 @@ impl<'a> LifecycleCountBatch<'a> { }; Ok(current.as_slice() == pending_key - && self - .db - .indexer - .pending - .get(current) - .into_diagnostic()? - .is_some()) + && self.db.indexer.is_pending(¤t).into_diagnostic()?) } fn stage_membership( @@ -251,7 +243,8 @@ mod tests { let mut batch = db.inner.batch(); batch.insert(&db.repos, &did_key, ser_repo_state(&state)?); batch.insert(&db.repo_metadata, metadata_key, ser_repo_meta(&metadata)?); - batch.insert(&db.indexer.pending, pending_key, did_key); + db.indexer + .stage_pending_insert(&mut batch, did, &pending_key); let mut lifecycle = db.lifecycle_counts(); lifecycle.transition(&mut batch, did, GaugeState::Pending)?; let lifecycle_reservation = lifecycle.stage(&mut batch); @@ -270,7 +263,8 @@ mod tests { GaugeState::Synced, )?; if applied { - batch.remove(&db.indexer.pending, pending_key); + db.indexer + .stage_pending_remove(&mut batch, did, &pending_key); } let lifecycle_reservation = lifecycle.stage(&mut batch); batch.commit().into_diagnostic()?; @@ -302,14 +296,15 @@ mod tests { let did = test_did()?; let stale_pending = enqueue_pending(&db, &did, 1)?; let current_pending = keys::pending_key(2); - let did_key = keys::repo_key(&did); let metadata_key = keys::repo_metadata_key(&did); let mut metadata = RepoMetadata::backfilling(2); metadata.tracked = true; let mut batch = db.inner.batch(); - batch.remove(&db.indexer.pending, stale_pending); - batch.insert(&db.indexer.pending, current_pending, &did_key); + db.indexer + .stage_pending_remove(&mut batch, &did, &stale_pending); + db.indexer + .stage_pending_insert(&mut batch, &did, ¤t_pending); batch.insert(&db.repo_metadata, metadata_key, ser_repo_meta(&metadata)?); let mut lifecycle = db.lifecycle_counts(); lifecycle.transition(&mut batch, &did, GaugeState::Pending)?; @@ -320,13 +315,7 @@ mod tests { assert_eq!(db.get_count_sync("pending"), 1); assert!(!complete_pending(&db, &did, stale_pending)?); assert_eq!(db.get_count_sync("pending"), 1); - assert!( - db.indexer - .pending - .get(current_pending) - .into_diagnostic()? - .is_some() - ); + assert!(db.indexer.is_pending(¤t_pending).into_diagnostic()?); Ok(()) } @@ -351,7 +340,8 @@ mod tests { GaugeState::Synced, )?; assert!(applied); - batch.remove(&db.indexer.pending, pending_key); + db.indexer + .stage_pending_remove(&mut batch, &did, &pending_key); let lifecycle_reservation = lifecycle.stage(&mut batch); batch.commit().into_diagnostic()?; db.persist()?; @@ -362,11 +352,9 @@ mod tests { let db = Db::open(&cfg)?; assert_eq!(db.get_count_sync("pending"), 0); assert!( - db.indexer - .pending - .get(keys::pending_key(1)) + !db.indexer + .is_pending(&keys::pending_key(1)) .into_diagnostic()? - .is_none() ); } diff --git a/src/ingest/indexer/shard.rs b/src/ingest/indexer/shard.rs index 3d082d0..1cee149 100644 --- a/src/ingest/indexer/shard.rs +++ b/src/ingest/indexer/shard.rs @@ -318,10 +318,8 @@ impl FirehoseWorker { let is_backfilling = if let Some(metadata_bytes) = metadata_bytes { let metadata = crate::db::deser_repo_meta(metadata_bytes.as_ref())?; db.indexer - .pending - .get(keys::pending_key(metadata.index_id)) + .is_pending(&keys::pending_key(metadata.index_id)) .into_diagnostic()? - .is_some() } else { false }; @@ -511,7 +509,6 @@ impl FirehoseWorker { if ctx.txn.lock_repo_and_is_excluded(did)? { return Ok(repo_state); } - let repo_key = keys::repo_key(did); let meta_key = keys::repo_metadata_key(did); let existing_metadata = db @@ -527,24 +524,19 @@ impl FirehoseWorker { }); let old_pkey = keys::pending_key(metadata.index_id); - let was_pending = had_metadata - && db - .indexer - .pending - .get(old_pkey.as_slice()) - .into_diagnostic()? - .is_some(); + let was_pending = had_metadata && db.indexer.is_pending(&old_pkey).into_diagnostic()?; // remove old pending entry and insert new one with fresh index_id if had_metadata { // only remove if we had one so we dont delete a random entry - ctx.txn.batch.remove(&db.indexer.pending, old_pkey); + db.indexer + .stage_pending_remove(&mut ctx.txn.batch, did, &old_pkey); } metadata.index_id = rand::random::(); - ctx.txn.batch.insert( - &db.indexer.pending, - keys::pending_key(metadata.index_id), - &repo_key, + db.indexer.stage_pending_insert( + &mut ctx.txn.batch, + did, + &keys::pending_key(metadata.index_id), ); ctx.txn .batch @@ -600,7 +592,10 @@ mod tests { keys::repo_metadata_key(&did), ser_repo_meta(&RepoMetadata::backfilling(index_id))?, ); - batch.insert(&state.db.indexer.pending, pending_key, keys::repo_key(&did)); + state + .db + .indexer + .stage_pending_insert(&mut batch, &did, &pending_key); batch.commit().into_diagnostic()?; let (tx, mut worker) = FirehoseWorker::new(state.clone(), 1); @@ -629,13 +624,11 @@ mod tests { assert_eq!(repo_state.status, RepoStatus::Synced); assert!(repo_state.active); assert!( - state + !state .db .indexer - .pending - .get(pending_key) + .is_pending(&pending_key) .into_diagnostic()? - .is_none() ); Ok(()) diff --git a/src/ops.rs b/src/ops.rs index 3278cdb..56edff1 100644 --- a/src/ops.rs +++ b/src/ops.rs @@ -85,8 +85,8 @@ pub(crate) fn delete_repo( let metadata_bytes = db.repo_metadata.get(&metadata_key).into_diagnostic()?; if let Some(metadata_bytes) = metadata_bytes { let metadata = db::deser_repo_meta(&metadata_bytes)?; - txn.batch - .remove(&db.indexer.pending, keys::pending_key(metadata.index_id)); + db.indexer + .stage_pending_remove(&mut txn.batch, did, &keys::pending_key(metadata.index_id)); } // we don't delete from repos, relay uses it as a tombstone @@ -135,7 +135,7 @@ pub fn transition_repo<'s>( match &new_status { RepoStatus::Synced => { lifecycle_transitions.push((did.clone().into_static(), GaugeState::Synced)); - batch.remove(&db.indexer.pending, pending_key.as_slice()); + db.indexer.stage_pending_remove(batch, did, &pending_key); batch.remove(&db.indexer.resync, &repo_key); } RepoStatus::Error(msg) => { @@ -144,7 +144,7 @@ pub fn transition_repo<'s>( did.clone().into_static(), GaugeState::Resync(Some(ResyncErrorKind::Generic)), )); - batch.remove(&db.indexer.pending, pending_key.as_slice()); + db.indexer.stage_pending_remove(batch, did, &pending_key); // TODO: we need to make errors have kind instead of "message" in repo status // and then pass it to resync error kind let resync_state = crate::types::ResyncState::Error { @@ -174,7 +174,7 @@ pub fn transition_repo<'s>( RepoStatus::Deleted => { lifecycle_transitions.push((did.clone().into_static(), GaugeState::Synced)); // terminal state: remove from queues, no resync entry needed - batch.remove(&db.indexer.pending, pending_key.as_slice()); + db.indexer.stage_pending_remove(batch, did, &pending_key); batch.remove(&db.indexer.resync, &repo_key); } RepoStatus::Desynchronized | RepoStatus::Throttled => { @@ -183,7 +183,7 @@ pub fn transition_repo<'s>( GaugeState::Resync(Some(ResyncErrorKind::Generic)), )); // like an error: remove from pending and schedule a resync attempt - batch.remove(&db.indexer.pending, pending_key.as_slice()); + db.indexer.stage_pending_remove(batch, did, &pending_key); let resync_state = crate::types::ResyncState::Error { kind: crate::types::ResyncErrorKind::Generic, retry_count: 0, -- 2.54.0