use fjall::OwnedWriteBatch; use jacquard_common::IntoStatic; use jacquard_common::types::did::Did; use miette::{Context, IntoDiagnostic, Result}; use tracing::debug; use crate::db::types::{DbAction, DbRkey, DbTid}; use crate::db::{self, Db, Txn, keys}; use crate::filter::FilterConfig; use crate::ingest::stream::Commit; use crate::ingest::validation::ValidatedCommit; use crate::state::AppState; use crate::types::{GaugeState, RepoState, RepoStatus, ResyncErrorKind, ResyncState}; #[cfg(feature = "indexer_stream")] use { crate::{db::outbox::StreamOutbox, types::IdentityEvt}, jacquard_common::types::string::Handle, tracing::error, }; pub(crate) mod backlink_ops; pub mod record_events; pub fn persist_to_resync_buffer( txn: &mut Txn<'_>, db: &Db, did: &Did, commit: &Commit, ) -> Result<()> { if txn.is_excluded(did)? { return Ok(()); } let key = keys::resync_buffer_key(did, DbTid::from(&commit.rev)); let value = rmp_serde::to_vec_named(commit).into_diagnostic()?; txn.batch.insert(&db.indexer.resync_buffer, key, value); debug!( did = %did, seq = commit.seq, "buffered commit to resync_buffer" ); Ok(()) } /// drops what `did` buffered at or below `rev`, for a backfill writing a car at `rev`. the car /// already has those commits, and the drain applying one on top would roll the car back pub fn stage_buffer_covered_by( batch: &mut OwnedWriteBatch, db: &Db, did: &Did, rev: DbTid, ) -> Result<()> { let covered = keys::resync_buffer_prefix(did)..=keys::resync_buffer_key(did, rev); for guard in db.indexer.resync_buffer.range(covered) { batch.remove(&db.indexer.resync_buffer, guard.key().into_diagnostic()?); } Ok(()) } /// broadcast identity and account events staged by `stage`. these are /// ephemeral: they take stream ids but are never persisted or replayed, so /// consumers that need current state fetch it themselves. #[cfg(feature = "indexer_stream")] pub async fn announce(db: &Db, stage: impl FnOnce(&mut StreamOutbox) + Send + 'static) { let result = db .run(move |db| { let mut txn = Txn::new(db); stage(&mut txn.outbox.stream); txn.commit() }) .await; if let Err(e) = result { error!(err = %e, "failed to announce stream events"); } } /// announce an identity the caller learned by re-resolving a doc rather than by /// receiving an event. #[cfg(feature = "indexer_stream")] pub async fn announce_identity(db: &Db, did: &Did, handle: Option) { let identity = IdentityEvt { did: did.clone().into_static(), handle, }; announce(db, move |events| events.push_identity(identity)).await; } /// the first txn of a deleted account's wipe: takes the repo off every queue and forgets what /// it buffered, then leaves the marker its owner shard moves the heads to history from, a /// step at a time. a running erase's marker stays, since the erase forgets the heads anyway pub(crate) fn start_wipe( txn: &mut Txn<'_>, db: &Db, did: &Did, repo_state: &RepoState, ) -> Result<()> { debug!(did = %did, "deleting repo"); let repo_key = keys::repo_key(did); let metadata_key = keys::repo_metadata_key(did); 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)?; 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 // todo: we should still delete it after some time txn.batch.remove(&db.indexer.resync, &repo_key); txn.batch.remove(&db.repo_metadata, &metadata_key); let resync_prefix = keys::resync_buffer_prefix(did); for guard in db.indexer.resync_buffer.prefix(&resync_prefix) { let k = guard.key().into_diagnostic()?; txn.batch.remove(&db.indexer.resync_buffer, k); } if matches!( db::read_bulk_job(db, did)?, Some(crate::types::BulkJobV1::Erase) ) { return Ok(()); } // the death rev sorts after every rev we have seen, including future-skewed ones, and the // marker keeps it so every step files the heads under the same one let death_rev = db::death_rev(repo_state.root.as_ref().map(|c| &c.rev)); db::stage_bulk_job( &mut txn.batch, db, did, &crate::types::BulkJobV1::Wipe { death_rev }, ) } pub fn transition_repo<'s>( txn: &mut Txn<'_>, lifecycle_transitions: &mut Vec<(Did, GaugeState)>, did: &Did, mut repo_state: RepoState<'s>, new_status: RepoStatus, ) -> Result> { debug!(did = %did, status = ?new_status, "updating repo status"); let db = txn.db; let batch = &mut txn.batch; let repo_key = keys::repo_key(did); let metadata_key = keys::repo_metadata_key(did); let metadata_bytes = db.repo_metadata.get(&metadata_key).into_diagnostic()?; if let Some(metadata_bytes) = metadata_bytes { let mut metadata = db::deser_repo_meta(&metadata_bytes)?; let pending_key = keys::pending_key(metadata.index_id); // the retry worker only requeues these, any other status would strand a deferred refetch let retries = repo_state.active && matches!( new_status, RepoStatus::Error(_) | RepoStatus::Desynchronized | RepoStatus::Throttled ); if !retries && std::mem::take(&mut metadata.refetch_deferred) { batch.insert( &db.repo_metadata, &metadata_key, db::ser_repo_meta(&metadata)?, ); } // manage queues match &new_status { RepoStatus::Synced => { lifecycle_transitions.push((did.clone().into_static(), GaugeState::Synced)); db.indexer.stage_pending_remove(batch, did, &pending_key); batch.remove(&db.indexer.resync, &repo_key); } // the relay writes these for inactive accounts too, and a pds won't hand those over // any more than a deactivated one, so a retry would only refetch into the same answer RepoStatus::Error(_) | RepoStatus::Desynchronized | RepoStatus::Throttled if !repo_state.active => { stage_gone(batch, db, lifecycle_transitions, did, &new_status)?; } RepoStatus::Error(msg) => { tracing::warn!("transitioning to error: {msg}"); lifecycle_transitions.push(( did.clone().into_static(), GaugeState::Resync(Some(ResyncErrorKind::Generic)), )); 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 { kind: crate::types::ResyncErrorKind::Generic, retry_count: 0, next_retry: chrono::Utc::now().timestamp(), }; batch.insert( &db.indexer.resync, &repo_key, rmp_serde::to_vec(&resync_state).into_diagnostic()?, ); } RepoStatus::Deactivated | RepoStatus::Takendown | RepoStatus::Suspended => { stage_gone(batch, db, lifecycle_transitions, did, &new_status)?; } RepoStatus::Deleted => { lifecycle_transitions.push((did.clone().into_static(), GaugeState::Synced)); // terminal state: remove from queues, no resync entry needed db.indexer.stage_pending_remove(batch, did, &pending_key); batch.remove(&db.indexer.resync, &repo_key); } RepoStatus::Desynchronized | RepoStatus::Throttled => { lifecycle_transitions.push(( did.clone().into_static(), GaugeState::Resync(Some(ResyncErrorKind::Generic)), )); // like an error: remove from pending and schedule a resync attempt db.indexer.stage_pending_remove(batch, did, &pending_key); let resync_state = crate::types::ResyncState::Error { kind: crate::types::ResyncErrorKind::Generic, retry_count: 0, next_retry: chrono::Utc::now().timestamp(), }; batch.insert( &db.indexer.resync, &repo_key, rmp_serde::to_vec(&resync_state).into_diagnostic()?, ); } } } repo_state.active = match new_status { RepoStatus::Synced => true, RepoStatus::Error(_) => repo_state.active, _ => false, }; repo_state.status = new_status; repo_state.touch(); txn.put_repo(did, &repo_state)?; Ok(repo_state) } /// parks a repo whose pds won't hand it over until its account comes back. the queue is the /// caller's to clear fn stage_gone( batch: &mut fjall::OwnedWriteBatch, db: &Db, lifecycle_transitions: &mut Vec<(Did, GaugeState)>, did: &Did, status: &RepoStatus, ) -> Result<()> { lifecycle_transitions.push((did.clone().into_static(), GaugeState::Resync(None))); let resync_state = ResyncState::Gone { status: status.clone(), }; batch.insert( &db.indexer.resync, keys::repo_key(did), rmp_serde::to_vec(&resync_state).into_diagnostic()?, ); Ok(()) } pub struct ApplyCommitResults<'s> { pub repo_state: RepoState<'s>, } pub fn apply_commit<'s>( txn: &mut crate::db::Txn<'_>, state: &AppState, mut repo_state: RepoState<'s>, validated: ValidatedCommit<'_>, filter: &FilterConfig, ) -> Result> { let commit = validated.commit; let parsed = validated.parsed_blocks; let did = &commit.repo; debug!(did = %did, commit = %commit.commit, "applying commit"); // the relay puts each commit's root on the row before it gets here, so the row can already // be a commit or more ahead, and the relay checks the next commit's since against it let root: crate::types::Commit = validated.commit_obj.into(); if repo_state .root .as_ref() .is_none_or(|current| current.rev < root.rev) { repo_state.root = Some(root); } repo_state.touch(); // the shard commits its batch even when this fails, so everything the commit itself can get // wrong is checked before the first write let mut writes = Vec::with_capacity(commit.ops.len()); for op in &commit.ops { let (collection, rkey) = parse_path(&op.path)?; if !filter.matches_collection(collection) { continue; } let rkey = DbRkey::new(rkey); let action = DbAction::try_from(op.action.as_str())?; let block = match action { DbAction::Create | DbAction::Update => { let Some(cid) = &op.cid else { continue; }; let cid_ipld = cid .to_ipld() .into_diagnostic() .wrap_err("expected valid cid from relay")?; let Some(block) = parsed.blocks.block(&cid_ipld) else { return Err(miette::miette!( "block {cid} not found in CAR for record {did}/{collection}/{rkey}" )); }; let block = block .checked() .map_err(|e| miette::miette!("record {did}/{collection}/{rkey}: {e}"))?; Some(block) } DbAction::Delete => None, }; writes.push((collection, rkey, action, block)); } let mut record_txn = txn.records(state, &DbTid::from(&commit.rev), did)?; for (collection, rkey, action, block) in &writes { match block { Some(block) => record_txn.put_record(collection, rkey, block, *action)?, None => record_txn.delete_record(collection, rkey)?, } } record_txn.update_repo_state(&repo_state)?; Ok(ApplyCommitResults { repo_state }) } pub fn parse_path(path: &str) -> Result<(&str, &str)> { let mut parts = path.splitn(2, '/'); let collection = parts.next().wrap_err("missing collection")?; let rkey = parts.next().wrap_err("missing rkey")?; if collection.is_empty() || jacquard_common::types::nsid::Nsid::new(collection).is_err() { miette::bail!("invalid collection NSID: {collection}"); } if rkey.is_empty() || jacquard_common::types::string::Rkey::new(rkey).is_err() { miette::bail!("invalid record key (rkey): {rkey}"); } Ok((collection, rkey)) } #[cfg(all(test, feature = "indexer_stream"))] mod tests { use super::*; use crate::config::Config; use crate::types::{AccountEvt, BroadcastEvent, MarshallableEvt}; use jacquard_common::{CowStr, types::string::Handle}; use tokio::sync::broadcast; fn next_ephemeral( rx: &mut broadcast::Receiver, ) -> Box> { let Ok(BroadcastEvent::Ephemeral(event)) = rx.try_recv() else { panic!("expected an ephemeral event"); }; event } #[tokio::test] async fn ephemeral_identity_and_account_json_contracts() { let tmp = tempfile::tempdir().unwrap(); let db = Db::open(&Config { database_path: tmp.path().to_path_buf(), ..Default::default() }) .unwrap(); let mut rx = db.stream.event_tx.subscribe(); let did = Did::new_static("did:web:example.com").unwrap(); announce_identity(&db, &did, Some(Handle::new_static("alice.test").unwrap())).await; let identity = next_ephemeral(&mut rx); let identity_json = serde_json::to_value(identity.as_ref()).unwrap(); assert_eq!(identity_json["type"], "identity"); assert_eq!(identity_json["identity"]["did"], "did:web:example.com"); assert_eq!(identity_json["identity"]["handle"], "alice.test"); assert!(identity_json.get("record").is_none()); assert!(identity_json.get("account").is_none()); let account = AccountEvt { did, active: false, status: Some(CowStr::Borrowed("stinkpot")), }; announce(&db, move |events| events.push_account(account)).await; let account = next_ephemeral(&mut rx); let account_json = serde_json::to_value(account.as_ref()).unwrap(); assert_eq!(account_json["id"].as_u64(), Some(identity.id + 1)); assert_eq!(account_json["type"], "account"); assert_eq!(account_json["account"]["did"], "did:web:example.com"); assert_eq!(account_json["account"]["active"], false); assert_eq!(account_json["account"]["status"], "stinkpot"); assert!(account_json.get("record").is_none()); assert!(account_json.get("identity").is_none()); // ephemeral events take ids but are never persisted assert_eq!(db.stream.ids.head(), Some(account.id)); assert!(db.stream.event_head().unwrap().is_none()); } }