haematite-cli 0.21.7

use std::collections::BTreeMap;
use std::error::Error;
use std::fs;
use std::path::Path;

use haematite::db::{DataDirLock, Database, DatabaseConfig};
use haematite::store::NodeCacheBudget;
use serde_json::{json, value::RawValue};

use super::{CandidateAction, MARKER, Storage, holder_name, load_marker, reads, revision_key};
use crate::serve::protocol::{
    Cardinality, Change, CollectionWrite, DocumentWrite, Edge, Fork, Mutation, Node, Relation,
    ReviewPage, Selection, Version,
};
use crate::serve::storage::{
    BranchDatabase, BranchRefStore, BranchRegistry, CANONICAL_REFS_DIR, Captured, MergeChoice,
    MergeChoose, MergeOwner, MergePin, branch_name, document_key, edges, history, receipt_key,
    review,
};

type TestResult<T = ()> = Result<T, Box<dyn Error>>;
type Rows = BTreeMap<Vec<u8>, Option<Vec<u8>>>;

struct Fixture {
    storage: Storage,
    directory: tempfile::TempDir,
    serial: usize,
}

struct FrozenFixture {
    directory: tempfile::TempDir,
    serial: usize,
    branches: Vec<Captured>,
}

impl FrozenFixture {
    fn new(fixture: Fixture) -> TestResult<Self> {
        let mut branches = vec![fixture.storage.capture(&Selection {
            branch: "main".to_owned(),
            version: None,
        })?];
        if fixture.storage.refs.get(&branch_name("feature")).is_some() {
            branches.push(fixture.storage.capture(&Selection {
                branch: "feature".to_owned(),
                version: None,
            })?);
        }
        let Fixture {
            storage,
            directory,
            serial,
        } = fixture;
        drop(storage);
        assert!(!DataDirLock::is_held(&directory.path().join("database"))?);
        Ok(Self {
            directory,
            serial,
            branches,
        })
    }

    fn copy(&self) -> TestResult<Fixture> {
        let source = self.directory.path().join("database");
        assert!(!DataDirLock::is_held(&source)?);
        let directory = tempfile::tempdir()?;
        let path = directory.path().join("database");
        let opened = (|| -> TestResult<Storage> {
            copy_fixture_directory(&source, &path)?;
            // The owning engine open preserves the seed's retention policy.
            let database = Database::open_path_writer(
                &path,
                haematite::RuntimeOptions {
                    long_wait_interval: std::time::Duration::from_secs(30),
                    reclaim_start: haematite::ReclaimStart::AtOpen,
                },
            )?;
            let writer = BranchDatabase::new(database, NodeCacheBudget::Unlimited)?;
            assert!(writer.writer_lock().is_for(&path)?);
            assert!(!writer.writer_lock().is_for(&source)?);
            let mut refs =
                BranchRefStore::open_writing(path.join(CANONICAL_REFS_DIR), writer.writer_lock())?;
            let kept = history::default_kept()?;
            refs.keep_versions(kept);
            let storage = Storage::finish_open(writer, refs, BranchRegistry::new(), kept)?;
            assert!(!DataDirLock::is_held(&source)?);
            for expected in &self.branches {
                let current = storage.capture(&Selection {
                    branch: expected.branch.clone(),
                    version: None,
                })?;
                assert_eq!(current.version, expected.version);
                assert_eq!(current.shards, expected.shards);
            }
            Ok(storage)
        })();
        match opened {
            Ok(storage) => Ok(Fixture {
                storage,
                directory,
                serial: self.serial,
            }),
            Err(error) => match directory.close() {
                Ok(()) => Err(error),
                Err(cleanup) => Err(format!(
                    "fixture copy/open failed: {error}; fixture cleanup failed: {cleanup}"
                )
                .into()),
            },
        }
    }

    fn close(self) -> TestResult {
        self.directory.close()?;
        Ok(())
    }
}

fn copy_fixture_directory(source: &Path, target: &Path) -> TestResult {
    let metadata = fs::symlink_metadata(source)?;
    if !metadata.is_dir() {
        return Err(format!("unsupported fixture directory: {}", source.display()).into());
    }
    fs::create_dir(target)?;
    fs::set_permissions(target, metadata.permissions())?;
    for entry in fs::read_dir(source)? {
        let entry = entry?;
        let kind = entry.file_type()?;
        let destination = target.join(entry.file_name());
        if kind.is_dir() {
            copy_fixture_directory(&entry.path(), &destination)?;
        } else if kind.is_file() {
            fs::copy(entry.path(), destination)?;
        } else {
            return Err(
                format!("unsupported fixture file kind: {}", entry.path().display()).into(),
            );
        }
    }
    Ok(())
}

impl Fixture {
    fn empty() -> TestResult<Self> {
        let directory = tempfile::tempdir()?;
        let storage = Storage::create(
            DatabaseConfig {
                data_dir: directory.path().join("database"),
                shard_count: 2,
                distributed: None,
                executor_threads: Some(2),
                node_cache_budget: Some(NodeCacheBudget::Unlimited),
                retention: haematite::Retention::KeepAll,
            },
            NodeCacheBudget::Unlimited,
            fixture_long_wait_interval(),
        )?;
        Ok(Self {
            storage,
            directory,
            serial: 0,
        })
    }

    fn seed(&mut self) -> TestResult {
        self.apply(
            "main",
            Change::Collection(CollectionWrite {
                collection: "records".to_owned(),
                schema: None,
            }),
        )?;
        self.put_many("main", &["kept", "conflict", "removed", "end"], 0)?;
        self.relation("main", "links")?;
        self.edge("main", "removed", "end")?;
        let from = self.live("main")?;
        self.apply(
            "main",
            Change::Fork(Fork {
                name: "feature".to_owned(),
                from,
            }),
        )
    }

    fn prepare(&mut self) -> TestResult<[String; 2]> {
        self.put("feature", "kept", 1)?;
        self.put("feature", "conflict", 1)?;
        let target_request_id = self.put_identified("main", "conflict", 2)?;
        self.put("feature", "created", 7)?;
        self.edge("feature", "created", "end")?;
        let source_request_id = self.relation_identified("feature", "new_links")?;
        self.remove_stored_endpoint()?;
        for id in ["ready", "missing"] {
            let source = MergePin {
                branch: "feature".to_owned(),
                version: self.live("feature")?,
            };
            let target = MergePin {
                branch: "main".to_owned(),
                version: self.live("main")?,
            };
            self.storage.merge_open(id, &source, &target)?;
        }
        for id in ["conflict", "removed"] {
            self.choose("ready", document(id), MergeChoice::TakeSource)?;
        }
        self.choose(
            "ready",
            MergeOwner::Edge {
                relation: "links".to_owned(),
                from: node("removed"),
                to: node("end"),
            },
            MergeChoice::RemoveEdge,
        )?;
        Ok([target_request_id, source_request_id])
    }

