haematite-cli 0.21.3

//! `changes.list`: the documents added, changed or removed between two kept
//! versions of one branch, in (collection, id) order, page by page.
//!
//! The changes come from the engine's diff of each shard's two roots over the
//! revision entries (`m\0`) alone, so no body is read unless values are asked,
//! and a page costs in proportion to the changes, never to the store. Each
//! shard's diff resumes after the last key the answer passed, and the shards'
//! changes are merged in key order, so the sealed `next` holds one key.
//!
//! A row's `before` and `after` are the versions its revisions were written
//! at. A record merged in from another branch keeps the version it was written
//! at on its own side; the version the change belongs to is the answer's `to`.
//!
//! A change to a record the caller may not read is absent and not counted, but
//! it is taken against the page's limit: a page that reaches its limit inside
//! a run of such changes gives no rows and a `next`. A caller can learn from
//! such pages that changes lie there which it was not given, and their number
//! bounds how many; never which records, or where.
//!
//! `to: {branch}` with no version is the head, captured on the first page;
//! the answer's `to` names it, and every later page reads at that version from
//! the token, so a writer between pages changes nothing. A live view asks again
//! with `from` set to the last `to` it was given. `from: {"fork": branch}` is
//! the moment that branch forked at, from its fork provenance; `to` must then
//! be on that branch.
//!
//! `from` may be later than `to`: the answer is then the reverse, what would
//! be undone going back, with added and removed exchanged and each row's
//! `before` and `after` exchanged.
//!
//! With values, each body is JSON as in query.run's rows: every number's text
//! as written, object keys in sorted order. The bytes exactly as stored are
//! `documents.get`'s `raw_json` at that version.

#[cfg(test)]
use std::cell::Cell;
use std::collections::{BTreeMap, VecDeque};
use std::num::{NonZeroU32, NonZeroU64, NonZeroUsize};

use haematite::db::{ObservedChange, ObserverError, ReadOnlyDatabase};
use haematite::tree::{DiffBudget, DiffError, DiffInterval};
use haematite::{Hash, TreeError};
use haematite_query::page::{Allowance, Origin, Resume};
use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;
use serde_json::{Value, json};

use super::access::Access;
use super::contracts::Caller;
use super::protocol::{QueryPage, Selection, Version};
use super::query::{Binding, Continuation, Engine, Rows, Seal};
use super::storage::{
    Captured, Catalogue, DocumentRevision, Failure, ForkProvenance, Held, document_key, reads,
    revision_key,
};

const REVISIONS: &[u8] = b"m\0";
const REVISIONS_END: &[u8] = b"m\x01";

#[cfg(test)]
thread_local! {
    /// Diff calls and body reads made on this thread, for tests.
    static DIFFS: Cell<u64> = const { Cell::new(0) };
    static BODIES: Cell<u64> = const { Cell::new(0) };
}

/// The most changes one shard's diff gives at once.
const BATCH: usize = 64;

/// The most bytes a collection name or record id may have.
const ID_BYTES: usize = 512;

/// The named limits of `changes.list`, each set in serve.json or defaulted.
#[derive(Clone, Copy, Debug)]
pub struct Settings {
    /// Changes one page may take, denied ones included; at least 2.
    pub scan_records: NonZeroU64,
    /// Tree nodes one shard's diff may load for one call.
    pub node_budget: NonZeroUsize,
    /// The most rows a page may ask for.
    pub max_page: NonZeroU32,
}

impl Default for Settings {
    fn default() -> Self {
        Self {
            scan_records: NonZeroU64::MIN.saturating_add(9_999),
            // As `branch.review`'s node budget (storage/review.rs).
            node_budget: NonZeroUsize::MIN.saturating_add(65_535),
            max_page: NonZeroU32::MIN.saturating_add(999),
        }
    }
}

/// A moment a caller names: a branch at a version, or its head (`to` only),
/// or the moment a branch forked at (`from` only).
#[derive(Debug, Serialize, Deserialize)]
#[serde(untagged)]
pub enum Moment {
    Fork(Fork),
    At(Selection),
}

/// The moment `fork` forked at.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Fork {
    pub fork: String,
}

/// The two moments, the page and the row form a caller asks for.
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ChangesRequest {
    pub from: Moment,
    pub to: Moment,
    /// `"origins"` (the default) or `"values"`.
    #[serde(default = "origins")]
    pub rows: Rows,
    pub page: QueryPage,
    #[serde(default)]
    pub continuation: Option<String>,
}

