use jacquard_common::IntoStatic;
use jacquard_common::types::did::Did;
use miette::{IntoDiagnostic, Result};
use std::sync::Arc;
use std::sync::atomic::Ordering;
use tokio::runtime::Handle as TokioHandle;
use tracing::{debug, error, warn};
use crate::db::types::DbTid;
use crate::db::{Txn, keys, ser_repo_meta};
use crate::ingest::cursor;
use crate::ingest::stream::types::AccountStatus;
use crate::ingest::stream::{Account, Commit, Identity};
use crate::ingest::validation;
use crate::ops;
use crate::state::AppState;
use crate::types::{GaugeState, RepoMetadata, RepoState, RepoStatus};
#[cfg(feature = "jetstream")]
use crate::{db::types::TrimmedDid, types::StoredJetstreamEvent};
#[cfg(feature = "indexer_stream")]
use {
crate::types::{AccountEvt, IdentityEvt},
jacquard_common::cowstr::ToCowStr,
};
use super::bulk::{BulkStep, Drained, ShardState};
use super::inbox::{BULK_FLOOR, Inbox, Next};
use super::message::{
BulkRx, IndexerAccountData, IndexerCommitData, IndexerEvent, IndexerEventData,
IndexerIdentityData, IndexerMessage, IndexerRx, RelayRow,
};
use super::refetch::{self, Charged};
use super::worker::{FirehoseWorker, IngestError, RepoProcessResult};
use crate::ingest::relay::overlay::{Landed, RowDelta};
/// what follows a live message whatever came of it, after its commit, or right away when it
/// stopped before one and wrote nothing: its frame is done, and the relay can read the row it
/// handed on from the db again
fn done(state: &AppState, frame: Option<(url::Url, cursor::Ticket)>, relayed: Option<(Did, u64)>) {
if let Some((firehose, ticket)) = frame {
cursor::done(state, &firehose, ticket);
}
if let Some((did, version)) = relayed {
state.relay_rows.applied(&did, version);
}
}
/// counts a shard thread in [`AppState::shards_running`] for as long as it lives, panics included
struct Running<'a>(&'a AppState);
impl<'a> Running<'a> {
fn new(state: &'a AppState) -> Self {
state.shards_running.fetch_add(1, Ordering::SeqCst);
Self(state)
}
}
impl Drop for Running<'_> {
fn drop(&mut self) {
self.0.shards_running.fetch_sub(1, Ordering::SeqCst);
}
}
pub(super) struct WorkerContext<'a> {
pub(super) state: &'a AppState,
pub(super) shard: &'a mut ShardState,
pub(super) txn: Txn<'a>,
pub(super) lifecycle_transitions: &'a mut Vec<(Did, GaugeState)>,
pub(super) notify_backfill: bool,
/// this txn queued the repo's backfill already. a new repo's row and its first event land
/// in one txn, and a second queueing there wouldn't see the first one's metadata
pub(super) backfill_queued: bool,
/// a repo whose bulk job this txn started, for the shard to run once it commits
pub(super) started_job: Option<Did>,
}
impl FirehoseWorker {
pub(crate) fn shard(
id: usize,
num_shards: usize,
rx: IndexerRx,
bulk_rx: BulkRx,
state: Arc<AppState>,
handle: TokioHandle,
) {
let _guard = handle.enter();
let _running = Running::new(&state);
debug!(shard = id, "shard started");
let mut lifecycle_transitions = Vec::new();
let mut inbox: Inbox<_, _, BulkStep> = Inbox::new(rx, bulk_rx, BULK_FLOOR);
let mut shard = ShardState::default();
for did in Self::resume_jobs(id, num_shards, &state, &mut shard) {
inbox.push_local(BulkStep::Job(did));
}
// dropping what's left at a stop is the same as a crash here, which a restart already
// recovers from
while let Some(next) = inbox.next(&state.stop, &state.bulk_permits, &handle) {
lifecycle_transitions.clear();
let shard = &mut shard;
let queued = match next {
Next::Live(msg) => {
let started =
Self::handle_live(id, &state, shard, msg, &mut lifecycle_transitions);
if let Some(did) = started
&& Self::start_job(&state, shard, &did)
{
inbox.push_local(BulkStep::Job(did));
}
for did in shard.take_marked() {
if Self::resume_job(&state, shard, &did) {
inbox.push_local(BulkStep::Job(did));
}
}
None
}
Next::Bulk(BulkStep::Chunk(chunk), _permit) => {
Self::write_chunk(&state, shard, chunk).map(BulkStep::Drain)
}
Next::Bulk(BulkStep::Drain(job), _permit) => {
Self::drain_step(id, &state, shard, job, &mut lifecycle_transitions)
.map(BulkStep::Drain)
}
Next::Bulk(BulkStep::Erase(erase), _permit) => {
let did = erase.did.clone();
Self::start_erase(&state, shard, erase).then_some(BulkStep::Job(did))
}
Next::Bulk(BulkStep::Job(did), _permit) => {
Self::job_step(&state, shard, &did).then_some(BulkStep::Job(did))
}
};
if let Some(step) = queued {
inbox.push_local(step);
}
}
}
/// serves one live message, and names a repo whose bulk job its txn started
fn handle_live(
id: usize,
state: &AppState,
shard: &mut ShardState,
msg: IndexerMessage,
lifecycle_transitions: &mut Vec<(Did, GaugeState)>,
) -> Option<Did> {
let txn = Txn::new(&state.db);
let mut ctx = WorkerContext {
state,
shard,
txn,
lifecycle_transitions,
notify_backfill: false,
backfill_queued: false,
started_job: None,
};
let (frame, relayed_version) = match msg {
IndexerMessage::OnShard(on) => {
let _span = tracing::info_span!("on_shard", did = %on.did).entered();
(on.work)(state, ctx.shard);
return None;
}
IndexerMessage::Row(relayed) => {
let RelayRow {
did,
row,
firehose,
ticket,
} = *relayed;
let _span = tracing::info_span!("ingest", did = %did).entered();
let frame = firehose.zip(ticket);
let relayed_version = Some((did.clone(), row.version));
if Self::relayed_row(&mut ctx, &did, Some(row)).is_none() {
done(state, frame, relayed_version);
return None;
}
(frame, relayed_version)
}
IndexerMessage::Event(e) => {
let IndexerEvent {
seq,
firehose,
ticket,
row,
data,
} = *e;
let _span =
tracing::info_span!("ingest", hose = %firehose, seq, did = tracing::field::Empty)
.entered();
let frame = ticket.map(|ticket| (firehose, ticket));
let did = match &data {
IndexerEventData::Commit(m) => &m.commit.repo,
IndexerEventData::Identity(m) => &m.identity.did,
IndexerEventData::Account(m) => &m.account.did,
IndexerEventData::Sync(did) => did,
};
_span.record("did", did.as_ref());
let relayed_version = row.as_ref().map(|row| (did.clone(), row.version));
let Some(repo_state) = Self::relayed_row(&mut ctx, did, row) else {
done(state, frame, relayed_version);
return None;
};
match data {
IndexerEventData::Commit(msg) => {
let IndexerCommitData {
commit,
chain_break,
parsed_blocks,
} = msg;
let try_persist = |ctx: &mut WorkerContext<'_>, commit: &Commit| {
if let Err(e) = ops::persist_to_resync_buffer(
&mut ctx.txn,
&state.db,
&commit.repo,
commit,
) {
error!(err = %e, "failed to persist commit to resync_buffer");
}
};
match Self::handle_commit(
&mut ctx,
repo_state,
&commit,
chain_break,
parsed_blocks,
) {
Ok(RepoProcessResult::Ok) => {}
Ok(RepoProcessResult::NeedsBackfill(commit)) => {
try_persist(&mut ctx, commit);
}
Err(e) => {
if let IngestError::Generic(ref r) = e {
state.db.poison.check_report(r);
}
error!(err = %e, "error processing commit");
try_persist(&mut ctx, &commit);
}
}
}
IndexerEventData::Identity(msg) => {
let IndexerIdentityData { identity, changed } = msg;
if let Err(e) =
Self::handle_identity(&mut ctx, repo_state, &identity, changed)
{
error!(err = %e, "error processing identity");
}
}
IndexerEventData::Account(msg) => {
let IndexerAccountData {
account,
was_active,
changed,
} = msg;
if let Err(e) = Self::handle_account(
&mut ctx, repo_state, changed, &account, was_active,
) {
error!(err = %e, "error processing account");
}
}
IndexerEventData::Sync(did) => {
match Self::requeue_after_steps(state, ctx.shard, &did) {
Ok(true) => warn!(
"sync event during the backfill's steps, backfilling again after them"
),
Ok(false) => {
warn!("sync event, triggering backfill");
if let Err(e) =
Self::trigger_backfill(&mut ctx, &did, repo_state, "sync")
{
error!(err = %e, "failed to trigger backfill on sync");
}
}
// a requeue now could throw away a half written repo
Err(e) => error!(err = %e, "failed to check the repo's backfill steps"),
}
}
}
(frame, relayed_version)
}
};
let started_job = ctx.started_job.take();
let committed = Self::commit(id, ctx);
// after the commit, so a crash before it reads the frame again. a commit that failed is
// done all the same, the next one would have moved the cursor past it anyway
done(state, frame, relayed_version);
let notify_backfill = committed?;
if notify_backfill {
state.notify_backfill();
}
started_job
}
/// the row the message for `did` applies to, with the relay's change to it put on it in
/// this txn, see [`RowDelta`]. `None` when the message has nothing to apply to: the repo is
/// excluded, an erase or a wipe is about to forget it, or it has no row
fn relayed_row(
ctx: &mut WorkerContext,
did: &Did,
row: Option<RowDelta>,
) -> Option<RepoState<'static>> {
match ctx.txn.is_excluded(did) {
// an erase or a wipe running for the repo is about to forget whatever this would
// write, the same as if it came after them and found no row
Ok(false) if !ctx.shard.drops_live(did) => {}
Ok(_) => return None,
Err(e) => {
error!(err = %e, "failed to check repository exclusion");
return None;
}
}
Self::land_row(ctx, did, row)
.inspect_err(|e| error!(err = %e, "failed to get repo state"))
.ok()
.flatten()
}
fn land_row(
ctx: &mut WorkerContext,
did: &Did,
row: Option<RowDelta>,
) -> Result<Option<RepoState<'static>>, IngestError> {
let current = ctx.txn.read_repo(did)?;
let Some(row) = row else {
return Ok(current);
};
let (landed, made) = match row.land(current) {
Landed::Row(landed) => (landed, false),
Landed::Made(landed) => (landed, true),
Landed::Gone => return Ok(None),
};
for host in ctx.txn.put_repo(did, &landed)?.hosts() {
let count = ctx
.txn
.counts
.projected_pds_account_count(&ctx.state.db, &host);
ctx.state
.apply_host_limit_status(&mut ctx.txn.batch, &host, count);
}
if made {
ctx.txn.counts.add_repos(1);
return Self::trigger_backfill(ctx, did, landed, "new repo")
.map(|triggered| Some(triggered.repo_state));
}
Ok(Some(landed))
}
/// stages the lifecycle moves `ctx` collected and commits. `None` when that failed,
/// otherwise whether a backfill was queued
pub(super) fn commit(id: usize, ctx: WorkerContext) -> Option<bool> {
let WorkerContext {
mut txn,
lifecycle_transitions,
notify_backfill,
..
} = ctx;
let staged = lifecycle_transitions
.drain(..)
.try_for_each(|(did, gauge)| txn.transition_lifecycle(&did, gauge).map(|_| ()));
if let Err(e) = staged {
error!(shard = id, err = %e, "failed to stage lifecycle transitions");
return None;
}
if let Err(e) = txn.commit() {
error!(shard = id, err = %e, "failed to commit transaction");
return None;
}
Some(notify_backfill)
}
fn handle_commit<'s, 'c>(
ctx: &mut WorkerContext,
mut repo_state: RepoState<'s>,
commit: &'c Commit<'c>,
chain_break: bool,
parsed_blocks: crate::car::CommitCar,
) -> Result<RepoProcessResult<'c>, IngestError> {
let db = &ctx.state.db;
let did = &commit.repo;
repo_state.advance_message_time(commit.time.0.timestamp_millis());
let metadata_key = keys::repo_metadata_key(did);
let metadata = db
.repo_metadata
.get(&metadata_key)
.into_diagnostic()?
.map(|bytes| crate::db::deser_repo_meta(&bytes))
.transpose()?;
// a new repo's row comes in this txn too, and the backfill it queued isn't in the db yet
let is_backfilling = ctx.backfill_queued
|| match &metadata {
Some(metadata) => db
.indexer
.is_pending(&keys::pending_key(metadata.index_id))
.into_diagnostic()?,
None => false,
};
if chain_break {
if Self::requeue_after_steps(ctx.state, ctx.shard, did)? {
debug!("chain break during the backfill's steps, backfilling again after them");
} else {
debug!("chain break detected, triggering backfill");
// don't buffer it, a parked refetch never drains the buffer
if Self::trigger_backfill(ctx, did, repo_state, "chain break")?.deferred {
return Ok(RepoProcessResult::Ok);
}
}
return Ok(RepoProcessResult::NeedsBackfill(commit));
}
// the parked refetch is a full fetch, it covers this commit
if metadata
.as_ref()
.is_some_and(|metadata| metadata.refetch_deferred)
{
return Ok(RepoProcessResult::Ok);
}
if is_backfilling {
return Ok(RepoProcessResult::NeedsBackfill(commit));
}
// the relay validated it before a backfill fetched past it, so the car already has it
if metadata.is_some_and(|metadata| metadata.car_covers(&commit.rev)) {
debug!(rev = %commit.rev, "commit is older than the car, skipping");
return Ok(RepoProcessResult::Ok);
}
Self::apply_checked_commit(ctx, repo_state, commit, parsed_blocks)
.map(|_| RepoProcessResult::Ok)
}
/// applies a commit that got past [`Self::handle_commit`]'s checks. the drain comes in here
/// directly, because its own backfill still holds the queue entry those checks buffer behind
fn apply_checked_commit<'s>(
ctx: &mut WorkerContext,
repo_state: RepoState<'s>,
commit: &Commit<'_>,
parsed_blocks: crate::car::CommitCar,
) -> Result<RepoState<'s>, IngestError> {
let root_bytes = parsed_blocks
.blocks
.get(&parsed_blocks.root)
.ok_or_else(|| IngestError::Generic(miette::miette!("root block missing from CAR")))?;
let commit_obj = crate::mst::AtpCommit::from_cbor(root_bytes)
.map_err(|e| IngestError::Generic(miette::miette!("invalid commit object: {e}")))?;
let validated = validation::ValidatedCommit {
commit,
parsed_blocks,
commit_obj,
chain_break: validation::ChainBreak::default(), // not used by apply_commit
};
let res = ops::apply_commit(
&mut ctx.txn,
ctx.state,
repo_state,
validated,
&ctx.state.filter.load(),
)?;
Ok(res.repo_state)
}
#[cfg_attr(not(feature = "indexer_stream"), allow(unused_variables))]
fn handle_identity<'s>(
ctx: &mut WorkerContext,
repo_state: RepoState<'s>,
identity: &Identity,
changed: bool,
) -> Result<(), IngestError> {
#[cfg(feature = "indexer_stream")]
{
let did = &identity.did;
#[cfg(feature = "jetstream")]
ctx.txn.outbox.jetstream.push(
StoredJetstreamEvent::Identity {
did: TrimmedDid::from(did).into_static(),
handle: identity.handle.clone().map(IntoStatic::into_static),
seq: identity.seq,
time: identity.time.clone(),
},
None,
);
if changed {
ctx.txn.outbox.stream.push_identity(IdentityEvt {
did: did.clone().into_static(),
handle: repo_state.handle.clone().map(IntoStatic::into_static),
});
}
}
Ok(())
}
fn handle_account<'s, 'c>(
ctx: &mut WorkerContext,
repo_state: RepoState<'s>,
changed: bool,
account: &'c Account<'c>,
was_active: bool,
) -> Result<(), IngestError> {
let db = &ctx.state.db;
let did = &account.did;
let is_inactive = !account.active;
#[cfg(feature = "indexer_stream")]
let evt = AccountEvt {
did: did.clone().into_static(),
active: account.active,
status: account.status.as_ref().map(|s| s.to_cowstr().into_static()),
};
#[cfg(feature = "jetstream")]
ctx.txn.outbox.jetstream.push(
StoredJetstreamEvent::Account {
did: TrimmedDid::from(did).into_static(),
active: account.active,
status: account.status.as_ref().map(|s| s.to_cowstr().into_static()),
seq: account.seq,
time: account.time.clone(),
},
None,
);
// before the deleted branch returns, since a wipe stages no record deletes and this is
// the only thing a /stream reader hears of it
#[cfg(feature = "indexer_stream")]
if changed {
ctx.txn.outbox.stream.push_account(evt);
}
if is_inactive {
match &account.status {
Some(AccountStatus::Deleted) => {
debug!("account deleted, wiping data");
ctx.lifecycle_transitions
.push((did.clone().into_static(), GaugeState::Synced));
crate::ops::start_wipe(&mut ctx.txn, db, did, &repo_state)?;
ctx.started_job = Some(did.clone().into_static());
return Ok(());
}
_ => {
// status update logic is now handled in RelayWorker;
// FirehoseWorker just needs to update gauges if status changed.
if changed && was_active {
ctx.lifecycle_transitions
.push((did.clone().into_static(), GaugeState::Resync(None)));
}
}
}
} else {
// if account became active, update gauges
if !was_active {
ctx.lifecycle_transitions
.push((did.clone().into_static(), GaugeState::Synced));
}
}
Ok(())
}
/// applies what a finished backfill's repo buffered during the fetch, while `pending_key` is
/// still what it's queued under, and takes the repo off the queue in the batch that empties
/// the buffer. `limit` caps the record writes this batch applies, so a big buffer goes in
/// several. `Done(false)` when the repo isn't queued under the key any more, or is excluded
/// or gone, so nothing was drained
pub(super) fn finish_backfill(
ctx: &mut WorkerContext,
did: &Did,
pending_key: &[u8],
limit: Option<usize>,
) -> Result<Drained, IngestError> {
if ctx.txn.is_excluded(did)?
|| !ctx
.state
.db
.indexer
.is_pending(pending_key)
.into_diagnostic()?
{
return Ok(Drained::Done(false));
}
// the buffer and the queue belong to the key the metadata names, so an orphan left
// queued next to it only takes itself off
if !ctx.txn.names_pending(did, pending_key)? {
ctx.state
.db
.indexer
.stage_pending_remove(&mut ctx.txn.batch, did, pending_key);
return Ok(Drained::Done(false));
}
let Some(repo_state) = ctx.txn.read_repo(did)? else {
// whatever erased the row is the last word, so the key it left behind goes now
// instead of on its next run
ctx.txn
.transition_pending_key(did, pending_key, GaugeState::Synced)?;
ctx.state
.db
.indexer
.stage_pending_remove(&mut ctx.txn.batch, did, pending_key);
return Ok(Drained::Done(false));
};
let metadata = ctx
.state
.db
.repo_metadata
.get(keys::repo_metadata_key(did))
.into_diagnostic()?
.map(|bytes| crate::db::deser_repo_meta(&bytes))
.transpose()?;
let (repo_state, emptied) =
Self::drain_resync_buffer(ctx, did, repo_state, metadata.as_ref(), limit)?;
if !emptied {
// the root moves with the commits, but the repo stays queued until the last batch
ctx.txn.put_repo(did, &repo_state)?;
return Ok(Drained::More);
}
// a repo that went inactive while it backfilled keeps its status, the backfill only
// caught its records up
let status = if repo_state.active {
RepoStatus::Synced
} else {
repo_state.status.clone()
};
ops::transition_repo(
&mut ctx.txn,
ctx.lifecycle_transitions,
did,
repo_state,
status,
)?;
// transition_repo leaves the queue alone for an inactive status
ctx.state
.db
.indexer
.stage_pending_remove(&mut ctx.txn.batch, did, pending_key);
Ok(Drained::Done(true))
}
/// applies the buffered commits in rev order until they've written `limit` records, at least
/// one commit, and says whether that emptied the buffer
fn drain_resync_buffer<'s>(
ctx: &mut WorkerContext,
did: &Did,
mut repo_state: RepoState<'s>,
metadata: Option<&RepoMetadata>,
limit: Option<usize>,
) -> Result<(RepoState<'s>, bool), IngestError> {
let db = &ctx.state.db;
let prefix = keys::resync_buffer_prefix(did);
let mut written = 0;
for guard in db.indexer.resync_buffer.prefix(&prefix) {
if limit.is_some_and(|limit| written >= limit) {
return Ok((repo_state, false));
}
let (key, value) = guard.into_inner().into_diagnostic()?;
// goes even when it fails to apply, because a failed drain sends the repo back for a
// whole refetch and that car has whatever this commit did. kept, a commit that can
// never apply would bring the refetch back forever
ctx.txn.batch.remove(&db.indexer.resync_buffer, key);
let ops;
(repo_state, ops) = Self::apply_buffered_commit(ctx, repo_state, &value, metadata)?;
// a commit without ops still costs a step something
written += ops.max(1);
}
Ok((repo_state, true))
}
fn apply_buffered_commit<'s>(
ctx: &mut WorkerContext,
mut repo_state: RepoState<'s>,
value: &[u8],
metadata: Option<&RepoMetadata>,
) -> Result<(RepoState<'s>, usize), IngestError> {
let commit: Commit = rmp_serde::from_slice(value).into_diagnostic()?;
// the car's persist dropped what was buffered up to it, but a commit the relay
// validated before the fetch can still reach the shard between the persist and here
if metadata.is_some_and(|metadata| metadata.car_covers(&commit.rev)) {
return Ok((repo_state, 0));
}
// it was checked against the root it came in on, which a car older than that root took
// back, and the firehose can skip one too. applied it would land on records it doesn't
// go with, so it fails the drain, which refetches. the relay puts each commit it forwards
// on the row, so one at or below the root came in during the fetch and only lacks its
// records. a chain break the relay let through there already queued another backfill
if let Some(root) = &repo_state.root
&& DbTid::from(&commit.rev) > root.rev
&& !validation::follows(&commit, root)
{
return Err(IngestError::Generic(miette::miette!(
"buffered commit {} doesn't follow rev {}",
commit.rev.as_str(),
root.rev.to_tid().as_str()
)));
}
// buffered commits were validated and source-checked on arrival, so the host check
// is skipped and the car is not rehashed as a whole. the check does not survive the
// buffer, so each record block is hashed once more as it is stored
let parsed_blocks = crate::car::parse_commit_car(commit.blocks.clone(), false)
.map_err(|e| IngestError::Generic(miette::miette!("malformed CAR: {e}")))?;
repo_state.advance_message_time(commit.time.0.timestamp_millis());
let ops = commit.ops.len();
Self::apply_checked_commit(ctx, repo_state, &commit, parsed_blocks)
.map(|repo_state| (repo_state, ops))
}
/// the first fetch is free, later ones are charged unless already queued, see [`refetch`]
pub(super) fn trigger_backfill<'s>(
ctx: &mut WorkerContext,
did: &Did,
mut repo_state: RepoState<'s>,
cause: &'static str,
) -> Result<Triggered<'s>, IngestError> {
let db = &ctx.state.db;
if ctx.backfill_queued || ctx.txn.is_excluded(did)? {
return Ok(Triggered {
repo_state,
deferred: false,
});
}
let meta_key = keys::repo_metadata_key(did);
let existing_metadata = db
.repo_metadata
.get(&meta_key)
.into_diagnostic()?
.map(|b| crate::db::deser_repo_meta(&b))
.transpose()?;
let had_metadata = existing_metadata.is_some();
// the index id is set later
let mut metadata = existing_metadata.unwrap_or(RepoMetadata::backfilling(0));
let old_pkey = keys::pending_key(metadata.index_id);
let was_pending = had_metadata && db.indexer.is_pending(&old_pkey).into_diagnostic()?;
// a queued fetch covers this request, unless its steps are already being written
let applying = was_pending && ctx.shard.is_applying(did, &old_pkey);
if had_metadata
&& (!was_pending || applying)
&& refetch::charge(
ctx.state,
&mut ctx.txn,
did,
&mut repo_state,
&mut metadata,
cause,
)? == Charged::Deferred
{
// parked, not queued. the applying backfill's entry still has to go
if applying {
db.indexer
.stage_pending_remove(&mut ctx.txn.batch, did, &old_pkey);
}
return Ok(Triggered {
repo_state,
deferred: true,
});
}
ctx.backfill_queued = true;
// remove old pending entry and insert new one with fresh index_id
if had_metadata {
// only remove if we had one so we dont delete a random entry
db.indexer
.stage_pending_remove(&mut ctx.txn.batch, did, &old_pkey);
}
metadata.index_id = rand::random::<u64>();
db.indexer.stage_pending_insert(
&mut ctx.txn.batch,
did,
&keys::pending_key(metadata.index_id),
);
ctx.txn
.batch
.insert(&db.repo_metadata, &meta_key, ser_repo_meta(&metadata)?);
if !was_pending {
ctx.txn.transition_lifecycle(did, GaugeState::Pending)?;
ctx.notify_backfill = true;
}
Ok(Triggered {
repo_state,
deferred: false,
})
}
}
pub(super) struct Triggered<'s> {
pub(super) repo_state: RepoState<'s>,
pub(super) deferred: bool,
}
#[cfg(test)]
mod tests {
use jacquard_common::{
CowStr,
types::{
cid::CidLink,
string::{Did, Tid},
},
};
use super::*;
use crate::{
config::Config,
db,
ingest::stream::{Datetime, RepoOp, RepoOpAction},
};
use std::sync::atomic::Ordering::SeqCst;
/// a repo row whose metadata names index `named` while the queue holds the repo under `queued`
fn queued_repo(
row: &RepoState<'_>,
named: u64,
queued: u64,
) -> Result<(tempfile::TempDir, Arc<AppState>, Did)> {
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();
let mut batch = state.db.inner.batch();
batch.insert(
&state.db.repos,
keys::repo_key(&did),
db::ser_repo_state(row)?,
);
batch.insert(
&state.db.repo_metadata,
keys::repo_metadata_key(&did),
ser_repo_meta(&RepoMetadata::backfilling(named))?,
);
state
.db
.indexer
.stage_pending_insert(&mut batch, &did, &keys::pending_key(queued));
batch.commit().into_diagnostic()?;
Ok((tmp, state, did))
}
/// a resync buffer entry that belongs to whichever backfill drains next, so a finish that
/// isn't that one has to leave it
fn buffer_an_entry(state: &AppState, did: &Did) -> Result<Vec<u8>> {
let key = keys::resync_buffer_key(
did,
crate::db::types::DbTid::new_from_bytes(4_u64.to_be_bytes()),
);
state
.db
.indexer
.resync_buffer
.insert(&key, b"for the next backfill")
.into_diagnostic()?;
Ok(key)
}
fn repo_row(state: &AppState, did: &Did) -> Result<RepoState<'static>> {
let bytes = state
.db
.repos
.get(keys::repo_key(did))
.into_diagnostic()?
.expect("repository state");
Ok(db::deser_repo_state(&bytes)?.into_static())
}
/// hands `did`'s finished backfill under `pending_key` to a shard and says whether it drained
async fn finish_on_a_shard(
state: &Arc<AppState>,
did: &Did,
pending_key: [u8; 8],
) -> Result<bool> {
let (tx, shard) = FirehoseWorker::spawn_one_shard(state.clone());
let finishing = did.clone();
let drained = tx
.on_owner(did, move |state, shard| {
FirehoseWorker::finish_for_test(state, shard, &finishing, &pending_key)
})
.await?;
drop(tx);
shard
.join()
.map_err(|_| miette::miette!("indexer shard panicked"))?;
drained
}
#[tokio::test]
async fn backfill_finished_persists_synced_repo_status() -> Result<()> {
let (_tmp, state, did) = queued_repo(&RepoState::backfilling(), 7, 7)?;
assert!(finish_on_a_shard(&state, &did, keys::pending_key(7)).await?);
let repo_state = repo_row(&state, &did)?;
assert_eq!(repo_state.status, RepoStatus::Synced);
assert!(repo_state.active);
assert!(
!state
.db
.indexer
.is_pending(&keys::pending_key(7))
.into_diagnostic()?
);
state.db.indexer.assert_pending_dids_match();
Ok(())
}
#[tokio::test]
async fn backfill_finished_leaves_a_repo_that_went_inactive_inactive() -> Result<()> {
// deactivated while its getRepo streamed
let mut deactivated = RepoState::backfilling();
deactivated.active = false;
deactivated.status = RepoStatus::Deactivated;
let (_tmp, state, did) = queued_repo(&deactivated, 7, 7)?;
assert!(finish_on_a_shard(&state, &did, keys::pending_key(7)).await?);
let repo_state = repo_row(&state, &did)?;
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(())
}
#[tokio::test]
async fn backfill_finished_parks_any_inactive_status_the_relay_writes() -> Result<()> {
for status in [
RepoStatus::Error("unknown".into()),
RepoStatus::Error("strange".into()),
RepoStatus::Desynchronized,
RepoStatus::Throttled,
] {
let mut inactive = RepoState::backfilling();
inactive.active = false;
inactive.status = status.clone();
let (_tmp, state, did) = queued_repo(&inactive, 7, 7)?;
assert!(finish_on_a_shard(&state, &did, keys::pending_key(7)).await?);
let repo_state = repo_row(&state, &did)?;
assert!(!repo_state.active, "{status:?} went active");
assert_eq!(repo_state.status, status);
let resync = state
.db
.indexer
.resync
.get(keys::repo_key(&did))
.into_diagnostic()?
.expect("a resync entry");
match rmp_serde::from_slice(&resync).into_diagnostic()? {
crate::types::ResyncState::Gone { status: gone } => assert_eq!(gone, status),
other => panic!("{status:?} left {other:?}, which refetches"),
}
}
Ok(())
}
#[tokio::test]
async fn a_backfill_whose_repo_was_requeued_drains_nothing() -> Result<()> {
// requeued under 8 while the backfill queued under 7 fetched
let (_tmp, state, did) = queued_repo(&RepoState::backfilling(), 8, 8)?;
let buffered = buffer_an_entry(&state, &did)?;
assert!(!finish_on_a_shard(&state, &did, keys::pending_key(7)).await?);
assert!(
state
.db
.indexer
.is_pending(&keys::pending_key(8))
.into_diagnostic()?
);
assert!(
state
.db
.indexer
.resync_buffer
.get(&buffered)
.into_diagnostic()?
.is_some()
);
assert_eq!(
repo_row(&state, &did)?.status,
RepoState::backfilling().status
);
state.db.indexer.assert_pending_dids_match();
Ok(())
}
#[tokio::test]
async fn an_orphaned_key_comes_off_the_queue_alone() -> Result<()> {
// 7 was left queued next to 8, which the metadata names and which still has to run
let (_tmp, state, did) = queued_repo(&RepoState::backfilling(), 8, 7)?;
let mut batch = state.db.inner.batch();
state
.db
.indexer
.stage_pending_insert(&mut batch, &did, &keys::pending_key(8));
batch.commit().into_diagnostic()?;
let buffered = buffer_an_entry(&state, &did)?;
assert!(!finish_on_a_shard(&state, &did, keys::pending_key(7)).await?);
let is_pending = |key| state.db.indexer.is_pending(&keys::pending_key(key));
assert!(!is_pending(7).into_diagnostic()?);
assert!(is_pending(8).into_diagnostic()?);
assert!(
state
.db
.indexer
.resync_buffer
.get(&buffered)
.into_diagnostic()?
.is_some()
);
assert_eq!(
repo_row(&state, &did)?.status,
RepoState::backfilling().status
);
state.db.indexer.assert_pending_dids_match();
Ok(())
}
#[tokio::test]
async fn a_backfill_whose_repo_row_is_gone_comes_off_the_queue() -> Result<()> {
let (_tmp, state, did) = queued_repo(&RepoState::backfilling(), 7, 7)?;
state
.db
.repos
.remove(keys::repo_key(&did))
.into_diagnostic()?;
assert!(!finish_on_a_shard(&state, &did, keys::pending_key(7)).await?);
assert!(
!state
.db
.indexer
.is_pending(&keys::pending_key(7))
.into_diagnostic()?
);
state.db.indexer.assert_pending_dids_match();
Ok(())
}
fn commit_at(did: &Did, seq: i64) -> Commit<'static> {
Commit {
blobs: Vec::new(),
blocks: bytes::Bytes::new(),
commit: CidLink::from(crate::car::compute_cid(b"commit")),
ops: Vec::new(),
prev_data: None,
rebase: false,
repo: did.clone(),
rev: Tid::new("3kzbif5moe22m").unwrap(),
seq,
since: None,
time: Datetime(chrono::Utc::now().into()),
too_big: false,
}
}
/// a live commit for a repo whose backfill is being written in steps goes to the resync
/// buffer, which is durable once that txn commits, so its frame is done then and not after
/// the drain. a frame for a repo the shard has no row for is done right away
#[tokio::test]
async fn a_frame_is_done_once_the_txn_that_buffered_it_commits() -> Result<()> {
let (_tmp, state, did) = queued_repo(&RepoState::synced(), 7, 7)?;
// the marker a backfill's first step leaves, so the shard starts with the repo busy
let mut batch = state.db.inner.batch();
crate::db::stage_bulk_job(
&mut batch,
&state.db,
&did,
&crate::types::BulkJobV1::Apply {
pending_key: keys::pending_key(7).to_vec(),
},
)?;
batch.commit().into_diagnostic()?;
let firehose = url::Url::parse("wss://relay.example/").into_diagnostic()?;
let _ = state
.firehose_cursors
.insert_sync(firehose.clone(), std::sync::atomic::AtomicI64::new(0));
let tracker = crate::ingest::cursor::SeqTracker::new(true);
let unknown = Did::new_static("did:plc:nobodyknowsthisrepoatall").into_diagnostic()?;
let (tx, shard) = FirehoseWorker::spawn_one_shard(state.clone());
for (did, seq) in [(&did, 5), (&unknown, 6)] {
tx.send(IndexerMessage::Event(Box::new(IndexerEvent {
seq,
firehose: firehose.clone(),
ticket: Some(tracker.ticket(seq, 0)),
row: None,
data: IndexerEventData::Commit(IndexerCommitData {
commit: commit_at(did, seq),
chain_break: false,
parsed_blocks: crate::car::CommitCar {
root: crate::car::compute_cid(b"commit"),
blocks: crate::car::Blocks::unverified(Default::default()),
},
}),
})))
.await
.into_diagnostic()?;
}
drop(tx);
shard
.join()
.map_err(|_| miette::miette!("indexer shard panicked"))?;
let cursor = state
.firehose_cursors
.peek_with(&firehose, |_, cursor| cursor.load(SeqCst))
.unwrap();
assert_eq!(cursor, 6);
assert_eq!(
state
.db
.indexer
.resync_buffer
.prefix(keys::repo_key(&did))
.count(),
1,
"the commit waits in the buffer"
);
assert!(
crate::db::read_bulk_job(&state.db, &did)?.is_some(),
"and its backfill is still running"
);
Ok(())
}
/// a new repo's row and its first commit land in one txn, and the commit waits for the
/// backfill that txn queued like every commit after it, so a car older than it can't undo it
#[tokio::test]
async fn a_new_repos_first_commit_waits_for_its_backfill() -> Result<()> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let state = Arc::new(AppState::new(&Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})?);
let did = Did::new_static("did:plc:anewaccountnobodyhasseen").into_diagnostic()?;
let signed = crate::mst::AtpCommit {
did: did.clone(),
version: 3,
data: crate::car::compute_cid(b"mst"),
rev: Tid::new("3kzbif5moe22m").into_diagnostic()?,
prev: None,
sig: bytes::Bytes::from_static(b"signature"),
};
let root = signed.to_cid()?;
let blocks = [(root, bytes::Bytes::from(signed.to_cbor()?))]
.into_iter()
.collect();
let mut commit = commit_at(&did, 5);
commit.commit = CidLink::from(root);
let (tx, shard) = FirehoseWorker::spawn_one_shard(state.clone());
tx.send(IndexerMessage::Event(Box::new(IndexerEvent {
seq: 5,
firehose: url::Url::parse("wss://relay.example/").into_diagnostic()?,
ticket: None,
row: Some(RowDelta {
before: None,
after: RepoState::backfilling(),
version: 0,
}),
data: IndexerEventData::Commit(IndexerCommitData {
commit,
chain_break: false,
parsed_blocks: crate::car::CommitCar {
root,
blocks: crate::car::Blocks::unverified(blocks),
},
}),
})))
.await
.into_diagnostic()?;
drop(tx);
shard
.join()
.map_err(|_| miette::miette!("indexer shard panicked"))?;
assert_eq!(
state
.db
.indexer
.resync_buffer
.prefix(keys::repo_key(&did))
.count(),
1
);
assert!(repo_row(&state, &did)?.root.is_none());
Ok(())
}
/// a wipe stages no record deletes, so the account event is all a /stream reader hears
#[cfg(feature = "indexer_stream")]
#[tokio::test]
async fn an_account_deletion_is_announced_on_the_stream() -> Result<()> {
let (_tmp, state, did) = queued_repo(&RepoState::synced(), 7, 7)?;
let mut stream = state.db.stream.event_tx.subscribe();
let (tx, shard) = FirehoseWorker::spawn_one_shard(state.clone());
tx.send(IndexerMessage::Event(Box::new(IndexerEvent {
seq: 5,
firehose: url::Url::parse("wss://relay.example/").into_diagnostic()?,
ticket: None,
row: None,
data: IndexerEventData::Account(IndexerAccountData {
account: Account {
active: false,
did: did.clone(),
seq: 5,
status: Some(AccountStatus::Deleted),
time: Datetime(chrono::Utc::now().into()),
},
was_active: true,
changed: true,
}),
})))
.await
.into_diagnostic()?;
drop(tx);
shard
.join()
.map_err(|_| miette::miette!("indexer shard panicked"))?;
let announced = std::iter::from_fn(|| stream.try_recv().ok())
.find_map(|event| match event {
crate::types::BroadcastEvent::Ephemeral(event) => Some(event),
_ => None,
})
.expect("an account event on the stream");
let json = serde_json::to_value(announced.as_ref()).into_diagnostic()?;
assert_eq!(json["type"], "account");
assert_eq!(json["account"]["did"], did.as_str());
assert_eq!(json["account"]["active"], false);
assert_eq!(json["account"]["status"], "deleted");
Ok(())
}
fn synced_repo(limit: u32) -> Result<(tempfile::TempDir, Arc<AppState>, Did)> {
let (tmp, state) = refetch::tests::state(Some(limit));
refetch::tests::seed(&state, true, None);
Ok((tmp, Arc::new(state), refetch::tests::did()))
}
fn event(data: IndexerEventData, seq: i64) -> Result<IndexerMessage> {
Ok(IndexerMessage::Event(Box::new(IndexerEvent {
seq,
firehose: url::Url::parse("wss://relay.example/").into_diagnostic()?,
ticket: None,
row: None,
data,
})))
}
fn live_commit(did: &Did, seq: i64, chain_break: bool) -> Result<IndexerMessage> {
event(
IndexerEventData::Commit(IndexerCommitData {
commit: commit_at(did, seq),
chain_break,
parsed_blocks: crate::car::CommitCar {
root: crate::car::compute_cid(b"commit"),
blocks: crate::car::Blocks::unverified(Default::default()),
},
}),
seq,
)
}
async fn through_a_shard(state: &Arc<AppState>, messages: Vec<IndexerMessage>) -> Result<()> {
let (tx, shard) = FirehoseWorker::spawn_one_shard(state.clone());
for message in messages {
tx.send(message).await.into_diagnostic()?;
}
drop(tx);
shard
.join()
.map_err(|_| miette::miette!("indexer shard panicked"))?;
Ok(())
}
fn fetch_done(state: &AppState, did: &Did) -> Result<()> {
let metadata = refetch::tests::metadata_of(state, did);
let mut batch = state.db.inner.batch();
state.db.indexer.stage_pending_remove(
&mut batch,
did,
&keys::pending_key(metadata.index_id),
);
batch.commit().into_diagnostic()
}
#[tokio::test]
async fn a_repo_that_keeps_sending_syncs_is_refetched_up_to_its_limit() -> Result<()> {
let (_tmp, state, did) = synced_repo(2)?;
let pending = || state.db.indexer.pending_iter().count();
for seq in 1..=2 {
through_a_shard(
&state,
vec![event(IndexerEventData::Sync(did.clone()), seq)?],
)
.await?;
assert_eq!(pending(), 1, "sync {seq} queues a refetch");
fetch_done(&state, &did)?;
}
for seq in 3..=4 {
through_a_shard(
&state,
vec![event(IndexerEventData::Sync(did.clone()), seq)?],
)
.await?;
assert_eq!(pending(), 0, "sync {seq} is over the limit");
}
let repo = repo_row(&state, &did)?;
assert!(repo.active);
assert_eq!(repo.status, RepoStatus::Throttled);
let metadata = refetch::tests::metadata_of(&state, &did);
assert!(metadata.refetch_deferred);
assert_eq!(metadata.refetch.map(|allowance| allowance.used), Some(2));
assert!(matches!(
refetch::tests::entry_of(&state, &did),
Some(crate::types::ResyncState::Error {
kind: crate::types::ResyncErrorKind::Ratelimited,
retry_count: 3,
..
})
));
state.db.indexer.assert_pending_dids_match();
Ok(())
}
#[tokio::test]
async fn a_repo_waiting_for_its_refetch_drops_live_commits_and_buffers_none() -> Result<()> {
let (_tmp, state, did) = synced_repo(1)?;
through_a_shard(&state, vec![event(IndexerEventData::Sync(did.clone()), 1)?]).await?;
fetch_done(&state, &did)?;
through_a_shard(&state, vec![event(IndexerEventData::Sync(did.clone()), 2)?]).await?;
assert!(refetch::tests::metadata_of(&state, &did).refetch_deferred);
// the relay's #sync already moved the root, commit 3 chains onto it
through_a_shard(
&state,
vec![live_commit(&did, 3, false)?, live_commit(&did, 4, true)?],
)
.await?;
assert_eq!(
state
.db
.indexer
.resync_buffer
.prefix(keys::repo_key(&did))
.count(),
0
);
assert_eq!(state.db.indexer.pending_iter().count(), 0);
assert_eq!(repo_row(&state, &did)?.status, RepoStatus::Throttled);
Ok(())
}
#[tokio::test]
async fn a_repos_first_fetch_costs_no_refetch() -> Result<()> {
let (_tmp, state, _) = synced_repo(1)?;
let new = Did::new_static("did:plc:anewaccountnobodyhasseen").into_diagnostic()?;
let mut message = event(IndexerEventData::Sync(new.clone()), 1)?;
if let IndexerMessage::Event(event) = &mut message {
event.row = Some(RowDelta {
before: None,
after: RepoState::backfilling(),
version: 0,
});
}
through_a_shard(&state, vec![message]).await?;
assert_eq!(state.db.indexer.pending_iter().count(), 1);
assert_eq!(refetch::tests::metadata_of(&state, &new).refetch, None);
Ok(())
}
#[test]
fn resync_buffer_commit_msgpack_round_trip_preserves_path() {
let cid = CidLink::from(crate::car::compute_cid(b"commit"));
let commit = Commit {
blobs: Vec::new(),
blocks: bytes::Bytes::new(),
commit: cid,
ops: vec![RepoOp {
action: RepoOpAction::Delete,
cid: None,
path: CowStr::Borrowed("app.bsky.feed.post/stinkpot"),
prev: None,
}],
prev_data: None,
rebase: false,
repo: Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap(),
rev: Tid::new("3kzbif5moe22m").unwrap(),
seq: 42,
since: None,
time: Datetime::try_from("2026-08-10T12:34:56Z".to_owned()).unwrap(),
too_big: false,
};
let encoded = rmp_serde::to_vec_named(&commit).unwrap();
let decoded: Commit = rmp_serde::from_slice(&encoded).unwrap();
assert_eq!(decoded.repo, commit.repo);
assert_eq!(decoded.rev, commit.rev);
assert_eq!(decoded.seq, commit.seq);
assert_eq!(decoded.ops.len(), 1);
assert_eq!(decoded.ops[0].path.as_ref(), "app.bsky.feed.post/stinkpot");
assert_eq!(decoded.ops[0].action, RepoOpAction::Delete);
}
}