    fn live(&self, branch: &str) -> TestResult<Version> {
        Ok(self
            .storage
            .capture(&Selection {
                branch: branch.to_owned(),
                version: None,
            })?
            .version)
    }

    fn request(&mut self, branch: &str, change: Change) -> TestResult<Mutation> {
        self.serial += 1;
        let expected = self.live(branch)?;
        Ok(Mutation {
            request_id: format!("candidate-{}", self.serial),
            branch: branch.to_owned(),
            expected,
            change,
        })
    }

    fn apply(&mut self, branch: &str, change: Change) -> TestResult {
        self.apply_identified(branch, change).map(|_| ())
    }

    fn apply_identified(&mut self, branch: &str, change: Change) -> TestResult<String> {
        let request = self.request(branch, change)?;
        self.storage.mutate(&request)?;
        Ok(request.request_id)
    }

    fn put_many(&mut self, branch: &str, ids: &[&str], value: u64) -> TestResult {
        let documents = ids
            .iter()
            .map(|id| {
                Ok(crate::serve::protocol::ImportedDocument {
                    document_id: (*id).to_owned(),
                    document: RawValue::from_string(json!({"n": value}).to_string())?,
                })
            })
            .collect::<Result<Vec<_>, serde_json::Error>>()?;
        let request = self.request(
            branch,
            Change::PutMany(crate::serve::protocol::DocumentImport {
                collection: "records".to_owned(),
                documents,
            }),
        )?;
        self.storage
            .mutate_bounded(&request, std::num::NonZeroU32::MIN)?;
        Ok(())
    }

    fn put(&mut self, branch: &str, id: &str, value: u64) -> TestResult {
        self.put_identified(branch, id, value).map(|_| ())
    }

    fn put_identified(&mut self, branch: &str, id: &str, value: u64) -> TestResult<String> {
        self.apply_identified(
            branch,
            Change::Put(DocumentWrite {
                collection: "records".to_owned(),
                document_id: id.to_owned(),
                document: RawValue::from_string(json!({"n": value}).to_string())?,
            }),
        )
    }

    fn relation(&mut self, branch: &str, name: &str) -> TestResult {
        self.relation_identified(branch, name).map(|_| ())
    }

    fn relation_identified(&mut self, branch: &str, name: &str) -> TestResult<String> {
        self.apply_identified(
            branch,
            Change::RelationPut(Relation {
                relation: name.to_owned(),
                from_collection: "records".to_owned(),
                to_collection: "records".to_owned(),
                cardinality: Cardinality::ManyToMany,
            }),
        )
    }

    fn edge(&mut self, branch: &str, from: &str, to: &str) -> TestResult {
        self.apply(
            branch,
            Change::EdgePut(Edge {
                relation: "links".to_owned(),
                from: node(from),
                to: node(to),
                properties: json!({}),
            }),
        )
    }

    fn choose(&mut self, id: &str, owner: MergeOwner, choice: MergeChoice) -> TestResult {
        let plan = self.storage.open_merge_plan(id)?;
        let inputs = self.storage.choice_inputs(&plan)?;
        let key = owner.key()?;
        let token = self
            .storage
            .choice_values(&inputs, &key)?
            .token(&inputs.pins, &key)?;
        self.storage.merge_choose(&MergeChoose {
            merge_id: id.to_owned(),
            owner,
            token,
            choice,
        })?;
        Ok(())
    }

    fn remove_stored_endpoint(&mut self) -> TestResult {
        use haematite::branch::{CommitDurability, CommitRequest, current_timestamp, open};
        let name = branch_name("feature");
        let record = self
            .storage
            .refs
            .get(&name)
            .ok_or("missing source")?
            .clone();
        self.storage.retain_current("feature", &record)?;
        let branch = open(&name, &self.storage.refs, &self.storage.registry)?;
        // Preserve an unchanged incident edge to exercise explicit removal admission.
        let key = document_key("records", "removed");
        branch.delete(self.storage.writer.shard_for(&key), key)?;
        self.storage.writer.commit(
            &branch,
            &self.storage.registry,
            CommitRequest {
                durability: CommitDurability::Durable {
                    refs: &mut self.storage.refs,
                },
                extra_parents: &[],
                timestamp: current_timestamp(),
            },
        )?;
        self.storage.publish_record(&name)?;
        Ok(())
    }

    fn candidate(&self, id: &str) -> TestResult<Captured> {
        Ok(self
            .storage
            .candidate_selection(&holder_name(id, "candidate"))
            .ok_or("missing candidate")?)
    }

    fn bytes(&self, id: &str, key: &[u8]) -> TestResult<Option<Vec<u8>>> {
        Ok(reads::get_bytes(
            &self.storage.reader,
            &self.candidate(id)?,
            key,
        )?)
    }

    fn visible(&self) -> TestResult<Vec<String>> {
        Ok(self
            .storage
            .visible
            .branches(None, 100)?
            .iter()
            .map(|at| format!("{}:{:?}:{:?}", at.branch, at.version, at.shards))
            .collect())
    }

    fn private(&self, id: &str, visible: &[String]) -> TestResult {
        assert_eq!(self.visible()?, visible);
        let name = holder_name(id, "candidate");
        assert!(self.storage.visible.history(&name, None, 10).is_err());
        assert!(
            self.storage
                .capture(&Selection {
                    branch: name.clone(),
                    version: None
                })
                .is_err()
        );
        assert!(
            review::page(
                &self.storage.reader,
                &self.storage.visible,
                &ReviewPage {
                    branch: name,
                    after: None,
                    limit: std::num::NonZeroU32::MIN,
                },
                std::num::NonZeroUsize::MIN
            )
            .is_err()
        );
        Ok(())
    }

    fn close(self) -> TestResult {
        let Self {
            storage,
            directory,
            serial: _,
        } = self;
        drop(storage);
        directory.close()?;
        Ok(())
    }

    fn reopen(self) -> TestResult<Self> {
        let Self {
            storage,
            directory,
            serial,
        } = self;
        drop(storage);
        let storage = Storage::open(
            &directory.path().join("database"),
            NodeCacheBudget::Unlimited,
            fixture_long_wait_interval(),
        )?;
        Ok(Self {
            storage,
            directory,
            serial,
        })
    }
}

fn counting_seed() -> TestResult<Fixture> {
    let mut fixture = Fixture::empty()?;
    fixture.apply(
        "main",
        Change::Collection(CollectionWrite {
            collection: "records".to_owned(),
            schema: None,
        }),
    )?;
    fixture.relation("main", "links")?;
    for id in ["base", "left", "right"] {
        fixture.put("main", id, 0)?;
    }
    Ok(fixture)
}

fn relationship_counting_seed() -> TestResult<Fixture> {
    let mut fixture = Fixture::empty()?;
    fixture.apply(
        "main",
        Change::Collection(CollectionWrite {
            collection: "records".to_owned(),
            schema: None,
        }),
    )?;
    fixture.relation("main", "links")?;
    fixture.put_many("main", &["base", "left", "right"], 0)?;
    Ok(fixture)
}