const fn origins() -> Rows {
    Rows::Origins
}

/// What one call of `changes.list` reads with, all owned by the service.
pub struct Context<'a> {
    pub access: &'a Access,
    pub reader: &'a ReadOnlyDatabase,
    pub catalogue: &'a Catalogue,
    pub settings: &'a Settings,
    pub seal: &'a Seal,
}

/// One page of the changes `request` asks for.
///
/// # Errors
///
/// A named `Failure` for every refusal; never a shortened answer.
pub fn list(
    context: &Context<'_>,
    request: &ChangesRequest,
    response_byte_limit: Option<NonZeroUsize>,
) -> Result<Box<RawValue>, Failure> {
    let asked = checked(request, context.settings)?;
    let token = request
        .continuation
        .as_deref()
        .map(|encoded| Continuation::decode(encoded, context.seal))
        .transpose()?;
    if let Some(token) = &token {
        token.check_moment(&asked.branch, asked.version)?;
    }
    // Both moments are held until this answer is built, on every path
    // (STOR-003 H1b).
    let mut from = match &request.from {
        Moment::At(selection) => context.catalogue.capture_held(selection)?,
        Moment::Fork(fork) => fork_moment(context, &fork.fork)?,
    };
    // The head is captured on the first page; a later page reads at the
    // version its token carries.
    let mut to = context.catalogue.capture_held(&Selection {
        branch: asked.branch.clone(),
        version: token.as_ref().map(|token| token.version).or(asked.version),
    })?;
    from.shards.sort_unstable_by_key(|shard| shard.shard_id);
    to.shards.sort_unstable_by_key(|shard| shard.shard_id);
    let binding = bind(request, from.version, context.settings)?;
    if let Some(token) = &token {
        token.check_binding(&binding)?;
    }
    let budget = budget(request.page.bytes, response_byte_limit)?;
    let widest = widest_next(&to, &binding, context.seal)?;
    let envelope = answer(&from, &to, &json!([]), Some(&widest), u64::MAX);
    let allowance = Allowance {
        rows: request.page.rows,
        bytes: rows_bytes(budget, &envelope)?,
        scan_records: context.settings.scan_records.get(),
    };
    let resume = token.as_ref().map(|token| &token.resume);
    let after = resume
        .map(|resume| revision_key(&resume.pending.collection, &resume.pending.id))
        .map(|mut key| {
            key.push(0); // Exclusive of the key given; ids never contain NUL.
            key
        });
    let read_before = resume.map_or(0, |resume| resume.records_read);
    let mut source = Changes::new(
        context.reader,
        &from,
        &to,
        after.as_deref(),
        context.settings,
    )?;
    drop(after);
    let reading = Reading {
        context,
        from: &from,
        to: &to,
        form: request.rows,
    };
    let page = page(&mut source, &reading, &allowance, read_before)?;
    let next = page
        .next
        .map(|place| {
            let resume = Resume {
                records_read: page.given,
                pending: place,
                discard_outputs: 0,
                passed: true,
                passed_counts: Vec::new(),
            };
            let now = raw("null")?;
            Continuation::new(to.branch.clone(), to.version, resume, now, binding)?
                .encode(context.seal)
        })
        .transpose()?;
    // NB1: the rows are written as text, each body as it is stored, and
    // never decoded into a value and encoded again.
    let rows = serde_json::value::to_raw_value(&page.rows).map_err(Failure::storage)?;
    let envelope = answer(&from, &to, &json!([]), next.as_deref(), page.given);
    let answer = with_rows(envelope, &rows)?;
    if answer.get().len() > budget.get() {
        return Err(Failure::new(
            "response_limit",
            "the answer exceeds its requested bytes",
        ));
    }
    Ok(answer)
}

/// An answer object's members beside a `rows` member written as text (NB1).
#[derive(Serialize)]
struct WithRows<'a> {
    #[serde(flatten)]
    envelope: &'a serde_json::Map<String, Value>,
    rows: &'a RawValue,
}

