haematite-cli 0.21.7

//! Reads at immutable captured roots, independent of mutation admission.

use std::num::NonZeroUsize;

use haematite::TreeError;
use haematite::db::{ObserverError, ReadOnlyDatabase};
use serde::Serialize;
use serde_json::value::RawValue;

use super::{
    Captured, Collection, DocumentRevision, Failure, collection_key, document_key, identifier,
    revision_key,
};
use crate::serve::access::Access;
use crate::serve::protocol::{CountReason, DocumentPage, KeptCount, Version};

#[derive(Debug, Serialize)]
pub struct Document {
    pub id: String,
    pub raw_json: String,
    pub revision: DocumentRevision,
}

#[derive(Debug, Serialize)]
#[serde(untagged)]
pub(in crate::serve) enum DocumentListEntry {
    Full(Document),
    Metadata {
        id: String,
        revision: DocumentRevision,
    },
}

#[derive(Clone, Copy)]
pub(in crate::serve) enum DocumentContent {
    Full,
    /// Makes zero logical reads of document, edge or vector keys and returns
    /// no document bytes on the wire. It makes no claim about which bytes a
    /// leaf read brings into memory.
    ///
    /// Lists the same document IDs and revisions as full where every document
    /// has its revision key and every revision key has its document. Each
    /// document write or removal path commits the pair in one commit.
    None,
}

#[derive(Debug, Serialize)]
pub struct CollectionEntry {
    pub name: String,
    pub changed_in: String,
    pub changed_at: Version,
    pub schema_json: Option<String>,
    pub documents: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub documents_reason: Option<CountReason>,
}

fn render_collection(row: Collection, total: &KeptCount) -> Result<CollectionEntry, Failure> {
    let schema_json = row
        .schema
        .as_ref()
        .map(serde_json::to_string)
        .transpose()
        .map_err(Failure::storage)?;
    Ok(CollectionEntry {
        name: row.name,
        changed_in: row.changed_in,
        changed_at: row.changed_at,
        schema_json,
        documents: total.value,
        documents_reason: total.reason,
    })
}

pub(in crate::serve) fn render_collections(
    access: &Access,
    reader: &ReadOnlyDatabase,
    at: &Captured,
    collections: Vec<Collection>,
) -> Result<Vec<CollectionEntry>, Failure> {
    let totals = collection_totals(access, reader, at, &collections)?;
    collections
        .into_iter()
        .zip(totals)
        .map(|(row, total)| render_collection(row, &total))
        .collect()
}

#[derive(Debug, Serialize)]
pub struct Page<T> {
    pub branch: String,
    pub version: Version,
    pub entries: Vec<T>,
    pub next_after: Option<String>,
}

#[derive(Serialize)]
pub(in crate::serve) struct CollectionCount {
    name: String,
    documents: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    documents_reason: Option<CountReason>,
}

#[derive(Serialize)]
pub(in crate::serve) struct EmbeddedFieldCount {
    collection: String,
    field: String,
    kind: super::embedded::VectorKind,
    storage: &'static str,
    count: Option<u64>,
    count_reason: &'static str,
}

pub(in crate::serve) struct Stats {
    pub(in crate::serve) collections: Vec<CollectionCount>,
    pub(in crate::serve) collection_count: KeptCount,
    pub(in crate::serve) relation_count: KeptCount,
    pub(in crate::serve) documents: KeptCount,
    pub(in crate::serve) edges: KeptCount,
    pub(in crate::serve) vectors: KeptCount,
    pub(in crate::serve) embedded_fields: Option<Vec<EmbeddedFieldCount>>,
    pub(in crate::serve) embedded_fields_reason: Option<CountReason>,
}

pub(in crate::serve) fn collection_totals(
    access: &Access,
    reader: &ReadOnlyDatabase,
    at: &Captured,
    collections: &[Collection],
) -> Result<Vec<KeptCount>, Failure> {
    let view = super::counts::View::new(reader, at)?;
    collections
        .iter()
        .map(|row| view.documents(access, &row.name))
        .collect()
}

