haematite-cli 0.21.7

#![forbid(unsafe_code)]
#![cfg(unix)]

#[path = "../../haematite/tests/common/runtime_options.rs"]
mod runtime_options;

use runtime_options::runtime_options;

use std::error::Error;
use std::io::Write;
use std::os::unix::fs::{DirBuilderExt, OpenOptionsExt};
use std::path::{Path, PathBuf};

use haematite::branch::history::{Key, Row, SERVICE_BRANCH, ShardRoot};
use haematite::branch::{
    CommitDurability, CommitRequest, Kind, RefRecord, RefStore, Registry, create_branch_with_kind,
    current_timestamp, fork_from_with_kind, open, remove,
};
use haematite::db::branches::BranchDatabase;
use haematite::db::vacuum::CANONICAL_REFS_DIR;
use haematite::db::{Database, DatabaseConfig};
use haematite::private_mode::{file_read_bytes, reset_file_read_bytes};
use haematite::store::{NodeCacheBudget, file_opens, reset_file_opens};
use haematite::tree::Hash;
use haematite::wal::Recovery;
use haematite_cli::serve::protocol::{Selection, Version};
use haematite_cli::serve::storage::Storage;

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

const SHARDS: usize = 2;
const LARGE: u64 = 20;
const SMALL: u64 = 2;
const LEGACY_SMALL: u64 = 32;
const LEGACY_LARGE: u64 = LEGACY_SMALL * 10;
const MAIN: &str = "@haem/branch/main";
// Configuration, the service preflight, two installed branch records and
// each shard's tiny WAL, checkpoint, empty node and framing prefixes fit
// within this fixed budget, without history nodes.
const OPEN_BYTES: u64 = 64 * 1024;

struct Fixture {
    directory: tempfile::TempDir,
    old: Version,
    predecessor: Version,
    current: Version,
    roots: Vec<(usize, Hash)>,
}

fn record(refs: &RefStore, name: &str) -> TestResult<RefRecord> {
    refs.get(name)
        .cloned()
        .ok_or_else(|| format!("reference missing: {name}").into())
}

fn row(record: &RefRecord, sequence: u64) -> Row {
    Row {
        key: Key {
            branch: Hash::of(b"main"),
            generation: record.created,
            sequence,
        },
        timestamp: record.timestamp,
        shards: record
            .shards
            .iter()
            .map(|shard| ShardRoot {
                shard_id: shard.shard_id,
                root: shard.head,
            })
            .collect(),
    }
}

fn seed_history(
    writer: &mut BranchDatabase,
    refs: &mut RefStore,
    registry: &Registry,
    records: (&RefRecord, &RefRecord),
    versions: u64,
) -> TestResult<u64> {
    let (previous, live) = records;
    let service = open(SERVICE_BRANCH, refs, registry)?;
    let old_count = versions
        .checked_sub(1)
        .ok_or("fixture needs a predecessor")?;
    let mut per_shard = [0_u64; SHARDS];
    let template = row(previous, 0);
    let value = template.encode_value()?;
    let kept = row(live, 0);
    let key = kept.key.encode();
    let kept_shard = writer.shard_for(&key);
    // The smallest fixture still puts history in both physical shards.
    let first_sequence = (0..1024)
        .find(|sequence| {
            let key = Key {
                sequence: *sequence,
                ..template.key
            }
            .encode();
            writer.shard_for(&key) != kept_shard
        })
        .ok_or("fixture cannot find history for the other shard")?;
    for offset in 0..old_count {
        let sequence = first_sequence + offset;
        let key = Key {
            sequence,
            ..template.key
        }
        .encode();
        let shard = writer.shard_for(&key);
        service.put(shard, key, &value)?;
        per_shard[shard] += 1;
    }
    service.put(kept_shard, key, kept.encode_value()?)?;
    per_shard[kept_shard] += 1;
    // This checks the fixture covers both shards.
    // A changed key selection must not silently leave a shard empty.
    assert!(per_shard.iter().all(|count| *count > 0));
    // One batch materialises the history; no commit follows each offered row.
    writer.commit(
        &service,
        registry,
        CommitRequest {
            durability: CommitDurability::Durable { refs },
            extra_parents: &[],
            timestamp: current_timestamp(),
        },
    )?;
    assert_eq!(record(refs, SERVICE_BRANCH)?.seq, 1);
    Ok(first_sequence)
}

