//! bulk work on the repo's owner shard, a step at a time so a big repo holds the shard for one
//! step and not its whole length: a backfill's records, which its task works out off the shard
//! and sends here, the drain of what it buffered meanwhile, an operator erase and a deleted
//! account's wipe. an erase or a wipe leaves a marker for as long as it runs, so a restart
//! resumes it, and so does a backfill's, so nothing can untrack a half written repo
use std::collections::HashMap;
use fjall::Slice;
use jacquard_common::types::did::Did;
use miette::{IntoDiagnostic, Result};
use tokio::sync::oneshot;
use tracing::{debug, error};
use super::message::{
BackfillChunk, BackfillTail, BulkMessage, ChunkReply, EraseRequest, shard_of,
};
use crate::backfill::worker::apply::DiffStep;
use super::shard::WorkerContext;
use super::worker::FirehoseWorker;
use crate::backfill::backfilled_state;
use crate::db::types::DbTid;
use crate::db::{self, Txn, keys};
use crate::ops;
use crate::state::AppState;
use crate::types::{BulkJobV1, DeleteBodiesReport, GaugeState, RepoStatus};
/// what a repo's owner shard is writing for it in steps, see [`ShardState`]
#[derive(Debug, Clone)]
pub(crate) enum Busy {
Apply(Applying),
Erase,
Wipe,
}
impl Busy {
/// an erase or a wipe, which the shard drives itself, so it drops the repo's live messages
/// until the last step. they'd only land on what's about to be gone
pub(crate) fn drops_live(&self) -> bool {
!matches!(self, Busy::Apply(_))
}
}
/// a repo whose backfill is being written in steps
#[derive(Debug, Clone)]
pub(crate) struct Applying {
pub(crate) pending_key: Slice,
/// a chain break or a sync came in during the steps, so the repo backfills again once the
/// last one commits
pub(crate) requeue_after: bool,
}
/// what an indexer shard keeps about the repos it's writing in steps. only that shard's thread
/// ever reads or writes it, so nothing needs a lock to agree with it
#[derive(Default)]
pub(crate) struct ShardState {
busy: HashMap<Did, Busy>,
jobs: Jobs,
/// repos whose job marker some work on the shard just left, see [`Self::mark`]
marked: Vec<Did>,
}
impl ShardState {
/// whether the repo's erase or wipe is running, see [`Busy::drops_live`]
pub(crate) fn drops_live(&self, did: &Did) -> bool {
self.busy.get(did).is_some_and(Busy::drops_live)
}
/// has the shard pick up the job whose marker `did` got from work running on it, right
/// after that work and before its next message, so the repo is busy from then on
pub(crate) fn mark(&mut self, did: Did) {
self.marked.push(did);
}
pub(super) fn take_marked(&mut self) -> Vec<Did> {
std::mem::take(&mut self.marked)
}
pub(crate) fn is_applying(&self, did: &Did, pending_key: &[u8]) -> bool {
matches!(
self.busy.get(did),
Some(Busy::Apply(applying)) if applying.pending_key.as_ref() == pending_key
)
}
/// see [`Busy::holds`]
pub(crate) fn is_busy(&self, state: &AppState, did: &Did) -> Result<bool> {
self.busy
.get(did)
.map_or(Ok(false), |busy| busy.holds(state))
}
}
impl Busy {
/// whether this would leave the repo half written if anything took it off the queue or
/// untracked it meanwhile. a backfill counts while its queue entry is there, since one
/// rebuilt from its marker at a start can fail before its first step and leave the entry
/// behind
pub(crate) fn holds(&self, state: &AppState) -> Result<bool> {
match self {
Busy::Apply(applying) => state
.db
.indexer
.is_pending(&applying.pending_key)
.into_diagnostic(),
Busy::Erase | Busy::Wipe => Ok(true),
}
}
}
/// what `did`'s owner shard is writing for it in steps. with no shard running, it's what the
/// repo's marker says, which is what a shard would pick up at its start
pub(crate) fn busy_of(
state: &AppState,
shard: Option<&ShardState>,
did: &Did,
) -> Result<Option<Busy>> {
if let Some(shard) = shard {
return Ok(shard.busy.get(did).cloned());
}
Ok(db::read_bulk_job(&state.db, did)?.map(|job| match job {
BulkJobV1::Apply { pending_key } => Busy::Apply(Applying {
pending_key: pending_key.into(),
requeue_after: false,
}),
BulkJobV1::Erase => Busy::Erase,
BulkJobV1::Wipe { .. } => Busy::Wipe,
}))
}
impl FirehoseWorker {
/// commits one backfill step. after the last one the repo's buffered commits still have to
/// be applied, which runs as steps of its own, so this returns the job that does it
pub(super) fn write_chunk(
state: &AppState,
shard: &mut ShardState,
chunk: BackfillChunk,
) -> Option<DrainJob> {
let BackfillChunk {
did,
pending_key,
rev,
writes,
tail,
reply,
} = chunk;
let _span = tracing::info_span!("backfill", did = %did).entered();
match Self::write_step(state, shard, &did, &pending_key, rev, writes, tail) {
Ok(Written::Records) => {
let _ = reply.send(ChunkReply::Written);
None
}
Ok(Written::Last) => Some(DrainJob {
did,
pending_key,
reply,
steps: 0,
}),
Ok(Written::Superseded) => {
forget_applying(state, shard, &did, &pending_key);
let _ = reply.send(ChunkReply::Superseded);
None
}
Err(e) => {
forget_applying(state, shard, &did, &pending_key);
error!(err = %e, "failed to write a backfill step");
let _ = reply.send(ChunkReply::failed(&e));
None
}
}
}
/// applies the next few commits the repo buffered while it backfilled. the step that empties
/// the buffer takes the repo off the queue and answers the backfill. a chain break that came
/// in meanwhile queues the repo again instead, and the rest of the buffer waits for that
/// backfill
pub(super) fn drain_step(
id: usize,
state: &AppState,
shard: &mut ShardState,
mut job: DrainJob,
lifecycle_transitions: &mut Vec<(Did, GaugeState)>,
) -> Option<DrainJob> {
let _span = tracing::info_span!("backfill", did = %job.did).entered();
let limit = (job.steps < DRAIN_STEPS_BEFORE_ALL).then_some(state.backfill_chunk_records);
let drained = Self::drain_some(
id,
state,
shard,
&job.did,
&job.pending_key,
limit,
lifecycle_transitions,
);
job.steps += 1;
#[cfg(test)]
if let Some(after_drain_step) = state.after_drain_step.lock().clone() {
after_drain_step(job.steps);
}
match drained {
Ok(Drained::More) => return Some(job),
Ok(Drained::Done(drained)) => {
let _ = job.reply.send(ChunkReply::Finished { drained });
}
Err(e) => {
error!(err = %e, "failed to drain what was buffered during the fetch");
let _ = job.reply.send(ChunkReply::failed(&e));
}
}
forget_applying(state, shard, &job.did, &job.pending_key);
None
}
fn drain_some(
id: usize,
state: &AppState,
shard: &mut ShardState,
did: &Did,
pending_key: &[u8],
limit: Option<usize>,
lifecycle_transitions: &mut Vec<(Did, GaugeState)>,
) -> Result<Drained> {
let requeue = matches!(
shard.busy.get(did),
Some(Busy::Apply(applying)) if applying.requeue_after
);
let mut ctx = WorkerContext {
state,
shard,
txn: Txn::new(&state.db),
lifecycle_transitions,
notify_backfill: false,
backfill_queued: false,
started_job: None,
};
let drained = if requeue {
debug!("chain break during the backfill, backfilling again");
match ctx.txn.read_repo(did) {
Ok(Some(repo_state)) => {
Self::trigger_backfill(&mut ctx, did, repo_state, "commit during the steps")
.map(|_| Drained::Done(false))
}
Ok(None) => Ok(Drained::Done(false)),
Err(e) => Err(e.into()),
}
} else {
Self::finish_backfill(&mut ctx, did, pending_key, limit)
};
if let Ok(Drained::Done(_)) = drained {
stage_forget_applying(&mut ctx.txn, state, &mut *ctx.shard, did, pending_key)?;
}
// a step that fails partway still commits what it got through, including taking the
// commit that failed off the buffer, since the refetch the failure leads to has it anyway
let notify_backfill = Self::commit(id, ctx)
.ok_or_else(|| miette::miette!("failed to commit a drain step"))?;
if notify_backfill {
state.notify_backfill();
}
Ok(drained?)
}
/// commits one step's writes, and on the last step the repo's row, metadata and counts
fn write_step(
state: &AppState,
shard: &mut ShardState,
did: &Did,
pending_key: &Slice,
rev: DbTid,
writes: DiffStep,
tail: Option<Box<BackfillTail>>,
) -> Result<Written> {
let db = &state.db;
let mut txn = Txn::new(db);
// an erase or a wipe running for the repo would forget these records behind its cursor
// or move them to history, and taking over its marker would stop a restart resuming it
if !txn.names_pending(did, pending_key)? || shard.drops_live(did) {
return Ok(Written::Superseded);
}
let running = matches!(
shard.busy.get(did),
Some(Busy::Apply(applying)) if applying.pending_key == *pending_key
);
if !running {
// read once, at the first step. an exclude that lands between steps doesn't stop
// them, since that would leave the repo half written, and the task takes it off the
// queue once they're done
if txn.is_excluded(did)? {
return Ok(Written::Superseded);
}
shard.busy.insert(
did.clone(),
Busy::Apply(Applying {
pending_key: pending_key.clone(),
requeue_after: false,
}),
);
db::stage_bulk_job(
&mut txn.batch,
db,
did,
&BulkJobV1::Apply {
pending_key: pending_key.to_vec(),
},
)?;
}
let mut records = txn.backfill_step_records(state, &rev, did);
records.apply(writes)?;
let Some(tail) = tail else {
txn.commit()?;
return Ok(Written::Records);
};
let BackfillTail {
before,
after,
collection_counts,
fetched_only,
} = *tail;
let Some(current) = records.read_repo()? else {
return Ok(Written::Superseded);
};
let after = backfilled_state(&before, after, current);
records.update_repo_state(&after)?;
ops::stage_buffer_covered_by(&mut txn.batch, db, did, rev)?;
let metadata_key = keys::repo_metadata_key(did);
let metadata_bytes = db
.repo_metadata
.get(&metadata_key)
.into_diagnostic()?
.ok_or_else(|| miette::miette!("repo metadata not found for {did}"))?;
let mut metadata = db::deser_repo_meta(&metadata_bytes)?;
metadata.tracked = true;
metadata.retry_count = 0;
metadata.backfilled_rev = Some(rev);
txn.batch.insert(
&db.repo_metadata,
&metadata_key,
db::ser_repo_meta(&metadata)?,
);
let counts = collection_counts
.iter()
.map(|(collection, count)| (collection.as_str(), *count));
match fetched_only {
Some(filter) => txn.replace_record_counts(
did,
|collection| filter.matches_collection(collection),
counts,
)?,
None => txn.replace_record_counts(did, |_| true, counts)?,
}
#[cfg(feature = "indexer_stream")]
crate::backfill::worker::process::BackfillImported { before, after }
.stage(did, &mut txn.outbox.stream);
txn.commit()?;
db.inner
.persist(fjall::PersistMode::Buffer)
.into_diagnostic()?;
Ok(Written::Last)
}
}
enum Written {
Records,
Last,
Superseded,
}
/// what waits on a shard's bulk queue: a backfill step its task sent or an erase the api asked
/// for, or a step of a job the shard queued for itself
pub(crate) enum BulkStep {
Chunk(BackfillChunk),
Erase(EraseRequest),
Drain(DrainJob),
/// the next step of the repo's job in [`Jobs`]
Job(Did),
}
impl From<BulkMessage> for BulkStep {
fn from(message: BulkMessage) -> Self {
match message {
BulkMessage::Chunk(chunk) => BulkStep::Chunk(chunk),
BulkMessage::Erase(erase) => BulkStep::Erase(erase),
}
}
}
/// the erases and wipes a shard is running, by repo. each has exactly one [`BulkStep::Job`]
/// waiting on the queue, so one that takes over from another keeps the step already there
type Jobs = HashMap<Did, Job>;
pub(crate) enum Job {
Erase(EraseJob),
Wipe(WipeJob),
}
/// a deleted account's heads moving to history at the rev its marker names
pub(crate) struct WipeJob {
death_rev: DbTid,
/// the last head the steps so far moved
after: Option<Vec<u8>>,
steps: usize,
}
/// an operator erase, and the callers waiting for its report
#[derive(Default)]
pub(crate) struct EraseJob {
cursor: db::EraseCursor,
report: DeleteBodiesReport,
replies: Vec<oneshot::Sender<DeleteBodiesReport>>,
steps: usize,
}
/// a backfill whose records are all written, applying what the repo buffered during the fetch
pub(crate) struct DrainJob {
did: Did,
pending_key: Slice,
reply: oneshot::Sender<ChunkReply>,
/// the drain steps done so far
steps: usize,
}
/// what a drain step came to
pub(crate) enum Drained {
/// the buffer has more, for the next step
More,
/// finished, and whether that took the repo off the queue
Done(bool),
}
/// drain steps that stop at a chunk's worth of record writes before one takes everything left.
/// an account that commits faster than its steps drain would keep the buffer from ever
/// emptying, so after these the last step takes whatever is there in one batch
pub(crate) const DRAIN_STEPS_BEFORE_ALL: usize = 64;
/// drops `did`'s backfill from the shard's busy repos and its marker, if they're still the
/// ones for `pending_key`. a stop leaves both, since the marker is what keeps a restart from
/// letting anything untrack the half written repo before the backfill runs again
pub(crate) fn forget_applying(
state: &AppState,
shard: &mut ShardState,
did: &Did,
pending_key: &[u8],
) {
if state.stop.is_set() {
return;
}
let mut txn = Txn::new(&state.db);
let forgotten = stage_forget_applying(&mut txn, state, shard, did, pending_key)
.and_then(|()| txn.commit().map(|_| ()));
if let Err(e) = forgotten {
error!(err = %e, "failed to forget a backfill's marker");
}
}
/// [`forget_applying`] in `txn`
fn stage_forget_applying(
txn: &mut Txn<'_>,
state: &AppState,
shard: &mut ShardState,
did: &Did,
pending_key: &[u8],
) -> Result<()> {
if matches!(
shard.busy.get(did),
Some(Busy::Apply(applying)) if applying.pending_key.as_ref() == pending_key
) {
shard.busy.remove(did);
}
// an erase or a wipe that took over left its own marker, which isn't this one to drop
if matches!(
db::read_bulk_job(&state.db, did)?,
Some(BulkJobV1::Apply { pending_key: ref marked }) if marked == pending_key
) {
txn.batch.remove(&state.db.crawler, keys::bulk_job_key(did));
}
Ok(())
}
impl FirehoseWorker {
/// marks `did` to backfill again once its running steps finish, if it has any. a chain break
/// can't swap its queue key now, since that would throw away a half written repo
pub(super) fn requeue_after_steps(
state: &AppState,
shard: &mut ShardState,
did: &Did,
) -> Result<bool> {
if !shard.is_busy(state, did)? {
return Ok(false);
}
if let Some(Busy::Apply(applying)) = shard.busy.get_mut(did) {
applying.requeue_after = true;
}
Ok(true)
}
}
impl FirehoseWorker {
/// picks up every job whose marker this shard owns, before it serves anything, and says
/// which repos have an erase or a wipe step to queue
pub(super) fn resume_jobs(
id: usize,
num_shards: usize,
state: &AppState,
shard: &mut ShardState,
) -> Vec<Did> {
let mut marked = Vec::new();
for guard in state.db.crawler.prefix(keys::BULK_JOB_PREFIX) {
let did = guard
.key()
.into_diagnostic()
.and_then(|key| keys::bulk_job_parse_key(&key).map(|did| did.to_did()));
match did {
Ok(did) if shard_of(&did, num_shards) == id => marked.push(did),
Ok(_) => {}
Err(e) => error!(err = %e, "failed to read a job marker"),
}
}
marked
.into_iter()
.filter(|did| Self::resume_job(state, shard, did))
.collect()
}
/// runs the job `did`'s marker names, if the state it was left for still holds, and drops
/// the marker otherwise. a backfill only counts as busy again, since its task does the
/// writing. true when an erase or a wipe step has to be queued
pub(super) fn resume_job(state: &AppState, shard: &mut ShardState, did: &Did) -> bool {
Self::run_marked_job(state, shard, did, false)
}
/// [`Self::resume_job`] for a job the shard's own txn just started, which nothing could
/// have moved on from yet. the account event that starts a wipe can come before anything
/// wrote its row as deleted
pub(super) fn start_job(state: &AppState, shard: &mut ShardState, did: &Did) -> bool {
Self::run_marked_job(state, shard, did, true)
}
fn run_marked_job(state: &AppState, shard: &mut ShardState, did: &Did, fresh: bool) -> bool {
let _span = tracing::info_span!("job", did = %did).entered();
Self::try_resume_job(state, shard, did, fresh).unwrap_or_else(|e| {
error!(err = %e, "failed to resume a job");
false
})
}
fn try_resume_job(
state: &AppState,
shard: &mut ShardState,
did: &Did,
fresh: bool,
) -> Result<bool> {
let db = &state.db;
let mut txn = Txn::new(db);
let Some(job) = db::read_bulk_job(db, did)? else {
return Ok(false);
};
let due = fresh
|| match &job {
BulkJobV1::Apply { pending_key } => txn.names_pending(did, pending_key)?,
BulkJobV1::Wipe { .. } => txn
.read_repo(did)?
.is_some_and(|repo| repo.status == RepoStatus::Deleted && !repo.active),
BulkJobV1::Erase => db
.repo_metadata
.get(keys::repo_metadata_key(did))
.into_diagnostic()?
.map(|bytes| db::deser_repo_meta(&bytes))
.transpose()?
.is_none_or(|metadata| !metadata.tracked),
};
if !due {
debug!(?job, "dropping a job marker whose state moved on");
txn.batch.remove(&db.crawler, keys::bulk_job_key(did));
txn.commit()?;
return Ok(false);
}
Ok(match job {
BulkJobV1::Apply { pending_key } => {
shard
.busy
.entry(did.clone())
.or_insert(Busy::Apply(Applying {
pending_key: pending_key.into(),
requeue_after: false,
}));
false
}
BulkJobV1::Wipe { death_rev } => {
if shard.jobs.contains_key(did) {
return Ok(false);
}
shard.busy.insert(did.clone(), Busy::Wipe);
shard.jobs.insert(
did.clone(),
Job::Wipe(WipeJob {
death_rev,
after: None,
steps: 0,
}),
);
true
}
BulkJobV1::Erase => {
shard.busy.insert(did.clone(), Busy::Erase);
match shard.jobs.get_mut(did) {
Some(Job::Erase(_)) => false,
// the erase forgets everything the wipe would have moved
Some(job) => {
*job = Job::Erase(EraseJob::default());
false
}
None => {
shard
.jobs
.insert(did.clone(), Job::Erase(EraseJob::default()));
true
}
}
}
})
}
/// untracks the repo and takes it off every queue in one txn, then erases what's stored
/// for it in steps, and answers `erase` after the last. true when a step has to be queued
pub(super) fn start_erase(
state: &AppState,
shard: &mut ShardState,
erase: EraseRequest,
) -> bool {
let EraseRequest {
did,
exclude,
reply,
} = erase;
let _span = tracing::info_span!("job", did = %did).entered();
let mut txn = Txn::new(&state.db);
let started = db::start_erase(&mut txn, &did, exclude).and_then(|()| txn.commit());
if let Err(e) = started {
// the api reads the dropped reply as a failed erase
error!(err = %e, "failed to start an erase");
return false;
}
let queue = Self::start_job(state, shard, &did);
if let Some(Job::Erase(job)) = shard.jobs.get_mut(&did) {
job.replies.push(reply);
}
queue
}
/// runs the next step of `did`'s job, and says whether another is due
pub(super) fn job_step(state: &AppState, shard: &mut ShardState, did: &Did) -> bool {
let _span = tracing::info_span!("job", did = %did).entered();
let Some(job) = shard.jobs.get_mut(did) else {
return false;
};
let (done, steps) = match job {
Job::Wipe(wipe) => {
wipe.steps += 1;
(Self::wipe_step(state, did, wipe), wipe.steps)
}
Job::Erase(erase) => {
erase.steps += 1;
let done = db::erase_step(
&state.db,
did,
&mut erase.cursor,
state.backfill_chunk_records,
&mut erase.report,
);
(done, erase.steps)
}
};
#[cfg(test)]
if let Some(after_job_step) = state.after_job_step.lock().clone() {
after_job_step(steps);
}
if let Ok(false) = done {
return true;
}
debug!(steps, "job done");
let job = shard.jobs.remove(did);
match (done, job) {
(Ok(_), job) => {
if shard.drops_live(did) {
shard.busy.remove(did);
}
if let Some(Job::Erase(erase)) = job {
for reply in erase.replies {
let _ = reply.send(erase.report.clone());
}
}
}
// the marker stays, so the next start or a retried erase runs the job again, and an
// erase's callers read their dropped replies as a failure. the repo stays busy till
// then, since a track or a live commit would land on a half erased repo
(Err(e), _) => error!(err = %e, "failed a job step"),
}
false
}
/// moves the next chunk of a deleted account's heads to history. the last step, which
/// found no more, also takes what's left of the repo's backlinks and its marker
fn wipe_step(state: &AppState, did: &Did, wipe: &mut WipeJob) -> Result<bool> {
let db = &state.db;
let mut txn = Txn::new(db);
wipe.after = db::wipe_heads(
&mut txn,
did,
&wipe.death_rev,
state.only_index_links,
wipe.after.as_deref(),
state.backfill_chunk_records,
)?;
let done = wipe.after.is_none();
if done {
ops::backlink_ops::delete_repo(&mut txn.batch, db, did)?;
txn.batch.remove(&db.crawler, keys::bulk_job_key(did));
}
txn.commit()?;
Ok(done)
}
/// applies everything `did` buffered and takes it off the queue the way the last backfill
/// step does, for a test that stored the records itself. whether that drained it
#[cfg(test)]
pub(crate) fn finish_for_test(
state: &AppState,
shard: &mut ShardState,
did: &Did,
pending_key: &[u8],
) -> Result<bool> {
let mut lifecycle_transitions = Vec::new();
loop {
let drained = Self::drain_some(
0,
state,
shard,
did,
pending_key,
None,
&mut lifecycle_transitions,
)?;
if let Drained::Done(drained) = drained {
return Ok(drained);
}
}
}
/// resumes `did`'s job from its marker and runs it to the end, the way its shard would
#[cfg(test)]
pub(crate) fn run_job(state: &AppState, did: &Did) {
let mut shard = ShardState::default();
if Self::resume_job(state, &mut shard, did) {
while Self::job_step(state, &mut shard, did) {}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn requeued_after_steps(
used: Option<(i64, u32)>,
) -> (tempfile::TempDir, AppState, ShardState, Did) {
use crate::ingest::indexer::refetch::tests as refetch;
let (tmp, state) = refetch::state(Some(1));
let did = refetch::did();
refetch::seed(&state, true, used);
let mut batch = state.db.inner.batch();
state
.db
.indexer
.stage_pending_insert(&mut batch, &did, &keys::pending_key(7));
batch.commit().unwrap();
let mut shard = ShardState::default();
shard.busy.insert(
did.clone(),
Busy::Apply(Applying {
pending_key: keys::pending_key(7).to_vec().into(),
requeue_after: true,
}),
);
(tmp, state, shard, did)
}
fn drain_requeue(state: &AppState, shard: &mut ShardState, did: &Did) -> Result<Drained> {
FirehoseWorker::drain_some(
0,
state,
shard,
did,
&keys::pending_key(7),
None,
&mut Vec::new(),
)
}
#[test]
fn a_chain_break_during_the_steps_costs_the_refetch_it_queues() -> Result<()> {
use crate::ingest::indexer::refetch::tests as refetch;
let (_tmp, state, mut shard, did) = requeued_after_steps(None);
drain_requeue(&state, &mut shard, &did)?;
let metadata = refetch::metadata_of(&state, &did);
assert_eq!(metadata.refetch.map(|allowance| allowance.used), Some(1));
assert_eq!(state.db.indexer.pending_iter().count(), 1);
assert_ne!(metadata.index_id, 7, "it queued a new fetch");
state.db.indexer.assert_pending_dids_match();
Ok(())
}
#[test]
fn a_chain_break_during_the_steps_with_no_refetch_left_parks_the_repo() -> Result<()> {
use crate::ingest::indexer::refetch::tests as refetch;
let now = chrono::Utc::now().timestamp();
let (_tmp, state, mut shard, did) = requeued_after_steps(Some((now, 1)));
drain_requeue(&state, &mut shard, &did)?;
assert_eq!(
state.db.indexer.pending_iter().count(),
0,
"the finished backfill's entry is gone, and nothing is queued"
);
assert_eq!(refetch::row(&state).status, RepoStatus::Throttled);
assert!(refetch::metadata_of(&state, &did).refetch_deferred);
assert!(refetch::entry_of(&state, &did).is_some());
assert!(!shard.is_applying(&did, &keys::pending_key(7)));
state.db.indexer.assert_pending_dids_match();
Ok(())
}
#[tokio::test]
async fn a_backfill_step_leaves_a_running_erase_alone() -> Result<()> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let state = AppState::new(&crate::config::Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})?;
let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").into_diagnostic()?;
// the erase untracked the repo, then something queued it again under 7
let mut metadata = crate::types::RepoMetadata::backfilling(7);
metadata.tracked = false;
let mut batch = state.db.inner.batch();
batch.insert(
&state.db.repo_metadata,
keys::repo_metadata_key(&did),
db::ser_repo_meta(&metadata)?,
);
state
.db
.indexer
.stage_pending_insert(&mut batch, &did, &keys::pending_key(7));
db::stage_bulk_job(&mut batch, &state.db, &did, &BulkJobV1::Erase)?;
batch.commit().into_diagnostic()?;
let mut shard = ShardState::default();
assert!(FirehoseWorker::resume_job(&state, &mut shard, &did));
let (reply, answer) = oneshot::channel();
let chunk = BackfillChunk {
did: did.clone(),
pending_key: keys::pending_key(7).to_vec().into(),
rev: DbTid::new_from_bytes(9_u64.to_be_bytes()),
writes: DiffStep::empty(),
tail: None,
reply,
};
assert!(FirehoseWorker::write_chunk(&state, &mut shard, chunk).is_none());
assert_eq!(answer.await.into_diagnostic()?, ChunkReply::Superseded);
assert!(shard.drops_live(&did));
assert_eq!(db::read_bulk_job(&state.db, &did)?, Some(BulkJobV1::Erase));
Ok(())
}
#[tokio::test]
async fn a_failed_erase_step_keeps_the_repo_busy() -> Result<()> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let state = AppState::new(&crate::config::Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})?;
let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").into_diagnostic()?;
// a head key with no collection/rkey separator, which the erase's first step fails on
let broken = [keys::record_prefix_did(&did), b"broken".to_vec()].concat();
let mut batch = state.db.inner.batch();
state.db.indexer.stage_head(&mut batch, &broken, b"body");
db::stage_bulk_job(&mut batch, &state.db, &did, &BulkJobV1::Erase)?;
batch.commit().into_diagnostic()?;
let mut shard = ShardState::default();
assert!(FirehoseWorker::resume_job(&state, &mut shard, &did));
assert!(!FirehoseWorker::job_step(&state, &mut shard, &did));
assert!(shard.drops_live(&did));
assert!(
!crate::ingest::indexer::repos::repo_op(
&state,
Some(&shard),
super::super::message::RepoOp::Track,
&did,
)?,
"a track went through on a half erased repo"
);
assert_eq!(db::read_bulk_job(&state.db, &did)?, Some(BulkJobV1::Erase));
Ok(())
}
/// a wipe some work on the shard started, like a crawler page's deleted listing, has the
/// repo busy before the shard's next message, so nothing lands on it in between
#[tokio::test]
async fn a_wipe_work_on_the_shard_started_is_busy_before_its_next_message() -> Result<()> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let state = std::sync::Arc::new(AppState::new(&crate::config::Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})?);
let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").into_diagnostic()?;
// no bulk step runs meanwhile, so the wipe can't finish before the check
let permits = state.bulk_permits.available_permits() as u32;
let held = state
.bulk_permits
.try_acquire_many(permits)
.into_diagnostic()?;
let (tx, shard) = FirehoseWorker::spawn_one_shard(state.clone());
let wiping = did.clone();
tx.on_owner(&did, move |state, shard| -> Result<()> {
let mut deleted = crate::types::RepoState::synced();
deleted.active = false;
deleted.status = RepoStatus::Deleted;
let mut txn = Txn::new(&state.db);
ops::start_wipe(&mut txn, &state.db, &wiping, &deleted)?;
txn.put_repo(&wiping, &deleted)?;
txn.commit()?;
shard.mark(wiping);
Ok(())
})
.await??;
let checking = did.clone();
let busy = tx
.on_owner(&did, move |_, shard| shard.drops_live(&checking))
.await?;
drop((held, tx));
tokio::task::spawn_blocking(move || shard.join())
.await
.into_diagnostic()?
.map_err(|_| miette::miette!("indexer shard panicked"))?;
assert!(busy, "the repo wasn't busy right after its wipe started");
Ok(())
}
#[tokio::test]
async fn forgetting_a_backfill_drops_only_its_own_marker() -> Result<()> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let state = AppState::new(&crate::config::Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
})?;
let did = Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").into_diagnostic()?;
let mark = |job: BulkJobV1| {
let mut batch = state.db.inner.batch();
db::stage_bulk_job(&mut batch, &state.db, &did, &job)?;
batch.commit().into_diagnostic()
};
let forget = || {
forget_applying(
&state,
&mut ShardState::default(),
&did,
&keys::pending_key(7),
)
};
let marker = || db::read_bulk_job(&state.db, &did);
// an erase that took over from the backfill keeps its marker
mark(BulkJobV1::Erase)?;
forget();
assert_eq!(marker()?, Some(BulkJobV1::Erase));
// and so does the backfill that runs under the next queue key
let next = BulkJobV1::Apply {
pending_key: keys::pending_key(8).to_vec(),
};
mark(next.clone())?;
forget();
assert_eq!(marker()?, Some(next));
mark(BulkJobV1::Apply {
pending_key: keys::pending_key(7).to_vec(),
})?;
forget();
assert_eq!(marker()?, None);
Ok(())
}
}