pub(in crate::serve) fn stats(
    access: &Access,
    reader: &ReadOnlyDatabase,
    at: &Captured,
) -> Result<Stats, Failure> {
    let view = super::counts::View::new(reader, at)?;
    let mut collections = std::collections::BTreeSet::new();
    scan_rows(reader, at, b"c\0", 1, |parts, _| {
        if !collections.insert(parts[0].to_owned()) {
            return Err(Failure::new(
                "corrupt_key",
                "collection key exists in multiple shards",
            ));
        }
        Ok(())
    })?;
    let allowed = access.can_read_store();
    let collection_count = declaration_total(allowed, collections.len())?;
    let mut declared_relations = std::collections::BTreeSet::new();
    if allowed {
        scan_rows(reader, at, b"n\0", 1, |parts, bytes| {
            super::relations::decode(parts[0], bytes)?;
            if !declared_relations.insert(parts[0].to_owned()) {
                return Err(Failure::new(
                    "counts_corrupt",
                    "duplicate relation declaration",
                ));
            }
            Ok(())
        })?;
    }
    let relation_count = declaration_total(allowed, declared_relations.len())?;
    let mut documents = 0_u64;
    let mut entries = Vec::new();
    for name in collections {
        let total = view.documents(access, &name)?;
        if let Some(count) = total.value {
            documents = documents
                .checked_add(count)
                .ok_or_else(|| Failure::new("count_overflow", "document total exceeds u64"))?;
        }
        entries.push(CollectionCount {
            name,
            documents: total.value,
            documents_reason: total.reason,
        });
    }
    if let Some(reason) = view.unavailable(access.can_read_store()) {
        return Ok(Stats {
            collections: entries,
            collection_count,
            relation_count,
            documents: KeptCount {
                value: None,
                reason: Some(reason),
            },
            edges: KeptCount {
                value: None,
                reason: Some(reason),
            },
            vectors: KeptCount {
                value: None,
                reason: Some(reason),
            },
            embedded_fields: if allowed {
                Some(embedded_field_counts(reader, at, None)?)
            } else {
                None
            },
            embedded_fields_reason: (!allowed).then_some(CountReason::Withheld),
        });
    }
    let (edges, relations) = super::counts::sum_rows(reader, at, b"k\0e\0", 1)?;
    let (vectors, fields) = super::counts::sum_rows(reader, at, b"k\0v\0", 2)?;
    for name in declared_relations {
        if !relations.contains(&name) {
            return Err(Failure::new(
                "counts_corrupt",
                "declared relation has no count",
            ));
        }
    }
    let embedded_fields = embedded_field_counts(reader, at, Some(&fields))?;
    Ok(Stats {
        collections: entries,
        collection_count,
        relation_count,
        documents: KeptCount {
            value: Some(documents),
            reason: None,
        },
        edges: KeptCount {
            value: Some(edges),
            reason: None,
        },
        vectors: KeptCount {
            value: Some(vectors),
            reason: None,
        },
        embedded_fields: Some(embedded_fields),
        embedded_fields_reason: None,
    })
}

fn embedded_field_counts(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    required_supplied: Option<&std::collections::BTreeSet<String>>,
) -> Result<Vec<EmbeddedFieldCount>, Failure> {
    let mut embedded_fields = Vec::new();
    scan_rows(reader, at, b"w\0", 2, |parts, bytes| {
        match super::embedded::read_field(bytes)? {
            super::embedded::Field::Supplied { .. } => {
                if required_supplied.is_some_and(|fields| !fields.contains(&parts.join("\0"))) {
                    return Err(Failure::new(
                        "counts_corrupt",
                        "declared field has no count",
                    ));
                }
            }
            super::embedded::Field::Embedded(declaration) => {
                // Companion vectors have no kept counter; never scan them for statistics.
                embedded_fields.push(EmbeddedFieldCount {
                    collection: parts[0].to_owned(),
                    field: parts[1].to_owned(),
                    kind: declaration.kind,
                    storage: "companion",
                    count: None,
                    count_reason: "companion_count_not_kept",
                });
            }
        }
        Ok(())
    })?;
    Ok(embedded_fields)
}

fn declaration_total(allowed: bool, count: usize) -> Result<KeptCount, Failure> {
    if !allowed {
        return Ok(KeptCount {
            value: None,
            reason: Some(CountReason::Withheld),
        });
    }
    let value =
        u64::try_from(count).map_err(|error| Failure::new("count_overflow", error.to_string()))?;
    Ok(KeptCount {
        value: Some(value),
        reason: None,
    })
}

#[cfg(test)]
pub(super) mod test_support {
    use super::{Captured, Failure, ReadOnlyDatabase, scan_rows};

    pub(in crate::serve::storage) fn count_keys(
        reader: &ReadOnlyDatabase,
        at: &Captured,
        prefix: &[u8],
        parts: usize,
    ) -> Result<u64, Failure> {
        let mut count = 0_u64;
        scan_rows(reader, at, prefix, parts, |_, _| {
            count = increment(count)?;
            Ok(())
        })?;
        Ok(count)
    }

    fn increment(value: u64) -> Result<u64, Failure> {
        value
            .checked_add(1)
            .ok_or_else(|| Failure::new("count_overflow", "stored key count exceeds u64"))
    }
}

/// Visit each live composite key once without retaining its stored value.
pub(super) fn scan_rows(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    prefix: &[u8],
    expected_parts: usize,
    mut visit: impl FnMut(&[&str], &[u8]) -> Result<(), Failure>,
) -> Result<(), Failure> {
    let rows = NonZeroUsize::new(64)
        .ok_or_else(|| Failure::new("page_limit", "key scan page size must be nonzero"))?;
    let mut from = prefix.to_vec();
    loop {
        let (_, next_after) = bounded_key_page(
            reader,
            at,
            prefix,
            &from,
            PageLimits {
                rows,
                bytes: NonZeroUsize::MAX,
            },
            |suffix, bytes| {
                let parts: Vec<_> = suffix.split('\0').collect();
                if parts.len() != expected_parts {
                    return Err(Failure::new(
                        "corrupt_key",
                        "stored key has the wrong number of parts",
                    ));
                }
                for part in &parts {
                    identifier(part)?;
                }
                visit(&parts, &bytes)
            },
            |&()| Ok(0),
        )?;
        let Some(next_after) = next_after else {
            return Ok(());
        };
        from = prefix.to_vec();
        from.extend_from_slice(next_after.as_bytes());
        from.push(0);
    }
}