/// `envelope`, an object, as text with `rows` in place of its own `rows`
/// member: every body inside `rows` goes out as the text it is.
///
/// # Errors
/// `storage` when `envelope` is not an object or cannot be written.
pub fn with_rows(envelope: Value, rows: &RawValue) -> Result<Box<RawValue>, Failure> {
    let Value::Object(mut envelope) = envelope else {
        return Err(Failure::new(
            "storage",
            "an answer envelope is not an object",
        ));
    };
    envelope.remove("rows");
    serde_json::value::to_raw_value(&WithRows {
        envelope: &envelope,
        rows,
    })
    .map_err(Failure::storage)
}

/// `to` as asked, once `to` is no fork, `from` names a version or a fork on
/// `to`'s branch, and the page is within the service's.
fn checked<'a>(request: &'a ChangesRequest, settings: &Settings) -> Result<&'a Selection, Failure> {
    if request.page.rows.get() > u64::from(settings.max_page.get()) {
        return Err(Failure::new(
            "page_limit",
            format!("maximum page size is {}", settings.max_page),
        ));
    }
    let Moment::At(to) = &request.to else {
        return Err(Failure::new(
            "changes_fork_as_to",
            "to names a branch and perhaps a version, never a fork",
        ));
    };
    let branch = match &request.from {
        Moment::At(from) if from.version.is_none() => {
            return Err(Failure::new(
                "changes_version_required",
                "from names a version or a fork",
            ));
        }
        Moment::At(from) => &from.branch,
        Moment::Fork(fork) => &fork.fork,
    };
    if *branch != to.branch {
        return Err(Failure::new(
            "changes_branch_mismatch",
            "from and to must be on the same branch",
        ));
    }
    Ok(to)
}

/// Where a branch's fork provenance is kept (storage/branches.rs).
const FORK_PROVENANCE: &[u8] = b"b\0fork";

/// The moment `branch` forked at: its parent at the version it forked from.
fn fork_moment(context: &Context<'_>, branch: &str) -> Result<Held, Failure> {
    let head = context.catalogue.capture_held(&Selection {
        branch: branch.to_owned(),
        version: None,
    })?;
    let bytes = reads::get_bytes(context.reader, &head, FORK_PROVENANCE)?.ok_or_else(|| {
        Failure::new(
            "changes_fork_missing",
            format!("branch {branch} has no recorded fork origin"),
        )
    })?;
    let fork: ForkProvenance = serde_json::from_slice(&bytes).map_err(Failure::storage)?;
    context.catalogue.capture_held(&Selection {
        branch: fork.branch,
        version: Some(fork.version),
    })
}

/// What a continuation is bound to: both moments as asked and the version
/// `from` was found at, the caller, the page's limits and the row form.
fn bind(request: &ChangesRequest, from: Version, settings: &Settings) -> Result<Binding, Failure> {
    let asked = (&request.from, &request.to, from);
    let moments = serde_json::to_string(&asked).map_err(Failure::storage)?;
    let engine = Engine {
        protocol: 1,
        worker: "haem".to_owned(),
        engine: "changes".to_owned(),
        jaq_core: String::new(),
        jaq_std: String::new(),
        jaq_json: String::new(),
    };
    let limits = Allowance {
        rows: request.page.rows,
        bytes: request.page.bytes,
        scan_records: settings.scan_records.get(),
    };
    Binding::new(
        &moments,
        &BTreeMap::new(),
        &engine,
        &Caller::local("standalone")?,
        &limits,
        request.rows,
    )
}

fn raw(json: &str) -> Result<Box<RawValue>, Failure> {
    RawValue::from_string(json.to_owned()).map_err(Failure::storage)
}

/// The longest `next` a page could give: a place with the longest collection
/// and id, every byte escaped, and the largest count.
fn widest_next(to: &Captured, binding: &Binding, seal: &Seal) -> Result<String, Failure> {
    let resume = Resume {
        records_read: u64::MAX,
        pending: Origin {
            collection: "\u{1}".repeat(ID_BYTES),
            id: "\u{1}".repeat(ID_BYTES),
            version: raw("null")?,
            score: None,
        },
        discard_outputs: u64::MAX,
        passed: false,
        passed_counts: Vec::new(),
    };
    let now = raw("null")?;
    Continuation::new(to.branch.clone(), to.version, resume, now, binding.clone())?.encode(seal)
}