fn fork_counting(mut fixture: Fixture, with_edge: bool) -> TestResult<Fixture> {
    if with_edge {
        fixture.edge("main", "left", "right")?;
    }
    let from = fixture.live("main")?;
    fixture.apply(
        "main",
        Change::Fork(Fork {
            name: "feature".to_owned(),
            from,
        }),
    )?;
    Ok(fixture)
}

fn counting_fixture(with_edge: bool) -> TestResult<Fixture> {
    fork_counting(counting_seed()?, with_edge)
}

fn open_counting(fixture: &mut Fixture) -> TestResult {
    let source = MergePin {
        branch: "feature".to_owned(),
        version: fixture.live("feature")?,
    };
    let target = MergePin {
        branch: "main".to_owned(),
        version: fixture.live("main")?,
    };
    fixture.storage.merge_open("counts", &source, &target)?;
    Ok(())
}

fn assert_counted(fixture: &Fixture, at: &Captured) -> TestResult {
    let owner = crate::serve::access::request(&crate::serve::access::Standalone)?;
    let answer = serde_json::to_value(Storage::audit_counts(&owner, &fixture.storage.reader, at)?)?;
    assert_eq!(answer["result"], "compared");
    assert!(
        answer["buckets"]
            .as_array()
            .ok_or("no audit buckets")?
            .iter()
            .all(|row| row["result"] == "agree"),
        "{answer}"
    );
    Ok(())
}

fn finish_counting(fixture: &mut Fixture) -> TestResult<Captured> {
    for _ in 0..256 {
        let page = fixture.storage.merge_candidate_page("counts")?;
        let at = fixture.candidate("counts")?;
        assert_counted(fixture, &at)?;
        if page.end {
            return Ok(at);
        }
    }
    Err("candidate exhausted bounded page count".into())
}

fn kept(fixture: &Fixture, at: &Captured, key: &[u8]) -> TestResult<u64> {
    let bytes = reads::get_bytes(&fixture.storage.reader, at, key)?.ok_or("no kept count")?;
    Ok(std::str::from_utf8(&bytes)?.parse()?)
}

#[test]
fn candidate_counts_distinct_and_same_key_additions_once() -> TestResult {
    let seed = FrozenFixture::new(counting_seed()?)?;
    for same in [false, true] {
        let mut fixture = fork_counting(seed.copy()?, false)?;
        fixture.put("main", "target-added", 1)?;
        fixture.put(
            "feature",
            if same { "target-added" } else { "source-added" },
            if same { 2 } else { 1 },
        )?;
        open_counting(&mut fixture)?;
        if same {
            fixture.choose("counts", document("target-added"), MergeChoice::TakeSource)?;
        }
        let at = finish_counting(&mut fixture)?;
        assert_eq!(
            kept(&fixture, &at, b"k\0d\0records")?,
            if same { 4 } else { 5 }
        );
        assert_eq!(
            reads::test_support::count_keys(&fixture.storage.reader, &at, b"d\0", 2)?,
            if same { 4 } else { 5 }
        );
        fixture.close()?;
    }
    seed.close()
}

#[test]
fn candidate_counts_delete_edit_each_choice() -> TestResult {
    let seed = FrozenFixture::new(counting_seed()?)?;
    for choice in [MergeChoice::TakeSource, MergeChoice::TakeTarget] {
        let mut fixture = fork_counting(seed.copy()?, false)?;
        fixture.apply(
            "feature",
            Change::Delete(crate::serve::protocol::DocumentDelete {
                collection: "records".to_owned(),
                document_id: "left".to_owned(),
            }),
        )?;
        fixture.put("main", "left", 1)?;
        open_counting(&mut fixture)?;
        fixture.choose("counts", document("left"), choice)?;
        let at = finish_counting(&mut fixture)?;
        let expected = if choice == MergeChoice::TakeSource {
            2
        } else {
            3
        };
        assert_eq!(kept(&fixture, &at, b"k\0d\0records")?, expected);
        fixture.close()?;
    }
    seed.close()
}

#[test]
fn candidate_relationship_counts_and_declared_zero_match_content() -> TestResult {
    let mut fixture = counting_fixture(false)?;
    fixture.edge("main", "base", "left")?;
    fixture.edge("feature", "base", "right")?;
    fixture.relation("feature", "empty")?;
    open_counting(&mut fixture)?;
    let at = finish_counting(&mut fixture)?;
    assert_eq!(kept(&fixture, &at, b"k\0e\0links")?, 2);
    assert_eq!(kept(&fixture, &at, b"k\0e\0empty")?, 0);
    assert_eq!(
        reads::test_support::count_keys(&fixture.storage.reader, &at, b"e\0", 5)?,
        2
    );
    Ok(())
}

#[test]
fn relationship_same_key_and_delete_edit_choices_count_once() -> TestResult {
    let seed = FrozenFixture::new(relationship_counting_seed()?)?;
    let mut fixture = fork_counting(seed.copy()?, false)?;
    fixture.edge("main", "base", "left")?;
    fixture.edge("feature", "base", "left")?;
    open_counting(&mut fixture)?;
    let at = finish_counting(&mut fixture)?;
    assert_eq!(kept(&fixture, &at, b"k\0e\0links")?, 1);
    fixture.close()?;
    for (choice, expected) in [(MergeChoice::TakeSource, 0), (MergeChoice::TakeTarget, 1)] {
        let mut fixture = fork_counting(seed.copy()?, true)?;
        fixture.apply(
            "feature",
            Change::EdgeDelete(crate::serve::protocol::EdgeDelete {
                relation: "links".to_owned(),
                from: node("left"),
                to: node("right"),
            }),
        )?;
        fixture.apply(
            "main",
            Change::EdgePut(Edge {
                relation: "links".to_owned(),
                from: node("left"),
                to: node("right"),
                properties: json!({"changed": true}),
            }),
        )?;
        open_counting(&mut fixture)?;
        fixture.choose(
            "counts",
            MergeOwner::Edge {
                relation: "links".to_owned(),
                from: node("left"),
                to: node("right"),
            },
            choice,
        )?;
        let at = finish_counting(&mut fixture)?;
        assert_eq!(kept(&fixture, &at, b"k\0e\0links")?, expected);
        fixture.close()?;
    }
    seed.close()
}

#[test]
fn undeclared_relationship_kind_keeps_its_candidate_counter() -> TestResult {
    let mut fixture = counting_fixture(false)?;
    fixture.apply(
        "feature",
        Change::EdgePut(Edge {
            relation: "unregistered".to_owned(),
            from: node("base"),
            to: node("left"),
            properties: json!({}),
        }),
    )?;
    open_counting(&mut fixture)?;
    let at = finish_counting(&mut fixture)?;
    assert_eq!(kept(&fixture, &at, b"k\0e\0unregistered")?, 1);
    Ok(())
}