/// # Errors
///
/// Returns `Failure` if `shard_for`, `find` or `get_at` fails; a node or root
/// of `at` that has gone is `version_released`, naming the version read.
pub fn get_bytes(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    key: &[u8],
) -> Result<Option<Vec<u8>>, Failure> {
    let shard_id = reader.shard_for(key);
    let shard = at
        .shards
        .iter()
        .find(|shard| shard.shard_id == shard_id)
        .ok_or_else(|| {
            Failure::new(
                "incomplete_roots",
                "captured version does not name the owning shard",
            )
        })?;
    #[cfg(test)]
    count_point_read();
    #[cfg(test)]
    count_content_read(key);
    reader
        .get_at(shard_id, shard.head, key)
        .map_err(|error| point_read_failure(at, error))
}

/// A point read whose node or root has gone names the version it read
/// (STOR-003 H1b2-A1): `version_released`, as a refused hold names it, never
/// `storage_unavailable`; diffs name theirs `..._moment_gone` the same way.
/// Anything else is the store's failure.
fn point_read_failure(at: &Captured, error: ObserverError) -> Failure {
    let gone = match error {
        ObserverError::Tree(TreeError::MissingNode { hash }) => format!("node {hash} is gone"),
        ObserverError::RootUnavailable { shard_id, root } => {
            format!("shard {shard_id} cannot read root {root}")
        }
        other => return Failure::observer(other),
    };
    Failure::new(
        "version_released",
        format!(
            "version {}.{} of branch {} was released by the store's tidying and cannot be \
             read: {gone}",
            at.version.generation, at.version.sequence, at.branch
        ),
    )
}

/// # Errors
///
/// Propagates any error from `identifier` or `get_bytes`.
pub fn collection(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    name: &str,
) -> Result<Option<Collection>, Failure> {
    identifier(name)?;
    get_bytes(reader, at, &collection_key(name))?
        .map(|bytes| serde_json::from_slice(&bytes).map_err(Failure::storage))
        .transpose()
}

/// # Errors
///
/// Propagates any error from `identifier` or `get_bytes`.
///
/// A point read cannot omit the request's authority. The first example is
/// the control (the same call with the authority compiles); the second is the
/// refusal.
///
/// ```no_run
/// use haematite_cli::serve::access::Access;
/// use haematite_cli::serve::storage::{Captured, reads};
/// use haematite::db::ReadOnlyDatabase;
/// fn read(access: &Access, reader: &ReadOnlyDatabase, at: &Captured) {
///     let result = reads::document(access, reader, at, "records", "one");
///     drop(result);
/// }
/// ```
///
/// ```compile_fail,E0061
/// use haematite_cli::serve::storage::{Captured, reads};
/// use haematite::db::ReadOnlyDatabase;
/// fn read(reader: &ReadOnlyDatabase, at: &Captured) {
///     let result = reads::document(reader, at, "records", "one");
///     drop(result);
/// }
/// ```
pub fn document(
    access: &crate::serve::access::Access,
    reader: &ReadOnlyDatabase,
    at: &Captured,
    collection_name: &str,
    id: &str,
) -> Result<Option<Document>, Failure> {
    identifier(collection_name)?;
    identifier(id)?;
    if !access.can_read_document(collection_name, id) {
        return Ok(None);
    }
    get_bytes(reader, at, &document_key(collection_name, id))?
        .map(|bytes| render_document(reader, at, collection_name, id.to_owned(), &bytes))
        .transpose()
}

pub(super) fn document_exists_for_constraint(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    collection_name: &str,
    id: &str,
) -> Result<bool, Failure> {
    identifier(collection_name)?;
    identifier(id)?;
    let Some(bytes) = get_bytes(reader, at, &document_key(collection_name, id))? else {
        return Ok(false);
    };
    render_document(reader, at, collection_name, id.to_owned(), &bytes)?;
    Ok(true)
}

/// # Errors
///
/// Propagates storage and identifier errors. Refuses a visibility scan that
/// exceeds the page examination limit with `access_scan_limit`.
pub fn documents(
    access: &Access,
    reader: &ReadOnlyDatabase,
    at: &Captured,
    collection_name: &str,
    after: Option<&str>,
    limit: NonZeroUsize,
) -> Result<Page<Document>, Failure> {
    document_page(
        access,
        reader,
        at,
        DocumentRange {
            collection: collection_name,
            prefix: None,
            after,
            metadata: false,
        },
        DocumentLimits {
            rows: limit,
            bytes: None,
        },
        render_document,
    )
}

/// Read a document page while bounding the cumulative serialized entry bytes.
///
/// The bound covers each document's JSON representation in the response entry
/// array, including JSON-string escaping of `raw_json`, plus commas between
/// entries. The response envelope is checked separately by the service. To
/// preserve global key order across shards, the implementation holds at most
/// one unread row per shard. A first row larger than `limits.bytes` is
/// refused with `response_limit`; it is never returned outside the budget.
///
/// # Errors
///
/// Propagates storage and identifier errors. Refuses a visibility scan that
/// exceeds the page examination limit with `access_scan_limit`.
pub fn documents_bounded(
    access: &Access,
    reader: &ReadOnlyDatabase,
    at: &Captured,
    collection_name: &str,
    after: Option<&str>,
    limits: PageLimits,
) -> Result<Page<Document>, Failure> {
    document_page(
        access,
        reader,
        at,
        DocumentRange {
            collection: collection_name,
            prefix: None,
            after,
            metadata: false,
        },
        DocumentLimits {
            rows: limits.rows,
            bytes: Some(limits.bytes),
        },
        render_document,
    )
}