/// The smaller of the request's bytes and the transport's limit.
fn budget(
    requested: NonZeroU64,
    response_byte_limit: Option<NonZeroUsize>,
) -> Result<NonZeroUsize, Failure> {
    let requested = NonZeroUsize::try_from(requested)
        .map_err(|_| Failure::new("page_limit", "the requested bytes do not fit this machine"))?;
    let bytes = response_byte_limit
        .map_or_else(|| requested.get(), |limit| limit.get().min(requested.get()));
    NonZeroUsize::new(bytes)
        .ok_or_else(|| Failure::new("page_limit", "selected byte limit is zero"))
}

/// The bytes left for the rows once the answer's `envelope` is counted.
fn rows_bytes(budget: NonZeroUsize, envelope: &Value) -> Result<NonZeroU64, Failure> {
    let envelope = serde_json::to_vec(envelope)
        .map_err(Failure::storage)?
        .len();
    // The envelope's empty `[]` is counted again by the page as its rows.
    // It holds room for the widest continuation the page could need, so the
    // refusal names the size: a caller asks once more, with enough.
    let needed = envelope - 2;
    u64::try_from(budget.get())
        .ok()
        .and_then(|budget| budget.checked_sub(u64::try_from(needed).ok()?))
        .and_then(NonZeroU64::new)
        .ok_or_else(|| {
            Failure::new(
                "response_limit",
                format!(
                    "the answer's envelope takes {needed} bytes before any row; ask for more than that"
                ),
            )
        })
}

fn moment(at: &Captured) -> Value {
    json!({"branch": at.branch, "version": at.version})
}

fn answer(from: &Captured, to: &Captured, rows: &Value, next: Option<&str>, read: u64) -> Value {
    json!({
        "from": moment(from),
        "to": moment(to),
        "rows": rows,
        "next": next,
        "changes_read": read,
    })
}

/// The moments and row form one page is read with.
struct Reading<'a> {
    context: &'a Context<'a>,
    from: &'a Captured,
    to: &'a Captured,
    form: Rows,
}

/// One page's rows, how many changes have been given in all, and where the
/// next page starts after.
struct Page {
    rows: Vec<Row>,
    given: u64,
    next: Option<Origin>,
}

/// The rows of one page: changes the caller may read, until the rows, the
/// bytes or the source's limit are reached.
fn page(
    source: &mut Changes<'_>,
    reading: &Reading<'_>,
    allowance: &Allowance,
    read_before: u64,
) -> Result<Page, Failure> {
    let mut page = Page {
        rows: Vec::new(),
        given: read_before,
        next: None,
    };
    // The rows array's brackets.
    let mut bytes = 2_u64;
    // The last key this page finished with, given or passed.
    let mut done: Option<Vec<u8>> = None;
    loop {
        let given = u64::try_from(page.rows.len()).map_err(Failure::storage)?;
        if given == allowance.rows.get() {
            page.next = done.as_deref().map(place).transpose()?;
            return Ok(page);
        }
        let change = match source.next()? {
            Next::Change(change) => change,
            Next::Stopped(key) => {
                page.next = Some(place(&key)?);
                return Ok(page);
            }
            Next::End => return Ok(page),
        };
        let key = changed_key(&change).to_vec();
        let Some(row) = row(reading, change)? else {
            done = Some(key);
            continue;
        };
        let size = u64::try_from(serde_json::to_vec(&row).map_err(Failure::storage)?.len())
            .map_err(Failure::storage)?;
        if size.saturating_add(2) > allowance.bytes.get() {
            return Err(Failure::new(
                "changes_row_too_large",
                format!("a change needs more than {} bytes", allowance.bytes),
            ));
        }
        let grown = bytes
            .saturating_add(u64::from(given > 0))
            .saturating_add(size);
        if grown > allowance.bytes.get() {
            // Not given: the next page reads it again.
            page.next = done.as_deref().map(place).transpose()?;
            return Ok(page);
        }
        bytes = grown;
        page.given = page.given.saturating_add(1);
        page.rows.push(row);
        done = Some(key);
    }
}

fn changed_key(change: &ObservedChange) -> &[u8] {
    match change {
        ObservedChange::Added { key, .. }
        | ObservedChange::Removed { key, .. }
        | ObservedChange::Modified { key, .. } => key,
    }
}

/// A revision key's collection and id.
fn parts(key: &[u8]) -> Result<(&str, &str), Failure> {
    key.strip_prefix(REVISIONS)
        .and_then(|rest| std::str::from_utf8(rest).ok())
        .and_then(|rest| rest.split_once('\0'))
        .ok_or_else(|| Failure::new("changes_key_corrupt", "a revision key names no record"))
}

