haematite-cli 0.21.7

//! Bounded attribution reads at captured roots, without a live session identity.

use std::num::NonZeroUsize;

use haematite::db::ReadOnlyDatabase;

use super::{Captured, Failure, reads};
use crate::serve::contracts::attribution::{Row, key, version_prefix};
use crate::serve::protocol::Version;

pub const MAX_VERSION_ROWS: usize = 256;

/// Read one author row at the captured roots.
///
/// # Errors
/// Refuses incomplete roots, invalid request ids, unreadable stored data or a malformed attribution row.
pub fn get(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    version: Version,
    request_id: &str,
) -> Result<Option<Row>, Failure> {
    validate_capture(reader, at)?;
    let key = key(version, request_id)?;
    reads::get_bytes(reader, at, &key)?
        .map(|bytes| Row::decode(&key, &bytes))
        .transpose()
}

/// Return all rows of one version or refuse if their count exceeds the cap.
/// A refusal never returns a truncated author list.
///
/// # Errors
/// Refuses incomplete roots, unreadable or invalid rows, more than 256 rows, or duplicate or unordered request identities.
pub fn version_rows(
    reader: &ReadOnlyDatabase,
    at: &Captured,
    version: Version,
) -> Result<Vec<Row>, Failure> {
    validate_capture(reader, at)?;
    let prefix = version_prefix(version);
    let probe = NonZeroUsize::new(MAX_VERSION_ROWS + 1).ok_or_else(|| {
        Failure::new(
            "change_record_version_limit",
            "version row probe must be nonzero",
        )
    })?;
    let (rows, continuation) = reads::bounded_key_page(
        reader,
        at,
        &prefix,
        &prefix,
        reads::PageLimits {
            rows: probe,
            bytes: NonZeroUsize::MAX,
        },
        |request_id, bytes| Row::decode(&key(version, &request_id)?, &bytes),
        // The codec bounds each row; this scan uses only the row-count budget.
        |_| Ok(0),
    )?;
    if rows.len() > MAX_VERSION_ROWS || continuation.is_some() {
        return Err(Failure::new(
            "change_record_version_limit",
            "version has more than 256 attribution rows",
        ));
    }
    if rows
        .windows(2)
        .any(|pair| pair[0].request_id() >= pair[1].request_id())
    {
        return Err(Failure::new(
            "change_record_key_invalid",
            "version contains duplicate or unordered attribution keys",
        ));
    }
    Ok(rows)
}

fn validate_capture(reader: &ReadOnlyDatabase, at: &Captured) -> Result<(), Failure> {
    if at.shards.len() != reader.shard_count()
        || at
            .shards
            .iter()
            .enumerate()
            .any(|(id, shard)| id != shard.shard_id)
    {
        return Err(Failure::new(
            "incomplete_roots",
            "attribution selection must name every physical shard in order",
        ));
    }
    Ok(())
}

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