/// Read only the document id range under the optional prefix.
///
/// # Errors
/// Refuses invalid prefixes, out-of-prefix continuations and the ordinary page failures.
pub fn documents_prefixed(
    access: &Access,
    reader: &ReadOnlyDatabase,
    at: &Captured,
    collection: &str,
    prefix: Option<&str>,
    after: Option<&str>,
    limit: NonZeroUsize,
) -> Result<Page<Document>, Failure> {
    document_page(
        access,
        reader,
        at,
        DocumentRange {
            collection,
            prefix,
            after,
            metadata: false,
        },
        DocumentLimits {
            rows: limit,
            bytes: None,
        },
        render_document,
    )
}

/// Read only the document id range while preserving the serialized entry budget.
///
/// # Errors
/// Refuses invalid prefixes, out-of-prefix continuations and the ordinary page failures.
pub fn documents_prefixed_bounded(
    access: &Access,
    reader: &ReadOnlyDatabase,
    at: &Captured,
    collection: &str,
    prefix: Option<&str>,
    after: Option<&str>,
    limits: PageLimits,
) -> Result<Page<Document>, Failure> {
    document_page(
        access,
        reader,
        at,
        DocumentRange {
            collection,
            prefix,
            after,
            metadata: false,
        },
        DocumentLimits {
            rows: limits.rows,
            bytes: Some(limits.bytes),
        },
        render_document,
    )
}

/// Validate the content mode before capturing a document version.
///
/// # Errors
/// Refuses every content string except full and none.
pub(in crate::serve) fn validate_document_content(
    content: Option<&str>,
) -> Result<DocumentContent, Failure> {
    match content {
        None | Some("full") => Ok(DocumentContent::Full),
        Some("none") => Ok(DocumentContent::None),
        Some(_) => Err(Failure::new(
            "invalid_document_content",
            "document content must be full or none",
        )),
    }
}

/// List full documents or their revision metadata at the captured roots.
///
/// # Errors
/// Refuses invalid selectors and propagates page, revision and storage failures.
pub(in crate::serve) fn list_documents(
    access: &Access,
    reader: &ReadOnlyDatabase,
    at: &Captured,
    page: &DocumentPage,
    limit: NonZeroUsize,
    byte_limit: Option<NonZeroUsize>,
) -> Result<Page<DocumentListEntry>, Failure> {
    let content = validate_document_content(page.content.as_deref())?;
    document_page(
        access,
        reader,
        at,
        DocumentRange {
            collection: &page.collection,
            prefix: page.prefix.as_deref(),
            after: page.after.as_deref(),
            metadata: matches!(content, DocumentContent::None),
        },
        DocumentLimits {
            rows: limit,
            bytes: byte_limit,
        },
        |reader, at, collection, id, bytes| match content {
            DocumentContent::Full => {
                render_document(reader, at, collection, id, bytes).map(DocumentListEntry::Full)
            }
            DocumentContent::None => Ok(DocumentListEntry::Metadata {
                id,
                revision: serde_json::from_slice(bytes).map_err(Failure::storage)?,
            }),
        },
    )
}

/// Validate the prefix before capturing a document version.
///
/// # Errors
/// Refuses invalid identifiers and continuations outside the prefix.
pub(in crate::serve) fn validate_document_prefix(
    prefix: Option<&str>,
    after: Option<&str>,
) -> Result<(), Failure> {
    if let Some(prefix) = prefix {
        identifier(prefix)?;
        if let Some(after) = after {
            identifier(after)?;
            if !after.as_bytes().starts_with(prefix.as_bytes()) {
                return Err(Failure::new(
                    "document_after_outside_prefix",
                    "document continuation does not start with the requested prefix",
                ));
            }
        }
    }
    Ok(())
}

#[derive(Clone, Copy)]
struct DocumentRange<'a> {
    collection: &'a str,
    prefix: Option<&'a str>,
    after: Option<&'a str>,
    metadata: bool,
}

#[derive(Clone, Copy)]
struct DocumentLimits {
    rows: NonZeroUsize,
    bytes: Option<NonZeroUsize>,
}

// A long denied run is refused by name rather than producing a hidden cursor
// or a short success; at most shards * (2 * row limit + 1) rows are examined.
const DOCUMENT_SCAN_ROWS_PER_LIMIT: usize = 2;