#[test]
fn new_collection_count_survives_declaration_and_document_page_order() -> TestResult {
    let mut fixture = counting_fixture(false)?;
    fixture.apply(
        "feature",
        Change::Collection(CollectionWrite {
            collection: "new".to_owned(),
            schema: None,
        }),
    )?;
    fixture.apply(
        "feature",
        Change::Put(DocumentWrite {
            collection: "new".to_owned(),
            document_id: "created".to_owned(),
            document: RawValue::from_string("{}".to_owned())?,
        }),
    )?;
    open_counting(&mut fixture)?;
    let at = finish_counting(&mut fixture)?;
    assert_eq!(kept(&fixture, &at, b"k\0d\0new")?, 1);
    Ok(())
}

#[test]
fn incident_relationship_visited_twice_counts_removal_once() -> TestResult {
    use haematite::branch::{CommitDurability, CommitRequest, current_timestamp, open};
    let mut fixture = counting_fixture(true)?;
    let name = branch_name("feature");
    let record = fixture.storage.refs.get(&name).ok_or("no source")?.clone();
    fixture.storage.retain_current("feature", &record)?;
    let branch = open(&name, &fixture.storage.refs, &fixture.storage.registry)?;
    for id in ["left", "right"] {
        for key in [document_key("records", id), revision_key("records", id)] {
            branch.delete(fixture.storage.writer.shard_for(&key), key)?;
        }
    }
    fixture.storage.writer.commit(
        &branch,
        &fixture.storage.registry,
        CommitRequest {
            durability: CommitDurability::Durable {
                refs: &mut fixture.storage.refs,
            },
            extra_parents: &[],
            timestamp: current_timestamp(),
        },
    )?;
    fixture.storage.publish_record(&name)?;
    open_counting(&mut fixture)?;
    fixture.choose("counts", document("left"), MergeChoice::TakeSource)?;
    fixture.choose("counts", document("right"), MergeChoice::TakeSource)?;
    fixture.choose(
        "counts",
        MergeOwner::Edge {
            relation: "links".to_owned(),
            from: node("left"),
            to: node("right"),
        },
        MergeChoice::RemoveEdge,
    )?;
    let at = finish_counting(&mut fixture)?;
    assert_eq!(kept(&fixture, &at, b"k\0e\0links")?, 0);
    assert_eq!(kept(&fixture, &at, b"k\0d\0records")?, 1);
    Ok(())
}

#[test]
fn empty_incident_scans_use_page_steps_instead_of_document_steps() -> TestResult {
    let additions = 130_usize;
    let mut fixture = counting_fixture(false)?;
    let documents = (0..additions)
        .map(|id| {
            Ok(crate::serve::protocol::ImportedDocument {
                document_id: format!("source-{id}"),
                document: RawValue::from_string("{}".to_owned())?,
            })
        })
        .collect::<TestResult<Vec<_>>>()?;
    let expected = fixture.live("feature")?;
    fixture.storage.mutate_bounded(
        &Mutation {
            request_id: "many".to_owned(),
            branch: "feature".to_owned(),
            expected,
            change: Change::PutMany(crate::serve::protocol::DocumentImport {
                collection: "records".to_owned(),
                documents,
            }),
        },
        std::num::NonZeroU32::new(64).ok_or("no page bound")?,
    )?;
    open_counting(&mut fixture)?;
    fixture.storage.merge_candidate_page("counts")?;
    let shards = fixture.candidate("counts")?.shards.len();
    // The fork and import add provenance, two receipts and one count change.
    // Each document adds content and revision keys; shard tails need partial pages.
    let changed_rows = additions
        .checked_mul(2)
        .and_then(|rows| rows.checked_add(4))
        .ok_or("difference count overflow")?;
    let max_pages = changed_rows
        .div_ceil(review::dangling::SCAN_ALLOWANCE)
        .checked_add(shards)
        .ok_or("page count overflow")?;
    let max_pages = u64::try_from(max_pages)?;
    for _ in 0..=max_pages {
        let page = fixture.storage.merge_candidate_page("counts")?;
        assert!(
            page.page <= max_pages,
            "empty incident phases exceeded the raw-page bound: {} > {max_pages}",
            page.page
        );
        let at = fixture.candidate("counts")?;
        assert_eq!(
            at.version.sequence, page.page,
            "a page must publish one durable version"
        );
        if page.end {
            assert_counted(&fixture, &at)?;
            assert_eq!(kept(&fixture, &at, b"k\0d\0records")?, 133);
            return Ok(());
        }
    }
    Err("candidate did not finish within the raw-page bound".into())
}

#[test]
fn candidate_pages_resume_reopen_and_restart_from_target_counts() -> TestResult {
    let mut fixture = counting_fixture(false)?;
    let documents = (0..130)
        .map(|id| {
            Ok(crate::serve::protocol::ImportedDocument {
                document_id: format!("source-{id}"),
                document: RawValue::from_string("{}".to_owned())?,
            })
        })
        .collect::<TestResult<Vec<_>>>()?;
    let expected = fixture.live("feature")?;
    fixture.storage.mutate_bounded(
        &Mutation {
            request_id: "many".to_owned(),
            branch: "feature".to_owned(),
            expected,
            change: Change::PutMany(crate::serve::protocol::DocumentImport {
                collection: "records".to_owned(),
                documents,
            }),
        },
        std::num::NonZeroU32::new(64).ok_or("no page bound")?,
    )?;
    open_counting(&mut fixture)?;
    fixture.storage.merge_candidate_page("counts")?;
    let page = fixture.storage.merge_candidate_page("counts")?;
    assert!(!page.end);
    let before = fixture.candidate("counts")?;
    assert_eq!(
        before.version.sequence, page.page,
        "a page must publish one durable version"
    );
    assert_counted(&fixture, &before)?;
    let count = kept(&fixture, &before, b"k\0d\0records")?;
    fixture = fixture.reopen()?;
    let confirmed = fixture.storage.merge_candidate_page("counts")?;
    assert!(matches!(confirmed.action, CandidateAction::Confirmed));
    assert_eq!(fixture.candidate("counts")?.version, before.version);
    assert_eq!(
        kept(&fixture, &fixture.candidate("counts")?, b"k\0d\0records")?,
        count
    );
    let at = finish_counting(&mut fixture)?;
    assert_eq!(kept(&fixture, &at, b"k\0d\0records")?, 133);
    fixture.storage.merge_candidate_restart("counts")?;
    fixture.storage.merge_candidate_page("counts")?;
    let restarted = fixture.candidate("counts")?;
    assert_eq!(kept(&fixture, &restarted, b"k\0d\0records")?, 3);
    assert_counted(&fixture, &restarted)?;
    fixture.storage.merge_abandon("counts")?;
    let target = fixture.storage.capture(&Selection {
        branch: "main".to_owned(),
        version: None,
    })?;
    assert_eq!(kept(&fixture, &target, b"k\0d\0records")?, 3);
    assert_counted(&fixture, &target)?;
    Ok(())
}

