haematite-cli 0.21.7

use std::error::Error;
use std::num::{NonZeroU64, NonZeroUsize};
use std::path::Path;

use super::{CommitFault, INTENT_PREFIX, Intent, PASS_REFUSED, ROW_COMMIT_FAULT, key, read_row};
use crate::serve::protocol::{
    BranchDiscard, Change, Fork, Mutation, ReceiptQuery, Selection, Version,
};
use crate::serve::storage::history::{MARK_READS, MARKER_READS, oldest_rows};
use crate::serve::storage::release::{HeldBy, Keep, PendingDiscards, Rules, Source};
use crate::serve::storage::{Failure, Receipt, SERVICE, Storage, branch_name, version};
use haematite::DiskStore;
use haematite::branch::{
    CommitDurability, CommitRequest, Kind, create_branch_with_kind, current_timestamp, open, remove,
};
use haematite::db::{DatabaseConfig, DatabaseError};
use haematite::store::NodeCacheBudget;
use haematite::tree::Hash;

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

fn setup() -> Result<(tempfile::TempDir, Storage), Box<dyn Error>> {
    setup_with(haematite::Retention::KeepAll)
}

fn setup_with(
    retention: haematite::Retention,
) -> Result<(tempfile::TempDir, Storage), Box<dyn Error>> {
    let directory = tempfile::tempdir()?;
    let mut 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,
        },
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )?;
    let main = storage
        .refs
        .get(&branch_name("main"))
        .ok_or("main missing")?
        .clone();
    let name = branch_name("work");
    create_branch_with_kind(
        &name,
        main.shards.iter().map(|shard| (shard.shard_id, shard.head)),
        Kind::Work,
        &mut storage.refs,
        &storage.registry,
        current_timestamp(),
    )?;
    storage.publish_record(&name)?;
    Ok((directory, storage))
}

fn intent(storage: &mut Storage) -> Result<(Intent, Receipt), Box<dyn Error>> {
    let record = storage
        .refs
        .get(&branch_name("work"))
        .ok_or("work missing")?
        .clone();
    let intent = Intent {
        branch: "work".to_owned(),
        request_id: "discard".to_owned(),
        version: version(&record),
    };
    create_branch_with_kind(
        &intent.name()?,
        record
            .shards
            .iter()
            .map(|shard| (shard.shard_id, shard.head)),
        Kind::Namespace,
        &mut storage.refs,
        &storage.registry,
        current_timestamp(),
    )?;
    let receipt = Receipt {
        retention: None,
        branch: intent.branch.clone(),
        request_id: intent.request_id.clone(),
        version: intent.version,
        fingerprint: Hash::of(b"request").to_string(),
    };
    Ok((intent, receipt))
}

fn reopen(directory: &Path) -> Result<Storage, Failure> {
    Storage::open(
        &directory.join("database"),
        NodeCacheBudget::Unlimited,
        fixture_long_wait_interval(),
    )
}

fn selection() -> Selection {
    Selection {
        branch: "work".to_owned(),
        version: None,
    }
}

fn put_row(storage: &mut Storage, key: Vec<u8>, bytes: Vec<u8>) -> TestResult {
    let service = open(SERVICE, &storage.refs, &storage.registry)?;
    service.put(storage.writer.shard_for(&key), key, bytes)?;
    storage.writer.commit(
        &service,
        &storage.registry,
        CommitRequest {
            durability: CommitDurability::Durable {
                refs: &mut storage.refs,
            },
            extra_parents: &[],
            timestamp: current_timestamp(),
        },
    )?;
    storage.publish_record(SERVICE)?;
    Ok(())
}

#[test]
fn removed_branch_with_pending_intent_reopens_with_only_the_receipt_row() -> TestResult {
    let (directory, mut storage) = setup()?;
    let (intent, receipt) = intent(&mut storage)?;
    storage.commit_discard_row(&receipt)?;
    remove(&branch_name("work"), &mut storage.refs)?;
    drop(storage);
    let mut storage = reopen(directory.path())?;
    assert_eq!(storage.refs.list_prefix(INTENT_PREFIX).count(), 0);
    assert_eq!(storage.refs.list_prefix("@haem/discard/").count(), 0);
    assert!(storage.capture(&selection()).is_err());
    let row = read_row(
        &storage.reader,
        &storage.refs,
        &key(&intent.branch, &intent.request_id),
    )?
    .ok_or("receipt row missing")?;
    assert_eq!(row.fingerprint, receipt.fingerprint);
    assert_eq!(
        storage
            .receipt(&ReceiptQuery {
                branch: intent.branch,
                request_id: intent.request_id
            })?
            .ok_or("replay missing")?
            .version,
        receipt.version
    );
    Ok(())
}