/// The place after which the next page starts.
fn place(key: &[u8]) -> Result<Origin, Failure> {
    let (collection, id) = parts(key)?;
    Ok(Origin {
        collection: collection.to_owned(),
        id: id.to_owned(),
        version: raw("null")?,
        score: None,
    })
}

fn written_at(revision: &[u8]) -> Result<Version, Failure> {
    serde_json::from_slice::<DocumentRevision>(revision)
        .map(|revision| revision.written_at)
        .map_err(|error| Failure::new("changes_revision_corrupt", error.to_string()))
}

/// A changed record's name.
#[derive(Serialize)]
struct Named {
    collection: String,
    id: String,
}

/// One change. Without values, `value_before` and `value_after` are absent;
/// with them, each is the body at its moment, or null where there is none.
#[derive(Serialize)]
struct Row {
    kind: &'static str,
    origin: Named,
    before: Option<Version>,
    after: Option<Version>,
    #[serde(skip_serializing_if = "Option::is_none")]
    value_before: Option<Box<RawValue>>,
    #[serde(skip_serializing_if = "Option::is_none")]
    value_after: Option<Box<RawValue>>,
}

/// The row for `change`, or none when the caller may not read its record.
fn row(reading: &Reading<'_>, change: ObservedChange) -> Result<Option<Row>, Failure> {
    let (key, kind, before, after) = match change {
        ObservedChange::Added { key, value } => (key, "added", None, Some(value)),
        ObservedChange::Removed { key, value } => (key, "removed", Some(value), None),
        ObservedChange::Modified {
            key,
            old_value,
            new_value,
        } => (key, "changed", Some(old_value), Some(new_value)),
    };
    let (collection, id) = parts(&key)?;
    if !reading.context.access.can_read_document(collection, id) {
        return Ok(None);
    }
    let before = before.as_deref().map(written_at).transpose()?;
    let after = after.as_deref().map(written_at).transpose()?;
    let mut row = Row {
        kind,
        origin: Named {
            collection: collection.to_owned(),
            id: id.to_owned(),
        },
        before,
        after,
        value_before: None,
        value_after: None,
    };
    if reading.form == Rows::Values {
        let body = document_key(collection, id);
        row.value_before = Some(body_at(reading, reading.from, &body, before.is_some())?);
        row.value_after = Some(body_at(reading, reading.to, &body, after.is_some())?);
    }
    Ok(Some(row))
}

/// The body at `at` of a record whose revision is there (`present`), else
/// null: the stored text as it is (NB1), checked to be JSON and never
/// decoded into a value, so every number keeps the text it was written with.
fn body_at(
    reading: &Reading<'_>,
    at: &Captured,
    key: &[u8],
    present: bool,
) -> Result<Box<RawValue>, Failure> {
    if !present {
        return raw("null");
    }
    #[cfg(test)]
    BODIES.with(|count| count.set(count.get() + 1));
    let bytes = reads::get_bytes(reading.context.reader, at, key)?
        .ok_or_else(|| Failure::new("changes_body_missing", "a revision has no body"))?;
    let text = std::str::from_utf8(&bytes).map_err(Failure::storage)?;
    RawValue::from_string(text.to_owned()).map_err(Failure::storage)
}

/// What the source gives next.
enum Next {
    Change(ObservedChange),
    /// The limit is reached with changes left: the last key taken.
    Stopped(Vec<u8>),
    End,
}

/// One shard's diff, read ahead after its last key.
struct Shard {
    id: usize,
    from: Hash,
    to: Hash,
    after: Option<Vec<u8>>,
    ahead: VecDeque<ObservedChange>,
    ended: bool,
}

/// The revision changes of every shard between two captures, in key order.
struct Changes<'a> {
    reader: &'a ReadOnlyDatabase,
    shards: Vec<Shard>,
    node_budget: NonZeroUsize,
    limit: u64,
    taken: u64,
    last: Option<Vec<u8>>,
}

