From 80f8e2a8582e35276f3180b064b23af21ef83ff2 Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Sat, 11 Jul 2026 18:44:22 +0300 Subject: [PATCH] [ops] extract record event emission and backlink hooks from apply_commit apply_commit's op loop is now unconditional: per-op stream/jetstream event emission lives in ops::record_events (RecordEmitter, a ZST no-op without indexer_stream) and backlinks indexing goes through ops::backlink_ops no-op-able hooks. ApplyCommitResults carries a single events: RecordEvents instead of three cfg'd fields. ops.rs drops from 28 inline cfg sites to 4. verified live: full-network ingest, live create events carry inline record payloads and cids over /stream, deletes flagged, ids monotonic. --- src/ingest/indexer/shard.rs | 6 +- src/ops.rs | 182 ++++++---------------------- src/ops/backlink_ops.rs | 77 ++++++++++++ src/ops/record_events.rs | 228 ++++++++++++++++++++++++++++++++++++ 4 files changed, 342 insertions(+), 151 deletions(-) create mode 100644 src/ops/backlink_ops.rs create mode 100644 src/ops/record_events.rs diff --git a/src/ingest/indexer/shard.rs b/src/ingest/indexer/shard.rs index a674c97..926f1d3 100644 --- a/src/ingest/indexer/shard.rs +++ b/src/ingest/indexer/shard.rs @@ -385,18 +385,18 @@ impl FirehoseWorker { { use std::sync::Arc; - let mut live_events = res.live_events; + let mut live_events = res.events.live_events; for evt in live_events.drain(..) { ctx.broadcast_events .push(BroadcastEvent::LiveRecord(Arc::new(evt))); } - if let Some(last_id) = res.last_event_id { + if let Some(last_id) = res.events.last_event_id { ctx.broadcast_events .push(BroadcastEvent::Persisted(last_id)); } } #[cfg(feature = "jetstream")] - ctx.jetstream_events.extend(res.jetstream_events); + ctx.jetstream_events.extend(res.events.jetstream_events); Ok(RepoProcessResult::Ok(repo_state)) } diff --git a/src/ops.rs b/src/ops.rs index bd88ff7..6ad9d06 100644 --- a/src/ops.rs +++ b/src/ops.rs @@ -1,8 +1,6 @@ use fjall::OwnedWriteBatch; use fjall::Slice; -#[cfg(feature = "backlinks")] -use jacquard_common::Data; use jacquard_common::IntoStatic; use jacquard_common::types::did::Did; use miette::{Context, IntoDiagnostic, Result}; @@ -19,16 +17,14 @@ use crate::types::{GaugeState, RepoState, RepoStatus, ResyncErrorKind, ResyncSta #[cfg(feature = "indexer_stream")] use { - crate::types::{ - AccountEvt, BroadcastEvent, IdentityEvt, LiveRecordEvent, MarshallableEvt, StoredData, - StoredEvent, - }, - jacquard_common::CowStr, + crate::types::{AccountEvt, BroadcastEvent, IdentityEvt, MarshallableEvt}, std::sync::atomic::Ordering, }; -#[cfg(feature = "jetstream")] -use crate::types::{JetstreamBroadcast, StoredJetstreamEvent}; +mod backlink_ops; +pub(crate) mod record_events; + +use record_events::{EmitOp, RecordEmitter, RecordEvents}; pub fn persist_to_resync_buffer(db: &Db, did: &Did, commit: &Commit) -> Result<()> { let key = keys::resync_buffer_key(did, DbTid::from(&commit.rev)); @@ -121,8 +117,7 @@ pub fn delete_repo( } // 5. remove backlinks for all records in this repo - #[cfg(feature = "backlinks")] - crate::backlinks::store::delete_repo(batch, &db.backlinks, did)?; + backlink_ops::delete_repo(batch, db, did)?; Ok(()) } @@ -223,12 +218,8 @@ pub struct ApplyCommitResults<'s> { pub repo_state: RepoState<'s>, pub records_delta: i64, pub blocks_count: i64, - #[cfg(feature = "indexer_stream")] - pub live_events: Vec, - #[cfg(feature = "indexer_stream")] - pub last_event_id: Option, - #[cfg(feature = "jetstream")] - pub jetstream_events: Vec, + #[cfg_attr(not(feature = "indexer_stream"), allow(dead_code))] + pub events: RecordEvents, } pub fn apply_commit<'s>( @@ -253,19 +244,7 @@ pub fn apply_commit<'s>( let mut records_delta = 0; let mut blocks_count = 0; let mut collection_deltas: HashMap<&str, i64> = HashMap::new(); - #[cfg(feature = "indexer_stream")] - let rev = DbTid::from(&commit.rev); - - #[cfg(feature = "indexer_stream")] - let should_broadcast_live = db.stream.event_tx.receiver_count() > 0; - #[cfg(feature = "indexer_stream")] - let mut live_events = Vec::new(); - #[cfg(feature = "indexer_stream")] - let mut last_event_id = None; - #[cfg(feature = "jetstream")] - let should_stage_jetstream = db.jetstream.tx.receiver_count() > 0; - #[cfg(feature = "jetstream")] - let mut jetstream_events = Vec::new(); + let mut emitter = RecordEmitter::new(state, &commit); for op in &commit.ops { let (collection, rkey) = parse_path(&op.path)?; @@ -279,12 +258,8 @@ pub fn apply_commit<'s>( let action = DbAction::try_from(op.action.as_str())?; - #[cfg(feature = "indexer_stream")] - let mut cid_for_event: Option = None; - #[cfg(feature = "indexer_stream")] - let mut block_inline_for_event: Option = None; - #[cfg(feature = "indexer_stream")] - let mut inline_block: Option = None; + let mut event_cid = None; + let mut event_block = None; match action { DbAction::Create | DbAction::Update => { @@ -295,16 +270,13 @@ pub fn apply_commit<'s>( .to_ipld() .into_diagnostic() .wrap_err("expected valid cid from relay")?; - #[cfg(feature = "indexer_stream")] - { - cid_for_event = Some(cid_ipld); - } - + event_cid = Some(cid_ipld); let Some(bytes) = parsed.blocks.get(&cid_ipld) else { return Err(miette::miette!( "block {cid} not found in CAR for record {did}/{collection}/{rkey}" )); }; + event_block = Some(bytes); let cid_raw = cid_ipld.to_bytes(); let block_key = Slice::from(keys::block_key(collection, &cid_raw)); @@ -319,28 +291,14 @@ pub fn apply_commit<'s>( records_delta += 1; *collection_deltas.entry(collection).or_default() += 1; } - #[cfg(feature = "backlinks")] - if let Ok(value) = serde_ipld_dagcbor::from_slice::(bytes.as_ref()) { - crate::backlinks::store::index_record( - batch, - &db.backlinks, - did.as_str(), - collection, - &rkey.to_smolstr(), - &value, - )?; - } - #[cfg(feature = "indexer_stream")] - if should_broadcast_live && !only_index_links { - // inline record bytes for live tailing so we don't have to load from blocks. - inline_block = Some(bytes.clone()); - } - } else { - #[cfg(feature = "indexer_stream")] - { - // in ephemeral mode, the event payload is the only place we persist the record. - block_inline_for_event = Some(bytes.clone()); - } + backlink_ops::index_record( + batch, + db, + did, + collection, + &rkey.to_smolstr(), + bytes.as_ref(), + )?; } } DbAction::Delete => { @@ -351,90 +309,23 @@ pub fn apply_commit<'s>( records_delta -= 1; *collection_deltas.entry(collection).or_default() -= 1; - #[cfg(feature = "backlinks")] - crate::backlinks::store::delete_record( - batch, - &db.backlinks, - did.as_str(), - collection, - &rkey.to_smolstr(), - )?; + backlink_ops::delete_record(batch, db, did, collection, &rkey.to_smolstr())?; } } }; - #[cfg(feature = "indexer_stream")] - { - let data = block_inline_for_event - .clone() - .map(StoredData::Block) - .or_else(|| { - (!only_index_links) - .then(|| cid_for_event.map(StoredData::Ptr)) - .flatten() - }) - .unwrap_or(StoredData::Nothing); - - let event_id = db.stream.next_event_id.fetch_add(1, Ordering::SeqCst); - last_event_id = Some(event_id); - let did_trimmed = TrimmedDid::from(did); - let collection = CowStr::Borrowed(collection); - - let evt = StoredEvent { - live: true, - did: did_trimmed.clone(), - rev, - collection: collection.clone(), - rkey: rkey.clone(), + emitter.emit( + batch, + db, + EmitOp { + did, + collection, + rkey: &rkey, action, - data: data.clone(), - }; - let bytes = rmp_serde::to_vec(&evt).into_diagnostic()?; - batch.insert(&db.stream.events, keys::event_key(event_id), bytes); - - #[cfg(feature = "jetstream")] - { - let jetstream = StoredJetstreamEvent::Commit { - did: did_trimmed.clone().into_static(), - collection: collection.clone().into_static(), - event_id, - live: true, - }; - let ephemeral = should_stage_jetstream - .then(|| { - crate::jetstream::build_ephemeral_from_stored( - did_trimmed.to_did().as_str(), - rev.to_tid().as_str(), - action.as_str(), - collection.as_str(), - rkey.to_smolstr().as_str(), - &data, - inline_block.as_ref(), - true, - ) - }) - .flatten(); - jetstream_events.push(crate::jetstream::stage_event( - batch, db, jetstream, ephemeral, - )?); - } - - if should_broadcast_live { - live_events.push(LiveRecordEvent { - id: event_id, - stored: StoredEvent { - live: evt.live, - did: did_trimmed.into_static(), - rev: evt.rev, - collection: collection.into_static(), - rkey, - action: evt.action, - data, - }, - inline_block, - }); - } - } + cid: event_cid, + block: event_block, + }, + )?; } // update counts @@ -448,12 +339,7 @@ pub fn apply_commit<'s>( repo_state, records_delta, blocks_count, - #[cfg(feature = "indexer_stream")] - live_events, - #[cfg(feature = "indexer_stream")] - last_event_id, - #[cfg(feature = "jetstream")] - jetstream_events, + events: emitter.finish(), }) } diff --git a/src/ops/backlink_ops.rs b/src/ops/backlink_ops.rs new file mode 100644 index 0000000..410ffb6 --- /dev/null +++ b/src/ops/backlink_ops.rs @@ -0,0 +1,77 @@ +//! backlinks index hooks for record ops. no-ops when the feature is off so +//! `apply_commit`/`delete_repo` stay unconditional. + +use fjall::OwnedWriteBatch; +use jacquard_common::types::did::Did; +use miette::Result; + +use crate::db::Db; + +#[cfg(feature = "backlinks")] +pub(crate) fn index_record( + batch: &mut OwnedWriteBatch, + db: &Db, + did: &Did<'_>, + collection: &str, + rkey: &str, + bytes: &[u8], +) -> Result<()> { + if let Ok(value) = serde_ipld_dagcbor::from_slice::(bytes) { + crate::backlinks::store::index_record( + batch, + &db.backlinks, + did.as_str(), + collection, + rkey, + &value, + )?; + } + Ok(()) +} + +#[cfg(not(feature = "backlinks"))] +#[inline(always)] +pub(crate) fn index_record( + _batch: &mut OwnedWriteBatch, + _db: &Db, + _did: &Did<'_>, + _collection: &str, + _rkey: &str, + _bytes: &[u8], +) -> Result<()> { + Ok(()) +} + +#[cfg(feature = "backlinks")] +pub(crate) fn delete_record( + batch: &mut OwnedWriteBatch, + db: &Db, + did: &Did<'_>, + collection: &str, + rkey: &str, +) -> Result<()> { + crate::backlinks::store::delete_record(batch, &db.backlinks, did.as_str(), collection, rkey) +} + +#[cfg(not(feature = "backlinks"))] +#[inline(always)] +pub(crate) fn delete_record( + _batch: &mut OwnedWriteBatch, + _db: &Db, + _did: &Did<'_>, + _collection: &str, + _rkey: &str, +) -> Result<()> { + Ok(()) +} + +#[cfg(feature = "backlinks")] +pub(crate) fn delete_repo(batch: &mut OwnedWriteBatch, db: &Db, did: &Did<'_>) -> Result<()> { + crate::backlinks::store::delete_repo(batch, &db.backlinks, did) +} + +#[cfg(not(feature = "backlinks"))] +#[inline(always)] +pub(crate) fn delete_repo(_batch: &mut OwnedWriteBatch, _db: &Db, _did: &Did<'_>) -> Result<()> { + Ok(()) +} diff --git a/src/ops/record_events.rs b/src/ops/record_events.rs new file mode 100644 index 0000000..20de4e0 --- /dev/null +++ b/src/ops/record_events.rs @@ -0,0 +1,228 @@ +//! per-op stream/jetstream event emission for `apply_commit`. +//! +//! when `indexer_stream` is enabled the emitter persists a `StoredEvent` per +//! record op, stages jetstream events, and collects live broadcasts; without +//! it the emitter is a zero-sized no-op, so `apply_commit` stays unconditional. + +#[cfg(feature = "indexer_stream")] +mod enabled { + use bytes::Bytes; + use fjall::OwnedWriteBatch; + use jacquard_common::CowStr; + use jacquard_common::IntoStatic; + use jacquard_common::types::cid::IpldCid; + use jacquard_common::types::did::Did; + use miette::{IntoDiagnostic, Result}; + use std::sync::atomic::Ordering; + + use crate::db::types::{DbAction, DbRkey, DbTid, TrimmedDid}; + use crate::db::{Db, keys}; + use crate::ingest::stream::Commit; + use crate::state::AppState; + use crate::types::{LiveRecordEvent, StoredData, StoredEvent}; + + /// one record op, as seen by the event emitter. + pub(crate) struct EmitOp<'a> { + pub(crate) did: &'a Did<'a>, + pub(crate) collection: &'a str, + pub(crate) rkey: &'a DbRkey, + pub(crate) action: DbAction, + /// record cid for create/update ops + pub(crate) cid: Option, + /// raw record block for create/update ops + pub(crate) block: Option<&'a Bytes>, + } + + pub(crate) struct RecordEmitter { + rev: DbTid, + ephemeral: bool, + only_index_links: bool, + should_broadcast_live: bool, + live_events: Vec, + last_event_id: Option, + #[cfg(feature = "jetstream")] + should_stage_jetstream: bool, + #[cfg(feature = "jetstream")] + jetstream_events: Vec, + } + + impl RecordEmitter { + pub(crate) fn new(state: &AppState, commit: &Commit) -> Self { + Self { + rev: DbTid::from(&commit.rev), + ephemeral: state.ephemeral, + only_index_links: state.only_index_links, + should_broadcast_live: state.db.stream.event_tx.receiver_count() > 0, + live_events: Vec::new(), + last_event_id: None, + #[cfg(feature = "jetstream")] + should_stage_jetstream: state.db.jetstream.tx.receiver_count() > 0, + #[cfg(feature = "jetstream")] + jetstream_events: Vec::new(), + } + } + + pub(crate) fn emit( + &mut self, + batch: &mut OwnedWriteBatch, + db: &Db, + op: EmitOp<'_>, + ) -> Result<()> { + // in ephemeral mode, the event payload is the only place we persist the record. + let block_inline_for_event = (self.ephemeral) + .then(|| op.block.cloned()) + .flatten(); + // inline record bytes for live tailing so we don't have to load from blocks. + let inline_block = (!self.ephemeral + && self.should_broadcast_live + && !self.only_index_links) + .then(|| op.block.cloned()) + .flatten(); + + let data = block_inline_for_event + .map(StoredData::Block) + .or_else(|| { + (!self.only_index_links) + .then(|| op.cid.map(StoredData::Ptr)) + .flatten() + }) + .unwrap_or(StoredData::Nothing); + + let event_id = db.stream.next_event_id.fetch_add(1, Ordering::SeqCst); + self.last_event_id = Some(event_id); + let did_trimmed = TrimmedDid::from(op.did); + let collection = CowStr::Borrowed(op.collection); + + let evt = StoredEvent { + live: true, + did: did_trimmed.clone(), + rev: self.rev, + collection: collection.clone(), + rkey: op.rkey.clone(), + action: op.action, + data: data.clone(), + }; + let bytes = rmp_serde::to_vec(&evt).into_diagnostic()?; + batch.insert(&db.stream.events, keys::event_key(event_id), bytes); + + #[cfg(feature = "jetstream")] + { + let jetstream = crate::types::StoredJetstreamEvent::Commit { + did: did_trimmed.clone().into_static(), + collection: collection.clone().into_static(), + event_id, + live: true, + }; + let ephemeral = self + .should_stage_jetstream + .then(|| { + crate::jetstream::build_ephemeral_from_stored( + did_trimmed.to_did().as_str(), + self.rev.to_tid().as_str(), + op.action.as_str(), + collection.as_str(), + op.rkey.to_smolstr().as_str(), + &data, + inline_block.as_ref(), + true, + ) + }) + .flatten(); + self.jetstream_events.push(crate::jetstream::stage_event( + batch, db, jetstream, ephemeral, + )?); + } + + if self.should_broadcast_live { + self.live_events.push(LiveRecordEvent { + id: event_id, + stored: StoredEvent { + live: evt.live, + did: did_trimmed.into_static(), + rev: evt.rev, + collection: collection.into_static(), + rkey: op.rkey.clone(), + action: evt.action, + data, + }, + inline_block, + }); + } + + Ok(()) + } + + pub(crate) fn finish(self) -> RecordEvents { + RecordEvents { + live_events: self.live_events, + last_event_id: self.last_event_id, + #[cfg(feature = "jetstream")] + jetstream_events: self.jetstream_events, + } + } + } + + /// events collected while applying one commit. + pub(crate) struct RecordEvents { + pub(crate) live_events: Vec, + pub(crate) last_event_id: Option, + #[cfg(feature = "jetstream")] + pub(crate) jetstream_events: Vec, + } +} + +#[cfg(feature = "indexer_stream")] +pub(crate) use enabled::*; + +#[cfg(not(feature = "indexer_stream"))] +mod noop { + use bytes::Bytes; + use fjall::OwnedWriteBatch; + use jacquard_common::types::cid::IpldCid; + use jacquard_common::types::did::Did; + use miette::Result; + + use crate::db::Db; + use crate::db::types::{DbAction, DbRkey}; + use crate::ingest::stream::Commit; + use crate::state::AppState; + + #[allow(dead_code)] + pub(crate) struct EmitOp<'a> { + pub(crate) did: &'a Did<'a>, + pub(crate) collection: &'a str, + pub(crate) rkey: &'a DbRkey, + pub(crate) action: DbAction, + pub(crate) cid: Option, + pub(crate) block: Option<&'a Bytes>, + } + + pub(crate) struct RecordEmitter; + + impl RecordEmitter { + #[inline(always)] + pub(crate) fn new(_state: &AppState, _commit: &Commit) -> Self { + Self + } + + #[inline(always)] + pub(crate) fn emit( + &mut self, + _batch: &mut OwnedWriteBatch, + _db: &Db, + _op: EmitOp<'_>, + ) -> Result<()> { + Ok(()) + } + + #[inline(always)] + pub(crate) fn finish(self) -> RecordEvents { + RecordEvents + } + } + + pub(crate) struct RecordEvents; +} + +#[cfg(not(feature = "indexer_stream"))] +pub(crate) use noop::*; -- 2.51.2