fn document_page<T: Serialize>(
    access: &Access,
    reader: &ReadOnlyDatabase,
    at: &Captured,
    range: DocumentRange<'_>,
    limits: DocumentLimits,
    make_entry: impl Fn(&ReadOnlyDatabase, &Captured, &str, String, &[u8]) -> Result<T, Failure>,
) -> Result<Page<T>, Failure> {
    let DocumentRange {
        collection,
        prefix: id_prefix,
        after,
        metadata,
    } = range;
    identifier(collection)?;
    validate_document_prefix(id_prefix, after)?;
    let family = if metadata { "m" } else { "d" };
    let prefix = format!("{family}\0{collection}\0").into_bytes();
    let mut range_prefix = prefix.clone();
    if let Some(id_prefix) = id_prefix {
        range_prefix.extend_from_slice(id_prefix.as_bytes());
    }
    let to = prefix_end(&range_prefix)?;
    let from = match after {
        Some(after) => exclusive_from(&prefix, Some(after))?,
        None => range_prefix,
    };
    let examined_limit = limits
        .rows
        .get()
        .checked_mul(DOCUMENT_SCAN_ROWS_PER_LIMIT)
        .and_then(|rows| rows.checked_add(1))
        .and_then(|rows| rows.checked_mul(at.shards.len()))
        .ok_or_else(|| Failure::new("limit_overflow", "document scan limit is too large"))?;
    let mut examined = 0;
    let mut candidates = Vec::with_capacity(at.shards.len());
    for shard in &at.shards {
        candidates.push((
            shard,
            document_candidate(reader, shard, &from, &to, &mut examined, examined_limit)?,
        ));
    }
    let mut entries = Vec::new();
    let mut last_id = None;
    let mut budget = limits.bytes.map(WireBudget::new);
    let next_after = loop {
        let Some(index) = candidates
            .iter()
            .enumerate()
            .filter_map(|(index, (_, row))| row.as_ref().map(|row| (index, &row.row.0)))
            .min_by(|(_, left), (_, right)| left.cmp(right))
            .map(|(index, _)| index)
        else {
            break None;
        };
        let DocumentCandidate {
            row: (key, value),
            next_from,
        } = candidates[index]
            .1
            .take()
            .ok_or_else(|| Failure::new("storage_unavailable", "page candidate disappeared"))?;
        let suffix = key
            .strip_prefix(prefix.as_slice())
            .ok_or_else(|| Failure::new("corrupt_key", "row is outside requested prefix"))?;
        let id = std::str::from_utf8(suffix).map_err(Failure::storage)?;
        if access.can_read_document(collection, id) {
            if entries.len() == limits.rows.get() {
                break last_id;
            }
            let entry = make_entry(reader, at, collection, id.to_owned(), &value)?;
            if let Some(budget) = budget.as_mut() {
                let bytes = serde_json::to_vec(&entry).map_err(Failure::storage)?.len();
                if !budget.admit_len(bytes)? {
                    break last_id;
                }
            }
            last_id = Some(id.to_owned());
            entries.push(entry);
        }
        candidates[index].1 = match next_from {
            Some(from) => document_candidate(
                reader,
                candidates[index].0,
                &from,
                &to,
                &mut examined,
                examined_limit,
            )?,
            None => None,
        };
    };
    Ok(Page {
        branch: at.branch.clone(),
        version: at.version,
        entries,
        next_after,
    })
}

struct DocumentCandidate {
    row: RawRow,
    next_from: Option<Vec<u8>>,
}

fn document_candidate(
    reader: &ReadOnlyDatabase,
    shard: &haematite::branch::ShardRef,
    from: &[u8],
    to: &[u8],
    examined: &mut usize,
    limit: usize,
) -> Result<Option<DocumentCandidate>, Failure> {
    // Refuse before fetching another candidate once the visibility scan cap is
    // spent; the internal next key is never used as the caller's continuation.
    if *examined == limit {
        return Err(Failure::new(
            "access_scan_limit",
            "document visibility scan exceeds the page examination limit",
        ));
    }
    #[cfg(test)]
    record_document_range(from, to);
    #[cfg(test)]
    count_range_read();
    #[cfg(test)]
    count_content_read(from);
    let page = reader
        .range_page_at(shard.shard_id, shard.head, from, to, NonZeroUsize::MIN)
        .map_err(Failure::observer)?;
    let Some(row) = page.entries.into_iter().next() else {
        return Ok(None);
    };
    *examined += 1;
    Ok(Some(DocumentCandidate {
        row,
        next_from: page.next_from,
    }))
}

/// # Errors
///
/// Propagates any error from `page`.
pub fn collections(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    after: Option<&str>,
    limit: NonZeroUsize,
) -> Result<Page<Collection>, Failure> {
    let (items, next_after) = page(reader, at, b"c\0", after, limit)?;
    let entries = items
        .into_iter()
        .map(|(_, bytes)| serde_json::from_slice(&bytes).map_err(Failure::storage))
        .collect::<Result<Vec<_>, _>>()?;
    Ok(Page {
        branch: at.branch.clone(),
        version: at.version,
        entries,
        next_after,
    })
}

/// Read a collection page with the same serialized-entry budget semantics as
/// [`documents_bounded`]. A single collection too large for the budget is an
/// explicit `response_limit` refusal.
///
/// # Errors
///
/// Propagates any error from `transpose`.
pub fn collections_bounded(
    access: &Access,
    reader: &ReadOnlyDatabase,
    at: &Captured,
    after: Option<&str>,
    limit: NonZeroUsize,
    entry_byte_limit: NonZeroUsize,
) -> Result<Page<CollectionEntry>, Failure> {
    let view = super::counts::View::new(reader, at)?;
    let (entries, next_after) = bounded_page(
        reader,
        at,
        b"c\0",
        after,
        PageLimits {
            rows: limit,
            bytes: entry_byte_limit,
        },
        |_, bytes| {
            let row: Collection = serde_json::from_slice(&bytes).map_err(Failure::storage)?;
            let total = view.documents(access, &row.name)?;
            render_collection(row, &total)
        },
        |entry| {
            serde_json::to_vec(entry)
                .map(|bytes| bytes.len())
                .map_err(Failure::storage)
        },
    )?;
    Ok(Page {
        branch: at.branch.clone(),
        version: at.version,
        entries,
        next_after,
    })
}