#[test]
fn vector_refusal_preserves_candidate_vector_count() -> TestResult {
    let mut fixture = counting_fixture(false)?;
    fixture.apply(
        "feature",
        Change::VectorPut(crate::serve::protocol::VectorWrite {
            collection: "records".to_owned(),
            field: "embedding".to_owned(),
            document_id: "left".to_owned(),
            vector: vec![1.0, 2.0],
        }),
    )?;
    open_counting(&mut fixture)?;
    fixture.storage.merge_candidate_page("counts")?;
    let key = b"k\0v\0records\0embedding";
    let target = fixture.storage.capture(&Selection {
        branch: "main".to_owned(),
        version: None,
    })?;
    let original = reads::get_bytes(&fixture.storage.reader, &target, key)?;
    let mut refused = false;
    for _ in 0..256 {
        match fixture.storage.merge_candidate_page("counts") {
            Ok(page) => assert!(!page.end),
            Err(error) => {
                assert_eq!(error.code, "candidate_vector_rebuild_required");
                refused = true;
                break;
            }
        }
    }
    assert!(refused);
    assert_eq!(
        reads::get_bytes(&fixture.storage.reader, &fixture.candidate("counts")?, key)?,
        original
    );
    Ok(())
}

fn clear_counts(fixture: &mut Fixture, branch_name_text: &str) -> TestResult {
    use haematite::branch::{CommitDurability, CommitRequest, current_timestamp, open};
    let name = branch_name(branch_name_text);
    let record = fixture.storage.refs.get(&name).ok_or("no target")?.clone();
    fixture.storage.retain_current(branch_name_text, &record)?;
    let branch = open(&name, &fixture.storage.refs, &fixture.storage.registry)?;
    for key in [b"k\0ready".as_slice(), b"k\0d\0records", b"k\0e\0links"] {
        branch.delete(fixture.storage.writer.shard_for(key), key)?;
    }
    fixture.storage.writer.commit(
        &branch,
        &fixture.storage.registry,
        CommitRequest {
            durability: CommitDurability::Durable {
                refs: &mut fixture.storage.refs,
            },
            extra_parents: &[],
            timestamp: current_timestamp(),
        },
    )?;
    fixture.storage.publish_record(&name)?;
    Ok(())
}

#[test]
fn unready_target_does_not_acquire_source_counts_or_readiness() -> TestResult {
    let mut fixture = counting_fixture(false)?;
    fixture.put("feature", "created", 1)?;
    clear_counts(&mut fixture, "main")?;
    open_counting(&mut fixture)?;
    let mut at = None;
    for _ in 0..256 {
        let page = fixture.storage.merge_candidate_page("counts")?;
        let captured = fixture.candidate("counts")?;
        for shard in &captured.shards {
            let rows = fixture.storage.reader.range_page_at(
                shard.shard_id,
                shard.head,
                b"k\0",
                b"k\x01",
                std::num::NonZeroUsize::MIN,
            )?;
            assert!(rows.entries.is_empty(), "candidate acquired count metadata");
        }
        if page.end {
            at = Some(captured);
            break;
        }
    }
    let at = at.ok_or("candidate did not end")?;
    let owner = crate::serve::access::request(&crate::serve::access::Standalone)?;
    let answer =
        serde_json::to_value(Storage::audit_counts(&owner, &fixture.storage.reader, &at)?)?;
    assert_eq!(answer["reason"], "counts_not_ready");
    assert_eq!(
        reads::test_support::count_keys(&fixture.storage.reader, &at, b"d\0", 2)?,
        4
    );
    Ok(())
}

#[test]
fn count_rows_are_not_coverage_choices_or_materialised_content() -> TestResult {
    let mut fixture = counting_fixture(false)?;
    fixture.put("feature", "created", 1)?;
    open_counting(&mut fixture)?;
    fixture.storage.merge_candidate_page("counts")?;
    let plan = fixture.storage.open_merge_plan("counts")?;
    let inputs = fixture.storage.choice_inputs(&plan)?;
    let service = fixture.storage.merge_service_selection()?;
    let candidate = fixture.candidate("counts")?;
    let rows = super::materialise(
        &fixture.storage,
        "counts",
        &inputs,
        &service,
        &candidate,
        &[b"k\0ready".to_vec(), b"k\0d\0records".to_vec()],
    )?;
    assert!(
        rows.is_empty(),
        "count rows were materialised from a merge input"
    );
    let mut position = None;
    for _ in 0..256 {
        let step = fixture.storage.coverage_step(
            "counts",
            position,
            std::num::NonZeroUsize::MIN.saturating_add(63),
        )?;
        assert!(step.keys.iter().all(|key| !key.starts_with(b"k\0")));
        position = step.next;
        if position.is_none() {
            break;
        }
    }
    assert!(position.is_none());
    let choices = fixture.storage.merge_choices(
        "counts",
        None,
        std::num::NonZeroUsize::MIN.saturating_add(63),
    )?;
    assert!(choices.entries.is_empty());
    assert!(
        serde_json::from_value::<MergeOwner>(json!({
            "kind":"count", "collection":"records",
        }))
        .is_err()
    );
    Ok(())
}

#[test]
fn counted_target_ignores_source_readiness() -> TestResult {
    let mut fixture = counting_fixture(false)?;
    fixture.put("feature", "created", 1)?;
    clear_counts(&mut fixture, "feature")?;
    open_counting(&mut fixture)?;
    let at = finish_counting(&mut fixture)?;
    assert_eq!(kept(&fixture, &at, b"k\0d\0records")?, 4);
    Ok(())
}

fn node(id: &str) -> Node {
    Node {
        collection: "records".to_owned(),
        document_id: id.to_owned(),
    }
}

fn document(id: &str) -> MergeOwner {
    MergeOwner::Document {
        collection: "records".to_owned(),
        document_id: id.to_owned(),
    }
}

fn expected_rows(fixture: &Fixture, request_ids: &[String; 2]) -> TestResult<Rows> {
    let plan = fixture.storage.open_merge_plan("ready")?;
    let inputs = fixture.storage.choice_inputs(&plan)?;
    let mut rows = BTreeMap::new();
    for id in ["kept", "conflict", "created"] {
        for key in [document_key("records", id), revision_key("records", id)] {
            let value = reads::get_bytes(&fixture.storage.reader, &inputs.source, &key)?;
            assert!(value.is_some());
            rows.insert(key, value);
        }
    }
    rows.insert(document_key("records", "removed"), None);
    rows.insert(revision_key("records", "removed"), None);
    for (from, removed) in [("removed", true), ("created", false)] {
        let both: [Vec<u8>; 2] = edges::keys(&node(from), "links", &node("end"))?.into();
        for key in both {
            let value = if removed {
                None
            } else {
                reads::get_bytes(&fixture.storage.reader, &inputs.source, &key)?
            };
            assert_eq!(value.is_none(), removed);
            rows.insert(key, value);
        }
    }
    let key = b"n\0new_links".to_vec();
    let value = reads::get_bytes(&fixture.storage.reader, &inputs.source, &key)?;
    assert!(value.is_some());
    rows.insert(key, value);
    let target_receipt = receipt_key("main", inputs.target.version.generation, &request_ids[0]);
    let value = reads::get_bytes(&fixture.storage.reader, &inputs.target, &target_receipt)?;
    assert!(value.is_some());
    rows.insert(target_receipt, value);
    let source_receipt = receipt_key("feature", inputs.source.version.generation, &request_ids[1]);
    assert!(reads::get_bytes(&fixture.storage.reader, &inputs.source, &source_receipt)?.is_some());
    rows.insert(source_receipt, None);
    let control = b"b\0fork".to_vec();
    assert!(reads::get_bytes(&fixture.storage.reader, &inputs.source, &control)?.is_some());
    rows.insert(
        control.clone(),
        reads::get_bytes(&fixture.storage.reader, &inputs.target, &control)?,
    );
    Ok(rows)
}