fn build(versions: u64) -> TestResult<Fixture> {
    if !matches!(versions, LARGE | SMALL) {
        return Err("fixture size must be one of the two declared cases".into());
    }
    build_versions(versions, None)
}

fn build_versions(versions: u64, stamps: Option<[u64; 3]>) -> TestResult<Fixture> {
    let directory = tempfile::tempdir()?;
    let data = directory.path().join("database");
    let database = Database::create(
        DatabaseConfig {
            data_dir: data.clone(),
            shard_count: SHARDS,
            distributed: None,
            executor_threads: Some(SHARDS),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::KeepAll,
        },
        runtime_options(),
    )?;
    let mut writer = BranchDatabase::new(database, NodeCacheBudget::Unlimited)?;
    let mut refs = RefStore::open(data.join(CANONICAL_REFS_DIR))?;
    let registry = Registry::new();
    let service = create_branch_with_kind(
        SERVICE_BRANCH,
        writer.base_roots()?,
        Kind::Namespace,
        &mut refs,
        &registry,
        stamps.map_or_else(current_timestamp, |[service, _, _]| service),
    )?;
    let previous_branch = fork_from_with_kind(
        &service,
        MAIN,
        Kind::Work,
        &mut refs,
        &registry,
        stamps.map_or_else(current_timestamp, |[_, previous, _]| previous),
    )?;
    let previous = record(&refs, MAIN)?;
    drop(previous_branch);
    assert!(remove(MAIN, &mut refs)?.is_some());
    let main = fork_from_with_kind(
        &service,
        MAIN,
        Kind::Work,
        &mut refs,
        &registry,
        stamps.map_or_else(current_timestamp, |[_, _, current]| current),
    )?;
    let live = record(&refs, MAIN)?;
    assert_ne!(live.created, previous.created);
    let old_sequence = seed_history(
        &mut writer,
        &mut refs,
        &registry,
        (&previous, &live),
        versions,
    )?;
    let key = b"fixture/live";
    main.put(writer.shard_for(key), key, b"current")?;
    writer.commit(
        &main,
        &registry,
        CommitRequest {
            durability: CommitDurability::Durable { refs: &mut refs },
            extra_parents: &[],
            timestamp: current_timestamp(),
        },
    )?;
    let current = record(&refs, MAIN)?;
    assert_eq!(current.seq, 1);
    assert_eq!(refs.list_prefix("@haem/history/").count(), 0);
    Ok(Fixture {
        directory,
        old: Version {
            generation: previous.created,
            sequence: old_sequence,
        },
        predecessor: Version {
            generation: live.created,
            sequence: 0,
        },
        current: Version {
            generation: current.created,
            sequence: current.seq,
        },
        roots: previous
            .shards
            .iter()
            .map(|shard| (shard.shard_id, shard.head))
            .collect(),
    })
}

fn check_open(fixture: &Fixture) -> TestResult {
    let data = fixture.directory.path().join("database");
    // The fixture is written under KeepAll, so its first open moves it to
    // reclaim (STOR-003 H1a) and owes a startup pass whose mark reads the
    // history tree. Both happen once in a store's life: move it, run that
    // pass to rest by its progress answers, and count the open after it.
    drop(Storage::open(
        &data,
        NodeCacheBudget::Unlimited,
        long_wait_interval(),
    )?);
    let moved = Database::open(&data, runtime_options())?;
    moved.quiesce_reclaim()?;
    drop(moved);
    // 6H (NB1): each counted open is logged by its call site and thread, and
    // printed when the count is wrong.
    haematite::private_mode::log_file_opens(true);
    reset_file_opens();
    reset_file_read_bytes();
    let storage = Storage::open(&data, NodeCacheBudget::Unlimited, long_wait_interval())?;
    let opened = file_opens();
    let log = haematite::private_mode::file_open_log();
    haematite::private_mode::log_file_opens(false);
    let bytes = file_read_bytes();
    // Fourteen fixed opens: five configuration reads, one writer lock,
    // one service-record preflight, three FORMAT reads, two root-source
    // records and two records loaded under the writer lock.
    // The preflight refuses an unserved store before retention can change;
    // its record is not reused as writable state. The fifth configuration
    // read is under the lock, so startup uses the current configuration.
    // Six per shard: migration probe, WAL recovery, tail reconciliation,
    // append framing, empty base node and index checkpoint.
    // This rows-only store has no legacy history files, snapshot registry or
    // reclaim resume record, and its startup pass has already settled.
    // Directory opens/listing, metadata calls and cache hits are not counted.
    assert_eq!(opened, 14 + 6 * u64::try_from(SHARDS)?, "{log:#?}");
    assert!(bytes > 0);
    assert!(bytes < OPEN_BYTES, "open read {bytes} bytes");
    let captured = storage.capture(&Selection {
        branch: "main".to_owned(),
        version: Some(fixture.old),
    })?;
    assert_eq!(captured.version, fixture.old);
    assert_eq!(
        captured
            .shards
            .iter()
            .map(|shard| (shard.shard_id, shard.head))
            .collect::<Vec<_>>(),
        fixture.roots
    );
    let page = storage.catalogue().history("main", None, 2)?;
    assert_eq!(
        page.iter()
            .map(|captured| captured.version)
            .collect::<Vec<_>>(),
        [fixture.predecessor, fixture.current]
    );
    assert_eq!(
        page[0]
            .shards
            .iter()
            .map(|shard| (shard.shard_id, shard.head))
            .collect::<Vec<_>>(),
        fixture.roots
    );
    Ok(())
}

