haematite-cli 0.21.3

use std::cell::Cell;
use std::collections::BTreeSet;
use std::error::Error;
use std::num::NonZeroU64;

use haematite::db::{DatabaseConfig, ReadOnlyDatabase};
use haematite::store::NodeCacheBudget;
use serde_json::value::RawValue;
use serde_json::{Value, json};

use super::{BODIES, Changes, ChangesRequest, Context, DIFFS, Fork, Moment, Settings, list};
use crate::serve::access::{self, Access, Standalone, TestSource};
use crate::serve::protocol::{
    Change, CollectionWrite, DocumentDelete, DocumentWrite, Fork as BranchFork, Mutation,
    QueryPage, Selection, Version,
};
use crate::serve::query::{Rows, Seal};
use crate::serve::storage::{Catalogue, Failure, Storage};

type TestResult = Result<(), Box<dyn Error>>;

enum Step<'a> {
    Put(&'a str, &'a str, &'a str),
    Delete(&'a str, &'a str),
}

/// A store of `shards` shards with `steps` written on main. `versions[0]` is
/// before the first step and `versions[n]` after step n.
struct Store {
    _directory: tempfile::TempDir,
    storage: Storage,
    reader: ReadOnlyDatabase,
    catalogue: Catalogue,
    versions: Vec<Version>,
}

fn main_at(version: Version) -> Selection {
    Selection {
        branch: "main".to_owned(),
        version: Some(version),
    }
}

fn mutate(
    storage: &mut Storage,
    request_id: String,
    expected: Version,
    change: Change,
) -> Result<Version, Box<dyn Error>> {
    let mutation = Mutation {
        request_id,
        branch: "main".to_owned(),
        expected,
        change,
    };
    Ok(storage.mutate(&mutation)?.version)
}

