//! events a transaction stages, written and broadcast when it commits.
//!
//! stream rows are staged here without a position. [`Outbox::commit`] gives
//! them one inside the [`Sequencer`](super::sequencer::Sequencer)'s critical
//! section, which is the only place stream positions come from.
use fjall::OwnedWriteBatch;
use miette::{IntoDiagnostic, Result};
use super::Db;
#[cfg(feature = "jetstream")]
mod jetstream;
#[cfg(feature = "relay")]
mod relay;
#[cfg(feature = "indexer_stream")]
mod stream;
#[cfg(feature = "jetstream")]
pub(crate) use jetstream::JetstreamOutbox;
#[cfg(feature = "relay")]
pub(crate) use relay::{RelayFrameRef, RelayOutbox};
#[cfg(feature = "indexer_stream")]
pub(crate) use stream::{StreamEventRef, StreamOutbox};
#[derive(Default)]
pub(crate) struct Outbox {
#[cfg(feature = "indexer_stream")]
pub(crate) stream: StreamOutbox,
#[cfg(feature = "relay")]
pub(crate) relay: RelayOutbox,
#[cfg(feature = "jetstream")]
pub(crate) jetstream: JetstreamOutbox,
}
/// the positions a commit gave its events.
#[derive(Default)]
pub(crate) struct Assigned {
#[cfg(feature = "indexer_stream")]
first_stream_id: Option<u64>,
}
#[cfg(feature = "indexer_stream")]
impl Assigned {
pub(crate) fn stream_id(&self, event: StreamEventRef) -> Option<u64> {
self.first_stream_id
.map(|first| first + event.index() as u64)
}
}
#[cfg(not(any(feature = "indexer_stream", feature = "relay")))]
impl Outbox {
pub(super) fn commit(
self,
_db: &Db,
batch: OwnedWriteBatch,
committed: impl FnOnce(),
) -> Result<Assigned> {
batch.commit().into_diagnostic()?;
committed();
Ok(Assigned::default())
}
}
#[cfg(any(feature = "indexer_stream", feature = "relay"))]
impl Outbox {
/// commit `batch` together with this outbox's rows, run `committed`, then
/// broadcast them.
pub(super) fn commit(
self,
db: &Db,
batch: OwnedWriteBatch,
committed: impl FnOnce(),
) -> Result<Assigned> {
if self.is_empty() {
batch.commit().into_diagnostic()?;
committed();
return Ok(Assigned::default());
}
db.sequencer.commit(
batch,
|assign, batch| self.stage(db, assign, batch),
|staged| {
committed();
staged.publish(db)
},
)
}
fn is_empty(&self) -> bool {
#[cfg(feature = "indexer_stream")]
if !self.stream.is_empty() {
return false;
}
#[cfg(feature = "relay")]
if !self.relay.is_empty() {
return false;
}
#[cfg(feature = "jetstream")]
if !self.jetstream.is_empty() {
return false;
}
true
}
fn stage<'a>(
self,
db: &'a Db,
assign: &mut super::sequencer::Assign<'a>,
batch: &mut OwnedWriteBatch,
) -> Result<Staged> {
#[cfg(feature = "indexer_stream")]
let stream = self.stream.stage(&db.stream, assign, batch)?;
#[cfg(feature = "relay")]
let relay = self.relay.stage(&db.relay, assign, batch)?;
// jetstream rows refer to the upstream rows staged just above
#[cfg(all(feature = "jetstream", feature = "indexer_stream"))]
let jetstream = self
.jetstream
.stage(&db.jetstream, assign, batch, &stream)?;
#[cfg(all(feature = "jetstream", feature = "relay"))]
let jetstream = self.jetstream.stage(&db.jetstream, assign, batch, &relay)?;
Ok(Staged {
#[cfg(feature = "indexer_stream")]
stream,
#[cfg(feature = "relay")]
relay,
#[cfg(feature = "jetstream")]
jetstream,
})
}
}
#[cfg(any(feature = "indexer_stream", feature = "relay"))]
struct Staged {
#[cfg(feature = "indexer_stream")]
stream: stream::StagedStream,
#[cfg(feature = "relay")]
relay: relay::StagedRelay,
#[cfg(feature = "jetstream")]
jetstream: jetstream::StagedJetstream,
}
#[cfg(any(feature = "indexer_stream", feature = "relay"))]
impl Staged {
fn publish(self, db: &Db) -> Assigned {
#[cfg(feature = "relay")]
self.relay.publish(&db.relay);
#[cfg(feature = "jetstream")]
self.jetstream.publish(&db.jetstream);
Assigned {
#[cfg(feature = "indexer_stream")]
first_stream_id: self.stream.publish(&db.stream),
}
}
}
#[cfg(all(test, feature = "indexer_stream"))]
mod tests {
use std::sync::Arc;
use jacquard_common::types::did::Did;
use jacquard_common::types::string::Tid;
use crate::config::Config;
use crate::db::Txn;
use crate::db::types::{DbAction, DbRkey, DbTid};
use crate::state::AppState;
use crate::types::BroadcastEvent;
const BACKFILLED: &str = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa";
const LIVE: &str = "did:plc:zzzzzzzzzzzzzzzzzzzzzzzz";
const COLLECTION: &str = "app.bsky.feed.post";
fn state() -> (tempfile::TempDir, Arc<AppState>) {
let tmp = tempfile::tempdir().unwrap();
let state = AppState::new(&Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})
.unwrap();
(tmp, Arc::new(state))
}
fn put(records: &mut crate::db::txn::RecordTxn<'_, '_, '_>, n: usize) {
let block = crate::car::CarBlock::from_body(
serde_ipld_dagcbor::to_vec(&serde_json::json!({ "$type": COLLECTION, "n": n }))
.unwrap(),
);
let rkey = DbRkey::Str(smol_str::format_smolstr!("r{n}"));
records
.put_record(COLLECTION, &rkey, &block, DbAction::Create)
.unwrap();
}
/// a backfill batch is staged, a live commit lands, then the backfill
/// commits: the order ingest produces when both run at once.
fn interleave(state: &AppState, backfilled: usize) {
let rev = DbTid::from(&Tid::now_0());
let did = Did::new_static(BACKFILLED).unwrap();
let mut backfill = Txn::new(&state.db);
let mut records = backfill.backfill_records(state, &rev, &did).unwrap();
(0..backfilled).for_each(|n| put(&mut records, n));
let did = Did::new_static(LIVE).unwrap();
let mut live = Txn::new(&state.db);
let mut records = live.records(state, &rev, &did).unwrap();
put(&mut records, backfilled);
live.commit().unwrap();
backfill.commit().unwrap();
}
fn event_did(state: &AppState, id: u64) -> String {
let row = state
.db
.stream
.event_range(crate::db::keys::event_key(id)..=crate::db::keys::event_key(id))
.next()
.unwrap()
.value()
.unwrap();
let event: crate::types::StoredEvent<'_> = rmp_serde::from_slice(&row).unwrap();
event.did.to_did().as_str().to_owned()
}
#[test]
fn stream_ids_follow_commit_order() {
let (_tmp, state) = state();
let mut rx = state.db.stream.event_tx.subscribe();
interleave(&state, 3);
// the live commit took the first id even though it staged last
let first = state.db.stream.ids.head().unwrap() - 3;
assert_eq!(event_did(&state, first), LIVE);
for id in first + 1..=first + 3 {
assert_eq!(event_did(&state, id), BACKFILLED);
}
let Ok(BroadcastEvent::LiveRecord(live)) = rx.try_recv() else {
panic!("expected the live record first");
};
assert_eq!(live.id, first);
let Ok(BroadcastEvent::Persisted(head)) = rx.try_recv() else {
panic!("expected the backfill's persisted marker");
};
assert_eq!(head, first + 3);
assert!(rx.try_recv().is_err());
}
#[test]
fn the_committed_hook_runs_before_the_broadcast() {
let (_tmp, state) = state();
let rx = state.db.stream.event_tx.subscribe();
let mut outbox = super::Outbox::default();
outbox.stream.push_identity(crate::types::IdentityEvt {
did: Did::new_static(LIVE).unwrap(),
handle: None,
});
let mut ran = false;
outbox
.commit(&state.db, state.db.inner.batch(), || {
assert!(rx.is_empty(), "broadcast before the hook ran");
ran = true;
})
.unwrap();
assert!(ran);
assert!(!rx.is_empty());
}
#[test]
fn stream_ids_are_not_reused_after_a_restart() {
let tmp = tempfile::tempdir().unwrap();
let config = Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
};
let head = {
let db = crate::db::Db::open(&config).unwrap();
// identity events take ids but leave no rows
let mut txn = Txn::new(&db);
txn.outbox.stream.push_identity(crate::types::IdentityEvt {
did: Did::new_static(LIVE).unwrap(),
handle: None,
});
txn.commit().unwrap();
db.persist().unwrap();
db.stream.ids.head()
};
let db = crate::db::Db::open(&config).unwrap();
let head = head.unwrap();
// past every id taken, and no further than one reserved block
let next = db.stream.ids.next();
assert!(head < next && next <= head + 1 + crate::db::sequencer::SEQUENCE_BLOCK);
}
#[cfg(feature = "jetstream")]
#[test]
fn jetstream_keys_follow_commit_order() {
use crate::types::StoredJetstreamEvent;
let (_tmp, state) = state();
interleave(&state, 3);
let rows: Vec<_> = state
.db
.jetstream
.events
.iter()
.map(|guard| {
let (key, value) = guard.into_inner().unwrap();
let position = crate::db::keys::parse_jetstream_event_key(&key).unwrap();
let event: StoredJetstreamEvent<'_> = rmp_serde::from_slice(&value).unwrap();
let StoredJetstreamEvent::Commit { event_id, live, .. } = event else {
panic!("expected commit rows");
};
(position, event_id, live)
})
.collect();
assert_eq!(rows.len(), 4);
assert!(rows.windows(2).all(|pair| {
let ((time_a, id_a), _, _) = pair[0];
let ((time_b, id_b), _, _) = pair[1];
time_a < time_b && id_a + 1 == id_b
}));
let (_, live_event, live) = rows[0];
assert!(live);
assert_eq!(event_did(&state, live_event), LIVE);
for &(_, event_id, live) in &rows[1..] {
assert!(!live);
assert_eq!(event_did(&state, event_id), BACKFILLED);
}
}
}
#[cfg(all(test, feature = "relay"))]
mod relay_tests {
use jacquard_common::types::string::Did;
use crate::config::Config;
use crate::db::{Db, Txn};
use crate::ingest::stream::{Datetime, Identity, SubscribeReposMessage, decode_frame};
use crate::types::RelayBroadcast;
const FIRST: &str = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa";
const SECOND: &str = "did:plc:zzzzzzzzzzzzzzzzzzzzzzzz";
fn identity(did: &'static str) -> Identity {
Identity {
did: Did::new_static(did).unwrap(),
handle: None,
seq: 0,
time: Datetime::try_from("2026-08-10T12:34:56Z".to_owned()).unwrap(),
}
}
fn stage<'a>(db: &'a Db, did: &'static str) -> Txn<'a> {
let mut txn = Txn::new(db);
let _frame = txn.outbox.relay.push_identity(identity(did));
#[cfg(feature = "jetstream")]
txn.outbox.jetstream.push(
crate::types::StoredJetstreamEvent::RelayIdentity {
did: crate::db::types::TrimmedDid::from(&Did::new_static(did).unwrap())
.into_static(),
relay_seq: _frame,
},
None,
);
txn
}
#[test]
fn relay_seqs_continue_after_their_rows_are_pruned() {
let tmp = tempfile::tempdir().unwrap();
let config = Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
};
let head = {
let db = Db::open(&config).unwrap();
stage(&db, FIRST).commit().unwrap();
stage(&db, SECOND).commit().unwrap();
// ttl pruning can remove every row
let mut batch = db.inner.batch();
for guard in db.relay.events.iter() {
batch.remove(&db.relay.events, guard.key().unwrap());
}
batch.commit().unwrap();
db.persist().unwrap();
db.relay.seqs.head()
};
let db = Db::open(&config).unwrap();
let head = head.unwrap();
// past every seq taken, and no further than one reserved block
let next = db.relay.seqs.next();
assert!(head < next && next <= head + 1 + crate::db::sequencer::SEQUENCE_BLOCK);
}
#[test]
fn a_frame_that_fails_to_encode_is_dropped_alone() {
let tmp = tempfile::tempdir().unwrap();
let db = Db::open(&Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})
.unwrap();
let mut rx = db.relay.broadcast_tx.subscribe();
let mut txn = stage(&db, FIRST);
txn.outbox.relay.push_unencodable();
let _ = txn.outbox.relay.push_identity(identity(SECOND));
txn.batch.insert(
&db.cursors,
b"test_cursor".as_slice(),
b"written".as_slice(),
);
txn.commit().unwrap();
// the rest of the transaction lands, and the lost frame's seq stays a hole
assert!(db.cursors.get(b"test_cursor").unwrap().is_some());
let seqs: Vec<u64> = db
.relay
.events
.iter()
.map(|guard| u64::from_be_bytes(guard.key().unwrap().as_ref().try_into().unwrap()))
.collect();
let head = db.relay.seqs.head().unwrap();
assert_eq!(seqs, [head - 2, head]);
let broadcast: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok()).collect();
assert!(matches!(
broadcast.as_slice(),
[
RelayBroadcast::Ephemeral(first, _),
RelayBroadcast::Persisted(hole),
RelayBroadcast::Ephemeral(second, _),
] if *first == head - 2 && *hole == head - 1 && *second == head
));
}
#[test]
fn relay_seqs_follow_commit_order() {
let tmp = tempfile::tempdir().unwrap();
let db = Db::open(&Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})
.unwrap();
let mut rx = db.relay.broadcast_tx.subscribe();
let first = stage(&db, FIRST);
stage(&db, SECOND).commit().unwrap();
first.commit().unwrap();
let frames: Vec<_> = db
.relay
.events
.iter()
.map(|guard| {
let (key, frame) = guard.into_inner().unwrap();
let seq = u64::from_be_bytes(key.as_ref().try_into().unwrap());
let SubscribeReposMessage::Identity(identity) = decode_frame(&frame).unwrap()
else {
panic!("expected identity frames");
};
assert_eq!(identity.seq as u64, seq, "a frame carries its own seq");
(seq, identity.did.as_str().to_owned())
})
.collect();
let head = db.relay.seqs.head().unwrap();
assert_eq!(
frames,
[(head - 1, SECOND.to_owned()), (head, FIRST.to_owned())]
);
let broadcast: Vec<_> = std::iter::from_fn(|| rx.try_recv().ok())
.map(|event| match event {
RelayBroadcast::Ephemeral(seq, _) => seq,
RelayBroadcast::Persisted(seq) => panic!("unexpected marker at {seq}"),
})
.collect();
assert_eq!(broadcast, [head - 1, head]);
#[cfg(feature = "jetstream")]
for guard in db.jetstream.events.iter() {
let (_, value) = guard.into_inner().unwrap();
let event: crate::types::StoredJetstreamEvent<'_> =
rmp_serde::from_slice(&value).unwrap();
let crate::types::StoredJetstreamEvent::RelayIdentity { did, relay_seq } = event else {
panic!("expected relay identity rows");
};
let frame = db
.relay
.events
.get(crate::db::keys::relay_event_key(relay_seq))
.unwrap()
.unwrap();
let SubscribeReposMessage::Identity(identity) = decode_frame(&frame).unwrap() else {
panic!("jetstream row points at a non-identity frame");
};
assert_eq!(identity.did.as_str(), did.to_did().as_str());
}
}
}