//! writing a fetched repo's records through its indexer shard. the merge against what's stored
//! and every record's key, cid and block are worked out here, off the shard, and the shard gets
//! them a step at a time, each step answered before the next goes
use std::collections::VecDeque;
use std::ops::Bound;
use std::sync::Arc;
use fjall::Slice;
use jacquard_common::types::cid::IpldCid;
use jacquard_common::types::did::Did;
use miette::{IntoDiagnostic, Result};
use tracing::trace;
use super::records::{CheckedRecords, NewRecord, StoredKey};
use crate::backfill::diff::{RecordOp, diff};
use crate::db::keys;
use crate::db::types::{DbRkey, DbTid};
use crate::filter::FilterConfig;
use crate::ingest::indexer::IndexerTx;
use crate::ingest::indexer::bulk::forget_applying;
use crate::ingest::indexer::message::{
BackfillChunk, BackfillTail, BulkMessage, ChunkReply, IndexerMessage, OnShard, RecordWrite,
};
use crate::state::AppState;
/// how a backfill ended, which says what its task still has to do with the queue entry
#[derive(Debug)]
pub(crate) enum Outcome {
/// the shard applied what was buffered during the fetch and took the repo off the queue
Drained,
/// the last step committed but the shard left the queue entry alone, because the repo is
/// excluded or gone, or a chain break queued it again, see [`ChunkReply::Finished`]
Skipped,
/// hydrant is stopping, so the queue entry stays for the next start to run again
Stopping,
/// the queue entry went away, or the repo was excluded before the first step
Superseded,
/// a signal filter found nothing it wants, so nothing was sent
Discarded,
/// the pds says the repo is inactive, which the fetch wrote down instead of any records
Inactive,
}
/// one step of writes minted by [`write`] from [`diff`] against [`StoredHeads`]: keys are
/// strictly increasing, blocks are cid-verified, and `Create` means `StoredHeads` saw no
/// stored head for the key
#[derive(Debug)]
pub(crate) struct DiffStep(Vec<RecordWrite>);
impl DiffStep {
pub(crate) fn into_writes(self) -> Vec<RecordWrite> {
self.0
}
#[cfg(test)]
pub(crate) fn empty() -> Self {
Self(Vec::new())
}
}
/// record bytes a step stops at, short of its chunk of records. the records come from the pds,
/// so a chunk of big ones would otherwise put far more than a step's share of the budget on the
/// heap and into one shard commit
pub(super) const STEP_BYTES: usize = 4 * 1024 * 1024;
/// sends `checked`'s writes to `did`'s shard, a chunk of records or [`STEP_BYTES`] per step,
/// and waits for each answer. the last step also carries `tail`, so a repo whose writes fit in
/// one step stores its records in one batch
#[allow(clippy::too_many_arguments)]
pub(super) fn write(
app_state: &AppState,
indexer: &IndexerTx,
did: &Did,
pending_key: &Slice,
rev: DbTid,
checked: &CheckedRecords,
tail: BackfillTail,
) -> Result<Outcome> {
let chunk = app_state.backfill_chunk_records;
let new = checked.new_records();
let stored = StoredHeads::new(app_state, did, chunk, tail.fetched_only.clone());
let mut writes = diff(new, stored)
.map(|op| op.and_then(record_write))
.peekable();
let mut tail = Some(tail);
let mut sent = 0;
loop {
let mut step = Vec::new();
let mut step_bytes = 0;
while step.len() < chunk
&& step_bytes < STEP_BYTES
&& let Some(write) = writes.next()
{
match write {
Ok(write) => {
if let RecordWrite::Create { block, .. } | RecordWrite::Update { block, .. } =
&write
{
step_bytes += block.size();
}
step.push(write);
}
// the shard forgets a backfill whose step failed on its own, but not one whose
// task stopped sending them
Err(e) if sent > 0 => {
let (did, pending_key) = (did.clone(), pending_key.clone());
let forget = OnShard {
did: did.clone(),
work: Box::new(move |state, shard| {
forget_applying(state, shard, &did, &pending_key)
}),
};
let _ = indexer.blocking_send(IndexerMessage::OnShard(forget));
return Err(e);
}
Err(e) => return Err(e),
}
}
let last = writes.peek().is_none();
if app_state.stop.is_set() {
return Ok(Outcome::Stopping);
}
let (reply, answer) = tokio::sync::oneshot::channel();
let chunk = BackfillChunk {
did: did.clone(),
pending_key: pending_key.clone(),
rev,
writes: DiffStep(step),
tail: if last {
tail.take().map(Box::new)
} else {
None
},
reply,
};
if indexer
.bulk
.blocking_send(BulkMessage::Chunk(chunk))
.is_err()
{
return stopping_or(app_state, "the indexer's shard is gone");
}
let Ok(answer) = answer.blocking_recv() else {
let why = if last {
"the indexer failed to write the last backfill step or drain what was buffered during the fetch"
} else {
"the indexer failed to write a backfill step"
};
return stopping_or(app_state, why);
};
sent += 1;
trace!(step = sent, ?answer, "wrote backfill step");
match answer {
ChunkReply::Written => {}
ChunkReply::Superseded => return Ok(Outcome::Superseded),
ChunkReply::Finished { drained: true } => return Ok(Outcome::Drained),
ChunkReply::Finished { drained: false } => return Ok(Outcome::Skipped),
ChunkReply::Failed(why) => return Err(miette::miette!("{why}")),
}
#[cfg(test)]
if let Some(after_chunk) = app_state.after_chunk.lock().clone() {
after_chunk(sent);
}
}
}
fn stopping_or(app_state: &AppState, why: &str) -> Result<Outcome> {
if app_state.stop.is_set() {
return Ok(Outcome::Stopping);
}
Err(miette::miette!("{why}"))
}
fn record_write(op: RecordOp<StoredKey, NewRecord>) -> Result<RecordWrite> {
Ok(match op {
RecordOp::Create { key, record } => {
let (collection, rkey) = key.split_at(record.collection_len);
RecordWrite::Create {
collection: collection.into(),
rkey: DbRkey::new(&rkey[1..]),
block: record.block,
}
}
RecordOp::Update { key, record } => {
let (collection, rkey) = key.split_at(record.collection_len);
RecordWrite::Update {
collection: collection.into(),
rkey: DbRkey::new(&rkey[1..]),
block: record.block,
}
}
RecordOp::Delete(stored) => {
let (collection, rkey) = keys::split_record_suffix(stored.as_ref())
.map_err(|e| miette::miette!("invalid record key '{:?}': {e}", stored.key))?;
RecordWrite::Delete {
collection: collection.into(),
rkey,
}
}
})
}
/// the repo's stored records in key order, with their cids, read a page at a time from just
/// after the last page. no snapshot stays open for the whole backfill, and a page read after
/// earlier steps committed is still right: those steps only wrote keys the merge had already
/// passed, and nothing else writes a queued repo's records but a redaction, which keeps the cid
struct StoredHeads<'a> {
app_state: &'a AppState,
did: &'a Did,
/// see [`BackfillTail::fetched_only`]
fetched_only: Option<Arc<FilterConfig>>,
prefix: Vec<u8>,
after: Option<Slice>,
page: VecDeque<(StoredKey, IpldCid)>,
page_len: usize,
done: bool,
}
impl<'a> StoredHeads<'a> {
fn new(
app_state: &'a AppState,
did: &'a Did,
page_len: usize,
fetched_only: Option<Arc<FilterConfig>>,
) -> Self {
Self {
app_state,
did,
fetched_only,
prefix: keys::record_prefix_did(did),
after: None,
page: VecDeque::new(),
page_len: page_len.clamp(1, 4096),
// ephemeral mode stores no records, so there's nothing to merge against
done: app_state.ephemeral,
}
}
fn read_page(&mut self) -> Result<()> {
let start = match &self.after {
Some(after) => Bound::Excluded(after.to_vec()),
None => Bound::Included(self.prefix.clone()),
};
// the prefix ends in the did's NUL separator, so one past it bounds the repo
let mut end = self.prefix.clone();
if let Some(sep) = end.last_mut() {
*sep += 1;
}
let range = self
.app_state
.db
.indexer
.head_range((start, Bound::Excluded(end)))
.take(self.page_len);
let mut read = 0;
for guard in range {
read += 1;
let (key, value) = guard.into_inner().into_diagnostic()?;
self.after = Some(key.clone());
if let Some(filter) = &self.fetched_only {
let (collection, _) = keys::split_record_suffix(&key[self.prefix.len()..])
.map_err(|e| miette::miette!("invalid record key '{key:?}': {e}"))?;
if !filter.matches_collection(collection) {
continue;
}
}
// heads keep bodies, so the cid comes from hashing the body (links-only mode keeps
// the cid itself)
let cid = crate::db::ipld_cid_from_record_value(&value)
.map_err(|e| miette::miette!("invalid record value for {}: {e}", self.did))?;
self.page.push_back((
StoredKey {
key,
suffix: self.prefix.len(),
},
cid,
));
}
self.done = read < self.page_len;
Ok(())
}
}
impl Iterator for StoredHeads<'_> {
type Item = Result<(StoredKey, IpldCid)>;
fn next(&mut self) -> Option<Self::Item> {
if self.page.is_empty()
&& !self.done
&& let Err(e) = self.read_page()
{
self.done = true;
return Some(Err(e));
}
self.page.pop_front().map(Ok)
}
}