fn store(shards: usize, steps: &[Step<'_>]) -> Result<Store, Box<dyn Error>> {
    let directory = tempfile::tempdir()?;
    let mut storage = Storage::create(
        DatabaseConfig {
            data_dir: directory.path().join("database"),
            shard_count: shards,
            distributed: None,
            executor_threads: Some(1),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::KeepAll,
        },
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?;
    let head = Selection {
        branch: "main".to_owned(),
        version: None,
    };
    let mut version = storage.capture(&head)?.version;
    let mut versions = vec![version];
    let mut declared = BTreeSet::new();
    for (number, step) in steps.iter().enumerate() {
        let collection = match step {
            Step::Put(collection, _, _) | Step::Delete(collection, _) => *collection,
        };
        if declared.insert(collection) {
            let declare = Change::Collection(CollectionWrite {
                collection: collection.to_owned(),
                schema: None,
            });
            version = mutate(
                &mut storage,
                format!("declare/{collection}"),
                version,
                declare,
            )?;
        }
        let change = change(step)?;
        version = mutate(&mut storage, format!("step/{number}"), version, change)?;
        versions.push(version);
    }
    let reader = storage.reader()?;
    let catalogue = storage.catalogue();
    Ok(Store {
        _directory: directory,
        storage,
        reader,
        catalogue,
        versions,
    })
}

fn change(step: &Step<'_>) -> Result<Change, serde_json::Error> {
    Ok(match step {
        Step::Put(collection, id, body) => Change::Put(DocumentWrite {
            collection: (*collection).to_owned(),
            document_id: (*id).to_owned(),
            document: RawValue::from_string((*body).to_owned())?,
        }),
        Step::Delete(collection, id) => Change::Delete(DocumentDelete {
            collection: (*collection).to_owned(),
            document_id: (*id).to_owned(),
        }),
    })
}

fn head_of(branch: &str) -> Selection {
    Selection {
        branch: branch.to_owned(),
        version: None,
    }
}

/// `change` written on `branch` at its head, after the store was made.
fn write(
    store: &mut Store,
    branch: &str,
    id: &str,
    change: Change,
) -> Result<Version, Box<dyn Error>> {
    let mutation = Mutation {
        request_id: id.to_owned(),
        branch: branch.to_owned(),
        expected: store.storage.capture(&head_of(branch))?.version,
        change,
    };
    Ok(store.storage.mutate(&mutation)?.version)
}

fn version(store: &Store, step: usize) -> Result<Version, Box<dyn Error>> {
    Ok(*store.versions.get(step).ok_or("no such step")?)
}

fn asked(from: Version, to: Version, rows: u64) -> Result<ChangesRequest, Box<dyn Error>> {
    Ok(ChangesRequest {
        from: Moment::At(main_at(from)),
        to: Moment::At(main_at(to)),
        rows: Rows::Origins,
        page: QueryPage {
            rows: NonZeroU64::new(rows).ok_or("zero rows")?,
            bytes: NonZeroU64::MIN.saturating_add((1 << 20) - 1),
        },
        continuation: None,
    })
}

/// Settings with a records-read limit of `scan` for each page.
fn limited(scan: u64) -> Result<Settings, Box<dyn Error>> {
    Ok(Settings {
        scan_records: NonZeroU64::new(scan).ok_or("zero scan")?,
        ..Settings::default()
    })
}

fn ask(
    store: &Store,
    access: &Access,
    settings: &Settings,
    seal: &Seal,
    request: &ChangesRequest,
) -> Result<Value, Failure> {
    let text = ask_text(store, access, settings, seal, request)?;
    serde_json::from_str(text.get()).map_err(Failure::storage)
}

/// The answer as the text the service writes (NB1).
fn ask_text(
    store: &Store,
    access: &Access,
    settings: &Settings,
    seal: &Seal,
    request: &ChangesRequest,
) -> Result<Box<serde_json::value::RawValue>, Failure> {
    let context = Context {
        access,
        reader: &store.reader,
        catalogue: &store.catalogue,
        settings,
        seal,
    };
    list(&context, request, None)
}

/// Every page from `from` to `to`, `rows` at a time, following `next`.
fn follow(
    store: &Store,
    access: &Access,
    settings: &Settings,
    (from, to, rows): (Version, Version, u64),
) -> Result<Vec<Value>, Box<dyn Error>> {
    let seal = Seal::new()?;
    let mut request = asked(from, to, rows)?;
    let mut pages = Vec::new();
    loop {
        let answer = ask(store, access, settings, &seal, &request)?;
        let next = answer["next"].as_str().map(str::to_owned);
        pages.push(answer);
        let Some(next) = next else {
            return Ok(pages);
        };
        request = asked(from, to, rows)?;
        request.continuation = Some(next);
    }
}

/// Every row of `pages`, as `kind collection/id`, none twice.
fn named(pages: &[Value]) -> Result<Vec<String>, Box<dyn Error>> {
    let mut names = Vec::new();
    let mut seen = BTreeSet::new();
    for page in pages {
        for row in page["rows"].as_array().ok_or("no rows array")? {
            let origin = &row["origin"];
            let collection = origin["collection"].as_str().ok_or("no collection")?;
            let id = origin["id"].as_str().ok_or("no id")?;
            if !seen.insert(format!("{collection}/{id}")) {
                return Err(format!("{collection}/{id} was given twice").into());
            }
            let kind = row["kind"].as_str().ok_or("no kind")?;
            names.push(format!("{kind} {collection}/{id}"));
        }
    }
    Ok(names)
}

fn everyone() -> Result<Access, Failure> {
    access::request(&Standalone)
}

fn denying(documents: &[(&str, &str)]) -> Result<Access, Failure> {
    access::request(&TestSource::denying(documents))
}

#[test]
fn one_write_between_two_versions_gives_that_change_and_reads_no_body() -> TestResult {
    let store = store(
        2,
        &[
            Step::Put("patients", "p1", r#"{"n":1}"#),
            Step::Put("patients", "p2", r#"{"n":1}"#),
            Step::Put("patients", "p1", r#"{"n":2}"#),
        ],
    )?;
    let request = asked(version(&store, 2)?, version(&store, 3)?, 10)?;
    BODIES.with(|count| count.set(0));
    let (access, seal) = (everyone()?, Seal::new()?);
    let answer = ask(&store, &access, &Settings::default(), &seal, &request)?;
    let expected = json!([{
        "kind": "changed",
        "origin": {"collection": "patients", "id": "p1"},
        "before": version(&store, 1)?,
        "after": version(&store, 3)?,
    }]);
    assert_eq!(answer["rows"], expected);
    let to = json!({"branch": "main", "version": version(&store, 3)?});
    assert_eq!(answer["to"], to);
    assert_eq!(answer["changes_read"], 1);
    assert_eq!(BODIES.with(Cell::get), 0, "a body was read");
    Ok(())
}

#[test]
fn added_changed_and_removed_are_each_named() -> TestResult {
    let store = store(
        2,
        &[
            Step::Put("alpha", "a1", "1"),
            Step::Put("alpha", "a2", "2"),
            Step::Put("alpha", "a3", "3"),
            Step::Put("alpha", "a1", "4"),
            Step::Delete("alpha", "a2"),
        ],
    )?;
    let span = (version(&store, 2)?, version(&store, 5)?, 10);
    let pages = follow(&store, &everyone()?, &Settings::default(), span)?;
    let expected = ["changed alpha/a1", "removed alpha/a2", "added alpha/a3"];
    assert_eq!(named(&pages)?, expected);
    Ok(())
}

/// Written out of order over two shards and two collections.
const SHUFFLED: [Step<'static>; 6] = [
    Step::Put("beta", "b2", "1"),
    Step::Put("alpha", "a3", "1"),
    Step::Put("beta", "b1", "1"),
    Step::Put("alpha", "a1", "1"),
    Step::Put("beta", "b3", "1"),
    Step::Put("alpha", "a2", "1"),
];

const ORDERED: [&str; 6] = [
    "added alpha/a1",
    "added alpha/a2",
    "added alpha/a3",
    "added beta/b1",
    "added beta/b2",
    "added beta/b3",
];

#[test]
fn rows_are_in_key_order_and_pages_cut_by_rows_or_limit_equal_one() -> TestResult {
    let store = store(2, &SHUFFLED)?;
    let (first, last) = (version(&store, 0)?, version(&store, 6)?);
    let (access, defaults) = (everyone()?, Settings::default());
    let one = follow(&store, &access, &defaults, (first, last, 100))?;
    assert_eq!(one.len(), 1);
    assert_eq!(named(&one)?, ORDERED);
    let by_rows = follow(&store, &access, &defaults, (first, last, 2))?;
    assert_eq!(named(&by_rows)?, ORDERED);
    assert!(by_rows.len() >= 3, "{} pages", by_rows.len());
    let by_limit = follow(&store, &access, &limited(2)?, (first, last, 100))?;
    assert_eq!(named(&by_limit)?, ORDERED);
    assert!(by_limit.len() >= 3, "{} pages", by_limit.len());
    Ok(())
}

#[test]
fn a_denied_change_is_absent_and_a_denied_run_is_crossed_with_empty_pages() -> TestResult {
    let store = store(2, &SHUFFLED)?;
    let span = (version(&store, 0)?, version(&store, 6)?, 100);
    let one_denied = denying(&[("alpha", "a2")])?;
    let one = follow(&store, &one_denied, &Settings::default(), span)?;
    let mut expected = ORDERED.to_vec();
    expected.retain(|name| *name != "added alpha/a2");
    assert_eq!(named(&one)?, expected);
    assert_eq!(one.last().ok_or("no page")?["changes_read"], 5);
    let run = [("alpha", "a1"), ("alpha", "a2"), ("alpha", "a3")];
    let pages = follow(&store, &denying(&run)?, &limited(2)?, span)?;
    let readable = ["added beta/b1", "added beta/b2", "added beta/b3"];
    assert_eq!(named(&pages)?, readable);
    let empty = pages
        .iter()
        .filter(|page| page["rows"] == json!([]) && page["next"].is_string())
        .count();
    assert!(empty > 0, "no page was cut inside the denied run");
    assert_eq!(pages.last().ok_or("no page")?["changes_read"], 3);
    Ok(())
}

#[test]
fn values_when_asked_are_each_body_at_its_own_moment() -> TestResult {
    let store = store(
        2,
        &[
            Step::Put("patients", "p1", r#"{"n":1}"#),
            Step::Put("patients", "p1", r#"{"n":2}"#),
        ],
    )?;
    let mut request = asked(version(&store, 1)?, version(&store, 2)?, 10)?;
    request.rows = Rows::Values;
    BODIES.with(|count| count.set(0));
    let (access, seal) = (everyone()?, Seal::new()?);
    let answer = ask(&store, &access, &Settings::default(), &seal, &request)?;
    let row = &answer["rows"][0];
    assert_eq!(row["value_before"], json!({"n": 1}));
    assert_eq!(row["value_after"], json!({"n": 2}));
    assert_eq!(BODIES.with(Cell::get), 2);
    Ok(())
}

#[test]
fn moments_not_kept_equal_versions_and_tokens_for_others_are_answered_by_name() -> TestResult {
    let store = store(
        2,
        &[Step::Put("alpha", "a1", "1"), Step::Put("alpha", "a2", "1")],
    )?;
    let (first, last) = (version(&store, 0)?, version(&store, 2)?);
    let seal = Seal::new()?;
    let access = everyone()?;
    let settings = Settings::default();
    let gone = Version {
        sequence: last.sequence + 100,
        ..last
    };
    let error = ask(&store, &access, &settings, &seal, &asked(first, gone, 10)?)
        .err()
        .ok_or("a version never written was answered")?;
    assert_eq!(error.code, "version_missing");
    DIFFS.with(|count| count.set(0));
    let same = ask(&store, &access, &settings, &seal, &asked(last, last, 10)?)?;
    assert_eq!(same["rows"], json!([]));
    assert!(same["next"].is_null());
    assert_eq!(DIFFS.with(Cell::get), 0, "equal versions were read");
    let page = ask(&store, &access, &settings, &seal, &asked(first, last, 1)?)?;
    let next = page["next"].as_str().ok_or("no next")?.to_owned();
    let mut other_moment = asked(first, version(&store, 1)?, 1)?;
    other_moment.continuation = Some(next.clone());
    let error = ask(&store, &access, &settings, &seal, &other_moment)
        .err()
        .ok_or("a token for other moments was answered")?;
    assert_eq!(error.code, "query_continuation_moment");
    let mut other_form = asked(first, last, 1)?;
    other_form.rows = Rows::Values;
    other_form.continuation = Some(next);
    let error = ask(&store, &access, &settings, &seal, &other_form)
        .err()
        .ok_or("a token for the other row form was answered")?;
    assert_eq!(error.code, "query_continuation_allowances");
    Ok(())
}

/// A body's numbers come back with their text as written, a large integer
/// whole, and the body equal as a value. NB1: read in the answer's text, as
/// the service writes it, never through a decoded value.
#[test]
fn a_body_keeps_every_number_as_it_was_written() -> TestResult {
    let body = r#"{"b":1.0,"a":1e2,"n":18446744073709551616}"#;
    let store = store(
        2,
        &[
            Step::Put("patients", "p1", "{}"),
            Step::Put("patients", "p1", body),
        ],
    )?;
    let mut request = asked(version(&store, 1)?, version(&store, 2)?, 10)?;
    request.rows = Rows::Values;
    let (access, seal) = (everyone()?, Seal::new()?);
    let answered = ask_text(&store, &access, &Settings::default(), &seal, &request)?;
    let text = answered.get();
    for written in [r#""b":1.0"#, r#""a":1e2"#, r#""n":18446744073709551616"#] {
        assert!(text.contains(written), "{written} is not in {text}");
    }
    let answer: Value = serde_json::from_str(text)?;
    let stored: Value = serde_json::from_str(body)?;
    assert_eq!(answer["rows"][0]["value_after"], stored);
    assert_eq!(answer["rows"][0]["value_before"], json!({}));
    Ok(())
}

/// `from` later than `to` is the reverse: the same records, added and
/// removed exchanged, and each row's before and after exchanged.
#[test]
fn from_later_than_to_gives_the_reverse() -> TestResult {
    let store = store(
        2,
        &[
            Step::Put("alpha", "a1", "1"),
            Step::Put("alpha", "a2", "2"),
            Step::Put("alpha", "a3", "3"),
            Step::Put("alpha", "a1", "4"),
            Step::Delete("alpha", "a2"),
        ],
    )?;
    let (earlier, later) = (version(&store, 2)?, version(&store, 5)?);
    let (access, defaults) = (everyone()?, Settings::default());
    let forward = follow(&store, &access, &defaults, (earlier, later, 10))?;
    let back = follow(&store, &access, &defaults, (later, earlier, 10))?;
    let ahead = ["changed alpha/a1", "removed alpha/a2", "added alpha/a3"];
    assert_eq!(named(&forward)?, ahead);
    let undone = ["changed alpha/a1", "added alpha/a2", "removed alpha/a3"];
    assert_eq!(named(&back)?, undone);
    let forward = forward.first().ok_or("no page")?["rows"].clone();
    let back = back.first().ok_or("no page")?["rows"].clone();
    let (forward, back) = (
        forward.as_array().ok_or("no rows array")?,
        back.as_array().ok_or("no rows array")?,
    );
    assert_eq!(forward.len(), back.len());
    for (one, other) in forward.iter().zip(back) {
        assert_eq!(one["origin"], other["origin"]);
        assert_eq!(one["before"], other["after"]);
        assert_eq!(one["after"], other["before"]);
    }
    Ok(())
}

/// Two moments whose shards pair by position but not by id are refused.
#[test]
fn moments_whose_shards_differ_are_refused_by_name() -> TestResult {
    let store = store(2, &[Step::Put("alpha", "a1", "1")])?;
    let from = store.catalogue.capture(&main_at(version(&store, 0)?))?;
    let mut to = store.catalogue.capture(&main_at(version(&store, 1)?))?;
    to.shards.reverse();
    let error = Changes::new(&store.reader, &from, &to, None, &Settings::default())
        .err()
        .ok_or("shards paired by position alone")?;
    assert_eq!(error.code, "changes_shards_differ");
    Ok(())
}

/// A request from `from` to `to`, `rows` at a time.
fn between(from: Moment, to: Moment, rows: u64) -> Result<ChangesRequest, Box<dyn Error>> {
    Ok(ChangesRequest {
        from,
        to,
        rows: Rows::Origins,
        page: QueryPage {
            rows: NonZeroU64::new(rows).ok_or("zero rows")?,
            bytes: NonZeroU64::MIN.saturating_add((1 << 20) - 1),
        },
        continuation: None,
    })
}

/// From `from` on main to main's head.
fn live(from: Version, rows: u64) -> Result<ChangesRequest, Box<dyn Error>> {
    between(Moment::At(main_at(from)), Moment::At(head_of("main")), rows)
}

/// The version an answer names for `end`, `from` or `to`.
fn version_of(answer: &Value, end: &str) -> Result<Version, serde_json::Error> {
    serde_json::from_value(answer[end]["version"].clone())
}

#[test]
fn a_live_view_asks_from_its_last_to_and_is_given_every_change_once() -> TestResult {
    let mut store = store(
        2,
        &[Step::Put("alpha", "a1", "1"), Step::Put("alpha", "a2", "1")],
    )?;
    let (access, seal, settings) = (everyone()?, Seal::new()?, Settings::default());
    let request = live(version(&store, 1)?, 100)?;
    let first = ask(&store, &access, &settings, &seal, &request)?;
    let seen = version_of(&first, "to")?;
    assert_eq!(seen, version(&store, 2)?);
    write(
        &mut store,
        "main",
        "later",
        change(&Step::Put("alpha", "a3", "1"))?,
    )?;
    let second = ask(&store, &access, &settings, &seal, &live(seen, 100)?)?;
    let head = store.storage.capture(&head_of("main"))?.version;
    assert_eq!(version_of(&second, "to")?, head);
    assert_eq!(
        named(&[first, second])?,
        ["added alpha/a2", "added alpha/a3"]
    );
    Ok(())
}

#[test]
fn a_head_that_moves_between_pages_does_not_change_later_pages() -> TestResult {
    let mut store = store(2, &SHUFFLED)?;
    let (access, seal, settings) = (everyone()?, Seal::new()?, Settings::default());
    let start = version(&store, 0)?;
    let first = ask(&store, &access, &settings, &seal, &live(start, 1)?)?;
    let at = version_of(&first, "to")?;
    write(
        &mut store,
        "main",
        "later",
        change(&Step::Put("alpha", "a9", "1"))?,
    )?;
    let mut pages = vec![first];
    while let Some(next) = pages.last().and_then(|page| page["next"].as_str()) {
        let mut later = live(start, 1)?;
        later.continuation = Some(next.to_owned());
        let answer = ask(&store, &access, &settings, &seal, &later)?;
        assert_eq!(version_of(&answer, "to")?, at);
        pages.push(answer);
    }
    assert_eq!(named(&pages)?, ORDERED);
    Ok(())
}

#[test]
fn replay_by_history_then_one_changes_list_a_step_gives_every_mutation_once() -> TestResult {
    let store = store(
        2,
        &[
            Step::Put("alpha", "a1", "1"),
            Step::Put("alpha", "a2", "1"),
            Step::Put("beta", "b1", "1"),
            Step::Put("alpha", "a1", "2"),
            Step::Delete("alpha", "a2"),
        ],
    )?;
    let (access, seal, settings) = (everyone()?, Seal::new()?, Settings::default());
    let rows = store.catalogue.history("main", None, 100)?;
    let mut versions: Vec<Version> = rows.iter().map(|row| row.version).collect();
    let head = store.storage.capture(&head_of("main"))?.version;
    if versions.last() != Some(&head) {
        versions.push(head);
    }
    let (mut given, mut quiet) = (Vec::new(), 0);
    for pair in versions.windows(2) {
        let [from, to] = pair else {
            return Err("a window is not a pair".into());
        };
        let answer = ask(&store, &access, &settings, &seal, &asked(*from, *to, 100)?)?;
        let step = named(std::slice::from_ref(&answer))?;
        if step.is_empty() {
            assert!(answer["next"].is_null(), "{answer}");
            quiet += 1;
        }
        given.extend(step);
    }
    let every = [
        "added alpha/a1",
        "added alpha/a2",
        "added beta/b1",
        "changed alpha/a1",
        "removed alpha/a2",
    ];
    assert_eq!(given, every);
    // The two collections' declarations changed no document.
    assert!(quiet >= 2, "{quiet} quiet steps");
    Ok(())
}

/// A store whose branch `b` forked from main after `alpha/a1`, then wrote
/// `alpha/a2`, while main went on to write `alpha/a3`.
fn forked() -> Result<Store, Box<dyn Error>> {
    let mut store = store(2, &[Step::Put("alpha", "a1", "1")])?;
    let from = store.storage.capture(&head_of("main"))?.version;
    let fork = Change::Fork(BranchFork {
        name: "b".to_owned(),
        from,
    });
    write(&mut store, "main", "fork-b", fork)?;
    write(
        &mut store,
        "b",
        "on-b",
        change(&Step::Put("alpha", "a2", "1"))?,
    )?;
    write(
        &mut store,
        "main",
        "on-main",
        change(&Step::Put("alpha", "a3", "1"))?,
    )?;
    Ok(store)
}

fn fork_of(branch: &str) -> Moment {
    Moment::Fork(Fork {
        fork: branch.to_owned(),
    })
}

#[test]
fn from_a_fork_gives_only_what_the_branch_wrote_after_it_forked() -> TestResult {
    let store = forked()?;
    let (access, seal, settings) = (everyone()?, Seal::new()?, Settings::default());
    let request = between(fork_of("b"), Moment::At(head_of("b")), 100)?;
    let answer = ask(&store, &access, &settings, &seal, &request)?;
    assert_eq!(named(std::slice::from_ref(&answer))?, ["added alpha/a2"]);
    Ok(())
}

#[test]
fn a_missing_fork_a_fork_as_to_and_a_fork_onto_another_branch_are_refused() -> TestResult {
    let store = forked()?;
    let (access, seal, settings) = (everyone()?, Seal::new()?, Settings::default());
    let cases = [
        (
            between(fork_of("main"), Moment::At(head_of("main")), 10)?,
            "changes_fork_missing",
        ),
        (
            between(Moment::At(main_at(version(&store, 1)?)), fork_of("b"), 10)?,
            "changes_fork_as_to",
        ),
        (
            between(fork_of("b"), Moment::At(head_of("main")), 10)?,
            "changes_branch_mismatch",
        ),
    ];
    for (request, code) in cases {
        let error = ask(&store, &access, &settings, &seal, &request)
            .err()
            .ok_or_else(|| format!("{code} was answered"))?;
        assert_eq!(error.code, code);
    }
    Ok(())
}

#[test]
fn nonzero_budget_readback() -> Result<(), Box<dyn std::error::Error>> {
    let maximum = u64::try_from(usize::MAX)?;
    let cases = [
        (7, None, 7),
        (7, Some(3), 3),
        (3, Some(7), 3),
        (7, Some(7), 7),
        (1, Some(1), 1),
        (maximum, Some(usize::MAX), usize::MAX),
    ];
    for (request, transport, expected) in cases {
        let request = std::num::NonZeroU64::new(request).ok_or("zero request")?;
        let transport = transport
            .map(|value| std::num::NonZeroUsize::new(value).ok_or("zero transport"))
            .transpose()?;
        let actual = super::budget(
            std::hint::black_box(request),
            std::hint::black_box(transport),
        )?;
        assert_eq!(actual.get(), expected);
    }
    Ok(())
}

#[test]
fn nonzero_changes_batch_readback() -> TestResult {
    let store = store(1, &[])?;
    for (remaining, expected) in [(0, 1), (1, 1), (17, 17), (u64::MAX, 64)] {
        let changes = Changes {
            reader: &store.reader,
            shards: Vec::new(),
            node_budget: std::num::NonZeroUsize::MIN,
            limit: std::hint::black_box(remaining),
            taken: 0,
            last: None,
        };
        assert_eq!(changes.batch().get(), expected);
    }
    Ok(())
}

#[test]
fn nonzero_changes_response_read_stops() -> TestResult {
    let store = store(
        1,
        &[
            Step::Put("records", "a", "{}"),
            Step::Put("records", "b", "{}"),
        ],
    )?;
    let access = everyone()?;
    let settings = Settings::default();
    let seal = Seal::new()?;
    let request = asked(version(&store, 0)?, version(&store, 2)?, 10)?;
    let context = Context {
        access: &access,
        reader: &store.reader,
        catalogue: &store.catalogue,
        settings: &settings,
        seal: &seal,
    };
    for given in [None, Some(65_536), Some(usize::MAX)] {
        let limit = given
            .map(|value| std::num::NonZeroUsize::new(value).ok_or("zero byte limit"))
            .transpose()?;
        let text = list(&context, &request, std::hint::black_box(limit))?;
        let answer: Value = serde_json::from_str(text.get())?;
        assert_eq!(answer["rows"].as_array().ok_or("rows absent")?.len(), 2);
        if let Some(limit) = limit {
            assert!(text.get().len() <= limit.get());
        }
    }
    let refused = list(&context, &request, Some(std::num::NonZeroUsize::MIN))
        .err()
        .ok_or("minimum returned an empty success")?;
    assert_eq!(refused.code, "response_limit");
    let needed: usize = refused
        .message
        .split_whitespace()
        .find_map(|word| word.parse().ok())
        .ok_or("the refusal names no size")?;
    let enough = std::num::NonZeroUsize::new(needed + 4096).ok_or("zero byte limit")?;
    let text = list(&context, &request, Some(enough))?;
    assert!(text.get().len() <= enough.get());
    Ok(())
}

/// STOR-003 MD4: a moment of a generation that has ended, with no row,
/// refuses the whole list by that name, at either end; nothing partial.
#[test]
fn a_moment_of_an_ended_generation_refuses_by_that_name() -> TestResult {
    let store = store(2, &[Step::Put("alpha", "a1", "1")])?;
    let last = version(&store, 1)?;
    let ended = Version {
        generation: last.generation - 1,
        ..last
    };
    let seal = Seal::new()?;
    let access = everyone()?;
    let settings = Settings::default();
    for (from, to) in [(ended, last), (last, ended)] {
        let error = ask(&store, &access, &settings, &seal, &asked(from, to, 10)?)
            .err()
            .ok_or("a moment of an ended generation was answered")?;
        assert_eq!(error.code, "version_generation_ended", "{}", error.message);
    }
    Ok(())
}

const fn fixture_long_wait_interval() -> std::time::Duration {
    std::time::Duration::from_secs(1)
}