use fjall::OwnedWriteBatch;
use jacquard_common::IntoStatic;
use miette::Result;
use url::Url;
use crate::ingest::cursor::Ticket;
use crate::ingest::indexer::{
IndexerAccountData, IndexerCommitData, IndexerEvent, IndexerEventData, IndexerIdentityData,
IndexerMessage, IndexerTx, RelayRow,
};
use crate::ingest::relay::overlay::RowDelta;
use crate::ingest::stream::{Account, Commit, Identity, Sync};
use crate::state::AppState;
use crate::types::{RepoState, RepoStatus};
use jacquard_common::types::did::Did;
use tracing::warn;
/// the indexer shards mark a frame done once they committed what it did
pub(crate) const PUBLISHES_WATERMARK: bool = true;
/// per-worker seed from which each shard builds its sink.
#[derive(Clone)]
pub(crate) struct SinkSeed {
hook: IndexerTx,
}
impl SinkSeed {
pub(crate) fn new(hook: IndexerTx) -> Self {
Self { hook }
}
}
/// forwards validated events to the indexer hook after batch commit.
pub(crate) struct EventSink {
hook: IndexerTx,
pending: Vec<Queued>,
/// the row change the current message made, handed on once its commit is through
row: Option<StagedRow>,
}
struct StagedRow {
did: Did,
/// `None` when the relay made the row up for a repo it hadn't seen
before: Option<RepoState<'static>>,
after: RepoState<'static>,
firehose: Option<Url>,
}
struct Queued {
msg: IndexerMessage,
/// the frame's own event gets the frame's ticket. the others are a synthetic identity
/// announce
own_frame: bool,
/// the relay row change it carries, see [`EventSink::hand_row`]
row: Option<(Did, u64)>,
}
fn event(firehose: &Url, seq: i64, data: IndexerEventData) -> IndexerMessage {
IndexerMessage::Event(Box::new(IndexerEvent {
seq,
firehose: firehose.clone(),
ticket: None,
row: None,
data,
}))
}
pub(crate) struct Staged;
pub(crate) struct AccountSnapshot {
status: RepoStatus,
}
impl EventSink {
pub(crate) fn new(
seed: &SinkSeed,
_stats: std::sync::Arc<crate::ingest::firehose_stats::RelayShardStats>,
) -> Self {
Self {
hook: seed.hook.clone(),
pending: Vec::with_capacity(2),
row: None,
}
}
fn push(&mut self, msg: IndexerMessage, own_frame: bool) {
self.pending.push(Queued {
msg,
own_frame,
row: None,
});
}
/// gives what the relay changed on `did`'s row to the frame's own event, or to the first
/// event it forwards, so it lands in the txn that applies that event. a frame with no event
/// sends it on its own
fn hand_row(&mut self, did: &Did, row: RowDelta, firehose: Option<&Url>) {
let relayed = Some((did.clone(), row.version));
let carrier = match self.pending.iter().position(|queued| queued.own_frame) {
Some(own) => Some(own),
None => self
.pending
.iter()
.position(|queued| matches!(queued.msg, IndexerMessage::Event(_))),
};
if let Some(queued) = carrier.map(|at| &mut self.pending[at])
&& let IndexerMessage::Event(event) = &mut queued.msg
{
event.row = Some(row);
queued.row = relayed;
return;
}
self.push(
IndexerMessage::Row(Box::new(RelayRow {
did: did.clone(),
row,
firehose: firehose.cloned(),
ticket: None,
})),
true,
);
if let Some(queued) = self.pending.last_mut() {
queued.row = relayed;
}
}
/// gives the frame's ticket to its own event, or the row it sent on its own, so the indexer
/// marks it done after its commit. hands it back when the frame forwarded nothing, for the
/// relay to mark done itself
pub(crate) fn hand_off(&mut self, ticket: Ticket) -> Option<Ticket> {
match self
.pending
.iter_mut()
.find(|queued| queued.own_frame)
.map(|queued| &mut queued.msg)
{
Some(IndexerMessage::Event(event)) => event.ticket = Some(ticket),
Some(IndexerMessage::Row(relayed)) => relayed.ticket = Some(ticket),
_ => return Some(ticket),
}
None
}
pub(crate) fn discard(&mut self) {
self.pending.clear();
self.row = None;
}
pub(crate) fn account_snapshot(&self, repo_state: &RepoState) -> AccountSnapshot {
AccountSnapshot {
status: repo_state.status.clone(),
}
}
pub(crate) fn commit(
&mut self,
_state: &AppState,
_batch: &mut OwnedWriteBatch,
firehose: &Url,
commit: Commit<'static>,
chain_break: bool,
parsed_blocks: crate::car::CommitCar,
) -> Result<()> {
let seq = commit.seq;
let commit = IndexerEventData::Commit(IndexerCommitData {
commit,
chain_break,
parsed_blocks,
});
self.push(event(firehose, seq, commit), true);
Ok(())
}
pub(crate) fn sync(
&mut self,
_state: &AppState,
_batch: &mut OwnedWriteBatch,
firehose: &Url,
sync: Sync<'static>,
) -> Result<()> {
let sync = event(
firehose,
sync.seq,
IndexerEventData::Sync(sync.did.into_static()),
);
self.push(sync, true);
Ok(())
}
pub(crate) fn identity(
&mut self,
_state: &AppState,
_batch: &mut OwnedWriteBatch,
firehose: &Url,
identity: Identity,
changed: bool,
own_frame: bool,
) -> Result<()> {
let seq = identity.seq;
let identity = event(
firehose,
seq,
IndexerEventData::Identity(IndexerIdentityData { identity, changed }),
);
self.push(identity, own_frame);
Ok(())
}
#[allow(clippy::too_many_arguments)]
pub(crate) fn account(
&mut self,
_state: &AppState,
_batch: &mut OwnedWriteBatch,
firehose: &Url,
account: Account<'static>,
repo_state: &RepoState,
snapshot: AccountSnapshot,
was_active: bool,
) -> Result<()> {
let changed = repo_state.active != was_active || repo_state.status != snapshot.status;
let seq = account.seq;
let account = IndexerEventData::Account(IndexerAccountData {
account,
was_active,
changed,
});
self.push(event(firehose, seq, account), true);
Ok(())
}
/// keeps the row the message staged for the indexer to put on the row, see
/// [`crate::ingest::relay::overlay`]
pub(crate) fn stage_row(
&mut self,
_state: &AppState,
_txn: &mut crate::db::Txn<'_>,
did: &Did,
loaded: Option<RepoState<'static>>,
staged: Option<RepoState<'static>>,
firehose: Option<&Url>,
) -> Result<()> {
self.row = staged.map(|after| StagedRow {
did: did.clone(),
before: loaded,
after,
firehose: firehose.cloned(),
});
Ok(())
}
/// once the commit is through, the row change goes to the overlay and onto the frame's
/// events. the txn is committed even though it's empty in indexer builds, because a
/// skipped one would silently drop whatever someone later puts in its outbox
pub(crate) fn commit_txn(
&mut self,
state: &AppState,
txn: crate::db::Txn<'_>,
) -> Result<(Staged, crate::db::TxnCommitTimings)> {
let (_, timings) = txn.commit_with(|_| Ok(()))?;
if let Some(StagedRow {
did,
before,
after,
firehose,
}) = self.row.take()
{
let row = state.relay_rows.stage(&did, before, after);
self.hand_row(&did, row, firehose.as_ref());
}
Ok((Staged, timings))
}
pub(crate) fn flush(&mut self, state: &AppState, _staged: Staged) {
for Queued { msg, row, .. } in self.pending.drain(..) {
if self.hook.blocking_send(msg).is_err() {
// the indexer shard is gone, so nothing will land the row change, and the relay
// has to read the durable row again
warn!("relay shard: the indexer shard is gone, dropping what it was handed");
if let Some((did, version)) = row {
state.relay_rows.applied(&did, version);
}
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::ingest::mailbox::ShardedSender;
use crate::ingest::stream::Datetime;
use jacquard_common::types::did::Did;
use miette::IntoDiagnostic;
#[test]
fn synthetic_identity_does_not_checkpoint_its_trigger() -> miette::Result<()> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let state = AppState::new(&crate::config::Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})?;
let (inner, _) = ShardedSender::channel(1);
let seed = SinkSeed::new(IndexerTx::live_only(inner));
let mut sink = EventSink::new(&seed, state.firehose_stats.relay_shard(0));
let mut batch = state.db.inner.batch();
let firehose = Url::parse("wss://pds.example.com").into_diagnostic()?;
let identity = || Identity {
did: Did::new_static("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa").unwrap(),
handle: None,
seq: 42,
time: Datetime(chrono::Utc::now().into()),
};
sink.identity(&state, &mut batch, &firehose, identity(), true, false)?;
sink.identity(&state, &mut batch, &firehose, identity(), true, true)?;
assert!(
sink.hand_off(crate::ingest::cursor::Ticket::detached(42))
.is_none()
);
let carries = sink
.pending
.iter()
.map(|queued| match &queued.msg {
IndexerMessage::Event(event) => event.ticket.is_some(),
_ => panic!("expected indexer event"),
})
.collect::<Vec<_>>();
assert_eq!(carries, [false, true]);
Ok(())
}
/// with the indexer shard gone nothing lands a row change, so the relay mustn't keep reading
/// it over the durable row
#[test]
fn a_row_change_the_indexer_never_gets_is_let_go() -> miette::Result<()> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let state = AppState::new(&crate::config::Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})?;
let (inner, receivers) = ShardedSender::channel(1);
drop(receivers);
let seed = SinkSeed::new(IndexerTx::live_only(inner));
let mut sink = EventSink::new(&seed, state.firehose_stats.relay_shard(0));
let did = Did::new_static("did:plc:aaaaaaaaaaaaaaaaaaaaaaaa").unwrap();
let row = state.relay_rows.stage(&did, None, RepoState::backfilling());
sink.hand_row(&did, row, None);
sink.flush(&state, Staged);
assert_eq!(state.relay_rows.len(), 0);
Ok(())
}
#[test]
fn a_frame_that_forwarded_nothing_keeps_its_ticket() -> miette::Result<()> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let state = AppState::new(&crate::config::Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})?;
let (inner, _) = ShardedSender::channel(1);
let seed = SinkSeed::new(IndexerTx::live_only(inner));
let mut sink = EventSink::new(&seed, state.firehose_stats.relay_shard(0));
assert!(
sink.hand_off(crate::ingest::cursor::Ticket::detached(42))
.is_some()
);
Ok(())
}
}