fn snapshot(fixture: &Fixture, expected: &Rows) -> TestResult<Rows> {
    expected
        .keys()
        .map(|key| Ok((key.clone(), fixture.bytes("ready", key)?)))
        .collect()
}

fn missing_page(fixture: &mut Fixture, visible: &[String]) -> TestResult {
    let provisioned = fixture.storage.merge_candidate_page("missing")?;
    assert!(matches!(provisioned.action, CandidateAction::Provisioned));
    assert_eq!(provisioned.page, 0);
    fixture.private("missing", visible)?;
    let mut refused = false;
    for _ in 0..128 {
        let before = fixture.candidate("missing")?.version;
        let marker = fixture.bytes("missing", MARKER)?;
        match fixture.storage.merge_candidate_page("missing") {
            Ok(page) => assert!(matches!(page.action, CandidateAction::Page) && !page.end),
            Err(error) => {
                assert_eq!(error.code, "choice_missing");
                let missing: super::MissingChoice = serde_json::from_str(&error.message)?;
                assert!(!missing.token.is_empty());
                assert_eq!(fixture.candidate("missing")?.version, before);
                assert_eq!(fixture.bytes("missing", MARKER)?, marker);
                refused = true;
            }
        }
        fixture.private("missing", visible)?;
        if refused {
            break;
        }
    }
    assert!(refused);
    Ok(())
}

fn finish_pages(
    fixture: &mut Fixture,
    visible: &[String],
    expected: &Rows,
    mut page: u64,
) -> TestResult {
    let mut settled = Rows::new();
    let mut ended = false;
    for _ in 0..128 {
        let advance = fixture.storage.merge_candidate_page("ready")?;
        assert!(matches!(advance.action, CandidateAction::Page));
        assert_eq!(advance.page, page + 1);
        page = advance.page;
        for (key, value) in expected {
            let actual = fixture.bytes("ready", key)?;
            if settled.contains_key(key) {
                assert_eq!(&actual, value);
            }
            if &actual == value {
                settled.insert(key.clone(), actual);
            }
        }
        fixture.private("ready", visible)?;
        if advance.end {
            ended = true;
            break;
        }
    }
    assert!(ended);
    assert_eq!(&settled, expected);
    let marker = fixture.bytes("ready", MARKER)?;
    let end = fixture.storage.merge_candidate_page("ready")?;
    assert!(matches!(end.action, CandidateAction::End));
    assert_eq!(end.page, page);
    assert_eq!(fixture.bytes("ready", MARKER)?, marker);
    Ok(())
}

#[derive(Debug, PartialEq, Eq)]
struct MergeControl {
    plan: Vec<u8>,
    journal: Vec<u8>,
    inputs: Vec<String>,
    service: Version,
}

fn merge_control(fixture: &Fixture) -> TestResult<MergeControl> {
    let plan = fixture.storage.open_merge_plan("ready")?;
    let journal = fixture.storage.merge_choices(
        "ready",
        None,
        std::num::NonZeroUsize::MIN.saturating_add(63),
    )?;
    assert!(journal.next_after.is_none());
    Ok(MergeControl {
        plan: serde_json::to_vec(&plan)?,
        journal: serde_json::to_vec(&journal)?,
        inputs: ["origin", "source", "target"]
            .iter()
            .map(|slot| {
                format!(
                    "{:?}",
                    fixture.storage.refs.get(&holder_name("ready", slot))
                )
            })
            .collect(),
        service: fixture.storage.merge_service_selection()?.version,
    })
}

fn restart_candidate(fixture: &mut Fixture, visible: &[String]) -> TestResult {
    let before = merge_control(fixture)?;
    let name = holder_name("ready", "candidate");
    assert!(fixture.storage.merge_candidate_restart("ready")?);
    assert!(fixture.storage.candidate_selection(&name).is_none());
    assert!(!fixture.storage.confirmed_candidates.contains_key(&name));
    assert_eq!(merge_control(fixture)?, before);
    fixture.private("ready", visible)?;
    assert!(!fixture.storage.merge_candidate_restart("ready")?);
    assert_eq!(merge_control(fixture)?, before);
    fixture.private("ready", visible)?;
    let provisioned = fixture.storage.merge_candidate_page("ready")?;
    assert!(matches!(provisioned.action, CandidateAction::Provisioned));
    assert_eq!(provisioned.page, 0);
    assert!(!provisioned.end);
    let at = fixture.candidate("ready")?;
    let marker = load_marker(&fixture.storage, &at)?;
    let journal = fixture
        .storage
        .merge_choices("ready", None, std::num::NonZeroUsize::MIN)?;
    assert_eq!(marker.revision, journal.revision);
    assert_eq!(marker.page, 0);
    assert!(!marker.end);
    assert!(marker.position.is_none());
    let plan = fixture.storage.open_merge_plan("ready")?;
    assert_eq!(marker.plan, plan);
    let target = fixture.storage.choice_inputs(&plan)?.target;
    let key = document_key("records", "kept");
    assert_eq!(
        fixture.bytes("ready", &key)?,
        reads::get_bytes(&fixture.storage.reader, &target, &key)?
    );
    assert_eq!(merge_control(fixture)?, before);
    fixture.private("ready", visible)?;
    Ok(())
}