type NamedBytes = Vec<(String, Vec<u8>)>;

fn page(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    prefix: &[u8],
    after: Option<&str>,
    limit: NonZeroUsize,
) -> Result<(NamedBytes, Option<String>), Failure> {
    let mut from = prefix.to_vec();
    if let Some(after) = after {
        identifier(after)?;
        from.extend_from_slice(after.as_bytes());
        from.push(0); // Exclusive last key; identifiers never contain NUL.
    }
    let mut to = prefix.to_vec();
    let last = to
        .last_mut()
        .ok_or_else(|| Failure::new("invalid_prefix", "empty key prefix"))?;
    *last = last
        .checked_add(1)
        .ok_or_else(|| Failure::new("invalid_prefix", "unbounded key prefix"))?;
    let fetch = limit
        .checked_add(1)
        .ok_or_else(|| Failure::new("limit_overflow", "page limit is too large"))?;
    let mut rows = Vec::new();
    for shard in &at.shards {
        let page = reader
            .range_page_at(shard.shard_id, shard.head, &from, &to, fetch)
            .map_err(Failure::observer)?;
        rows.extend(page.entries);
    }
    rows.sort_unstable_by(|left, right| left.0.cmp(&right.0));
    let more = rows.len() > limit.get();
    rows.truncate(limit.get());
    let items = rows
        .into_iter()
        .map(|(key, value)| {
            let suffix = key
                .strip_prefix(prefix)
                .ok_or_else(|| Failure::new("corrupt_key", "row is outside requested prefix"))?;
            let name = std::str::from_utf8(suffix)
                .map_err(Failure::storage)?
                .to_owned();
            Ok((name, value))
        })
        .collect::<Result<NamedBytes, Failure>>()?;
    let next_after = if more {
        items.last().map(|(id, _)| id.clone())
    } else {
        None
    };
    Ok((items, next_after))
}

#[derive(Clone, Copy)]
pub struct PageLimits {
    pub rows: NonZeroUsize,
    pub bytes: NonZeroUsize,
}

fn bounded_page<T>(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    prefix: &[u8],
    after: Option<&str>,
    limits: PageLimits,
    decode: impl FnMut(String, Vec<u8>) -> Result<T, Failure>,
    serialized_len: impl FnMut(&T) -> Result<usize, Failure>,
) -> Result<(Vec<T>, Option<String>), Failure> {
    let from = exclusive_from(prefix, after)?;
    bounded_key_page(reader, at, prefix, &from, limits, decode, serialized_len)
}

/// Merge captured shard pages from a caller-validated composite key boundary.
/// The boundary must stay in the requested prefix; callers validate its parts.
pub(super) fn bounded_key_page<T>(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    prefix: &[u8],
    from: &[u8],
    limits: PageLimits,
    mut decode: impl FnMut(String, Vec<u8>) -> Result<T, Failure>,
    mut serialized_len: impl FnMut(&T) -> Result<usize, Failure>,
) -> Result<(Vec<T>, Option<String>), Failure> {
    let to = prefix_end(prefix)?;
    if !from.starts_with(prefix) || from >= to.as_slice() {
        return Err(Failure::new(
            "invalid_continuation",
            "page boundary is outside the requested prefix",
        ));
    }
    let mut candidates = Vec::with_capacity(at.shards.len());
    for shard in &at.shards {
        candidates.push((
            shard.shard_id,
            first_row(reader, shard.shard_id, shard.head, from, &to)?,
        ));
    }

    let mut entries = Vec::new();
    let mut last_id = None;
    let mut budget = WireBudget::new(limits.bytes);
    loop {
        let Some(index) = candidates
            .iter()
            .enumerate()
            .filter_map(|(index, (_, row))| row.as_ref().map(|row| (index, &row.0)))
            .min_by(|(_, left), (_, right)| left.cmp(right))
            .map(|(index, _)| index)
        else {
            return Ok((entries, None));
        };
        let (key, value) = candidates[index]
            .1
            .take()
            .ok_or_else(|| Failure::new("storage_unavailable", "page candidate disappeared"))?;
        let suffix = key
            .strip_prefix(prefix)
            .ok_or_else(|| Failure::new("corrupt_key", "row is outside requested prefix"))?;
        let id = std::str::from_utf8(suffix)
            .map_err(Failure::storage)?
            .to_owned();
        let entry = decode(id.clone(), value)?;
        if !budget.admit_len(serialized_len(&entry)?)? {
            return Ok((entries, last_id));
        }
        entries.push(entry);
        last_id = Some(id);

        let shard_id = candidates[index].0;
        let shard = at
            .shards
            .iter()
            .find(|shard| shard.shard_id == shard_id)
            .ok_or_else(|| Failure::new("incomplete_roots", "page shard disappeared"))?;
        let mut next_from = key;
        next_from.push(0);
        candidates[index].1 = first_row(reader, shard_id, shard.head, &next_from, &to)?;

        if entries.len() == limits.rows.get() {
            let next_after = if candidates.iter().any(|(_, row)| row.is_some()) {
                last_id
            } else {
                None
            };
            return Ok((entries, next_after));
        }
    }
}