#[test]
fn recovered_old_intent_preserves_a_new_generation() -> TestResult {
    let (directory, mut storage) = setup()?;
    let (intent, receipt) = intent(&mut storage)?;
    storage.commit_discard_row(&receipt)?;
    remove(&branch_name("work"), &mut storage.refs)?;
    let roots = storage
        .refs
        .get(SERVICE)
        .ok_or("service missing")?
        .shards
        .clone();
    let created = intent
        .version
        .generation
        .checked_add(1)
        .ok_or("generation overflow")?;
    create_branch_with_kind(
        &branch_name("work"),
        roots.iter().map(|shard| (shard.shard_id, shard.head)),
        Kind::Work,
        &mut storage.refs,
        &storage.registry,
        created,
    )?;
    drop(storage);
    let mut storage = reopen(directory.path())?;
    assert_eq!(storage.capture(&selection())?.version.generation, created);
    assert_eq!(storage.refs.list_prefix(INTENT_PREFIX).count(), 0);
    assert_eq!(
        storage
            .receipt(&ReceiptQuery {
                branch: intent.branch,
                request_id: intent.request_id
            })?
            .ok_or("old replay missing")?
            .version,
        receipt.version
    );
    assert_eq!(storage.capture(&selection())?.version.generation, created);
    Ok(())
}

#[test]
fn legacy_discard_file_without_row_is_refused_by_name() -> TestResult {
    let (directory, mut storage) = setup()?;
    let name = format!("@haem/discard/{}/{}", Hash::of(b"work"), Hash::of(b"old"));
    let roots = storage
        .refs
        .get(SERVICE)
        .ok_or("service missing")?
        .shards
        .clone();
    create_branch_with_kind(
        &name,
        roots.iter().map(|shard| (shard.shard_id, shard.head)),
        Kind::Work,
        &mut storage.refs,
        &storage.registry,
        current_timestamp(),
    )?;
    drop(storage);
    let error = reopen(directory.path())
        .err()
        .ok_or("legacy store accepted")?;
    assert_eq!(error.code, "discard_rows_incomplete");
    assert!(error.message.contains(&name));
    Ok(())
}

#[test]
fn receipt_row_identity_mismatch_never_removes_a_live_branch() -> TestResult {
    let (directory, mut storage) = setup()?;
    let original = storage.capture(&selection())?.version;
    let receipt = Receipt {
        retention: None,
        branch: "another".to_owned(),
        request_id: "discard".to_owned(),
        version: original,
        fingerprint: Hash::of(b"request").to_string(),
    };
    put_row(
        &mut storage,
        key("work", "discard"),
        serde_json::to_vec(&receipt)?,
    )?;
    let error = storage
        .receipt(&ReceiptQuery {
            branch: "work".to_owned(),
            request_id: "discard".to_owned(),
        })
        .err()
        .ok_or("misbound receipt accepted")?;
    assert_eq!(error.code, "discard_record_corrupt");
    assert_eq!(storage.capture(&selection())?.version, original);
    assert_eq!(storage.refs.list_prefix(INTENT_PREFIX).count(), 0);
    drop(storage);
    assert_eq!(
        reopen(directory.path())?.capture(&selection())?.version,
        original
    );
    Ok(())
}

#[test]
fn committed_row_with_advanced_target_refuses_recovery() -> TestResult {
    let (directory, mut storage) = setup()?;
    let (_, receipt) = intent(&mut storage)?;
    storage.commit_discard_row(&receipt)?;
    let branch = open(&branch_name("work"), &storage.refs, &storage.registry)?;
    branch.put(storage.writer.shard_for(b"test"), b"test", b"value")?;
    storage.writer.commit(
        &branch,
        &storage.registry,
        CommitRequest {
            durability: CommitDurability::Durable {
                refs: &mut storage.refs,
            },
            extra_parents: &[],
            timestamp: current_timestamp(),
        },
    )?;
    drop(storage);
    let error = reopen(directory.path())
        .err()
        .ok_or("advanced target discarded")?;
    assert_eq!(error.code, "discard_conflict");
    Ok(())
}

