From 15190a2cec9cace7b7eda9ecbbe2c0989635e9a9 Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Mon, 5 Oct 2026 20:04:17 +0300 Subject: [PATCH] [backfill] keep what changed on a repo row while its backfill fetched a backfill read the repo row before resolving and fetching, then wrote that copy back with its new root, so a deactivation, takedown or identity change that landed in between was lost, and BackfillFinished then marked the repo synced and active. both persists now re-read the row under the repo lock they already hold and put only the backfill's own changes onto it. RepoState::rebased keeps the status and identity a writer didn't touch, never moves a root back to an older rev and only moves event times forward. the doc a backfill resolved is older than any identity written during the fetch, so that identity wins. BackfillFinished leaves a repo that went inactive with its status and a gone resync entry, like a backfill that found the account inactive. --- src/backfill/mod.rs | 25 ++++++++ src/backfill/sparse.rs | 25 +++++++- src/backfill/worker/process.rs | 7 +- src/backfill/worker/task.rs | 87 +++++++++++++++++++++++++ src/ingest/indexer/shard.rs | 80 ++++++++++++++++++++++- src/types.rs | 113 +++++++++++++++++++++++++++++++++ src/types/v2.rs | 2 +- 7 files changed, 332 insertions(+), 7 deletions(-) diff --git a/src/backfill/mod.rs b/src/backfill/mod.rs index 9511dda..dd432eb 100644 --- a/src/backfill/mod.rs +++ b/src/backfill/mod.rs @@ -8,6 +8,11 @@ pub mod worker; pub use worker::BackfillWorker; +use jacquard_common::IntoStatic; +use jacquard_common::types::did::Did; +use miette::IntoDiagnostic; + +use crate::db::{Db, keys}; use crate::types::RepoState; /// what writing a fetched repo came to @@ -22,3 +27,23 @@ pub(crate) enum Persisted { /// an untrack, a deletion or a newer requeue gets the last word Superseded, } + +/// the repo state a backfill writes, read under the repo lock its persist holds. `after` is +/// `before` plus the doc resolved before the fetch and the fetched root, put onto the row as +/// it is now so a status change made during the fetch survives. that doc is older than any +/// identity written during the fetch, so such an identity stays too. `None` if the row is gone +pub(crate) fn backfilled_state( + db: &Db, + did: &Did, + before: &RepoState<'static>, + mut after: RepoState<'static>, +) -> miette::Result>> { + let Some(bytes) = db.repos.get(keys::repo_key(did)).into_diagnostic()? else { + return Ok(None); + }; + let current = crate::db::deser_repo_state(&bytes)?.into_static(); + if current.identity() != before.identity() { + after.set_identity(current.identity()); + } + Ok(Some(after.rebased(before, current))) +} diff --git a/src/backfill/sparse.rs b/src/backfill/sparse.rs index 1a60aba..7347066 100644 --- a/src/backfill/sparse.rs +++ b/src/backfill/sparse.rs @@ -1,6 +1,6 @@ -use crate::backfill::Persisted; use crate::backfill::client::ThrottledHttpClient; use crate::backfill::error::BackfillError; +use crate::backfill::{Persisted, backfilled_state}; use crate::car::{Blocks, CarBlock, CommitBlocks, CommitCar}; use crate::config::{BackfillStrategy, RateTier}; use crate::db::types::{DbAction, DbRkey}; @@ -79,6 +79,7 @@ pub(crate) async fn process_did_sparse( did: &Did, pending_key: &[u8], pds: &url::Url, + before: &RepoState<'static>, state: RepoState<'static>, verify_signatures: bool, strategy: BackfillStrategy, @@ -293,8 +294,16 @@ pub(crate) async fn process_did_sparse( }) .collect::, _>>()?; - let result = - persist_sparse_backfill(app_state, did, pending_key, state, root_commit, records).await?; + let result = persist_sparse_backfill( + app_state, + did, + pending_key, + before.clone(), + state, + root_commit, + records, + ) + .await?; let (records, state) = match result { Persisted::Stored { records, state } => (records, *state), Persisted::Discarded => return Ok(SparseBackfillResult::Discarded), @@ -550,6 +559,7 @@ async fn persist_sparse_backfill( app_state: &Arc, did: &Did, pending_key: &[u8], + before: RepoState<'static>, mut state: RepoState<'static>, root_commit: Commit, records: Vec<(SmolStr, CarBlock)>, @@ -648,6 +658,9 @@ async fn persist_sparse_backfill( state.root = Some(root_commit); state.touch(); + let Some(state) = backfilled_state(db, &did, &before, state)? else { + return Ok(Persisted::Superseded); + }; record_txn.update_repo_state(&state)?; record_txn.finish()?; @@ -985,6 +998,11 @@ mod tests { state.filter.store(Arc::new(filter(&["sh.tangled.*"], &[]))); let did = Did::new_static(PROOF_DID).unwrap(); let mut batch = state.db.inner.batch(); + batch.insert( + &state.db.repos, + keys::repo_key(&did), + db::ser_repo_state(&RepoState::backfilling()).unwrap(), + ); batch.insert( &state.db.repo_metadata, keys::repo_metadata_key(&did), @@ -1006,6 +1024,7 @@ mod tests { &did, &pending_key, &pds, + &RepoState::backfilling(), RepoState::backfilling(), false, strategy, diff --git a/src/backfill/worker/process.rs b/src/backfill/worker/process.rs index 6b9f635..c1fad03 100644 --- a/src/backfill/worker/process.rs +++ b/src/backfill/worker/process.rs @@ -14,7 +14,6 @@ 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; @@ -22,6 +21,7 @@ use crate::backfill::filtered_car::{FilteredCar, FilteredRepo}; use crate::backfill::sparse::{ SparseBackfillResult, check_commit_did, fetch_blocks, process_did_sparse, }; +use crate::backfill::{Persisted, backfilled_state}; use crate::car::{Blocks, CarBlock}; use crate::config::{BackfillStrategy, RateTier}; use crate::db::types::{DbAction, DbRkey}; @@ -154,6 +154,7 @@ pub(super) async fn process_did( did, &pending_key, &pds, + &previous_state, state.clone(), verify_signatures, strategy, @@ -388,6 +389,7 @@ pub(super) async fn process_did( let did = did.clone(); let rev = root_commit.rev; let pending_key = pending_key.clone(); + let before = previous_state.clone(); tokio::task::spawn_blocking(move || { let mut count = 0; @@ -476,6 +478,9 @@ pub(super) async fn process_did( } state.root = Some(root_commit); state.touch(); + let Some(state) = backfilled_state(&app_state.db, &did, &before, state)? else { + return Ok(Persisted::Superseded); + }; record_txn.update_repo_state(&state)?; record_txn.finish()?; diff --git a/src/backfill/worker/task.rs b/src/backfill/worker/task.rs index abd8f4e..c08e139 100644 --- a/src/backfill/worker/task.rs +++ b/src/backfill/worker/task.rs @@ -839,6 +839,16 @@ mod tests { Ok(result) } + fn put_repo(&self, repo: &RepoState<'_>) -> miette::Result<()> { + let mut batch = self.state.db.inner.batch(); + batch.insert( + &self.state.db.repos, + keys::repo_key(&self.did), + db::ser_repo_state(repo)?, + ); + batch.commit().into_diagnostic() + } + fn repo(&self) -> miette::Result> { let bytes = self .state @@ -1403,6 +1413,83 @@ mod tests { Ok(()) } + #[tokio::test] + async fn a_deactivation_during_the_fetch_survives_it() -> 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 result = fixture + .run_changing(keys::pending_key(7), async { + // what the relay writes for an account event that lands while getRepo streams + let mut row = fixture.repo()?; + row.active = false; + row.status = RepoStatus::Deactivated; + fixture.put_repo(&row) + }) + .await?; + + assert!( + matches!(result, Ok(TaskDisposition::Finished)), + "{result:?}" + ); + let after = fixture.repo()?; + assert_eq!(after.status, RepoStatus::Deactivated); + assert!(!after.active); + assert_eq!( + after.root.map(|root| root.rev), + Some(crate::db::types::DbTid::from(&car.rev)) + ); + assert_eq!( + after.pds.as_deref(), + Some(fixture.pds.base.as_str()), + "nothing else touched the identity, so the doc the backfill resolved lands" + ); + assert_eq!(fixture.record_rkeys()?, ["a", "b"]); + Ok(()) + } + + #[tokio::test] + async fn an_identity_written_during_the_fetch_outlives_the_backfills_doc() -> miette::Result<()> + { + let fixture = Fixture::new().await?; + let seeded = rooted("3jzfcijpj2z2a")?; + fixture.seed(&seeded, 7, &[])?; + let body = post("a"); + let car = build_car("3jzfcijpj2z2b", &[("a", body.as_slice())]).await?; + fixture.pds.serve_car(car.bytes).await; + + let result = fixture + .run_changing(keys::pending_key(7), async { + let mut row = fixture.repo()?; + row.handle = Some(Handle::new_static("bob.test").into_diagnostic()?); + fixture.put_repo(&row) + }) + .await?; + + assert!( + matches!(result, Ok(TaskDisposition::Finished)), + "{result:?}" + ); + let after = fixture.repo()?; + assert_eq!(after.handle.as_ref().map(Handle::as_str), Some("bob.test")); + assert_eq!(after.pds.as_deref(), seeded.pds.as_deref()); + assert_eq!( + after.root.map(|root| root.rev), + Some(crate::db::types::DbTid::from(&car.rev)) + ); + assert_eq!(after.status, seeded.status); + 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/ingest/indexer/shard.rs b/src/ingest/indexer/shard.rs index 04a05d4..7804363 100644 --- a/src/ingest/indexer/shard.rs +++ b/src/ingest/indexer/shard.rs @@ -80,16 +80,23 @@ impl FirehoseWorker { match Self::drain_resync_buffer(&mut ctx, &did, repo_state) { Ok(RepoProcessResult::Ok(s)) => { + // a repo that went inactive while it backfilled keeps + // its status, the backfill only caught its records up + let status = if s.active { + RepoStatus::Synced + } else { + s.status.clone() + }; let res = ops::transition_repo( &mut ctx.txn.batch, &state.db, ctx.lifecycle_transitions, &did, s, - RepoStatus::Synced, + status, ); if let Err(e) = res { - error!(err = %e, "failed to transition to synced"); + error!(err = %e, "failed to transition the backfilled repo"); } } Ok(RepoProcessResult::NeedsBackfill(_)) => {} @@ -635,6 +642,75 @@ mod tests { Ok(()) } + #[tokio::test] + async fn backfill_finished_leaves_a_repo_that_went_inactive_inactive() -> Result<()> { + let tmp = tempfile::tempdir().into_diagnostic()?; + let config = Config { + database_path: tmp.path().to_path_buf(), + ..Default::default() + }; + let state = Arc::new(AppState::new(&config)?); + let did = Did::new("did:plc:backfillfinishedstatus") + .into_diagnostic()? + .into_static(); + + // deactivated while its getRepo streamed, so the task already dropped its queue entry + let mut deactivated = RepoState::backfilling(); + deactivated.active = false; + deactivated.status = RepoStatus::Deactivated; + let mut batch = state.db.inner.batch(); + batch.insert( + &state.db.repos, + keys::repo_key(&did), + db::ser_repo_state(&deactivated)?, + ); + batch.insert( + &state.db.repo_metadata, + keys::repo_metadata_key(&did), + ser_repo_meta(&RepoMetadata::backfilling(7))?, + ); + batch.commit().into_diagnostic()?; + + let (tx, mut worker) = FirehoseWorker::new(state.clone(), 1); + let rx = worker.rxs.pop().expect("single shard receiver"); + let shard_state = state.clone(); + let runtime = TokioHandle::current(); + let shard = std::thread::spawn(move || { + FirehoseWorker::shard(0, rx, shard_state, runtime); + }); + tx.send(IndexerMessage::BackfillFinished(did.clone())) + .await + .into_diagnostic()?; + drop(tx); + shard + .join() + .map_err(|_| miette::miette!("indexer shard panicked"))?; + + let state_bytes = state + .db + .repos + .get(keys::repo_key(&did)) + .into_diagnostic()? + .expect("repository state"); + let repo_state = db::deser_repo_state(&state_bytes)?; + assert_eq!(repo_state.status, RepoStatus::Deactivated); + assert!(!repo_state.active); + let resync = state + .db + .indexer + .resync + .get(keys::repo_key(&did)) + .into_diagnostic()? + .expect("a gone resync entry, like a backfill that found the account inactive"); + assert!(matches!( + rmp_serde::from_slice(&resync).into_diagnostic()?, + crate::types::ResyncState::Gone { + status: RepoStatus::Deactivated + } + )); + Ok(()) + } + #[test] fn resync_buffer_commit_msgpack_round_trip_preserves_path() { let cid = CidLink::from(jacquard_repo::mst::util::compute_cid(b"commit").unwrap()); diff --git a/src/types.rs b/src/types.rs index 5ae6e7f..d7bce11 100644 --- a/src/types.rs +++ b/src/types.rs @@ -249,6 +249,57 @@ impl<'i> RepoState<'i> { self.signing_key = doc.key.map(From::from); self.identity() != previous } + + #[cfg(feature = "indexer")] + pub(crate) fn set_identity(&mut self, identity: DocIdentity<'i>) { + let DocIdentity { + handle, + pds, + signing_key, + } = identity; + self.handle = handle; + self.pds = pds; + self.signing_key = signing_key; + } + + /// puts what one writer changed, going from `before` to `self`, onto `current`, the + /// row as it is now, so whatever another writer changed in between survives. if both + /// changed the same part `self` wins, but a root never goes back to an older rev and + /// the event times only move forward + #[cfg(feature = "indexer")] + pub(crate) fn rebased(self, before: &Self, current: Self) -> Self { + let account_changed = (self.active, &self.status) != (before.active, &before.status); + let identity_changed = self.identity() != before.identity(); + let root_changed = self.root != before.root; + + let (active, status) = if account_changed { + (self.active, self.status) + } else { + (current.active, current.status) + }; + let (handle, pds, signing_key) = if identity_changed { + (self.handle, self.pds, self.signing_key) + } else { + (current.handle, current.pds, current.signing_key) + }; + let root = match (self.root, current.root) { + (_, theirs) if !root_changed => theirs, + (Some(mine), Some(theirs)) if theirs.rev > mine.rev => Some(theirs), + (mine, _) => mine, + }; + Self { + active, + status, + root, + last_message_time: self.last_message_time.max(current.last_message_time), + last_identity_time: self.last_identity_time.max(current.last_identity_time), + last_account_time: self.last_account_time.max(current.last_account_time), + last_updated_at: self.last_updated_at.max(current.last_updated_at), + signing_key, + pds, + handle, + } + } } impl<'i> IntoStatic for RepoState<'i> { @@ -405,6 +456,68 @@ mod tests { Ok(()) } + #[cfg(feature = "indexer")] + fn commit_at(rev: &str) -> Commit { + Commit { + version: 3, + rev: crate::db::types::DbTid::from( + &jacquard_common::types::tid::Tid::new(rev).unwrap(), + ), + data: jacquard_repo::mst::util::compute_cid(rev.as_bytes()).unwrap(), + prev: None, + sig: bytes::Bytes::from_static(b"signature"), + } + } + + #[cfg(feature = "indexer")] + #[test] + fn a_rebase_keeps_what_only_the_other_writer_changed() -> miette::Result<()> { + let mut before = RepoState::synced(); + before.root = Some(commit_at("3jzfcijpj2z2a")); + before.handle = Some(Handle::new_static("alice.test").into_diagnostic()?); + before.last_message_time = Some(10); + + let mut mine = before.clone(); + mine.root = Some(commit_at("3jzfcijpj2z2b")); + mine.last_message_time = Some(30); + + let mut theirs = before.clone(); + theirs.active = false; + theirs.status = RepoStatus::Deactivated; + theirs.handle = Some(Handle::new_static("bob.test").into_diagnostic()?); + theirs.last_message_time = Some(20); + theirs.last_account_time = Some(20); + + let merged = mine.rebased(&before, theirs); + assert_eq!(merged.root, Some(commit_at("3jzfcijpj2z2b"))); + assert!(!merged.active); + assert_eq!(merged.status, RepoStatus::Deactivated); + assert_eq!(merged.handle.as_ref().map(Handle::as_str), Some("bob.test")); + assert_eq!(merged.last_message_time, Some(30)); + assert_eq!(merged.last_account_time, Some(20)); + Ok(()) + } + + #[cfg(feature = "indexer")] + #[test] + fn a_rebase_lets_this_writer_win_a_conflict_but_never_moves_the_root_back() { + let before = RepoState::synced(); + + let mut mine = before.clone(); + mine.status = RepoStatus::Takendown; + mine.active = false; + mine.root = Some(commit_at("3jzfcijpj2z2a")); + + let mut theirs = before.clone(); + theirs.status = RepoStatus::Throttled; + theirs.root = Some(commit_at("3jzfcijpj2z2b")); + + let merged = mine.rebased(&before, theirs); + assert_eq!(merged.status, RepoStatus::Takendown); + assert!(!merged.active); + assert_eq!(merged.root, Some(commit_at("3jzfcijpj2z2b"))); + } + #[test] fn commit_and_repo_status_conversion_contracts() { let did: Did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(); diff --git a/src/types/v2.rs b/src/types/v2.rs index a914fdb..d143a22 100644 --- a/src/types/v2.rs +++ b/src/types/v2.rs @@ -16,7 +16,7 @@ pub enum RepoStatus { Suspended, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] pub(crate) struct Commit { pub version: i64, pub rev: DbTid, -- 2.51.2