use std::collections::HashMap;
use std::sync::Arc;
use super::bulk::ShardState;
use crate::backfill::worker::apply::DiffStep;
use crate::car::CarBlock;
use crate::db::types::{DbRkey, DbTid};
use crate::filter::FilterConfig;
use crate::ingest::cursor::Ticket;
use crate::ingest::mailbox::{ShardedMessage, ShardedReceiver, ShardedSender};
use crate::ingest::relay::overlay::RowDelta;
use crate::ingest::stream;
use crate::state::AppState;
use crate::types::{DeleteBodiesReport, RepoState};
use fjall::Slice;
use jacquard_common::types::did::Did;
use miette::Result;
use smol_str::SmolStr;
use tokio::sync::oneshot;
use url::Url;
#[derive(Debug)]
pub struct IndexerCommitData {
pub commit: stream::Commit<'static>,
/// true if the relay detected a gap (missing seq) before this commit
/// and the indexer should trigger a backfill.
pub chain_break: bool,
/// the car the relay already parsed and validated, so the indexer does not re-parse it.
pub parsed_blocks: crate::car::CommitCar,
}
#[derive(Debug)]
pub struct IndexerIdentityData {
pub identity: stream::Identity,
/// whether the identity actually changed (handle or key).
pub changed: bool,
}
#[derive(Debug)]
pub struct IndexerAccountData {
pub account: stream::Account<'static>,
/// whether the repo was active prior to this event.
pub was_active: bool,
/// whether any state actually changed (active or status).
pub changed: bool,
}
#[derive(Debug)]
#[allow(clippy::large_enum_variant)]
pub enum IndexerEventData {
Commit(IndexerCommitData),
Identity(IndexerIdentityData),
Account(IndexerAccountData),
Sync(Did),
}
#[derive(Debug)]
pub struct IndexerEvent {
pub seq: i64,
pub firehose: Url,
/// the frame this event came from, done once the shard committed what the event did. a
/// synthetic event paired with the real one at the same upstream sequence has none
pub ticket: Option<Ticket>,
/// what the relay changed on the repo row for this event, which only lands with it
pub(crate) row: Option<RowDelta>,
pub data: IndexerEventData,
}
/// a relay row change whose frame had no event of its own to carry it
#[derive(Debug)]
pub struct RelayRow {
pub(crate) did: Did,
pub(crate) row: RowDelta,
pub(crate) firehose: Option<Url>,
/// the frame, done once the shard committed the row
pub(crate) ticket: Option<Ticket>,
}
/// message sent from `relay_worker` to the indexer (`FirehoseWorker`) after
/// validation and repo-state management are done.
#[derive(Debug)]
pub enum IndexerMessage {
/// a firehose event that passed relay-side validation.
Event(Box<IndexerEvent>),
/// a relay row change with no event, see [`RelayRow`]
Row(Box<RelayRow>),
/// work only the repo's owner shard may do, see [`OnShard`]
OnShard(OnShard),
}
/// what the api asks of a repo, see [`crate::control::repos::ReposControl`]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RepoOp {
Track,
Untrack,
Resync,
/// a crawl's resync costs a refetch, a plain one is free
ResyncCrawled,
}
/// work only `did`'s owner shard may do, run between its live messages. it holds the shard while
/// it runs and live messages wait behind it, so two rules keep it short: it only does db work on
/// the shard's state and a txn of its own, with no network, no waiting and no sleeping, and the
/// caller bounds its size to one repo's op or one batch of at most 256 repos, so nothing the
/// size of a whole repo comes through here. erases, wipes and backfill steps are bulk kinds of
/// their own, written a step at a time
pub struct OnShard {
pub(crate) did: Did,
pub(crate) work: ShardWork,
}
pub(crate) type ShardWork = Box<dyn FnOnce(&AppState, &mut ShardState) + Send + Sync>;
impl OnShard {
/// `work` for `did`'s owner, and where its result comes back
fn answering<T: Send + 'static>(
did: &Did,
work: impl FnOnce(&AppState, &mut ShardState) -> T + Send + Sync + 'static,
) -> (Self, oneshot::Receiver<T>) {
let (reply, answer) = oneshot::channel();
let on = OnShard {
did: did.clone(),
work: Box::new(move |state, shard| {
let _ = reply.send(work(state, shard));
}),
};
(on, answer)
}
}
fn shard_gone() -> miette::Report {
miette::miette!("the repo's indexer shard is gone")
}
fn shard_stopped() -> miette::Report {
miette::miette!("the repo's indexer shard stopped")
}
impl std::fmt::Debug for OnShard {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("OnShard").field("did", &self.did).finish()
}
}
/// one record write a backfill worked out off the shard, for the shard to store
#[derive(Debug)]
pub(crate) enum RecordWrite {
Create {
collection: SmolStr,
rkey: DbRkey,
block: CarBlock,
},
Update {
collection: SmolStr,
rkey: DbRkey,
block: CarBlock,
},
Delete {
collection: SmolStr,
rkey: DbRkey,
},
}
/// what a backfill writes after its records: the repo's row and the counts its first pass took
#[derive(Debug)]
pub(crate) struct BackfillTail {
/// the row as the backfill read it before the fetch
pub before: RepoState<'static>,
/// `before` with the doc the backfill resolved and the root it fetched
pub after: RepoState<'static>,
pub collection_counts: HashMap<SmolStr, u64>,
/// the filter whose collections a sparse backfill fetched. it diffs, deletes and counts
/// only records in those, since it never saw the rest. `None` for a getRepo, which saw
/// the whole repo
pub fetched_only: Option<Arc<FilterConfig>>,
}
/// one step of a backfill's record writes, for the repo's owner shard to commit. the task sends
/// the next step only once this one is answered. the step with a tail is the last one, and it
/// finishes the backfill
#[derive(Debug)]
pub(crate) struct BackfillChunk {
pub did: Did,
pub pending_key: Slice,
/// the fetched car's rev, which history entries are keyed by
pub rev: DbTid,
pub writes: DiffStep,
pub tail: Option<Box<BackfillTail>>,
/// dropped unanswered when the step failed or hydrant is stopping
pub reply: oneshot::Sender<ChunkReply>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) enum ChunkReply {
/// written, so the next step can come
Written,
/// nothing written: the queue entry the backfill ran for went away, or the repo was
/// excluded before its first step
Superseded,
/// the last step committed. `drained` when the shard also applied what was buffered during
/// the fetch and took the repo off the queue, which it doesn't when the queue entry went
/// away in between or a chain break came in during the steps and queued the repo again
Finished { drained: bool },
/// the step failed, and why, so the backfill retries with that error. a drain step that
/// failed partway keeps what it got through, like any drain step
Failed(String),
}
impl ChunkReply {
/// [`ChunkReply::Failed`] with `e` and everything it was caused by
pub(crate) fn failed(e: &miette::Report) -> Self {
let why = e.chain().map(ToString::to_string).collect::<Vec<_>>();
ChunkReply::Failed(why.join(": "))
}
}
/// an operator erase of everything stored for `did`, see [`crate::db::erase_step`]
#[derive(Debug)]
pub(crate) struct EraseRequest {
pub did: Did,
/// also exclude the repo, so it isn't indexed again until it's taken back out
pub exclude: bool,
/// answered once the last step committed, dropped if a step failed or hydrant is stopping
pub reply: oneshot::Sender<DeleteBodiesReport>,
}
/// what the bulk channel carries to a shard, which writes it in steps between live messages
#[derive(Debug)]
pub(crate) enum BulkMessage {
Chunk(BackfillChunk),
Erase(EraseRequest),
}
#[derive(Clone, Debug)]
pub struct IndexerTx {
pub(crate) inner: ShardedSender<IndexerMessage>,
pub(crate) bulk: ShardedSender<BulkMessage>,
}
pub type IndexerRx = ShardedReceiver<IndexerMessage>;
pub(crate) type BulkRx = ShardedReceiver<BulkMessage>;
impl IndexerTx {
#[cfg(test)]
pub async fn send(
&self,
msg: IndexerMessage,
) -> Result<(), tokio::sync::mpsc::error::SendError<IndexerMessage>> {
self.inner.send(msg).await
}
pub fn blocking_send(
&self,
msg: IndexerMessage,
) -> Result<(), tokio::sync::mpsc::error::SendError<IndexerMessage>> {
self.inner.blocking_send(msg)
}
/// runs `work` on `did`'s owner shard, see [`OnShard`], and returns `work`'s result.
/// fails when the shard is gone, or stopped before it ran
pub(crate) async fn on_owner<T: Send + 'static>(
&self,
did: &Did,
work: impl FnOnce(&AppState, &mut ShardState) -> T + Send + Sync + 'static,
) -> miette::Result<T> {
let (on, answer) = OnShard::answering(did, work);
self.inner
.send(IndexerMessage::OnShard(on))
.await
.map_err(|_| shard_gone())?;
answer.await.map_err(|_| shard_stopped())
}
/// [`Self::on_owner`] from a thread that may block
pub(crate) fn on_owner_blocking<T: Send + 'static>(
&self,
did: &Did,
work: impl FnOnce(&AppState, &mut ShardState) -> T + Send + Sync + 'static,
) -> miette::Result<T> {
let (on, answer) = OnShard::answering(did, work);
self.inner
.blocking_send(IndexerMessage::OnShard(on))
.map_err(|_| shard_gone())?;
answer.blocking_recv().map_err(|_| shard_stopped())
}
/// a sender for live messages alone, with nothing behind its backfill steps
#[cfg(test)]
pub(crate) fn live_only(inner: ShardedSender<IndexerMessage>) -> Self {
Self {
inner,
bulk: ShardedSender::channel(1).0,
}
}
}
impl ShardedMessage for BulkMessage {
fn shard_idx(&self, num_shards: usize) -> usize {
let did = match self {
BulkMessage::Chunk(chunk) => &chunk.did,
BulkMessage::Erase(erase) => &erase.did,
};
shard_of(did, num_shards)
}
}
pub(crate) fn shard_of(did: &Did, num_shards: usize) -> usize {
(crate::util::hash(did) as usize) % num_shards
}
impl IndexerMessage {
/// what the relay sends for an account it hadn't seen and that came with no event
#[cfg(test)]
pub(crate) fn new_repo_for_test(did: Did) -> Self {
Self::Row(Box::new(RelayRow {
did,
row: RowDelta {
before: None,
after: RepoState::backfilling(),
version: 0,
},
firehose: None,
ticket: None,
}))
}
}
impl ShardedMessage for IndexerMessage {
fn shard_idx(&self, num_shards: usize) -> usize {
// keep every message for a did on one ordered path.
let did = match self {
IndexerMessage::Event(e) => match &e.data {
IndexerEventData::Commit(m) => &m.commit.repo,
IndexerEventData::Identity(m) => &m.identity.did,
IndexerEventData::Account(m) => &m.account.did,
IndexerEventData::Sync(did) => did,
},
IndexerMessage::Row(relayed) => &relayed.did,
IndexerMessage::OnShard(on) => &on.did,
};
shard_of(did, num_shards)
}
}