#[test]
fn clean_open_counts_every_regular_file_open_independently_of_kept_versions() -> TestResult {
    let large = build(LARGE)?;
    let small = build(SMALL)?;
    check_open(&large)?;
    check_open(&small)?;
    Ok(())
}

fn legacy_record_bytes(record: &RefRecord) -> TestResult<Vec<u8>> {
    let mut bytes = b"HBR1".to_vec();
    bytes.extend_from_slice(&u64::try_from(record.name.len())?.to_le_bytes());
    bytes.extend_from_slice(record.name.as_bytes());
    for value in [
        record.created,
        record.seq,
        record.timestamp,
        u64::try_from(record.shards.len())?,
    ] {
        bytes.extend_from_slice(&value.to_le_bytes());
    }
    for shard in &record.shards {
        bytes.extend_from_slice(&u64::try_from(shard.shard_id)?.to_le_bytes());
        bytes.extend_from_slice(shard.fork_anchor.as_bytes());
        bytes.extend_from_slice(shard.head.as_bytes());
    }
    bytes.extend_from_slice(&u64::try_from(record.parents.len())?.to_le_bytes());
    for parent in &record.parents {
        bytes.extend_from_slice(parent.as_bytes());
    }
    Ok(bytes)
}

fn build_legacy(versions: u64) -> TestResult<Fixture> {
    if !matches!(versions, LEGACY_SMALL | LEGACY_LARGE) {
        return Err("legacy fixture size must be one of the two declared cases".into());
    }
    let fixture = build_versions(
        versions.checked_add(1).ok_or("fixture size overflow")?,
        Some([10, 20, 30]),
    )?;
    let refs_dir = fixture
        .directory
        .path()
        .join("database")
        .join(CANONICAL_REFS_DIR);
    assert!(!refs_dir.join("FORMAT").exists());
    for offset in 0..versions {
        let sequence = fixture
            .old
            .sequence
            .checked_add(offset)
            .ok_or("legacy history sequence overflow")?;
        let name = format!(
            "@haem/history/{}/{}/{sequence:020}",
            Hash::of(b"main"),
            fixture.old.generation,
        );
        let reference = RefRecord {
            name: name.clone(),
            created: fixture.old.generation,
            kind: Kind::Work,
            namespace_lineage: None,
            seq: sequence,
            timestamp: fixture.old.generation,
            shards: fixture
                .roots
                .iter()
                .map(|(shard_id, root)| haematite::branch::ShardRef {
                    shard_id: *shard_id,
                    fork_anchor: *root,
                    head: *root,
                })
                .collect(),
            parents: Vec::new(),
        };
        let digest = Hash::of(name.as_bytes()).to_string();
        let prefix = digest.get(..32).ok_or("reference hash is too short")?;
        let mut file = std::fs::OpenOptions::new()
            .create_new(true)
            .write(true)
            .mode(0o600)
            .open(refs_dir.join(format!("{prefix}.ref")))?;
        file.write_all(&legacy_record_bytes(&reference)?)?;
    }
    std::fs::File::open(&refs_dir)?.sync_all()?;
    let refs = RefStore::open(&refs_dir)?;
    assert_eq!(
        u64::try_from(refs.list_prefix("@haem/history/").count())?,
        versions,
    );
    Ok(fixture)
}