#[test]
fn bounded_candidate_pages_keep_private_rows_and_resume_exactly() -> TestResult {
    let mut fixture = Fixture::empty()?;
    fixture.seed()?;
    let ids = (0..=super::coverage::PAGE_OWNER_LIMIT)
        .map(|index| format!("paged-{index:03}"))
        .collect::<Vec<_>>();
    let documents = ids
        .iter()
        .map(|id| {
            Ok(crate::serve::protocol::ImportedDocument {
                document_id: id.clone(),
                document: RawValue::from_string("{}".to_owned())?,
            })
        })
        .collect::<Result<Vec<_>, serde_json::Error>>()?;
    fixture.storage.mutate_bounded(
        &Mutation {
            request_id: "paged-source".to_owned(),
            branch: "feature".to_owned(),
            expected: fixture.live("feature")?,
            change: Change::PutMany(crate::serve::protocol::DocumentImport {
                collection: "records".to_owned(),
                documents,
            }),
        },
        std::num::NonZeroU32::new(64).ok_or("zero import page bound")?,
    )?;
    let request_ids = fixture.prepare()?;
    let visible = fixture.visible()?;
    assert_eq!(visible.len(), 2);
    let mut expected = expected_rows(&fixture, &request_ids)?;
    let plan = fixture.storage.open_merge_plan("ready")?;
    let source = fixture.storage.choice_inputs(&plan)?.source;
    for id in &ids {
        for key in [document_key("records", id), revision_key("records", id)] {
            let bytes = reads::get_bytes(&fixture.storage.reader, &source, &key)?
                .ok_or("paged source row missing")?;
            expected.insert(key, Some(bytes));
        }
    }
    missing_page(&mut fixture, &visible)?;
    let provisioned = fixture.storage.merge_candidate_page("ready")?;
    assert!(matches!(provisioned.action, CandidateAction::Provisioned));
    assert_eq!(provisioned.page, 0);
    fixture.private("ready", &visible)?;
    let first = fixture.storage.merge_candidate_page("ready")?;
    assert!(matches!(first.action, CandidateAction::Page));
    assert_eq!(first.page, 1);
    assert!(!first.end);
    fixture.private("ready", &visible)?;
    let marker = fixture.bytes("ready", MARKER)?;
    let before_restart = snapshot(&fixture, &expected)?;
    fixture = fixture.reopen()?;
    let confirmed = fixture.storage.merge_candidate_page("ready")?;
    assert!(matches!(confirmed.action, CandidateAction::Confirmed));
    assert_eq!(confirmed.page, first.page);
    assert_eq!(fixture.bytes("ready", MARKER)?, marker);
    assert_eq!(snapshot(&fixture, &expected)?, before_restart);
    fixture.private("ready", &visible)?;
    finish_pages(&mut fixture, &visible, &expected, first.page)?;
    let marker = fixture.bytes("ready", MARKER)?;
    let at = fixture.candidate("ready")?.version;
    fixture.choose("ready", document("created"), MergeChoice::TakeTarget)?;
    let error = fixture
        .storage
        .merge_candidate_page("ready")
        .err()
        .ok_or("journal move accepted")?;
    assert_eq!(error.code, "choice_journal_moved");
    assert_eq!(fixture.candidate("ready")?.version, at);
    assert_eq!(fixture.bytes("ready", MARKER)?, marker);
    fixture.private("ready", &visible)?;
    restart_candidate(&mut fixture, &visible)?;
    fixture.storage.merge_abandon("ready")?;
    assert!(
        fixture
            .storage
            .candidate_selection(&holder_name("ready", "candidate"))
            .is_none()
    );
    assert!(
        !fixture
            .storage
            .confirmed_candidates
            .contains_key(&holder_name("ready", "candidate"))
    );
    fixture.private("ready", &visible)?;
    Ok(())
}

fn declaration_fixture() -> TestResult<Fixture> {
    let mut fixture = Fixture::empty()?;
    fixture.apply(
        "main",
        Change::Collection(CollectionWrite {
            collection: "records".to_owned(),
            schema: None,
        }),
    )?;
    fixture.apply(
        "main",
        Change::Fork(Fork {
            name: "feature".to_owned(),
            from: fixture.live("main")?,
        }),
    )?;
    Ok(fixture)
}

fn embedded_declaration(model: &str) -> serde_json::Value {
    json!({"kind":"dense", "source":"body", "model": {
        "name":model, "revision":"v1", "manifest_sha256":"a".repeat(64)
    }, "dimensions":2, "distance":"cosine"})
}

fn declare_field(fixture: &mut Fixture, branch: &str, model: &str) -> TestResult {
    fixture.apply(
        branch,
        serde_json::from_value(json!({"kind":"vector.declare", "value": {
            "collection":"records", "field":"embedding", "declaration":embedded_declaration(model)
        }}))?,
    )
}

fn declaration_materialisation(fixture: &Fixture, keys: &[Vec<u8>]) -> TestResult<Rows> {
    let plan = fixture.storage.open_merge_plan("counts")?;
    let inputs = fixture.storage.choice_inputs(&plan)?;
    let service = fixture.storage.merge_service_selection()?;
    Ok(super::materialise(
        &fixture.storage,
        "counts",
        &inputs,
        &service,
        &fixture.candidate("counts")?,
        keys,
    )?)
}

fn finish_declaration_candidate(fixture: &mut Fixture) -> TestResult<(Captured, usize)> {
    let mut commits = 0;
    for _ in 0..16 {
        let page = fixture.storage.merge_candidate_page("counts")?;
        if matches!(
            page.action,
            CandidateAction::Provisioned | CandidateAction::Page
        ) {
            commits += 1;
        }
        if page.end {
            return Ok((fixture.candidate("counts")?, commits));
        }
    }
    Err("candidate exhausted the fixture's bounded page count".into())
}

fn assert_declaration_candidate(fixture: &Fixture, at: &Captured, model: &str) -> TestResult {
    let bytes = reads::get_bytes(&fixture.storage.reader, at, b"w\0records\0embedding")?
        .ok_or("candidate declaration missing")?;
    assert_eq!(
        serde_json::from_slice::<serde_json::Value>(&bytes)?,
        embedded_declaration(model)
    );
    assert!(reads::get_bytes(&fixture.storage.reader, at, b"k\0v\0records\0embedding")?.is_none());
    assert_eq!(
        reads::test_support::count_keys(&fixture.storage.reader, at, b"v\0", 3)?,
        0
    );
    assert_counted(fixture, at)?;
    Ok(())
}

#[test]
fn source_declaration_reaches_candidate_end_with_ordinary_commit_count() -> TestResult {
    let seed = FrozenFixture::new(declaration_fixture()?)?;
    let mut control = seed.copy()?;
    control.apply(
        "feature",
        Change::Collection(CollectionWrite {
            collection: "records".to_owned(),
            schema: None,
        }),
    )?;
    open_counting(&mut control)?;
    let (_, ordinary_commits) = finish_declaration_candidate(&mut control)?;
    control.close()?;
    let mut fixture = seed.copy()?;
    declare_field(&mut fixture, "feature", "local-model")?;
    let target = fixture.storage.capture(&Selection {
        branch: "main".to_owned(),
        version: None,
    })?;
    open_counting(&mut fixture)?;
    let (candidate, declaration_commits) = finish_declaration_candidate(&mut fixture)?;
    assert_eq!(ordinary_commits, 2);
    assert_eq!(declaration_commits, ordinary_commits);
    assert_declaration_candidate(&fixture, &candidate, "local-model")?;
    let owner = crate::serve::access::request(&crate::serve::access::Standalone)?;
    let listing =
        crate::serve::storage::vectors::fields(&owner, &fixture.storage.reader, &candidate)?;
    assert!(listing.fields.is_empty());
    let embedded = serde_json::to_value(listing.embedded_fields)?;
    assert_eq!(embedded[0]["model"]["name"], "local-model");
    assert_eq!(embedded[0]["status"], "embed_not_configured");
    assert!(embedded[0].get("count").is_none());
    assert_eq!(fixture.live("main")?, target.version);
    assert!(
        reads::get_bytes(&fixture.storage.reader, &target, b"w\0records\0embedding")?.is_none()
    );
    fixture.close()?;
    seed.close()
}

