From 542c37c0c7beae2ca4b7e84f30800fb979490348 Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Wed, 5 Aug 2026 18:23:14 +0300 Subject: [PATCH] [db] enforce stripe lock ordering and exclusion re-checks record mutations hold their DID's write-stripe lock from read through commit, and every multi-DID path acquires its complete lock set in one canonical order before lifecycle counts: the crawler groups each page by lock index instead of capturing nearly all 256 stripes behind one long redaction, and resync/track pre-acquire sorted stripe sets. writers re-check filter exclusion under the lock so a drop_repo racing in-flight work cannot be undone by a stale write; backfill tasks drop orphan pending keys for excluded DIDs before doing resolver or network work, and the gone/error retry scans take a per-repo transaction instead of one global pass. issue: hydrant-vvq, hydrant-ara --- src/backfill/manager.rs | 354 +++++++++++++++++++-------------- src/backfill/worker/process.rs | 9 + src/backfill/worker/task.rs | 117 ++++++++++- src/control/repos/indexer.rs | 12 +- src/crawler/worker.rs | 133 ++++++++----- src/db/txn.rs | 2 - 6 files changed, 424 insertions(+), 203 deletions(-) diff --git a/src/backfill/manager.rs b/src/backfill/manager.rs index b16baa4..f4e690d 100644 --- a/src/backfill/manager.rs +++ b/src/backfill/manager.rs @@ -2,18 +2,63 @@ use crate::db::types::TrimmedDid; use crate::db::{self, keys}; use crate::state::AppState; use crate::types::{GaugeState, ResyncState}; +use jacquard_common::types::did::Did; use miette::{IntoDiagnostic, Result}; use std::sync::Arc; use std::time::Duration; use tracing::{debug, error, info}; -pub fn queue_gone_backfills(state: &Arc) -> Result<()> { - debug!("scanning for deactivated/takendown repos to retry..."); - let mut transitions = 0usize; - +fn queue_resync_if( + state: &AppState, + did: &Did<'_>, + eligible: impl FnOnce(&ResyncState) -> bool, +) -> Result { let mut txn = db::Txn::new(&state.db); + if txn.lock_repo_and_is_excluded(did)? { + return Ok(false); + } + let did_key = keys::repo_key(did); + let Some(value) = state.db.indexer.resync.get(&did_key).into_diagnostic()? else { + return Ok(false); + }; + let resync_state: ResyncState = rmp_serde::from_slice(&value).into_diagnostic()?; + if !eligible(&resync_state) { + return Ok(false); + } + + let metadata_key = keys::repo_metadata_key(did); + let metadata_bytes = state + .db + .repo_metadata + .get(&metadata_key) + .into_diagnostic()? + .ok_or_else(|| miette::miette!("repo metadata not found"))?; + let mut metadata = crate::db::deser_repo_meta(&metadata_bytes)?; + let old_pending = keys::pending_key(metadata.index_id); + 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, + ); + txn.batch.insert( + &state.db.repo_metadata, + metadata_key, + crate::db::ser_repo_meta(&metadata)?, + ); + txn.transition_lifecycle(did, GaugeState::Pending)?; + txn.commit()?; + Ok(true) +} + +pub fn queue_gone_backfills(state: &Arc) -> Result<()> { + debug!("scanning for deactivated/takendown repos to retry..."); + let mut candidates = Vec::new(); for guard in state.db.indexer.resync.iter() { let (key, val) = guard.into_inner().into_diagnostic()?; let did = match TrimmedDid::try_from(key.as_ref()) { @@ -27,55 +72,28 @@ pub fn queue_gone_backfills(state: &Arc) -> Result<()> { if let Ok(resync_state) = rmp_serde::from_slice::(&val) && matches!(resync_state, ResyncState::Gone { .. }) { - debug!(did = %did, "queuing retry for gone repo"); - - let metadata_key = keys::repo_metadata_key(&did); - let metadata_bytes = match state - .db - .repo_metadata - .get(&metadata_key) - .map(|b| b.ok_or_else(|| miette::miette!("repo metadata not found"))) - .into_diagnostic() - .flatten() - { - Ok(b) => b, - Err(e) => { - error!(did = %did, err = %e, "failed to get repo metadata"); - continue; - } - }; - let mut metadata = crate::db::deser_repo_meta(&metadata_bytes)?; - - // move from resync back into pending - txn.batch.remove(&state.db.indexer.resync, key.clone()); - let old_pending = keys::pending_key(metadata.index_id); - txn.batch.remove(&state.db.indexer.pending, old_pending); - metadata.index_id = rand::random::(); - txn.batch.insert( - &state.db.indexer.pending, - keys::pending_key(metadata.index_id), - key.clone(), - ); - txn.batch.insert( - &state.db.repo_metadata, - &metadata_key, - crate::db::ser_repo_meta(&metadata)?, - ); - - txn.transition_lifecycle(&did, GaugeState::Pending)?; - transitions += 1; + candidates.push(did); } } - if transitions == 0 { - return Ok(()); + let mut transitions = 0usize; + for did in candidates { + debug!(did = %did, "queuing retry for gone repo"); + match queue_resync_if(state, &did, |resync| { + matches!(resync, ResyncState::Gone { .. }) + }) { + Ok(true) => transitions += 1, + Ok(false) => {} + Err(e) => { + error!(did = %did, err = %e, "failed to queue gone repo"); + db::check_poisoned_report(&e); + } + } + } + if transitions > 0 { + state.notify_backfill(); + info!(count = transitions, "queued gone backfills"); } - - txn.commit()?; - - state.notify_backfill(); - - info!(count = transitions, "queued gone backfills"); Ok(()) } @@ -83,14 +101,10 @@ pub fn retry_worker(state: Arc) { let db = &state.db; info!("retry worker started"); loop { - // sleep first (e.g., check every minute) std::thread::sleep(Duration::from_secs(60)); let now = chrono::Utc::now().timestamp(); - let mut transitions = 0usize; - - let mut txn = db::Txn::new(&state.db); - + let mut candidates = Vec::new(); for guard in db.indexer.resync.iter() { let (key, value) = match guard.into_inner() { Ok(t) => t, @@ -109,109 +123,153 @@ pub fn retry_worker(state: Arc) { }; match rmp_serde::from_slice::(&value) { - Ok(ResyncState::Error { next_retry, .. }) => { - if next_retry <= now { - let did_key = keys::repo_key(&did); - let is_pds_throttled = if let Ok(Some(state_bytes)) = db.repos.get(&did_key) - { - if let Ok(repo_state) = - rmp_serde::from_slice::(&state_bytes) - { - if let Some(pds_str) = &repo_state.pds { - if let Ok(pds_url) = url::Url::parse(pds_str.as_ref()) { - let now_ts = chrono::Utc::now().timestamp(); - state.throttler.snapshot(&pds_url).is_throttled(now_ts) - } else { - false - } - } else { - false - } - } else { - false - } - } else { - false - }; - - if is_pds_throttled { - continue; - } - - debug!(did = %did, "retrying backfill"); - - let metadata_key = keys::repo_metadata_key(&did); - let metadata_bytes = match state - .db - .repo_metadata - .get(&metadata_key) - .map(|b| b.ok_or_else(|| miette::miette!("repo metadata not found"))) - .into_diagnostic() - .flatten() - { - Ok(b) => b, - Err(e) => { - error!(did = %did, err = %e, "failed to get repo metadata"); - continue; - } - }; - let mut metadata = match crate::db::deser_repo_meta(metadata_bytes.as_ref()) - { - Ok(m) => m, - Err(e) => { - error!(did = %did, err = %e, "failed to deserialize repo metadata"); - continue; - } - }; - - let old_pending = keys::pending_key(metadata.index_id); - metadata.index_id = rand::random::(); - let new_pending = keys::pending_key(metadata.index_id); - let serialized_metadata = match crate::db::ser_repo_meta(&metadata) { - Ok(s) => s, - Err(e) => { - error!(did = %did, err = %e, "failed to serialize repo metadata"); - continue; - } - }; - if let Err(e) = txn.transition_lifecycle(&did, GaugeState::Pending) { - error!(did = %did, err = %e, "failed to transition lifecycle"); - continue; - } - - // move from resync back into pending - txn.batch.remove(&state.db.indexer.resync, key.clone()); - txn.batch.remove(&state.db.indexer.pending, old_pending); - txn.batch - .insert(&state.db.indexer.pending, new_pending, key.clone()); - txn.batch.insert( - &state.db.repo_metadata, - &metadata_key, - serialized_metadata, - ); - transitions += 1; - } - } - Ok(_) => { - // not an error state, do nothing + Ok(ResyncState::Error { next_retry, .. }) if next_retry <= now => { + candidates.push(did); } + Ok(_) => {} Err(e) => { error!(did = %did, err = %e, "failed to deserialize resync state"); - continue; } } } - if transitions == 0 { - continue; + let mut transitions = 0usize; + for did in candidates { + let did_key = keys::repo_key(&did); + let is_pds_throttled = db + .repos + .get(&did_key) + .ok() + .flatten() + .and_then(|bytes| { + let repo = rmp_serde::from_slice::(&bytes).ok()?; + repo.pds + .as_ref() + .and_then(|pds| url::Url::parse(pds.as_ref()).ok()) + }) + .is_some_and(|pds| { + state + .throttler + .snapshot(&pds) + .is_throttled(chrono::Utc::now().timestamp()) + }); + if is_pds_throttled { + continue; + } + + debug!(did = %did, "retrying backfill"); + match queue_resync_if( + &state, + &did, + |resync| matches!(resync, ResyncState::Error { next_retry, .. } if *next_retry <= now), + ) { + Ok(true) => transitions += 1, + Ok(false) => {} + Err(e) => { + error!(did = %did, err = %e, "failed to queue retry"); + db::check_poisoned_report(&e); + } + } + } + if transitions > 0 { + state.notify_backfill(); + info!(count = transitions, "queued retries"); } + } +} - if let Err(e) = txn.commit() { - error!(err = %e, "failed to commit batch"); - db::check_poisoned_report(&e); - continue; +#[cfg(test)] +mod tests { + use super::*; + use crate::config::Config; + use crate::types::{RepoMetadata, RepoState}; + + fn state() -> (tempfile::TempDir, AppState) { + let tmp = tempfile::tempdir().unwrap(); + let config = Config { + database_path: tmp.path().to_path_buf(), + ..Default::default() + }; + let state = AppState::new(&config).unwrap(); + (tmp, state) + } + + fn did() -> Did<'static> { + Did::new("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap() + } + + fn stage_retry(state: &AppState, excluded: bool) { + let did = did(); + let mut batch = state.db.inner.batch(); + batch.insert( + &state.db.repos, + keys::repo_key(&did), + crate::db::ser_repo_state(&RepoState::synced()).unwrap(), + ); + batch.insert( + &state.db.repo_metadata, + keys::repo_metadata_key(&did), + crate::db::ser_repo_meta(&RepoMetadata::backfilling(7)).unwrap(), + ); + batch.insert( + &state.db.indexer.resync, + keys::repo_key(&did), + rmp_serde::to_vec(&ResyncState::Gone { + status: crate::types::RepoStatus::Deactivated, + }) + .unwrap(), + ); + if excluded { + batch.insert( + &state.db.filter, + crate::db::filter::exclude_key(did.as_str()).unwrap(), + [], + ); } - state.notify_backfill(); - info!(count = transitions, "queued retries"); + batch.commit().unwrap(); + } + + #[test] + fn excluded_repo_is_not_requeued() { + let (_tmp, state) = state(); + stage_retry(&state, true); + + assert!( + !queue_resync_if(&state, &did(), |resync| { + matches!(resync, ResyncState::Gone { .. }) + }) + .unwrap() + ); + assert!( + state + .db + .indexer + .resync + .contains_key(keys::repo_key(&did())) + .unwrap() + ); + assert!(state.db.indexer.pending.is_empty().unwrap()); + } + + #[test] + fn eligible_repo_moves_from_resync_to_pending() { + let (_tmp, state) = state(); + stage_retry(&state, false); + + assert!( + queue_resync_if(&state, &did(), |resync| { + matches!(resync, ResyncState::Gone { .. }) + }) + .unwrap() + ); + assert!( + !state + .db + .indexer + .resync + .contains_key(keys::repo_key(&did())) + .unwrap() + ); + assert_eq!(state.db.indexer.pending.len().unwrap(), 1); } } diff --git a/src/backfill/worker/process.rs b/src/backfill/worker/process.rs index 56c6edd..961de55 100644 --- a/src/backfill/worker/process.rs +++ b/src/backfill/worker/process.rs @@ -180,6 +180,9 @@ pub(super) async fn process_did( FullRepoOutcome::NotFound => { warn!("repo not found, deleting"); let mut txn = DbTxn::new(db); + if txn.lock_repo_and_is_excluded(did)? { + return Ok(None); + } let applied = txn.transition_pending_key(did, pending_key.as_ref(), GaugeState::Synced)?; txn.batch.remove(&db.indexer.pending, pending_key.clone()); @@ -209,6 +212,9 @@ pub(super) async fn process_did( .db .run(move |db| { let mut txn = DbTxn::new(db); + if txn.lock_repo_and_is_excluded(&did)? { + return Ok::<_, miette::Report>(()); + } let applied = txn.transition_pending_key( &did, pending_key.as_ref(), @@ -466,6 +472,9 @@ async fn remove_discarded_repo( .db .run(move |db| { let mut txn = DbTxn::new(db); + if txn.lock_repo_and_is_excluded(&did)? { + return Ok(()); + } let applied = txn.transition_pending_key(&did, pending_key.as_ref(), GaugeState::Synced)?; txn.batch.remove(&db.indexer.pending, pending_key); diff --git a/src/backfill/worker/task.rs b/src/backfill/worker/task.rs index a722f91..0d8be2d 100644 --- a/src/backfill/worker/task.rs +++ b/src/backfill/worker/task.rs @@ -22,6 +22,50 @@ pub(super) enum TaskDisposition { Deferred, } +fn stage_excluded_pending_cleanup( + txn: &mut DbTxn<'_>, + db: &crate::db::Db, + did: &Did<'_>, + pending_key: &[u8], +) -> Result { + if !txn.lock_repo_and_is_excluded(did)? { + return Ok(false); + } + txn.transition_pending_key(did, pending_key, GaugeState::Synced)?; + txn.batch.remove(&db.indexer.pending, pending_key); + Ok(true) +} + +async fn remove_pending_if_excluded( + state: &AppState, + did: &Did<'static>, + pending_key: Slice, +) -> Result { + let did = did.clone(); + state + .db + .run(move |db| { + let mut txn = DbTxn::new(db); + if !stage_excluded_pending_cleanup(&mut txn, db, &did, pending_key.as_ref())? { + return Ok(false); + } + txn.commit()?; + Ok(true) + }) + .await +} + +async fn remove_pending(state: &AppState, pending_key: Slice) -> Result<()> { + state + .db + .run(move |db| { + let mut batch = db.inner.batch(); + batch.remove(&db.indexer.pending, pending_key); + batch.commit().into_diagnostic() + }) + .await +} + pub(super) async fn did_task( state: &Arc, http: ThrottledHttpClient, @@ -35,6 +79,13 @@ pub(super) async fn did_task( ) -> Result { let db = &state.db; + // Drop stale/excluded work before resolver, admission, or network I/O. A + // repository drop may race task spawning, and orphan pending keys are not + // necessarily the one named by current metadata. + if remove_pending_if_excluded(state, did, pending_key.clone()).await? { + return Ok(TaskDisposition::Finished); + } + match process_did( state, &http, @@ -55,6 +106,11 @@ pub(super) async fn did_task( move |db| { let did_key = keys::repo_key(&did); let mut txn = DbTxn::new(db); + if stage_excluded_pending_cleanup(&mut txn, db, &did, pending_key.as_ref())? + { + txn.commit()?; + return Ok::<_, miette::Report>(false); + } let applied = txn.transition_pending_key( &did, pending_key.as_ref(), @@ -96,7 +152,10 @@ pub(super) async fn did_task( } Ok(TaskDisposition::Finished) } - Ok(None) => Ok(TaskDisposition::Finished), + Ok(None) => { + remove_pending(state, pending_key).await?; + Ok(TaskDisposition::Finished) + } Err(BackfillError::PdsBusy) => Ok(TaskDisposition::Deferred), Err(BackfillError::Deleted) => { warn!("orphaned pending entry, cleaning up"); @@ -107,6 +166,11 @@ pub(super) async fn did_task( let pending_key = pending_key.clone(); move |db| { let mut txn = DbTxn::new(db); + if stage_excluded_pending_cleanup(&mut txn, db, &did, pending_key.as_ref())? + { + txn.commit()?; + return Ok::<_, miette::Report>(()); + } txn.transition_pending_key(&did, pending_key.as_ref(), GaugeState::Synced)?; txn.batch.remove(&db.indexer.pending, pending_key); txn.commit()?; @@ -163,7 +227,10 @@ pub(super) async fn did_task( let mut retry_count = match existing_state { Some(ResyncState::Error { retry_count, .. }) => retry_count, - Some(ResyncState::Gone { .. }) => return Ok(TaskDisposition::Finished), // should handle gone? original code didn't really? + Some(ResyncState::Gone { .. }) => { + remove_pending(state, pending_key).await?; + return Ok(TaskDisposition::Finished); + } None => 0, }; @@ -205,6 +272,11 @@ pub(super) async fn did_task( None }; let mut txn = DbTxn::new(db); + if stage_excluded_pending_cleanup(&mut txn, db, &did, pending_key.as_ref())? + { + txn.commit()?; + return Ok::<_, miette::Report>(false); + } let applied = txn.transition_pending_key( &did, pending_key.as_ref(), @@ -234,3 +306,44 @@ pub(super) async fn did_task( } } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::Config; + + #[tokio::test] + async fn excluded_orphan_pending_entry_is_removed_before_work() -> miette::Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let config = Config { + database_path: tmp.path().to_path_buf(), + ..Default::default() + }; + let state = AppState::new(&config)?; + let did = Did::new("did:plc:ewvi7nxzyoun6zhxrhs64oiz")?.into_static(); + let pending_key = keys::pending_key(7); + let mut batch = state.db.inner.batch(); + batch.insert( + &state.db.filter, + crate::db::filter::exclude_key(did.as_str())?, + [], + ); + batch.insert(&state.db.indexer.pending, pending_key, keys::repo_key(&did)); + batch.commit().into_diagnostic()?; + + assert!( + remove_pending_if_excluded(&state, &did, fjall::Slice::from(pending_key.as_slice()),) + .await? + ); + assert!( + state + .db + .indexer + .pending + .get(pending_key) + .into_diagnostic()? + .is_none() + ); + Ok(()) + } +} diff --git a/src/control/repos/indexer.rs b/src/control/repos/indexer.rs index a406d34..b83b006 100644 --- a/src/control/repos/indexer.rs +++ b/src/control/repos/indexer.rs @@ -101,6 +101,9 @@ impl ReposControl { } pub(crate) fn _resync(db: &Db, did: &Did<'_>, txn: &mut crate::db::Txn<'_>) -> Result { + if txn.lock_repo_and_is_excluded(did)? { + return Ok(false); + } let did_key = keys::repo_key(did); let metadata_key = keys::repo_metadata_key(did); @@ -169,6 +172,10 @@ impl ReposControl { .db .run(move |db| { let mut txn = crate::db::Txn::new(db); + txn.hold_record_lock_indexes_sorted( + dids.iter() + .map(|did| crate::db::record_lock_index_for_did(did)), + ); let mut queued: Vec> = Vec::new(); for did in dids { @@ -206,6 +213,10 @@ impl ReposControl { .db .run(move |db| { let mut txn = crate::db::Txn::new(db); + txn.hold_record_lock_indexes_sorted( + dids.iter() + .map(|did| crate::db::record_lock_index_for_did(did)), + ); let mut queued: Vec> = Vec::new(); for did in dids { @@ -740,7 +751,6 @@ mod tests { Ok(()) } - #[tokio::test] async fn drop_repo_excludes_untracks_and_redacts_all_body_storage() -> miette::Result<()> { let tmp = tempfile::tempdir().into_diagnostic()?; diff --git a/src/crawler/worker.rs b/src/crawler/worker.rs index fe22612..93fc6ca 100644 --- a/src/crawler/worker.rs +++ b/src/crawler/worker.rs @@ -66,8 +66,9 @@ pub(crate) struct RepoListing { /// a batch of confirmed repos from any crawler source ready to enqueue. /// /// `guards` hold the DIDs (via `Deref`) and keep their in-flight slots occupied -/// until the batch is committed to the database. cursor updates are committed -/// atomically with the repos/pending inserts. +/// until their lock group is committed to the database. cursor updates commit +/// only after every idempotent group, so a crash can replay but never skip a +/// partially applied page. pub(crate) struct CrawlerBatch { pub(super) guards: Vec, pub(super) listings: Vec, @@ -168,66 +169,93 @@ impl CrawlerWorker { return Ok(()); } - // filter already-known repos, build and commit the write batch, then return - // the surviving guards so they are dropped on the async side after commit. + // Group by the one record lock each DID needs. Acquiring a whole crawler + // page at once usually captures nearly all 256 locks and turns one long + // operator redaction into a global ingestion convoy. let app_state = self.state.clone(); - let surviving = tokio::time::timeout( - BLOCKING_TASK_TIMEOUT, - app_state.db.run(move |db| { + let surviving = app_state + .db + .run(move |db| { let mut rng: SmallRng = rand::make_rng(); - let mut txn = crate::db::Txn::new(db); - let mut surviving = Vec::new(); - let mut changed_listings = Vec::new(); + let mut listing_groups = std::collections::BTreeMap::>::new(); for listing in listings { - if reconcile_listing(&mut txn, &listing)?.emits_account() { - changed_listings.push(listing); - } + listing_groups + .entry(crate::db::record_lock_index_for_did(&listing.did)) + .or_default() + .push(listing); } + let mut guard_groups = std::collections::BTreeMap::>::new(); for guard in guards { - let did_key = keys::repo_key(&guard); - let metadata_key = keys::repo_metadata_key(&guard); - if db.repos.contains_key(&did_key).into_diagnostic()? { - continue; + guard_groups + .entry(crate::db::record_lock_index_for_did(&guard)) + .or_default() + .push(guard); + } + let lock_indices = listing_groups + .keys() + .chain(guard_groups.keys()) + .copied() + .collect::>(); + let mut surviving = Vec::new(); + let mut changed_listings = Vec::new(); + + for lock_index in lock_indices { + let mut txn = crate::db::Txn::new(db); + txn.hold_record_lock_index(lock_index); + for listing in listing_groups.remove(&lock_index).unwrap_or_default() { + if reconcile_listing(&mut txn, &listing)?.emits_account() { + changed_listings.push(listing); + } } - let state = RepoState::backfilling(); - let metadata = RepoMetadata::backfilling(rng.next_u64()); - txn.batch - .insert(&db.repos, &did_key, ser_repo_state(&state)?); - txn.batch.insert( - &db.repo_metadata, - &metadata_key, - crate::db::ser_repo_meta(&metadata)?, - ); - #[cfg(feature = "indexer")] - txn.batch.insert( - &db.indexer.pending, - keys::pending_key(metadata.index_id), - &did_key, - ); - // clear any stale retry entry, this DID is confirmed and being enqueued - txn.batch - .remove(&db.crawler, keys::crawler_retry_key(&guard)); - trace!(did = %*guard, "enqueuing repo"); - txn.counts.add_repos(1); - #[cfg(feature = "indexer")] - txn.transition_lifecycle(&guard, crate::types::GaugeState::Pending)?; - surviving.push(guard); + for guard in guard_groups.remove(&lock_index).unwrap_or_default() { + if db + .filter + .contains_key(crate::db::filter::exclude_key(guard.as_str())?) + .into_diagnostic()? + { + continue; + } + let did_key = keys::repo_key(&guard); + let metadata_key = keys::repo_metadata_key(&guard); + if db.repos.contains_key(&did_key).into_diagnostic()? { + continue; + } + let state = RepoState::backfilling(); + let metadata = RepoMetadata::backfilling(rng.next_u64()); + txn.batch + .insert(&db.repos, &did_key, ser_repo_state(&state)?); + txn.batch.insert( + &db.repo_metadata, + &metadata_key, + crate::db::ser_repo_meta(&metadata)?, + ); + #[cfg(feature = "indexer")] + txn.batch.insert( + &db.indexer.pending, + keys::pending_key(metadata.index_id), + &did_key, + ); + txn.batch + .remove(&db.crawler, keys::crawler_retry_key(&guard)); + trace!(did = %*guard, "enqueuing repo"); + txn.counts.add_repos(1); + #[cfg(feature = "indexer")] + txn.transition_lifecycle(&guard, crate::types::GaugeState::Pending)?; + surviving.push(guard); + } + txn.commit()?; } + + let mut txn = crate::db::Txn::new(db); for cursor in cursor_updates { stage_cursor(&mut txn.batch, db, cursor); } - // todo: repo state overwrites here are acceptable? txn.commit()?; Ok((surviving, changed_listings)) - }), - ) - .await - .map_err(|_| { - error!("enqueue batch timed out after {BLOCKING_TASK_TIMEOUT:?}"); - miette::miette!("enqueue batch timed out") - })? - .inspect_err(|e| error!(err = ?e, "enqueue batch commit failed")) - .unwrap_or_default(); + }) + .await + .inspect_err(|e| error!(err = ?e, "enqueue batch commit failed")) + .unwrap_or_default(); let (surviving, changed_listings) = surviving; let count = surviving.len(); @@ -304,7 +332,12 @@ fn reconcile_listing( txn: &mut crate::db::Txn<'_>, listing: &RepoListing, ) -> Result { + // production callers pre-acquire this listing's one lock group; this local + // check keeps direct callers on the same record-lock -> lifecycle hierarchy. let db = txn.db; + if txn.lock_repo_and_is_excluded(&listing.did)? { + return Ok(ReconcileOutcome::Unchanged); + } let did_key = keys::repo_key(&listing.did); let Some(state_bytes) = db.repos.get(&did_key).into_diagnostic()? else { if listing.active || !listing.create_if_missing { diff --git a/src/db/txn.rs b/src/db/txn.rs index 3206db7..04d1806 100644 --- a/src/db/txn.rs +++ b/src/db/txn.rs @@ -153,8 +153,6 @@ impl<'db> Txn<'db> { /// callers that can touch several DIDs must do this before initializing /// lifecycle counts; dynamic encounter-order locking can deadlock against /// another multi-repository operation. - // bulk-mutation callers adopt this with the lock-ordering commit - #[allow(dead_code)] #[cfg(feature = "indexer")] pub(crate) fn hold_record_lock_indexes_sorted(&mut self, locks: impl IntoIterator) { assert!( -- 2.51.2