struct OpenCost {
    files: u64,
    bytes: u64,
    wal_file_bytes: [u64; SHARDS],
}

fn wal_path(data: &Path, shard: usize) -> PathBuf {
    data.join(format!("shard-{shard}")).join("shard.wal")
}

fn wal_lengths(data: &Path) -> TestResult<[u64; SHARDS]> {
    let mut lengths = [0; SHARDS];
    for (shard, length) in lengths.iter_mut().enumerate() {
        *length = std::fs::metadata(wal_path(data, shard))?.len();
    }
    Ok(lengths)
}

fn first_open_cost(fixture: &Fixture) -> TestResult<OpenCost> {
    let data = fixture.directory.path().join("database");
    let wal_file_bytes = wal_lengths(&data)?;
    reset_file_opens();
    reset_file_read_bytes();
    let storage = Storage::open(&data, NodeCacheBudget::Unlimited, long_wait_interval())?;
    let cost = OpenCost {
        files: file_opens(),
        bytes: file_read_bytes(),
        wal_file_bytes,
    };
    drop(storage);
    Ok(cost)
}

fn sum_bytes(bytes: &[u64]) -> TestResult<u64> {
    bytes
        .iter()
        .try_fold(0_u64, |sum, count| sum.checked_add(*count))
        .ok_or_else(|| "fixture byte count overflow".into())
}

fn print_open_cost(versions: u64, phase: &str, cost: &OpenCost) -> TestResult {
    let wal_file_bytes = sum_bytes(&cost.wal_file_bytes)?;
    println!(
        "BOUND_MEASURE legacy_refs={versions} phase={phase} file_opens={} bytes_read={} \
         wal_file_bytes={wal_file_bytes} wal_file_bytes_by_shard={:?}",
        cost.files, cost.bytes, cost.wal_file_bytes,
    );
    Ok(())
}

fn copy_fixture_directory(source: &Path, destination: &Path) -> TestResult {
    std::fs::DirBuilder::new().mode(0o700).create(destination)?;
    for entry in std::fs::read_dir(source)? {
        let entry = entry?;
        let kind = entry.file_type()?;
        let copied = destination.join(entry.file_name());
        if kind.is_dir() {
            copy_fixture_directory(&entry.path(), &copied)?;
        } else if kind.is_file() {
            std::fs::copy(entry.path(), copied)?;
        } else {
            return Err("fixture copy contains a nonregular entry".into());
        }
    }
    Ok(())
}

fn recovery_probe(fixture: &Fixture, versions: u64) -> TestResult {
    let directory = tempfile::tempdir()?;
    let data = directory.path().join("database");
    copy_fixture_directory(&fixture.directory.path().join("database"), &data)?;
    let wal_file_bytes = wal_lengths(&data)?;
    let mut probe_bytes = [0; SHARDS];
    for (shard, bytes) in probe_bytes.iter_mut().enumerate() {
        let recovered = Recovery::open(wal_path(&data, shard))?.recover_unverified()?;
        *bytes = recovered.bytes_read();
    }
    println!(
        "BOUND_MEASURE legacy_refs={versions} phase=recovery_probe wal_file_bytes={} \
         wal_file_bytes_by_shard={wal_file_bytes:?} \
         recovery_probe_bytes_read (not served-open)={} \
         recovery_probe_bytes_read_by_shard (not served-open)={probe_bytes:?}",
        sum_bytes(&wal_file_bytes)?,
        sum_bytes(&probe_bytes)?,
    );
    Ok(())
}

#[test]
fn legacy_refs_startup_counts_and_wal_probe() -> TestResult {
    let small = build_legacy(LEGACY_SMALL)?;
    let small_first = first_open_cost(&small)?;
    let small_reopen = first_open_cost(&small)?;
    let large = build_legacy(LEGACY_LARGE)?;
    let large_first = first_open_cost(&large)?;
    let large_reopen = first_open_cost(&large)?;
    print_open_cost(LEGACY_SMALL, "first_open", &small_first)?;
    print_open_cost(LEGACY_SMALL, "reopen", &small_reopen)?;
    print_open_cost(LEGACY_LARGE, "first_open", &large_first)?;
    print_open_cost(LEGACY_LARGE, "reopen", &large_reopen)?;
    recovery_probe(&small, LEGACY_SMALL)?;
    recovery_probe(&large, LEGACY_LARGE)?;
    Ok(())
}

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