#[test]
fn malformed_pending_intent_is_named_at_open() -> TestResult {
    let (directory, mut storage) = setup()?;
    let roots = storage
        .refs
        .get(SERVICE)
        .ok_or("service missing")?
        .shards
        .clone();
    create_branch_with_kind(
        &format!("{INTENT_PREFIX}invalid"),
        roots.iter().map(|shard| (shard.shard_id, shard.head)),
        Kind::Namespace,
        &mut storage.refs,
        &storage.registry,
        current_timestamp(),
    )?;
    drop(storage);
    let error = reopen(directory.path())
        .err()
        .ok_or("malformed intent accepted")?;
    assert_eq!(error.code, "discard_intent_corrupt");
    Ok(())
}

#[test]
fn malformed_receipt_row_preserves_the_target_and_pending_intent() -> TestResult {
    let (directory, mut storage) = setup()?;
    let (intent, _) = intent(&mut storage)?;
    let original = storage.capture(&selection())?.version;
    put_row(
        &mut storage,
        key(&intent.branch, &intent.request_id),
        b"{".to_vec(),
    )?;
    let error = storage
        .receipt(&ReceiptQuery {
            branch: intent.branch,
            request_id: intent.request_id,
        })
        .err()
        .ok_or("malformed receipt accepted")?;
    assert_eq!(error.code, "discard_record_corrupt");
    assert_eq!(storage.capture(&selection())?.version, original);
    assert_eq!(storage.refs.list_prefix(INTENT_PREFIX).count(), 1);
    drop(storage);
    assert_eq!(
        reopen(directory.path())
            .err()
            .ok_or("malformed pending row opened")?
            .code,
        "discard_record_corrupt"
    );
    Ok(())
}

#[test]
fn pending_target_cannot_be_recreated_before_cleanup() -> TestResult {
    let (directory, mut storage) = setup()?;
    let (intent, _) = intent(&mut storage)?;
    storage.pending_discards.insert(branch_name(&intent.branch));
    let origin = storage
        .capture(&Selection {
            branch: "main".to_owned(),
            version: None,
        })?
        .version;
    let mutation = crate::serve::protocol::Mutation {
        branch: "main".to_owned(),
        request_id: "fork-pending".to_owned(),
        expected: origin,
        change: crate::serve::protocol::Change::Fork(crate::serve::protocol::Fork {
            name: intent.branch,
            from: origin,
        }),
    };
    // Removing the public name models an unlink whose intent cleanup failed.
    remove(&branch_name("work"), &mut storage.refs)?;
    storage.visible.remove(&branch_name("work"))?;
    assert_eq!(
        storage
            .mutate(&mutation)
            .err()
            .ok_or("pending target recreated")?
            .code,
        "discard_pending"
    );
    assert!(storage.refs.get(&branch_name("work")).is_none());
    assert_eq!(
        haematite::branch::RefStore::open(
            directory
                .path()
                .join("database")
                .join(haematite::db::vacuum::CANONICAL_REFS_DIR)
        )?
        .list_prefix(INTENT_PREFIX)
        .count(),
        1
    );
    Ok(())
}

#[test]
fn duplicate_target_intents_refuse_before_any_cleanup() -> TestResult {
    let (directory, mut storage) = setup()?;
    let (first, _) = intent(&mut storage)?;
    let second = Intent {
        request_id: "another".to_owned(),
        ..first
    };
    let roots = storage
        .refs
        .get(&branch_name("work"))
        .ok_or("work missing")?
        .shards
        .clone();
    create_branch_with_kind(
        &second.name()?,
        roots.iter().map(|shard| (shard.shard_id, shard.head)),
        Kind::Namespace,
        &mut storage.refs,
        &storage.registry,
        current_timestamp(),
    )?;
    drop(storage);
    assert_eq!(
        reopen(directory.path())
            .err()
            .ok_or("duplicate intents accepted")?
            .code,
        "discard_intent_corrupt"
    );
    let refs = haematite::branch::RefStore::open(
        directory
            .path()
            .join("database")
            .join(haematite::db::vacuum::CANONICAL_REFS_DIR),
    )?;
    assert_eq!(refs.list_prefix(INTENT_PREFIX).count(), 2);
    assert!(refs.get(&branch_name("work")).is_some());
    Ok(())
}