impl<'a> Changes<'a> {
    fn new(
        reader: &'a ReadOnlyDatabase,
        from: &Captured,
        to: &Captured,
        after: Option<&[u8]>,
        settings: &Settings,
    ) -> Result<Self, Failure> {
        if from.shards.len() != to.shards.len() {
            return Err(Failure::new(
                "changes_shards_differ",
                "the two moments do not have the same shards",
            ));
        }
        let mut shards = Vec::with_capacity(to.shards.len());
        for (before, now) in from.shards.iter().zip(&to.shards) {
            if before.shard_id != now.shard_id {
                return Err(Failure::new(
                    "changes_shards_differ",
                    "the two moments do not name the same shards",
                ));
            }
            shards.push(Shard {
                id: now.shard_id,
                from: before.head,
                to: now.head,
                after: after.map(<[u8]>::to_vec),
                ahead: VecDeque::new(),
                // Equal roots have no changes: nothing is read.
                ended: before.head == now.head,
            });
        }
        Ok(Self {
            reader,
            shards,
            node_budget: settings.node_budget,
            limit: settings.scan_records.get(),
            taken: 0,
            last: None,
        })
    }

    /// As many changes as this page may still take, at most `BATCH`, at least one.
    fn batch(&self) -> NonZeroUsize {
        let left = usize::try_from(self.limit.saturating_sub(self.taken)).unwrap_or(usize::MAX);
        NonZeroUsize::new(left.min(BATCH)).unwrap_or(NonZeroUsize::MIN)
    }

    /// Reads on every shard whose read-ahead is empty until it has a change or
    /// its diff is done; a diff page may be empty with more to come.
    fn fill(&mut self) -> Result<(), Failure> {
        let budget = DiffBudget {
            max_changes: self.batch(),
            max_nodes: self.node_budget,
        };
        let interval = DiffInterval {
            lower: Some(REVISIONS),
            upper: Some(REVISIONS_END),
        };
        for shard in &mut self.shards {
            while shard.ahead.is_empty() && !shard.ended {
                #[cfg(test)]
                DIFFS.with(|count| count.set(count.get() + 1));
                let page = self
                    .reader
                    .changes_page_in_range(
                        shard.id,
                        shard.from,
                        shard.to,
                        interval,
                        shard.after.as_deref(),
                        budget,
                    )
                    .map_err(diff_failure)?;
                shard.ended = page.next_after.is_none();
                shard.after = page.next_after;
                shard.ahead.extend(page.entries);
            }
        }
        Ok(())
    }

    fn next(&mut self) -> Result<Next, Failure> {
        if self.taken >= self.limit {
            if self.shards.iter().all(|shard| shard.ahead.is_empty()) {
                self.fill()?;
            }
            if self.shards.iter().all(|shard| shard.ahead.is_empty()) {
                return Ok(Next::End);
            }
            let last = self.last.clone().ok_or_else(|| {
                Failure::new(
                    "changes_scan_limit",
                    "a page that may take no changes moves nowhere",
                )
            })?;
            return Ok(Next::Stopped(last));
        }
        self.fill()?;
        let Some(shard) = self
            .shards
            .iter_mut()
            .filter(|shard| !shard.ahead.is_empty())
            .min_by(|left, right| {
                let left = left.ahead.front().map(changed_key);
                let right = right.ahead.front().map(changed_key);
                left.cmp(&right)
            })
        else {
            return Ok(Next::End);
        };
        let Some(change) = shard.ahead.pop_front() else {
            return Ok(Next::End);
        };
        self.taken += 1;
        self.last = Some(changed_key(&change).to_vec());
        Ok(Next::Change(change))
    }
}

/// A moment whose nodes are gone is named, never read around.
fn diff_failure(error: ObserverError) -> Failure {
    let classification = match &error {
        ObserverError::Tree(TreeError::MissingNode { hash }) => Failure::new(
            "changes_moment_gone",
            format!("the moment no longer holds node {hash}"),
        ),
        ObserverError::RootUnavailable { shard_id, root } => Failure::new(
            "changes_moment_gone",
            format!("shard {shard_id} cannot read root {root}"),
        ),
        ObserverError::Diff(error @ DiffError::NodeBudgetTooSmall { .. }) => {
            Failure::new("changes_node_budget", error.to_string())
        }
        other => Failure::storage(other),
    };
    let mut failure = Failure::observer(error);
    failure.code = classification.code;
    failure.message = classification.message;
    failure.path = classification.path;
    failure
}

#[cfg(test)]
#[path = "changes_tests.rs"]
mod tests;

#[cfg(test)]
#[path = "held_reads_tests.rs"]
mod held_reads_tests;