#[derive(Clone, Copy)]
pub(super) struct VisibleKeyPage<'a> {
    pub access: &'a Access,
    pub collection: &'a str,
    pub prefix: &'a [u8],
    pub from: &'a [u8],
    pub limits: PageLimits,
}

/// Page only permitted document-owned rows, with a bounded visibility scan.
pub(super) fn visible_key_page<T>(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    request: VisibleKeyPage<'_>,
    decode: impl FnMut(String, Vec<u8>) -> Result<T, Failure>,
    serialized_len: impl FnMut(&T) -> Result<usize, Failure>,
) -> Result<(Vec<T>, Option<String>), Failure> {
    visible_row_page(
        reader,
        at,
        VisibleRowPage {
            prefix: request.prefix,
            from: request.from,
            limits: request.limits,
        },
        |id| {
            identifier(id)?;
            Ok(request.access.can_read_document(request.collection, id))
        },
        decode,
        serialized_len,
    )
}

#[derive(Clone, Copy)]
pub(super) struct VisibleRowPage<'a> {
    pub prefix: &'a [u8],
    pub from: &'a [u8],
    pub limits: PageLimits,
}

/// The continuation comes from an accepted row, never a hidden scan boundary.
pub(super) fn visible_row_page<T>(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    request: VisibleRowPage<'_>,
    mut visible: impl FnMut(&str) -> Result<bool, Failure>,
    mut decode: impl FnMut(String, Vec<u8>) -> Result<T, Failure>,
    mut serialized_len: impl FnMut(&T) -> Result<usize, Failure>,
) -> Result<(Vec<T>, Option<String>), Failure> {
    let to = prefix_end(request.prefix)?;
    if !request.from.starts_with(request.prefix) || request.from >= to.as_slice() {
        return Err(Failure::new(
            "invalid_continuation",
            "page boundary is outside the requested prefix",
        ));
    }
    let examined_limit = request
        .limits
        .rows
        .get()
        .checked_mul(DOCUMENT_SCAN_ROWS_PER_LIMIT)
        .and_then(|rows| rows.checked_add(1))
        .and_then(|rows| rows.checked_mul(at.shards.len()))
        .ok_or_else(|| Failure::new("limit_overflow", "visibility scan limit is too large"))?;
    let mut examined = 0;
    let mut candidates = Vec::with_capacity(at.shards.len());
    for shard in &at.shards {
        candidates.push((
            shard,
            document_candidate(
                reader,
                shard,
                request.from,
                &to,
                &mut examined,
                examined_limit,
            )?,
        ));
    }
    let mut entries = Vec::new();
    let mut last_id = None;
    let mut budget = WireBudget::new(request.limits.bytes);
    loop {
        let Some(index) = candidates
            .iter()
            .enumerate()
            .filter_map(|(index, (_, row))| row.as_ref().map(|row| (index, &row.row.0)))
            .min_by(|(_, left), (_, right)| left.cmp(right))
            .map(|(index, _)| index)
        else {
            return Ok((entries, None));
        };
        let DocumentCandidate {
            row: (key, value),
            next_from,
        } = candidates[index]
            .1
            .take()
            .ok_or_else(|| Failure::new("storage_unavailable", "page candidate disappeared"))?;
        let suffix = key
            .strip_prefix(request.prefix)
            .ok_or_else(|| Failure::new("corrupt_key", "row is outside requested prefix"))?;
        let id = std::str::from_utf8(suffix).map_err(Failure::storage)?;
        if visible(id)? {
            if entries.len() == request.limits.rows.get() {
                return Ok((entries, last_id));
            }
            let entry = decode(id.to_owned(), value)?;
            if !budget.admit_len(serialized_len(&entry)?)? {
                return Ok((entries, last_id));
            }
            last_id = Some(id.to_owned());
            entries.push(entry);
        }
        candidates[index].1 = match next_from {
            Some(from) => document_candidate(
                reader,
                candidates[index].0,
                &from,
                &to,
                &mut examined,
                examined_limit,
            )?,
            None => None,
        };
    }
}

/// Read sorted keys without fetching a successor after the requested bound.
pub(super) fn prefix_keys(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    prefix: &[u8],
    limit: NonZeroUsize,
) -> Result<Vec<Vec<u8>>, Failure> {
    let to = prefix_end(prefix)?;
    let mut candidates = Vec::with_capacity(at.shards.len());
    for shard in &at.shards {
        candidates.push((
            shard,
            first_row(reader, shard.shard_id, shard.head, prefix, &to)?,
        ));
    }
    let mut keys = Vec::new();
    while keys.len() < limit.get() {
        let Some(index) = candidates
            .iter()
            .enumerate()
            .filter_map(|(index, (_, row))| row.as_ref().map(|row| (index, &row.0)))
            .min_by(|(_, left), (_, right)| left.cmp(right))
            .map(|(index, _)| index)
        else {
            break;
        };
        let (key, _) = candidates[index].1.take().ok_or_else(|| {
            Failure::new("storage_unavailable", "retention key candidate disappeared")
        })?;
        if !key.starts_with(prefix) || keys.last() == Some(&key) {
            return Err(Failure::new(
                "corrupt_key",
                "retention key is outside its scope or duplicated",
            ));
        }
        let mut from = key.clone();
        from.push(0);
        keys.push(key);
        if keys.len() == limit.get() {
            break;
        }
        let shard = candidates[index].0;
        candidates[index].1 = first_row(reader, shard.shard_id, shard.head, &from, &to)?;
    }
    Ok(keys)
}