#[test]
fn pre_row_intent_can_be_aborted_by_receipt_replay() -> TestResult {
    let (directory, mut storage) = setup()?;
    let (intent, _) = intent(&mut storage)?;
    storage.pending_discards.insert(branch_name(&intent.branch));
    storage.visible.remove(&branch_name(&intent.branch))?;
    assert!(
        storage
            .receipt(&ReceiptQuery {
                branch: intent.branch.clone(),
                request_id: intent.request_id
            })?
            .is_none()
    );
    assert_eq!(storage.capture(&selection())?.version, intent.version);
    assert!(storage.pending_discards.is_empty());
    assert_eq!(storage.refs.list_prefix(INTENT_PREFIX).count(), 0);
    drop(storage);
    assert_eq!(
        reopen(directory.path())?.capture(&selection())?.version,
        intent.version
    );
    Ok(())
}

fn discard_mutation(storage: &Storage) -> Result<Mutation, Failure> {
    Ok(Mutation {
        branch: "work".to_owned(),
        request_id: "row-commit-failure".to_owned(),
        expected: storage.capture(&selection())?.version,
        change: Change::Discard(BranchDiscard {
            name: "work".to_owned(),
        }),
    })
}

fn fork_mutation(from: crate::serve::protocol::Version) -> Mutation {
    Mutation {
        branch: "work".to_owned(),
        request_id: "fork-after-refusal".to_owned(),
        expected: from,
        change: Change::Fork(Fork {
            name: "child".to_owned(),
            from,
        }),
    }
}

#[test]
fn definite_row_refusal_restores_visibility_fork_admission_and_request_retry() -> TestResult {
    let (directory, mut storage) = setup()?;
    let request = discard_mutation(&storage)?;
    ROW_COMMIT_FAULT.with(|slot| slot.set(Some(CommitFault::RefuseBeforeCommit)));
    let refused = storage
        .mutate(&request)
        .err()
        .ok_or("row commit was not refused")?;
    assert_eq!(refused.code, "write_refused");
    assert_eq!(
        refused.message,
        "injected refusal before discard row commit"
    );
    assert_eq!(storage.capture(&selection())?.version, request.expected);
    assert!(storage.pending_discards.is_empty());
    assert_eq!(storage.refs.list_prefix(INTENT_PREFIX).count(), 0);
    assert!(
        read_row(
            &storage.reader,
            &storage.refs,
            &key(&request.branch, &request.request_id)
        )?
        .is_none()
    );
    let child = storage.mutate(&fork_mutation(request.expected))?;
    assert_eq!(child.branch, "child");
    assert_eq!(
        storage
            .fork_provenance(&storage.capture(&Selection {
                branch: "child".to_owned(),
                version: None,
            })?)?
            .version,
        request.expected
    );
    let receipt = storage.mutate(&request)?;
    assert_eq!(receipt.version, request.expected);
    assert!(storage.capture(&selection()).is_err());
    assert!(storage.pending_discards.is_empty());
    assert_eq!(storage.refs.list_prefix(INTENT_PREFIX).count(), 0);
    assert_eq!(
        read_row(
            &storage.reader,
            &storage.refs,
            &key(&request.branch, &request.request_id)
        )?
        .ok_or("retry row missing")?
        .fingerprint,
        receipt.fingerprint
    );
    drop(storage);
    assert_eq!(
        reopen(directory.path())?.mutate(&request)?.fingerprint,
        receipt.fingerprint
    );
    Ok(())
}

#[test]
fn unknown_row_commit_keeps_the_intent_and_refuses_fork_until_replay() -> TestResult {
    let (directory, mut storage) = setup()?;
    let request = discard_mutation(&storage)?;
    ROW_COMMIT_FAULT.with(|slot| slot.set(Some(CommitFault::LoseCommittedAnswer)));
    let unknown = storage
        .mutate(&request)
        .err()
        .ok_or("commit answer was not lost")?;
    assert_eq!(unknown.code, "durability_unknown");
    assert_eq!(
        unknown.message,
        "injected lost answer after discard row commit"
    );
    assert!(storage.capture(&selection()).is_err());
    assert!(storage.pending_discards.contains(&branch_name("work")));
    assert_eq!(storage.refs.list_prefix(INTENT_PREFIX).count(), 1);
    let row = read_row(
        &storage.reader,
        &storage.refs,
        &key(&request.branch, &request.request_id),
    )?
    .ok_or("committed row missing")?;
    assert_eq!(row.version, request.expected);
    assert_eq!(
        row.fingerprint,
        Hash::of(&serde_json::to_vec(&request)?).to_string()
    );
    assert_eq!(
        storage
            .mutate(&fork_mutation(request.expected))
            .err()
            .ok_or("pending source admitted a fork")?
            .code,
        "discard_pending"
    );
    assert!(storage.refs.get(&branch_name("child")).is_none());
    let receipt = storage.mutate(&request)?;
    assert_eq!(receipt.fingerprint, row.fingerprint);
    assert!(storage.pending_discards.is_empty());
    assert_eq!(storage.refs.list_prefix(INTENT_PREFIX).count(), 0);
    assert!(storage.capture(&selection()).is_err());
    drop(storage);
    assert_eq!(
        reopen(directory.path())?.mutate(&request)?.fingerprint,
        receipt.fingerprint
    );
    Ok(())
}