#[test]
fn equal_declarations_complete_without_staging_the_field_again() -> TestResult {
    let mut fixture = declaration_fixture()?;
    declare_field(&mut fixture, "feature", "local-model")?;
    declare_field(&mut fixture, "main", "local-model")?;
    let target = fixture.live("main")?;
    open_counting(&mut fixture)?;
    fixture.storage.merge_candidate_page("counts")?;
    let rows = declaration_materialisation(&fixture, &[b"w\0records\0embedding".to_vec()])?;
    assert!(rows.is_empty(), "equal declaration was staged again");
    let (candidate, remaining_commits) = finish_declaration_candidate(&mut fixture)?;
    assert_eq!(remaining_commits, 1);
    assert_declaration_candidate(&fixture, &candidate, "local-model")?;
    assert_eq!(fixture.live("main")?, target);
    Ok(())
}

#[test]
fn different_declarations_refuse_without_changing_the_target() -> TestResult {
    let mut fixture = declaration_fixture()?;
    declare_field(&mut fixture, "feature", "source-model")?;
    declare_field(&mut fixture, "main", "target-model")?;
    let target = fixture.storage.capture(&Selection {
        branch: "main".to_owned(),
        version: None,
    })?;
    let before = reads::get_bytes(&fixture.storage.reader, &target, b"w\0records\0embedding")?;
    open_counting(&mut fixture)?;
    fixture.storage.merge_candidate_page("counts")?;
    let failure = fixture
        .storage
        .merge_candidate_page("counts")
        .err()
        .ok_or("model conflict accepted")?;
    assert_eq!(failure.code, "embed_model_change_not_available");
    assert_eq!(fixture.live("main")?, target.version);
    let current = fixture.storage.capture(&Selection {
        branch: "main".to_owned(),
        version: None,
    })?;
    assert_eq!(
        reads::get_bytes(&fixture.storage.reader, &current, b"w\0records\0embedding")?,
        before
    );
    Ok(())
}

#[test]
fn embedded_against_supplied_refuses_in_either_direction() -> TestResult {
    let seed = FrozenFixture::new(declaration_fixture()?)?;
    for embedded_branch in ["feature", "main"] {
        let mut fixture = seed.copy()?;
        let supplied_branch = if embedded_branch == "feature" {
            "main"
        } else {
            "feature"
        };
        declare_field(&mut fixture, embedded_branch, "local-model")?;
        fixture.put(supplied_branch, "one", 0)?;
        fixture.apply(
            supplied_branch,
            Change::VectorPut(crate::serve::protocol::VectorWrite {
                collection: "records".to_owned(),
                field: "embedding".to_owned(),
                document_id: "one".to_owned(),
                vector: vec![1.0, 2.0],
            }),
        )?;
        let target = fixture.live("main")?;
        open_counting(&mut fixture)?;
        fixture.storage.merge_candidate_page("counts")?;
        let failure = declaration_materialisation(&fixture, &[b"w\0records\0embedding".to_vec()])
            .err()
            .ok_or("supplied declaration conflict accepted")?;
        let failure = failure
            .downcast_ref::<crate::serve::storage::Failure>()
            .ok_or("unexpected failure type")?;
        assert_eq!(failure.code, "vector_field_supplied");
        let failure = fixture
            .storage
            .merge_candidate_page("counts")
            .err()
            .ok_or("mixed field candidate accepted")?;
        assert_eq!(failure.code, "vector_field_supplied");
        assert_eq!(fixture.live("main")?, target);
        fixture.close()?;
    }
    seed.close()
}

#[test]
fn declaration_review_offers_no_token_or_decodable_choice_owner() -> TestResult {
    let mut fixture = declaration_fixture()?;
    declare_field(&mut fixture, "feature", "local-model")?;
    open_counting(&mut fixture)?;
    let page = review::prepare::page(
        &fixture.storage.reader,
        &fixture.storage.visible,
        &ReviewPage {
            branch: "feature".to_owned(),
            after: None,
            limit: std::num::NonZeroU32::new(8).ok_or("zero page")?,
        },
        std::num::NonZeroUsize::new(8).ok_or("zero page")?,
    )?;
    let lines = serde_json::to_value(page.lines)?;
    let line = lines
        .as_array()
        .ok_or("review lines absent")?
        .iter()
        .find(|line| line["what"] == "vector field embedding")
        .ok_or("declaration line absent")?;
    assert!(
        line.get("choice_token").is_none(),
        "declaration offered a choice token"
    );
    assert_eq!(
        serde_json::from_str::<serde_json::Value>(
            line["source"].as_str().ok_or("source text absent")?
        )?,
        embedded_declaration("local-model")
    );
    assert!(serde_json::from_value::<MergeChoose>(json!({
        "merge_id":"counts", "owner":{"kind":"vector_field", "collection":"records", "field":"embedding"},
        "token":"merge-choice-v1:".to_owned()+&"a".repeat(64), "choice":"take_source"
    })).is_err());
    let choices = fixture
        .storage
        .merge_choices("counts", None, std::num::NonZeroUsize::MIN)?;
    assert_eq!(choices.revision, 0);
    assert!(choices.entries.is_empty());
    Ok(())
}

#[test]
fn supplied_vector_and_dimension_rows_keep_their_rebuild_refusal() -> TestResult {
    let mut fixture = declaration_fixture()?;
    fixture.put("feature", "one", 0)?;
    fixture.apply(
        "feature",
        Change::VectorPut(crate::serve::protocol::VectorWrite {
            collection: "records".to_owned(),
            field: "embedding".to_owned(),
            document_id: "one".to_owned(),
            vector: vec![1.0, 2.0],
        }),
    )?;
    let target = fixture.live("main")?;
    open_counting(&mut fixture)?;
    fixture.storage.merge_candidate_page("counts")?;
    for key in [
        b"v\0records\0embedding\0one".as_slice(),
        b"w\0records\0embedding".as_slice(),
    ] {
        let failure = declaration_materialisation(&fixture, &[key.to_vec()])
            .err()
            .ok_or("supplied vector row accepted")?;
        let failure = failure
            .downcast_ref::<crate::serve::storage::Failure>()
            .ok_or("unexpected refusal type")?;
        assert_eq!(failure.code, "candidate_vector_rebuild_required");
    }
    assert_eq!(fixture.live("main")?, target);
    Ok(())
}

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