fn exclusive_from(prefix: &[u8], after: Option<&str>) -> Result<Vec<u8>, Failure> {
    let mut from = prefix.to_vec();
    if let Some(after) = after {
        identifier(after)?;
        from.extend_from_slice(after.as_bytes());
        from.push(0);
    }
    Ok(from)
}

fn prefix_end(prefix: &[u8]) -> Result<Vec<u8>, Failure> {
    let mut to = prefix.to_vec();
    let last = to
        .last_mut()
        .ok_or_else(|| Failure::new("invalid_prefix", "empty key prefix"))?;
    *last = last
        .checked_add(1)
        .ok_or_else(|| Failure::new("invalid_prefix", "unbounded key prefix"))?;
    Ok(to)
}

type RawRow = (Vec<u8>, Vec<u8>);

fn first_row(
    reader: &ReadOnlyDatabase,
    shard_id: usize,
    root: haematite::tree::Hash,
    from: &[u8],
    to: &[u8],
) -> Result<Option<RawRow>, Failure> {
    #[cfg(test)]
    count_range_read();
    #[cfg(test)]
    count_content_read(from);
    Ok(reader
        .range_page_at(shard_id, root, from, to, NonZeroUsize::MIN)
        .map_err(Failure::observer)?
        .entries
        .into_iter()
        .next())
}

#[cfg(test)]
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(in crate::serve) struct PageReadCounts {
    pub points: usize,
    pub ranges: usize,
}

#[cfg(test)]
thread_local! {
    static DOCUMENT_RANGES: std::cell::RefCell<Vec<(Vec<u8>, Vec<u8>)>> = const {
        std::cell::RefCell::new(Vec::new())
    };
    static CONTENT_READS: std::cell::Cell<usize> = const { std::cell::Cell::new(0) };
    static PAGE_READ_COUNTS: std::cell::Cell<PageReadCounts> = const {
        std::cell::Cell::new(PageReadCounts { points: 0, ranges: 0 })
    };
}

#[cfg(test)]
pub(super) fn take_content_reads() -> usize {
    CONTENT_READS.with(std::cell::Cell::take)
}

#[cfg(test)]
fn count_content_read(key: &[u8]) {
    if key.starts_with(b"d\0") || key.starts_with(b"e\0") || key.starts_with(b"v\0") {
        CONTENT_READS.with(|reads| reads.set(reads.get() + 1));
    }
}

#[cfg(test)]
pub(in crate::serve) fn reset_page_read_counts() {
    PAGE_READ_COUNTS.with(|counts| counts.set(PageReadCounts::default()));
}

#[cfg(test)]
pub(in crate::serve) fn page_read_counts() -> PageReadCounts {
    PAGE_READ_COUNTS.with(std::cell::Cell::get)
}

#[cfg(test)]
fn count_point_read() {
    PAGE_READ_COUNTS.with(|counts| {
        let mut current = counts.get();
        current.points += 1;
        counts.set(current);
    });
}

#[cfg(test)]
fn count_range_read() {
    PAGE_READ_COUNTS.with(|counts| {
        let mut current = counts.get();
        current.ranges += 1;
        counts.set(current);
    });
}

struct WireBudget {
    limit: usize,
    used: usize,
}

impl WireBudget {
    const fn new(limit: NonZeroUsize) -> Self {
        Self {
            limit: limit.get(),
            used: 0,
        }
    }

    #[cfg(test)]
    fn admit(&mut self, entry: &impl Serialize) -> Result<bool, Failure> {
        let bytes = serde_json::to_vec(entry).map_err(Failure::storage)?.len();
        self.admit_len(bytes)
    }

    fn admit_len(&mut self, bytes: usize) -> Result<bool, Failure> {
        let separator = usize::from(self.used != 0);
        let required = bytes
            .checked_add(separator)
            .ok_or_else(|| Failure::new("response_limit", "response byte count overflow"))?;
        if required > self.limit.saturating_sub(self.used) {
            if self.used == 0 {
                return Err(Failure::new(
                    "response_limit",
                    "one row exceeds the response byte budget",
                ));
            }
            return Ok(false);
        }
        self.used += required;
        Ok(true)
    }
}

pub(super) fn render_document(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    collection: &str,
    id: String,
    bytes: &[u8],
) -> Result<Document, Failure> {
    let raw: Box<RawValue> = serde_json::from_slice(bytes).map_err(Failure::storage)?;
    let revision = get_bytes(reader, at, &revision_key(collection, &id))?.ok_or_else(|| {
        Failure::new(
            "revision_missing",
            "document has no committed schema revision",
        )
    })?;
    Ok(Document {
        id,
        raw_json: raw.get().to_owned(),
        revision: serde_json::from_slice(&revision).map_err(Failure::storage)?,
    })
}

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

#[cfg(test)]
fn record_document_range(from: &[u8], to: &[u8]) {
    DOCUMENT_RANGES.with(|ranges| ranges.borrow_mut().push((from.to_vec(), to.to_vec())));
}

#[cfg(test)]
fn take_document_ranges() -> Vec<(Vec<u8>, Vec<u8>)> {
    DOCUMENT_RANGES.with(|ranges| std::mem::take(&mut *ranges.borrow_mut()))
}

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