#[test]
fn served_discard_refusal_restores_the_branch_before_receipt_reconciliation() -> TestResult {
    use std::num::NonZeroU32;

    use crate::serve::protocol::{
        BranchPage, CollectionWrite, DocumentRead, DocumentWrite, Operation,
    };
    use crate::serve::service::Service;
    use serde_json::value::RawValue;

    let (_directory, mut storage) = setup()?;
    let initial = storage.capture(&selection())?.version;
    let declared = storage.mutate(&Mutation {
        request_id: "declare-kept-records".to_owned(),
        branch: "work".to_owned(),
        expected: initial,
        change: Change::Collection(CollectionWrite {
            collection: "records".to_owned(),
            schema: None,
        }),
    })?;
    storage.mutate(&Mutation {
        request_id: "write-kept-record".to_owned(),
        branch: "work".to_owned(),
        expected: declared.version,
        change: Change::Put(DocumentWrite {
            collection: "records".to_owned(),
            document_id: "one".to_owned(),
            document: RawValue::from_string(r#"{"kept":true}"#.to_owned())?,
        }),
    })?;
    let request = discard_mutation(&storage)?;
    let request_bytes = serde_json::to_vec(&request)?;
    let service = Service::new(storage, NonZeroU32::MIN.saturating_add(7))?;
    let branches = || {
        service.dispatch(Operation::Branches(BranchPage {
            after: None,
            limit: NonZeroU32::MIN.saturating_add(7),
        }))
    };
    let record = || {
        service.dispatch(Operation::Get(DocumentRead {
            selection: Selection {
                branch: "work".to_owned(),
                version: Some(request.expected),
            },
            collection: "records".to_owned(),
            document_id: "one".to_owned(),
        }))
    };
    let receipt = || {
        service.dispatch(Operation::Receipt(ReceiptQuery {
            branch: request.branch.clone(),
            request_id: request.request_id.clone(),
        }))
    };
    let before_branches = branches()?;
    let before_record = record()?;
    assert_eq!(before_record["document"]["raw_json"], r#"{"kept":true}"#);
    let before_work = before_branches["entries"]
        .as_array()
        .ok_or("branch list is not an array")?
        .iter()
        .find(|entry| entry["branch"] == "work")
        .ok_or("branch was absent before discard")?;
    assert_eq!(
        before_work["version"],
        serde_json::to_value(request.expected)?
    );

    ROW_COMMIT_FAULT.with(|slot| slot.set(Some(CommitFault::RefuseBeforeCommit)));
    let refusal = service
        .dispatch(Operation::Mutate(serde_json::from_slice(&request_bytes)?))
        .err()
        .ok_or("discard row commit was not refused")?;
    let refusal = serde_json::to_value(refusal)?;
    assert_eq!(refusal["code"], "write_refused");
    assert_eq!(refusal["outcome"], "not_applied");
    assert_eq!(refusal["request_id"], request.request_id);
    assert_eq!(refusal["receipt_branch"], request.branch);

    // A receipt read can repair an unfinished abort, so observe visibility first.
    assert_eq!(branches()?, before_branches);
    assert_eq!(record()?, before_record);
    assert!(receipt()?.is_null());

    let applied = service.dispatch(Operation::Mutate(serde_json::from_slice(&request_bytes)?))?;
    assert_eq!(applied["request_id"], request.request_id);
    assert_eq!(applied["branch"], request.branch);
    assert_eq!(applied["version"], serde_json::to_value(request.expected)?);
    assert_eq!(receipt()?, applied);
    let after = branches()?;
    assert!(
        !after["entries"]
            .as_array()
            .ok_or("branch list is not an array")?
            .iter()
            .any(|entry| entry["branch"] == "work")
    );
    Ok(())
}

/// The release floor's pending discards, read as production reads them
/// (STOR-003 H1b2).
struct Intents;

impl PendingDiscards for Intents {
    fn pending(&self, storage: &Storage) -> Result<Vec<(String, Version)>, Failure> {
        storage.pending_discards()
    }
}

/// Fails while the floor reads no pending discard: an unfinished discard of
/// `work` is listed from its intent, and `work`'s floor relies on its
/// version; `main`'s floor gains nothing from it.
#[test]
fn a_pending_discard_is_read_from_its_intent_into_the_floor() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let (intent, _receipt) = intent(&mut storage)?;
    assert_eq!(
        storage.pending_discards()?,
        vec![("work".to_owned(), intent.version)]
    );
    let floor = storage.relied_on("work", &Intents)?;
    assert!(floor.held_by.contains(&HeldBy {
        source: Source::Discard,
        name: "work".to_owned(),
        version: intent.version,
    }));
    let main = storage.relied_on("main", &Intents)?;
    assert!(
        main.held_by
            .iter()
            .all(|held| held.source != Source::Discard)
    );
    Ok(())
}

/// Fails while an intent that does not decode is passed over: the listing
/// is refused by name, so no floor is given without it.
#[test]
fn an_intent_that_does_not_decode_refuses_the_listing_by_name() -> TestResult {
    let (_directory, mut storage) = setup()?;
    let roots = storage
        .refs
        .get(SERVICE)
        .ok_or("service missing")?
        .shards
        .clone();
    create_branch_with_kind(
        &format!("{INTENT_PREFIX}invalid"),
        roots.iter().map(|shard| (shard.shard_id, shard.head)),
        Kind::Namespace,
        &mut storage.refs,
        &storage.registry,
        current_timestamp(),
    )?;
    let error = storage
        .pending_discards()
        .err()
        .ok_or("an intent that does not decode was passed over")?;
    assert_eq!(error.code, "discard_intent_corrupt");
    assert!(error.message.contains("invalid"), "{}", error.message);
    Ok(())
}

/// Whether shard `shard`'s store, read fresh from disk, still holds `node`.
fn on_disk(directory: &Path, shard: usize, node: &Hash) -> Result<bool, Box<dyn Error>> {
    let store = DiskStore::new(
        directory
            .join("database")
            .join(format!("shard-{shard}"))
            .join("store"),
        NodeCacheBudget::Unlimited,
    )?;
    Ok(store.get(node)?.is_some())
}

/// Writes `entries` on every shard of `branch`, one durable commit, published.
fn commit_on(storage: &mut Storage, branch: &str, entries: &[(Vec<u8>, Vec<u8>)]) -> TestResult {
    let name = branch_name(branch);
    let handle = open(&name, &storage.refs, &storage.registry)?;
    for shard in 0..2 {
        for (key, value) in entries {
            handle.put(shard, key, value)?;
        }
    }
    storage.writer.commit(
        &handle,
        &storage.registry,
        CommitRequest {
            durability: CommitDurability::Durable {
                refs: &mut storage.refs,
            },
            extra_parents: &[],
            timestamp: current_timestamp(),
        },
    )?;
    storage.publish_record(&name)?;
    Ok(())
}

/// About 128 KiB on main that does not compress (hex of distinct hashes),
/// so each shard keeps far more than a discard writes, and a small value only
/// `work` holds; answers `work`'s head on each shard.
fn hold_apart(storage: &mut Storage) -> Result<Vec<(usize, Hash)>, Box<dyn Error>> {
    let kept: Vec<u8> = (0..2048_u32)
        .flat_map(|index| Hash::of(&index.to_le_bytes()).to_string().into_bytes())
        .collect();
    commit_on(storage, "main", &[(b"kept-on-main".to_vec(), kept)])?;
    commit_on(
        storage,
        "work",
        &[(b"only-work".to_vec(), b"held by work alone".to_vec())],
    )?;
    let record = storage
        .refs
        .get(&branch_name("work"))
        .ok_or("work missing")?;
    Ok(record
        .shards
        .iter()
        .map(|shard| (shard.shard_id, shard.head))
        .collect())
}

/// A discard asks for a reclaim pass once its branch and intent are gone, and
/// the asked pass does not give the branch back: before the removal the
/// discard writes a history row for the head it removes (discards.rs
/// `rows_before_discard`, STOR-003 MD5a/MD5c), and every pass marks every
/// history row's roots (crates/haematite/src/shard/actor/native/reclaim.rs
/// page). So the row names work's heads, and they stay on disk after a pass:
/// the discard's own rows were written since the last one, so quiesce asks one
/// (reclaim.rs:1188-1194). That no release ever takes this row is the leak
/// docs/design/HAEM-DISCARDED-ROWS-RELEASE-2026-10-06.md answers.
#[test]
fn a_discard_asks_for_a_pass_and_its_history_row_keeps_the_heads() -> TestResult {
    let (directory, mut storage) = setup_with(haematite::Retention::reclaim(0))?;
    let heads = hold_apart(&mut storage)?;
    let discarded = version(
        storage
            .refs
            .get(&branch_name("work"))
            .ok_or("work missing")?,
    );
    storage.activate_reclaim()?;
    storage.quiesce_reclaim()?;
    let request = discard_mutation(&storage)?;
    storage.mutate(&request)?;
    let said = storage.discard_passes().status();
    assert_eq!(
        (said["branch"].as_str(), said["asked"].as_bool()),
        (Some("work"), Some(true)),
        "{said}"
    );
    let service = storage.refs.get(SERVICE).ok_or("service missing")?;
    let next = Version {
        sequence: discarded.sequence + 1,
        ..discarded
    };
    let rows = oldest_rows(
        &storage.reader,
        service,
        "work",
        (discarded, next),
        NonZeroUsize::MIN,
    )?;
    let named: Vec<(usize, Hash)> = rows
        .iter()
        .flat_map(|row| row.shards.iter().map(|root| (root.shard_id, root.root)))
        .collect();
    assert_eq!(named, heads, "the discard's history row names work's heads");
    storage.activate_reclaim()?;
    storage.quiesce_reclaim()?;
    for (shard, head) in &heads {
        assert!(
            on_disk(directory.path(), *shard, head)?,
            "the row keeps work's head on shard {shard} after a pass"
        );
    }
    assert!(storage.capture(&selection()).is_err());
    Ok(())
}

/// A refused ask is said by name in `store.stats`; the discard stands, and
/// its receipt is the stored row.
#[test]
fn a_refused_pass_leaves_the_discard_done_and_names_the_refusal() -> TestResult {
    let (_directory, mut storage) = setup()?;
    PASS_REFUSED.with(|slot| {
        slot.set(Some(DatabaseError::ReclaimFailed {
            shard: 1,
            reason: "pass_refused_for_test".to_owned(),
        }));
    });
    let request = discard_mutation(&storage)?;
    let receipt = storage.mutate(&request)?;
    let said = storage.discard_passes().status();
    assert_eq!(said["asked"].as_bool(), Some(false));
    assert!(
        said["refused"]
            .as_str()
            .is_some_and(|refused| refused.contains("pass_refused_for_test")),
        "{said}"
    );
    assert!(storage.capture(&selection()).is_err());
    let stored = storage
        .receipt(&ReceiptQuery {
            branch: "work".to_owned(),
            request_id: request.request_id.clone(),
        })?
        .ok_or("no stored receipt")?;
    assert_eq!(
        serde_json::to_value(&receipt)?,
        serde_json::to_value(&stored)?
    );
    Ok(())
}

/// The window keeping one version and no days: a discarded generation's last
/// version may go at once, the discard counted as the version after it.
fn window() -> Rules {
    Rules {
        keep: Some(Keep {
            versions: NonZeroU64::new(1),
            days: Some(0),
        }),
        max_rows: NonZeroUsize::new(64).unwrap_or(NonZeroUsize::MIN),
        now: current_timestamp(),
    }
}

/// A discarded generation's rows are released under the same window rules as
/// a live branch's, and its heads leave the disk; the released version then
/// answers `version_released`
/// (docs/design/HAEM-DISCARDED-ROWS-RELEASE-2026-10-06.md).
#[test]
fn a_discarded_generation_is_released_and_its_heads_leave_the_disk() -> TestResult {
    let (directory, mut storage) = setup_with(haematite::Retention::reclaim(0))?;
    let heads = hold_apart(&mut storage)?;
    let discarded = version(
        storage
            .refs
            .get(&branch_name("work"))
            .ok_or("work missing")?,
    );
    let request = discard_mutation(&storage)?;
    storage.mutate(&request)?;
    let page = NonZeroUsize::new(64).unwrap_or(NonZeroUsize::MIN);
    let marked = storage.dead_page(None, page)?.names;
    assert_eq!(marked, ["work"], "the discard marks its generation");
    let released = storage.release("work", None, &window(), &Intents);
    assert!(
        released.is_ok(),
        "a discarded generation is released: {released:?}"
    );
    let span = released?.released.ok_or("nothing was released")?;
    assert_eq!(
        span.last, discarded,
        "the release ends at the discarded version"
    );
    let marked = storage.dead_page(None, page)?.names;
    assert!(marked.is_empty(), "the marker went with the last row");
    storage.activate_reclaim()?;
    storage.quiesce_reclaim()?;
    for (shard, head) in &heads {
        assert!(
            !on_disk(directory.path(), *shard, head)?,
            "work's head left shard {shard}"
        );
    }
    let read = storage
        .capture(&Selection {
            branch: "work".to_owned(),
            version: Some(discarded),
        })
        .err()
        .ok_or("the released version was read")?;
    assert_eq!(read.code, "version_released", "{}", read.message);
    Ok(())
}

/// A fork taken from the version a discard removed still holds it: release
/// of the discarded generation stops below it, and the version reads back
/// (STOR-003 HR1).
#[test]
fn a_fork_from_a_discarded_version_still_holds_it() -> TestResult {
    let (directory, mut storage) = setup_with(haematite::Retention::reclaim(0))?;
    let heads = hold_apart(&mut storage)?;
    let discarded = version(
        storage
            .refs
            .get(&branch_name("work"))
            .ok_or("work missing")?,
    );
    storage.mutate(&fork_mutation(discarded))?;
    let request = discard_mutation(&storage)?;
    storage.mutate(&request)?;
    let answer = storage.release("work", None, &window(), &Intents);
    assert!(
        answer.is_ok(),
        "a discarded generation is released: {answer:?}"
    );
    let answer = answer?;
    assert!(
        answer
            .held_by
            .iter()
            .any(|held| held.source == Source::Fork && held.version == discarded),
        "the fork holds the discarded version: {:?}",
        answer.held_by
    );
    assert!(
        answer.released.is_none_or(|span| span.last != discarded),
        "the held version is not released: {:?}",
        answer.released
    );
    storage.activate_reclaim()?;
    storage.quiesce_reclaim()?;
    for (shard, head) in &heads {
        assert!(
            on_disk(directory.path(), *shard, head)?,
            "kept on shard {shard}"
        );
    }
    let read = storage.capture(&Selection {
        branch: "work".to_owned(),
        version: Some(discarded),
    })?;
    assert_eq!(read.version, discarded);
    Ok(())
}

/// DR-6: a fork that holds an older discarded generation of `work` holds that
/// generation only. The name made again and discarded again is released past
/// it: one marker page, the held generation's mark looked at and passed over.
#[test]
fn a_held_old_generation_does_not_hold_a_newer_one_of_its_name() -> TestResult {
    let (_directory, mut storage) = setup_with(haematite::Retention::reclaim(0))?;
    let first = storage.capture(&selection())?.version;
    storage.mutate(&fork_mutation(first))?;
    let request = discard_mutation(&storage)?;
    storage.mutate(&request)?;
    let main = storage
        .capture(&Selection {
            branch: "main".to_owned(),
            version: None,
        })?
        .version;
    storage.mutate(&Mutation {
        branch: "main".to_owned(),
        request_id: "recreate-work".to_owned(),
        expected: main,
        change: Change::Fork(Fork {
            name: "work".to_owned(),
            from: main,
        }),
    })?;
    let second = storage.capture(&selection())?.version;
    storage.mutate(&Mutation {
        branch: "work".to_owned(),
        request_id: "discard-work-again".to_owned(),
        expected: second,
        change: Change::Discard(BranchDiscard {
            name: "work".to_owned(),
        }),
    })?;

    MARK_READS.with(|reads| reads.set(0));
    MARKER_READS.with(|reads| reads.set(0));
    let answer = storage.release("work", None, &window(), &Intents)?;
    assert_eq!(
        answer.released.map(|span| span.last),
        Some(second),
        "the newer generation goes past the held one: {answer:?}"
    );
    // Two generations' marks (each its own key, then the name's old key),
    // and the released generation's marker as its last row goes.
    assert_eq!(MARK_READS.with(std::cell::Cell::get), 4, "mark reads");
    assert_eq!(MARKER_READS.with(std::cell::Cell::get), 1, "marker reads");
    let kept = storage
        .capture(&Selection {
            branch: "work".to_owned(),
            version: Some(first),
        })
        .map(|capture| capture.version);
    assert_eq!(
        kept.ok(),
        Some(first),
        "the held generation still reads back"
    );
    Ok(())
}

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