use fjall::{Keyspace, OwnedWriteBatch};
#[cfg(feature = "indexer_stream")]
use jacquard_common::types::cid::IpldCid;
use jacquard_common::types::string::Did;
use miette::{IntoDiagnostic, Result, WrapErr};
use smol_str::SmolStr;
use url::Url;
use crate::db::counts::apply_count_delta;
use crate::db::types::DbTid;
use crate::db::{CountDeltas, Db, Txn, keys};
use crate::types::{BulkJob, BulkJobV1, DeleteBodiesReport, DeleteBodyTarget};
use std::collections::BTreeMap;
use std::ops::Bound;
impl Db {
/// resolve one stored event pointer from the body-bearing layouts that can
/// legitimately own it. every candidate is content-verified before use.
#[cfg(feature = "indexer_stream")]
pub(crate) fn resolve_event_record_body(
&self,
record_key: &[u8],
event_rev: &DbTid,
cid: &IpldCid,
) -> Result<Option<fjall::Slice>> {
let matches =
|body: &[u8]| !is_cid_record_value(body) && crate::car::compute_cid(body) == *cid;
if let Some(body) = self.indexer.head(record_key).into_diagnostic()?
&& matches(&body)
{
return Ok(Some(body));
}
for guard in self
.indexer
.history_range(keys::history_range_after(record_key, event_rev))
{
let (key, body) = guard.into_inner().into_diagnostic()?;
if keys::history_hit_for(record_key, &key) && !body.is_empty() && matches(&body) {
return Ok(Some(body));
}
}
// last, because only events from before v10 keep a body here. a database started at v10 or
// later has none
let archive_key = keys::event_body_key(record_key, cid);
let Some(body) = self
.stream
.event_bodies
.get(archive_key)
.into_diagnostic()?
else {
return Ok(None);
};
if !matches(&body) {
miette::bail!("compatibility event body at {cid} is corrupt");
}
Ok(Some(body))
}
}
/// [`ipld_cid_from_record_value`] as a string
#[cfg(feature = "backlinks")]
pub(crate) fn cid_from_record_value(value: &[u8]) -> Result<smol_str::SmolStr> {
use smol_str::ToSmolStr;
Ok(ipld_cid_from_record_value(value)?.to_smolstr())
}
/// cid of a records-keyspace value: recomputed from the body in the current
/// format, or parsed directly in links-only mode and on pre-v10 databases,
/// where the value is still the raw cid bytes.
///
/// the shapes cannot be confused: a dag-cbor record is always a map and can
/// never start with the cidv1-dagcbor-sha256 prefix.
pub(crate) fn ipld_cid_from_record_value(
value: &[u8],
) -> Result<jacquard_common::types::cid::IpldCid> {
if is_cid_record_value(value) {
return cid::Cid::read_bytes(value).into_diagnostic();
}
Ok(crate::car::compute_cid(value))
}
/// whether a body-bearing keyspace value has already been reduced to its raw
/// CID marker. DAG-CBOR records are maps, so this prefix cannot be confused
/// with a valid record body.
pub(crate) fn is_cid_record_value(value: &[u8]) -> bool {
value.len() == 36 && value.starts_with(&[0x01, 0x71, 0x12, 0x20])
}
struct RedactedValue {
cid: jacquard_common::types::cid::IpldCid,
body_bytes: Option<u64>,
}
fn redact_value(
batch: &mut OwnedWriteBatch,
keyspace: &Keyspace,
key: &[u8],
value: &[u8],
) -> Result<Option<RedactedValue>> {
if value.is_empty() {
return Ok(None);
}
if is_cid_record_value(value) {
return Ok(Some(RedactedValue {
cid: ipld_cid_from_record_value(value)?,
body_bytes: None,
}));
}
let cid = crate::car::compute_cid(value);
batch.insert(keyspace, key, cid.to_bytes());
Ok(Some(RedactedValue {
cid,
body_bytes: Some(value.len() as u64),
}))
}
fn stage_redaction(
batch: &mut OwnedWriteBatch,
db: &Db,
record_key: &[u8],
cid: &jacquard_common::types::cid::IpldCid,
) {
batch.insert(
&db.indexer.redactions,
keys::redaction_key(record_key, cid),
[],
);
}
#[cfg(feature = "indexer_stream")]
fn delete_event_body(
batch: &mut OwnedWriteBatch,
db: &Db,
key: &[u8],
value: &[u8],
) -> Result<Option<u64>> {
let Some((record_key, cid)) = keys::split_record_cid_key(key) else {
miette::bail!("invalid compatibility event-body key");
};
let computed = crate::car::compute_cid(value);
if computed != cid {
miette::bail!("compatibility event body does not match its cid");
}
stage_redaction(batch, db, record_key, &cid);
batch.remove(db.stream.event_bodies.raw(), key);
Ok(Some(value.len() as u64))
}
const REDACTION_CHUNK_ENTRIES: usize = 4_096;
const REDACTION_CHUNK_BYTES: u64 = 32 * 1024 * 1024;
/// only the owner shard runs this, which keeps inserts from landing behind the cursor.
/// every committed batch stands alone and a retry after a crash is fine.
fn rewrite_prefix<F>(
db: &Db,
did: &Did,
keyspace: &Keyspace,
prefix: &[u8],
mut rewrite: F,
) -> Result<(u64, u64)>
where
F: FnMut(&mut OwnedWriteBatch, &[u8], &[u8]) -> Result<Option<u64>>,
{
let mut redacted = 0_u64;
let mut bytes = 0_u64;
let mut cursor: Option<Vec<u8>> = None;
loop {
let start = cursor
.as_ref()
.map_or(Bound::Included(prefix.to_vec()), |key| {
Bound::Excluded(key.clone())
});
let mut batch = db.inner.batch();
let mut scanned = 0_usize;
let mut staged_bytes = 0_u64;
let mut last = None;
let mut rewrote = false;
for guard in keyspace.range((start, Bound::<Vec<u8>>::Unbounded)) {
let (key, value) = guard.into_inner().into_diagnostic()?;
if !key.starts_with(prefix) {
break;
}
if scanned > 0
&& (scanned >= REDACTION_CHUNK_ENTRIES || staged_bytes >= REDACTION_CHUNK_BYTES)
{
break;
}
if let Some(len) = rewrite(&mut batch, &key, &value)? {
rewrote = true;
redacted += 1;
bytes += len;
staged_bytes += len;
}
scanned += 1;
last = Some(key.to_vec());
}
let Some(last) = last else { break };
if rewrote {
mark_chunks(&mut batch, db, did);
}
batch.commit().into_diagnostic()?;
cursor = Some(last);
}
Ok((redacted, bytes))
}
#[cfg_attr(not(feature = "indexer_stream"), allow(unused_variables))]
fn touch_chunks(db: &Db, did: &Did) {
#[cfg(feature = "indexer_stream")]
if let Some(chunks) = db.stream.chunks() {
chunks.touch(did.as_str());
}
}
#[cfg_attr(not(feature = "indexer_stream"), allow(unused_variables))]
fn mark_chunks(batch: &mut OwnedWriteBatch, db: &Db, did: &Did) {
#[cfg(feature = "indexer_stream")]
if let Some(chunks) = db.stream.chunks() {
chunks.mark(batch, did.as_str());
}
}
/// erase one record's body bytes without changing its CID identity, counts,
/// or the repository MST root. history tombstones are metadata and remain.
/// runs on the repo's owner shard, so it lands in order with the repo's live
/// commits, and the caller persists once it's back
pub(crate) fn redact_record_bodies(
db: &Db,
did: &Did,
collection: &str,
rkey: &crate::db::types::DbRkey,
target: DeleteBodyTarget,
) -> Result<DeleteBodiesReport> {
let record_key = keys::record_key(did, collection, rkey);
let mut report = DeleteBodiesReport::default();
touch_chunks(db, did);
if target.includes_head() {
let mut batch = db.inner.batch();
if let Some(value) = db.indexer.head(&record_key).into_diagnostic()?
&& let Some(redacted) =
redact_value(&mut batch, db.indexer.heads.raw(), &record_key, &value)?
{
stage_redaction(&mut batch, db, &record_key, &redacted.cid);
mark_chunks(&mut batch, db, did);
if let Some(len) = redacted.body_bytes {
report.heads_redacted = 1;
report.body_bytes_redacted += len;
}
crate::ops::backlink_ops::delete_record(
&mut batch,
db,
did,
collection,
&rkey.to_smolstr(),
)?;
}
batch.commit().into_diagnostic()?;
}
if target.includes_history() {
let mut prefix = record_key.clone();
prefix.push(keys::indexer::REC_SEP);
let (count, bytes) = rewrite_prefix(
db,
did,
db.indexer.history.raw(),
&prefix,
|batch, key, value| {
let Some(redacted) = redact_value(batch, db.indexer.history.raw(), key, value)?
else {
return Ok(None);
};
stage_redaction(batch, db, &record_key, &redacted.cid);
Ok(redacted.body_bytes)
},
)?;
report.history_redacted = count;
report.body_bytes_redacted += bytes;
#[cfg(feature = "indexer_stream")]
{
let (count, bytes) = rewrite_prefix(
db,
did,
db.stream.event_bodies.raw(),
&prefix,
|batch, key, value| delete_event_body(batch, db, key, value),
)?;
report.event_bodies_redacted = count;
report.body_bytes_redacted += bytes;
}
}
touch_chunks(db, did);
Ok(report)
}
/// forget everything stored for a repository in one go, the way the steps of an erase do it a
/// chunk at a time. the caller untracks the repo first, see [`start_erase`]
#[cfg(test)]
pub(crate) fn erase_repo(db: &Db, did: &Did) -> Result<DeleteBodiesReport> {
let mut cursor = EraseCursor::default();
let mut report = DeleteBodiesReport::default();
while !erase_step(db, did, &mut cursor, REDACTION_CHUNK_ENTRIES, &mut report)? {}
db.persist()?;
Ok(report)
}
/// the first txn of an erase: excludes the repo if asked, untracks it and takes it off every
/// queue, so nothing ingests it again while the steps go, and leaves the marker a restart
/// resumes the erase from
pub(crate) fn start_erase(txn: &mut Txn<'_>, did: &Did, exclude: bool) -> Result<()> {
let db = txn.db;
if exclude {
txn.batch.insert(
&db.filter,
crate::db::filter::exclude_key(did.as_str())?,
[],
);
}
let metadata_key = keys::repo_metadata_key(did);
if let Some(bytes) = db.repo_metadata.get(&metadata_key).into_diagnostic()? {
let mut metadata = crate::db::deser_repo_meta(&bytes)?;
db.indexer
.stage_pending_remove(&mut txn.batch, did, &keys::pending_key(metadata.index_id));
metadata.tracked = false;
metadata.refetch_deferred = false;
txn.batch.insert(
&db.repo_metadata,
metadata_key,
crate::db::ser_repo_meta(&metadata)?,
);
}
let repo_key = keys::repo_key(did);
txn.batch.remove(&db.indexer.resync, &repo_key);
txn.batch.remove(&db.crawler, keys::crawler_retry_key(did));
if db.repos.contains_key(&repo_key).into_diagnostic()? {
txn.transition_lifecycle(did, crate::types::GaugeState::Synced)?;
}
stage_bulk_job(&mut txn.batch, db, did, &BulkJobV1::Erase)
}
/// where an erase has got to, so each of its steps picks up after the last
#[derive(Debug, Default)]
pub(crate) struct EraseCursor {
phase: ErasePhase,
after: Option<Vec<u8>>,
}
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
enum ErasePhase {
#[default]
Records,
History,
#[cfg(feature = "indexer_stream")]
EventBodies,
Redactions,
Buffer,
Repo,
}
impl ErasePhase {
/// erasing these takes bodies a replay sends. their chunk gets marked dirty.
fn takes_bodies(self) -> bool {
match self {
Self::Records | Self::History => true,
#[cfg(feature = "indexer_stream")]
Self::EventBodies => true,
Self::Redactions | Self::Buffer | Self::Repo => false,
}
}
fn next(self) -> Self {
match self {
Self::Records => Self::History,
#[cfg(feature = "indexer_stream")]
Self::History => Self::EventBodies,
#[cfg(feature = "indexer_stream")]
Self::EventBodies => Self::Redactions,
#[cfg(not(feature = "indexer_stream"))]
Self::History => Self::Redactions,
Self::Redactions => Self::Buffer,
Self::Buffer | Self::Repo => Self::Repo,
}
}
}
/// one step of forgetting everything stored for a repository: record bodies along with their
/// cids, history, buffered commits, backlinks, counts and the repo's own state, so tracking it
/// again backfills it from nothing instead of skipping records it still has cids for. a step
/// erases at most `limit` entries of one kind, and the last one, which says `true`, removes the
/// repo itself and its marker. stream events only point at bodies, so they stay and replaying
/// them resolves no body
pub(crate) fn erase_step(
db: &Db,
did: &Did,
cursor: &mut EraseCursor,
limit: usize,
report: &mut DeleteBodiesReport,
) -> Result<bool> {
let limit = limit.min(REDACTION_CHUNK_ENTRIES);
let prefix = keys::record_prefix_did(did);
let after = cursor.after.as_deref();
let bodies = cursor.phase.takes_bodies().then_some(did);
if bodies.is_some() {
touch_chunks(db, did);
}
let erased = match cursor.phase {
ErasePhase::Records => {
let erased = erase_chunk(
db,
db.indexer.heads.raw(),
&prefix,
after,
limit,
bodies,
|batch, counts, key, value| {
let (collection, rkey) = keys::split_record_suffix(&key[prefix.len()..])?;
crate::ops::backlink_ops::delete_record(
batch,
db,
did,
collection,
&rkey.to_smolstr(),
)?;
counts.add_records(-1);
Ok(body_len(value))
},
)?;
report.heads_redacted += erased.count;
report.body_bytes_redacted += erased.bytes;
erased
}
ErasePhase::History => {
let erased = erase_chunk(
db,
db.indexer.history.raw(),
&prefix,
after,
limit,
bodies,
|_, _, _, value| Ok(body_len(value)),
)?;
report.history_redacted += erased.count;
report.body_bytes_redacted += erased.bytes;
erased
}
#[cfg(feature = "indexer_stream")]
ErasePhase::EventBodies => {
let erased = erase_chunk(
db,
db.stream.event_bodies.raw(),
&prefix,
after,
limit,
bodies,
|_, _, _, value| Ok(value.len() as u64),
)?;
report.event_bodies_redacted += erased.count;
report.body_bytes_redacted += erased.bytes;
erased
}
// markers would keep the same versions bodiless once the repo is indexed again
ErasePhase::Redactions => erase_chunk(
db,
db.indexer.redactions.raw(),
&prefix,
after,
limit,
bodies,
|_, _, _, _| Ok(0),
)?,
ErasePhase::Buffer => {
let erased = erase_chunk(
db,
db.indexer.resync_buffer.raw(),
&keys::resync_buffer_prefix(did),
after,
limit,
bodies,
|_, _, _, value| Ok(value.len() as u64),
)?;
report.buffered_commits_deleted += erased.count;
report.buffered_commit_bytes_deleted += erased.bytes;
erased
}
ErasePhase::Repo => {
let mut txn = Txn::new(db);
erase_repo_row(&mut txn, did)?;
txn.commit()?;
return Ok(true);
}
};
if bodies.is_some() {
touch_chunks(db, did);
}
match erased.more_after {
Some(last) => cursor.after = Some(last),
None => {
cursor.phase = cursor.phase.next();
cursor.after = None;
}
}
Ok(false)
}
/// the last step of an erase, once everything hanging off the repo is gone
fn erase_repo_row(txn: &mut Txn<'_>, did: &Did) -> Result<()> {
let db = txn.db;
// the gauge is read from the pending and resync entries, so it goes first
txn.forget_lifecycle(did)?;
let metadata_key = keys::repo_metadata_key(did);
if let Some(bytes) = db.repo_metadata.get(&metadata_key).into_diagnostic()? {
let metadata = crate::db::deser_repo_meta(&bytes)?;
db.indexer
.stage_pending_remove(&mut txn.batch, did, &keys::pending_key(metadata.index_id));
txn.batch.remove(&db.repo_metadata, metadata_key);
}
let repo_key = keys::repo_key(did);
if db.repos.contains_key(&repo_key).into_diagnostic()? {
txn.remove_repo(did)?;
txn.counts.add_repos(-1);
}
txn.batch.remove(&db.indexer.resync, &repo_key);
txn.batch.remove(&db.crawler, keys::crawler_retry_key(did));
txn.batch.remove(&db.crawler, keys::bulk_job_key(did));
for guard in db.counts.prefix(keys::did_collection_prefix(did)) {
txn.batch.remove(&db.counts, guard.key().into_diagnostic()?);
}
Ok(())
}
/// bytes of record content in a stored value, which a redacted cid marker has none of
fn body_len(value: &[u8]) -> u64 {
if is_cid_record_value(value) {
0
} else {
value.len() as u64
}
}
/// what one [`erase_chunk`] took
struct Erased {
count: u64,
/// the bytes `stage` reported
bytes: u64,
/// the last key it deleted, when it stopped at a cap and the next step goes on from there
more_after: Option<Vec<u8>>,
}
/// everything staged, counts and the chunk mark included, commits with the delete.
fn erase_chunk<F>(
db: &Db,
keyspace: &Keyspace,
prefix: &[u8],
after: Option<&[u8]>,
limit: usize,
takes_bodies_of: Option<&Did>,
mut stage: F,
) -> Result<Erased>
where
F: FnMut(&mut OwnedWriteBatch, &mut CountDeltas, &[u8], &[u8]) -> Result<u64>,
{
// starting after the last key the step before erased, because starting over at `prefix`
// walks every tombstone the earlier steps left until a compaction drops them
let start = after.map_or_else(
|| Bound::Included(prefix.to_vec()),
|after| Bound::Excluded(after.to_vec()),
);
let mut batch = db.inner.batch();
let mut counts = CountDeltas::default();
let mut staged = 0_usize;
let mut staged_bytes = 0_u64;
let mut bytes = 0_u64;
let mut last = None;
let mut more = false;
for guard in keyspace.range((start, Bound::Unbounded)) {
// either cap, because a step cut short by bytes leaves the rest for the next one too
if staged >= limit || staged_bytes >= REDACTION_CHUNK_BYTES {
more = true;
break;
}
let (key, value) = guard.into_inner().into_diagnostic()?;
if !key.starts_with(prefix) {
break;
}
bytes += stage(&mut batch, &mut counts, &key, &value)?;
batch.remove(keyspace, key.clone());
last = Some(key.to_vec());
staged += 1;
staged_bytes += value.len() as u64;
}
if last.is_none() {
return Ok(Erased {
count: 0,
bytes: 0,
more_after: None,
});
}
if let Some(did) = takes_bodies_of {
mark_chunks(&mut batch, db, did);
}
let reservation = db.stage_count_deltas(&mut batch, &counts);
batch.commit().into_diagnostic()?;
db.apply_count_deltas(&counts);
drop(reservation);
Ok(Erased {
count: staged as u64,
bytes,
more_after: last.filter(|_| more),
})
}
/// the rev an account deletion files a repo's heads under in history, after every rev seen
/// for it, including ones from a clock running ahead
pub(crate) fn death_rev(root_rev: Option<&DbTid>) -> DbTid {
let local_now = DbTid::from(&jacquard_common::types::string::Tid::now_0());
match root_rev {
Some(r) => local_now.max(r.add_micros(1)),
None => local_now,
}
}
/// moves every one of the repo's heads to history at once, see [`wipe_heads`]
#[cfg(test)]
pub(crate) fn delete_repo_records(
txn: &mut Txn<'_>,
did: &Did,
root_rev: Option<&DbTid>,
links_only: bool,
) -> Result<()> {
wipe_heads(txn, did, &death_rev(root_rev), links_only, None, usize::MAX)?;
Ok(())
}
/// moves at most `limit` of the repo's heads that sort after `after` into history at
/// `death_rev`, so replays of earlier events can still resolve their bodies, and takes their
/// backlinks and counts with them. a links-only database has no bodies to keep, so there it
/// only deletes the heads. returns the last key it moved when it stopped at `limit`, and `None`
/// once it got to the end of the repo's heads
pub(crate) fn wipe_heads(
txn: &mut Txn<'_>,
did: &Did,
death_rev: &DbTid,
links_only: bool,
after: Option<&[u8]>,
limit: usize,
) -> Result<Option<Vec<u8>>> {
let db = txn.db;
let mut moved_heads: i64 = 0;
let mut last = None;
let mut stopped = false;
let prefix = keys::record_prefix_did(did);
let start = after.map_or_else(
|| Bound::Included(prefix.clone()),
|after| Bound::Excluded(after.to_vec()),
);
for guard in db.indexer.heads.range((start, Bound::Unbounded)) {
if moved_heads as usize >= limit {
stopped = true;
break;
}
let (key, value) = guard.into_inner().into_diagnostic()?;
if !key.starts_with(&prefix) {
break;
}
moved_heads += 1;
let suffix = &key[prefix.len()..];
if let Ok((collection, rkey)) = keys::split_record_suffix(suffix) {
txn.add_record_count(did, collection, -1);
crate::ops::backlink_ops::delete_record(
&mut txn.batch,
db,
did,
collection,
&rkey.to_smolstr(),
)?;
}
if !links_only && !value.is_empty() {
txn.batch.insert(
&db.indexer.history,
keys::history_key(&key, death_rev),
value.as_ref(),
);
}
txn.batch.remove(&db.indexer.heads, key.clone());
last = Some(key.to_vec());
}
// deferred to the txn's commit so the in-memory projection never leads the durable batch
txn.counts.add_records(-moved_heads);
Ok(last.filter(|_| stopped))
}
/// the bulk job the repo's owner shard is running for it, if any
pub(crate) fn read_bulk_job(db: &Db, did: &Did) -> Result<Option<BulkJobV1>> {
let Some(bytes) = db.crawler.get(keys::bulk_job_key(did)).into_diagnostic()? else {
return Ok(None);
};
let BulkJob::V1(job) = rmp_serde::from_slice(&bytes).into_diagnostic()?;
Ok(Some(job))
}
pub(crate) fn stage_bulk_job(
batch: &mut OwnedWriteBatch,
db: &Db,
did: &Did,
job: &BulkJobV1,
) -> Result<()> {
let bytes = rmp_serde::to_vec(&BulkJob::V1(job.clone())).into_diagnostic()?;
batch.insert(&db.crawler, keys::bulk_job_key(did), bytes);
Ok(())
}
pub fn set_record_count(
batch: &mut OwnedWriteBatch,
db: &Db,
did: &Did,
collection: &str,
count: u64,
) {
let key = keys::count_collection_key(did, collection);
batch.insert(&db.counts, key, count.to_be_bytes());
}
/// what one transaction does to its repos' collection counts, staged into its batch when it
/// commits, so each count is read and written once however many record writes moved it
#[derive(Default)]
pub(crate) struct RecordCounts {
by_repo: BTreeMap<Did, BTreeMap<SmolStr, RecordCount>>,
}
#[derive(Clone, Copy)]
enum RecordCount {
/// moves the stored count
Delta(i64),
Set(u64),
Remove,
}
impl RecordCounts {
pub(crate) fn add(&mut self, did: &Did, collection: &str, delta: i64) {
let entry = self.entry(did, collection);
*entry = match *entry {
RecordCount::Delta(moved) => RecordCount::Delta(moved + delta),
RecordCount::Set(count) => RecordCount::Set(apply_count_delta(count, delta)),
RecordCount::Remove => RecordCount::Set(apply_count_delta(0, delta)),
};
}
/// replaces the repo's counts of the collections `matches_collection` takes, stored or moved
/// in this transaction, with `counts`. writes after this move the replaced counts
pub(crate) fn replace<'a>(
&mut self,
db: &Db,
did: &Did,
matches_collection: impl Fn(&str) -> bool,
counts: impl IntoIterator<Item = (&'a str, u64)>,
) -> Result<()> {
let repo = self.by_repo.entry(did.clone()).or_default();
repo.iter_mut()
.filter(|(collection, _)| matches_collection(collection))
.for_each(|(_, count)| *count = RecordCount::Remove);
let prefix = keys::did_collection_prefix(did);
for guard in db.counts.prefix(&prefix) {
let key = guard.key().into_diagnostic()?;
let collection = key
.strip_prefix(prefix.as_slice())
.ok_or_else(|| miette::miette!("invalid collection count key: {key:?}"))
.and_then(|raw| std::str::from_utf8(raw).into_diagnostic())?;
if matches_collection(collection) {
repo.insert(SmolStr::new(collection), RecordCount::Remove);
}
}
for (collection, count) in counts {
repo.insert(SmolStr::new(collection), RecordCount::Set(count));
}
Ok(())
}
pub(crate) fn stage(self, batch: &mut OwnedWriteBatch, db: &Db) -> Result<()> {
for (did, repo) in &self.by_repo {
for (collection, count) in repo {
match *count {
RecordCount::Delta(0) => {}
RecordCount::Delta(delta) => {
update_record_count(batch, db, did, collection, delta)?
}
RecordCount::Set(count) => set_record_count(batch, db, did, collection, count),
RecordCount::Remove => {
batch.remove(&db.counts, keys::count_collection_key(did, collection))
}
}
}
}
Ok(())
}
fn entry(&mut self, did: &Did, collection: &str) -> &mut RecordCount {
self.by_repo
.entry(did.clone())
.or_default()
.entry(SmolStr::new(collection))
.or_insert(RecordCount::Delta(0))
}
}
pub fn update_record_count(
batch: &mut OwnedWriteBatch,
db: &Db,
did: &Did,
collection: &str,
delta: i64,
) -> Result<()> {
let key = keys::count_collection_key(did, collection);
let count = db
.counts
.get(&key)
.into_diagnostic()?
.map(|v| -> Result<_> {
Ok(u64::from_be_bytes(
v.as_ref()
.try_into()
.into_diagnostic()
.wrap_err("expected to be count (8 bytes)")?,
))
})
.transpose()?
.unwrap_or(0);
let new_count = if delta >= 0 {
count.saturating_add(delta as u64)
} else {
count.saturating_sub(delta.unsigned_abs())
};
batch.insert(&db.counts, key, new_count.to_be_bytes());
Ok(())
}
pub fn get_record_count(db: &Db, did: &Did, collection: &str) -> Result<u64> {
let key = keys::count_collection_key(did, collection);
let count = db
.counts
.get(&key)
.into_diagnostic()?
.map(|v| -> Result<_> {
Ok(u64::from_be_bytes(
v.as_ref()
.try_into()
.into_diagnostic()
.wrap_err("expected to be count (8 bytes)")?,
))
})
.transpose()?;
Ok(count.unwrap_or(0))
}
pub fn load_persisted_crawler_sources(
db: &crate::db::Db,
) -> Result<Vec<crate::config::CrawlerSource>> {
use crate::db::keys::CRAWLER_SOURCE_PREFIX;
let mut sources = Vec::new();
for entry in db.crawler.prefix(CRAWLER_SOURCE_PREFIX) {
let (key, val) = entry.into_inner().into_diagnostic()?;
let url_bytes = &key[CRAWLER_SOURCE_PREFIX.len()..];
let url_str = std::str::from_utf8(url_bytes).into_diagnostic()?;
let url = Url::parse(url_str).into_diagnostic()?;
let mode: crate::config::CrawlerMode = rmp_serde::from_slice(&val).into_diagnostic()?;
sources.push(crate::config::CrawlerSource { url, mode });
}
Ok(sources)
}
#[cfg(test)]
mod tests {
use super::*;
const DID: &str = "did:plc:yk4q3id7id6p5z3bypvshc64";
const OTHER_DID: &str = "did:plc:ewvi7nxzyoun6zhxrhs64oiz";
const COLLECTION: &str = "app.bsky.feed.post";
const RKEY: &str = "3kzbif5moe22m";
fn test_db() -> Result<(tempfile::TempDir, Db)> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let cfg = crate::config::Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
};
let db = Db::open(&cfg)?;
Ok((tmp, db))
}
fn body(text: &str) -> Vec<u8> {
serde_ipld_dagcbor::to_vec(&serde_json::json!({
"$type": COLLECTION,
"text": text,
}))
.unwrap()
}
fn did(value: &'static str) -> Did {
Did::new_static(value).unwrap()
}
fn record_key(did: &Did) -> Vec<u8> {
keys::record_key(did, COLLECTION, &crate::db::types::DbRkey::new(RKEY))
}
fn stage_body_copies(db: &Db, did: &Did) -> Result<(Vec<u8>, Vec<u8>, Vec<u8>)> {
let head = body("head");
let old_a = body("old a");
let old_b = body("old b");
#[cfg(feature = "indexer_stream")]
let archive = body("archive");
#[cfg(feature = "indexer_stream")]
let archive_cid = crate::car::compute_cid(&archive);
let key = record_key(did);
let mut batch = db.inner.batch();
batch.insert(&db.indexer.heads, &key, &head);
batch.insert(
&db.indexer.history,
keys::history_key(&key, &DbTid::new_from_bytes(1_u64.to_be_bytes())),
&old_a,
);
batch.insert(
&db.indexer.history,
keys::history_key(&key, &DbTid::new_from_bytes(2_u64.to_be_bytes())),
&old_b,
);
batch.insert(
&db.indexer.history,
keys::history_key(&key, &DbTid::new_from_bytes(3_u64.to_be_bytes())),
[],
);
#[cfg(feature = "indexer_stream")]
batch.insert(
&db.stream.event_bodies,
keys::event_body_key(&key, &archive_cid),
&archive,
);
batch.commit().into_diagnostic()?;
Ok((head, old_a, old_b))
}
#[test]
fn record_head_redaction_is_idempotent_and_keeps_history() -> Result<()> {
let (_tmp, db) = test_db()?;
let did = did(DID);
let (head, old_a, _) = stage_body_copies(&db, &did)?;
let report = redact_record_bodies(
&db,
&did,
COLLECTION,
&crate::db::types::DbRkey::new(RKEY),
DeleteBodyTarget::Head,
)?;
assert_eq!(report.heads_redacted, 1);
assert_eq!(report.history_redacted, 0);
assert_eq!(report.body_bytes_redacted, head.len() as u64);
let key = record_key(&did);
let marker = db.indexer.head(&key).into_diagnostic()?.unwrap();
assert!(is_cid_record_value(&marker));
assert_eq!(
db.indexer
.history
.get(keys::history_key(
&key,
&DbTid::new_from_bytes(1_u64.to_be_bytes())
))
.into_diagnostic()?
.as_deref(),
Some(old_a.as_slice())
);
let again = redact_record_bodies(
&db,
&did,
COLLECTION,
&crate::db::types::DbRkey::new(RKEY),
DeleteBodyTarget::Head,
)?;
assert_eq!(again, DeleteBodiesReport::default());
Ok(())
}
#[test]
fn record_history_redaction_keeps_tombstones_and_removes_archive() -> Result<()> {
let (_tmp, db) = test_db()?;
let did = did(DID);
let (head, old_a, old_b) = stage_body_copies(&db, &did)?;
let report = redact_record_bodies(
&db,
&did,
COLLECTION,
&crate::db::types::DbRkey::new(RKEY),
DeleteBodyTarget::History,
)?;
assert_eq!(report.heads_redacted, 0);
assert_eq!(report.history_redacted, 2);
#[cfg(feature = "indexer_stream")]
assert_eq!(report.event_bodies_redacted, 1);
assert_eq!(
report.body_bytes_redacted,
(old_a.len()
+ old_b.len()
+ if cfg!(feature = "indexer_stream") {
body("archive").len()
} else {
0
}) as u64
);
let key = record_key(&did);
assert_eq!(
db.indexer.head(&key).into_diagnostic()?.as_deref(),
Some(head.as_slice())
);
assert!(
db.indexer
.history
.get(keys::history_key(
&key,
&DbTid::new_from_bytes(1_u64.to_be_bytes())
))
.into_diagnostic()?
.is_some_and(|value| is_cid_record_value(&value))
);
assert!(
db.indexer
.history
.get(keys::history_key(
&key,
&DbTid::new_from_bytes(3_u64.to_be_bytes())
))
.into_diagnostic()?
.is_some_and(|value| value.is_empty())
);
#[cfg(feature = "indexer_stream")]
assert!(
db.stream
.event_bodies
.prefix([key.as_slice(), &[keys::indexer::REC_SEP]].concat())
.next()
.is_none()
);
Ok(())
}
#[test]
fn record_redaction_does_not_discard_repo_resync_buffer() -> Result<()> {
let (_tmp, db) = test_db()?;
let did = did(DID);
stage_body_copies(&db, &did)?;
let buffer_key = keys::resync_buffer_key(&did, DbTid::new_from_bytes(4_u64.to_be_bytes()));
db.indexer
.resync_buffer
.insert(&buffer_key, b"unrelated buffered commit")
.into_diagnostic()?;
redact_record_bodies(
&db,
&did,
COLLECTION,
&crate::db::types::DbRkey::new(RKEY),
DeleteBodyTarget::All,
)?;
assert_eq!(
db.indexer
.resync_buffer
.get(buffer_key)
.into_diagnostic()?
.as_deref(),
Some(b"unrelated buffered commit".as_slice())
);
Ok(())
}
#[test]
fn repo_erase_is_did_prefix_isolated() -> Result<()> {
let (_tmp, db) = test_db()?;
let target = did(DID);
let other = did(OTHER_DID);
stage_body_copies(&db, &target)?;
let (other_head, other_old, _) = stage_body_copies(&db, &other)?;
// leaves a redaction marker behind, which the erase has to take too
redact_record_bodies(
&db,
&target,
COLLECTION,
&crate::db::types::DbRkey::new(RKEY),
DeleteBodyTarget::All,
)?;
let report = erase_repo(&db, &target)?;
assert_eq!(report.heads_redacted, 1);
assert_eq!(report.history_redacted, 3);
let prefix = keys::record_prefix_did(&target);
assert!(
db.indexer
.head(record_key(&target))
.into_diagnostic()?
.is_none()
);
assert!(db.indexer.history.prefix(&prefix).next().is_none());
assert!(db.indexer.redactions.prefix(&prefix).next().is_none());
let other_key = record_key(&other);
assert_eq!(
db.indexer.head(&other_key).into_diagnostic()?.as_deref(),
Some(other_head.as_slice())
);
assert_eq!(
db.indexer
.history
.get(keys::history_key(
&other_key,
&DbTid::new_from_bytes(1_u64.to_be_bytes())
))
.into_diagnostic()?
.as_deref(),
Some(other_old.as_slice())
);
Ok(())
}
#[test]
fn redaction_chunks_large_history_prefixes() -> Result<()> {
let (_tmp, db) = test_db()?;
let did = did(DID);
let key = record_key(&did);
let body = body("version");
let mut batch = db.inner.batch();
for i in 0..(REDACTION_CHUNK_ENTRIES as u64 + 1) {
batch.insert(
&db.indexer.history,
keys::history_key(&key, &DbTid::new_from_bytes(i.to_be_bytes())),
&body,
);
}
batch.commit().into_diagnostic()?;
let report = redact_record_bodies(
&db,
&did,
COLLECTION,
&crate::db::types::DbRkey::new(RKEY),
DeleteBodyTarget::History,
)?;
assert_eq!(report.history_redacted, REDACTION_CHUNK_ENTRIES as u64 + 1);
assert!(
db.indexer
.history
.prefix([key.as_slice(), &[keys::indexer::REC_SEP]].concat())
.all(|item| item.value().is_ok_and(|value| is_cid_record_value(&value)))
);
Ok(())
}
#[test]
fn erase_goes_past_one_chunk_but_not_past_its_did() -> Result<()> {
let (_tmp, db) = test_db()?;
// the did whose keys come right after `did`'s, where a scan that missed the end of
// the prefix would carry on
let next = did("did:plc:zzzzzzzzzzzzzzzzzzzzzzzz");
let did = did(DID);
assert!(keys::record_prefix_did(&next) > keys::record_prefix_did(&did));
let key = record_key(&did);
let next_version = keys::history_key(
&record_key(&next),
&DbTid::new_from_bytes(0_u64.to_be_bytes()),
);
let mut batch = db.inner.batch();
for i in 0..(REDACTION_CHUNK_ENTRIES as u64 + 1) {
batch.insert(
&db.indexer.history,
keys::history_key(&key, &DbTid::new_from_bytes(i.to_be_bytes())),
body("version"),
);
}
batch.insert(&db.indexer.history, &next_version, body("version"));
batch.commit().into_diagnostic()?;
let report = erase_repo(&db, &did)?;
assert_eq!(report.history_redacted, REDACTION_CHUNK_ENTRIES as u64 + 1);
assert!(db.indexer.history.prefix(&key).next().is_none());
assert!(
db.indexer
.history
.get(&next_version)
.into_diagnostic()?
.is_some()
);
Ok(())
}
#[test]
fn erase_goes_on_after_a_step_the_byte_cap_cut_short() -> Result<()> {
let (_tmp, db) = test_db()?;
let did = did(DID);
let key = record_key(&did);
// past the byte cap well before the entry cap
let versions = REDACTION_CHUNK_BYTES / (1 << 20) + 8;
let big = vec![0xa1_u8; 1 << 20];
let mut batch = db.inner.batch();
for i in 0..versions {
batch.insert(
&db.indexer.history,
keys::history_key(&key, &DbTid::new_from_bytes(i.to_be_bytes())),
&big,
);
}
batch.commit().into_diagnostic()?;
let report = erase_repo(&db, &did)?;
assert_eq!(report.history_redacted, versions);
assert!(db.indexer.history.prefix(&key).next().is_none());
Ok(())
}
#[cfg(feature = "backlinks")]
#[test]
fn only_head_redaction_removes_outgoing_backlinks() -> Result<()> {
let (_tmp, db) = test_db()?;
let did = did(DID);
let rkey = crate::db::types::DbRkey::new(RKEY);
let value = serde_json::json!({
"$type": "app.bsky.feed.like",
"subject": {
"uri": "at://did:plc:target/app.bsky.feed.post/3jz",
"cid": "bafyreig2toyab2fv7kqv3wcldlgl63p2oqxv54htv23s3kbsnshs7j2yli"
},
"createdAt": "2026-08-04T00:00:00Z"
});
let body = serde_ipld_dagcbor::to_vec(&value).into_diagnostic()?;
let data = serde_ipld_dagcbor::from_slice(&body).into_diagnostic()?;
let mut batch = db.inner.batch();
db.indexer.stage_head(
&mut batch,
keys::record_key(&did, "app.bsky.feed.like", &rkey),
&body,
);
crate::backlinks::store::index_record(
&mut batch,
&db.backlinks,
did.as_str(),
"app.bsky.feed.like",
RKEY,
&data,
)?;
batch.commit().into_diagnostic()?;
let forward =
crate::backlinks::store::forward_key(did.as_str(), "app.bsky.feed.like", RKEY);
assert!(db.backlinks.contains_key(&forward).into_diagnostic()?);
redact_record_bodies(
&db,
&did,
"app.bsky.feed.like",
&rkey,
DeleteBodyTarget::History,
)?;
assert!(db.backlinks.contains_key(&forward).into_diagnostic()?);
redact_record_bodies(
&db,
&did,
"app.bsky.feed.like",
&rkey,
DeleteBodyTarget::Head,
)?;
assert!(!db.backlinks.contains_key(&forward).into_diagnostic()?);
Ok(())
}
#[test]
fn replace_record_counts_clears_stale_collections() -> Result<()> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let cfg = crate::config::Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
};
let db = Db::open(&cfg)?;
let did = Did::new_static("did:plc:yk4q3id7id6p5z3bypvshc64").into_diagnostic()?;
let mut batch = db.inner.batch();
set_record_count(&mut batch, &db, &did, "app.bsky.feed.post", 3);
set_record_count(&mut batch, &db, &did, "app.bsky.feed.like", 2);
batch.commit().into_diagnostic()?;
let mut txn = Txn::new(&db);
txn.replace_record_counts(
&did,
|_| true,
[("app.bsky.feed.like", 7), ("app.bsky.actor.profile", 1)],
)?;
txn.commit()?;
assert_eq!(get_record_count(&db, &did, "app.bsky.feed.post")?, 0);
assert_eq!(get_record_count(&db, &did, "app.bsky.feed.like")?, 7);
assert_eq!(get_record_count(&db, &did, "app.bsky.actor.profile")?, 1);
assert_eq!(
db.counts.prefix(keys::did_collection_prefix(&did)).count(),
2
);
Ok(())
}
#[test]
fn replace_record_counts_matching_keeps_unmatched_collections() -> Result<()> {
let tmp = tempfile::tempdir().into_diagnostic()?;
let cfg = crate::config::Config {
database_path: tmp.path().to_path_buf(),
..Default::default()
};
let db = Db::open(&cfg)?;
let did = Did::new_static("did:plc:yk4q3id7id6p5z3bypvshc64").into_diagnostic()?;
let mut batch = db.inner.batch();
set_record_count(&mut batch, &db, &did, "sh.tangled.repo", 3);
set_record_count(&mut batch, &db, &did, "app.bsky.feed.post", 2);
batch.commit().into_diagnostic()?;
let mut txn = Txn::new(&db);
txn.replace_record_counts(
&did,
|collection| collection.starts_with("sh.tangled."),
[("sh.tangled.repo.issue", 7)],
)?;
txn.commit()?;
assert_eq!(get_record_count(&db, &did, "sh.tangled.repo")?, 0);
assert_eq!(get_record_count(&db, &did, "sh.tangled.repo.issue")?, 7);
assert_eq!(get_record_count(&db, &did, "app.bsky.feed.post")?, 2);
assert_eq!(
db.counts.prefix(keys::did_collection_prefix(&did)).count(),
2
);
Ok(())
}
#[test]
fn record_count_codec_updates_and_rejects_invalid_bytes() -> Result<()> {
let (_tmp, db) = test_db()?;
let did = did(DID);
let key = keys::count_collection_key(&did, COLLECTION);
let mut batch = db.inner.batch();
set_record_count(&mut batch, &db, &did, COLLECTION, 7);
batch.commit().into_diagnostic()?;
assert_eq!(
db.counts.get(&key).into_diagnostic()?.unwrap().as_ref(),
7_u64.to_be_bytes()
);
assert_eq!(get_record_count(&db, &did, COLLECTION)?, 7);
let mut batch = db.inner.batch();
update_record_count(&mut batch, &db, &did, COLLECTION, -9)?;
batch.commit().into_diagnostic()?;
assert_eq!(get_record_count(&db, &did, COLLECTION)?, 0);
db.counts.insert(&key, [0_u8; 7]).into_diagnostic()?;
assert!(get_record_count(&db, &did, COLLECTION).is_err());
let mut batch = db.inner.batch();
assert!(update_record_count(&mut batch, &db, &did, COLLECTION, 1).is_err());
Ok(())
}
#[test]
fn persisted_crawler_sources_decode_mode_and_url() -> Result<()> {
let (_tmp, db) = test_db()?;
let fixtures = [
(
Url::parse("https://relay.example/x").into_diagnostic()?,
crate::config::CrawlerMode::RelayFirst,
),
(
Url::parse("https://collection.example/").into_diagnostic()?,
crate::config::CrawlerMode::ByCollection,
),
];
for (url, mode) in &fixtures {
db.crawler
.insert(
keys::crawler_source_key(url.as_str()),
rmp_serde::to_vec(mode).into_diagnostic()?,
)
.into_diagnostic()?;
}
let mut decoded = load_persisted_crawler_sources(&db)?;
decoded.sort_by(|a, b| a.url.as_str().cmp(b.url.as_str()));
let mut expected = fixtures
.into_iter()
.map(|(url, mode)| crate::config::CrawlerSource { url, mode })
.collect::<Vec<_>>();
expected.sort_by(|a, b| a.url.as_str().cmp(b.url.as_str()));
assert_eq!(decoded.len(), expected.len());
for (decoded, expected) in decoded.iter().zip(expected) {
assert_eq!(decoded.url, expected.url);
assert_eq!(decoded.mode, expected.mode);
}
db.crawler
.insert(keys::crawler_source_key("not a url"), [0xc0])
.into_diagnostic()?;
assert!(load_persisted_crawler_sources(&db).is_err());
Ok(())
}
}