liminal-rs 0.16.2

// The fixtures writer has no tests of its own (Cargo.toml: test = false); the gate's explicit `--bins` still
// builds its test harness, which must then hold nothing, not a fixture writer whose `land_on` is `cfg(not(test))`.
#![cfg(not(test))]

use std::fs;
use std::io::Write;
use std::path::Path;
use std::sync::Arc;

use std::num::NonZeroU64;

use haematite::{CommitPolicy, Database, DatabaseConfig, EventStore, NodeCacheBudget, Retention};
use liminal::durability::bridge::block_on;
use liminal::durability::{
    ClaimEvidence, CountingStore, DedupCache, DedupDecision, DurabilityError, DurableStore,
    HaematiteStore, ProcessingReceipt, RoutedBatchEntry, RoutedBatchItem, ScanCursor, ScanPage,
    StoredEntry,
};

#[path = "../durability/tests_memory_store.rs"]
mod tests_memory_store;

use tests_memory_store::MemoryStore;

type FixtureResult<T> = Result<T, Box<dyn std::error::Error>>;

const SETTLED: usize = 1280;
const NAMESPACE: &str = "liminal:in-flight-test";

fn evidence() -> ClaimEvidence {
    ClaimEvidence::new("orders", vec![0])
}

/// The commit policy a fixture image is built under: liminal-server's spine
/// policy (buffered at 1 MiB, `SPINE_COMMIT_POLICY` in its connection
/// services). Each `append_all` is permanent through its WAL frames and one
/// device flush, and the tree commits once, at the land's single `flush`,
/// where the per-append policy committed the tree and its marker for every
/// stream. The largest image (ceiling-reclaimed, 171,516 bytes over 1,282
/// streams) stays under the threshold, so no image commits before its flush.
const FIXTURE_COMMIT_POLICY: CommitPolicy =
    CommitPolicy::buffered(match NonZeroU64::new(1 << 20) {
        Some(threshold) => threshold,
        None => unreachable!(),
    });

/// A fixture image being built: the counted store the land writes through,
/// and its event store, whose database runs reclaim once the land is written.
struct Image {
    store: CountingStore<HaematiteStore>,
    events: Arc<EventStore>,
}

impl Image {
    /// Releases the reclaim held since open and runs it to rest once, so the
    /// image is left as a running store would leave it: every root but the
    /// last reclaimed, one pass for the whole land rather than one pass per
    /// stream committed.
    fn settle(&self) -> FixtureResult<()> {
        let database = self.events.database();
        database.release_reclaim()?;
        database.quiesce_reclaim()?;
        Ok(())
    }
}

/// Creates an empty one-shard image at `path`, its reclaim held until
/// [`Image::settle`] and its appends buffered under [`FIXTURE_COMMIT_POLICY`].
fn create(path: &Path) -> FixtureResult<Image> {
    fs::create_dir(path)?;
    let database = Database::create(
        DatabaseConfig {
            data_dir: path.to_path_buf(),
            shard_count: 1,
            distributed: None,
            executor_threads: None,
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: Retention::reclaim(0),
        },
        haematite::RuntimeOptions {
            long_wait_interval: std::time::Duration::from_secs(5),
            reclaim_start: haematite::ReclaimStart::Held,
        },
    )?;
    let events = Arc::new(EventStore::with_commit_policy(
        database,
        FIXTURE_COMMIT_POLICY,
    ));
    Ok(Image {
        store: CountingStore::new(HaematiteStore::new(Arc::clone(&events))),
        events,
    })
}

fn check_counts(
    name: &str,
    image: &Image,
    batches: u64,
    rows: u64,
    values: u64,
) -> FixtureResult<()> {
    image.settle()?;
    let counts = image.store.snapshot();
    if counts.append_calls != 0
        || counts.append_all_calls != batches
        || counts.append_all_rows != rows
        || counts.cas_calls != values
        || counts.flushes != 1
    {
        return Err(format!("durability fixture {name} write count mismatch: {counts:?}").into());
    }
    let database = image.events.database();
    println!(
        "fixture {name}: {counts:?}; device flushes {}; {:?}",
        database.device_flushes(),
        database.sync_counts()
    );
    Ok(())
}

fn ceiling(root: &Path, reclaimed: bool) -> FixtureResult<()> {
    let seeded = Arc::new(MemoryStore::default());
    let running = DedupCache::new(Arc::clone(&seeded) as Arc<dyn DurableStore>, NAMESPACE);
    if reclaimed {
        block_on(running.claim_with_evidence("k", evidence(), 1))??;
        block_on(running.release_claim("k"))??;
        block_on(running.claim_with_evidence("k", evidence(), 2))??;
    } else {
        block_on(running.claim_with_evidence("held", evidence(), 1))??;
    }
    for index in 0..SETTLED {
        let key = format!("settled-{index}");
        let decision = block_on(running.claim_with_evidence(&key, evidence(), 1))??;
        if !reclaimed {
            assert_eq!(decision, DedupDecision::Claimed);
        }
        if reclaimed || index % 2 == 0 {
            block_on(running.complete_receipt(&key, ProcessingReceipt::for_position(1)))??;
        } else {
            block_on(running.release_claim(&key))??;
        }
    }
    drop(running);
    let name = if reclaimed {
        "ceiling-reclaimed"
    } else {
        "ceiling-held"
    };
    let image = create(&root.join(name))?;
    block_on(seeded.land_on(&image.store))??;
    check_counts(name, &image, 1282, if reclaimed { 3847 } else { 3844 }, 1)?;
    drop(image);
    Ok(())
}

fn pages(root: &Path) -> FixtureResult<()> {
    let image = create(&root.join("paged-read"))?;
    let payloads = (0_u64..256)
        .map(|sequence| sequence.to_be_bytes().to_vec())
        .collect();
    block_on(
        image
            .store
            .append_all("liminal/p0-60/paged-read-shape", payloads, 0),
    )??;
    block_on(image.store.flush())??;
    check_counts("paged-read", &image, 1, 256, 0)?;
    drop(image);

    let image = create(&root.join("paged-scan"))?;
    for (stream, rows) in [
        ("liminal/scan-bound/alpha", 7),
        ("liminal/scan-bound/bravo", 11),
        ("liminal/scan-bound/charlie", 5),
    ] {
        let payloads = (0..rows)
            .map(|sequence| format!("{stream}#{sequence}").into_bytes())
            .collect();
        block_on(image.store.append_all(stream, payloads, 0))??;
    }
    block_on(image.store.flush())??;
    check_counts("paged-scan", &image, 3, 23, 0)?;
    drop(image);

    let image = create(&root.join("paged-walk"))?;
    for index in 0..48 {
        let stream = format!("liminal/scan-bound/walk-{index:03}");
        block_on(
            image
                .store
                .append_all(&stream, vec![format!("{stream}#0").into_bytes()], 0),
        )??;
    }
    block_on(image.store.flush())??;
    check_counts("paged-walk", &image, 48, 48, 0)?;
    drop(image);
    Ok(())
}

fn full_services(root: &Path) -> FixtureResult<()> {
    let path = root.join("full-services");
    fs::create_dir(&path)?;
    let database = Database::create(
        DatabaseConfig {
            data_dir: path.join("durability"),
            shard_count: 8,
            distributed: None,
            executor_threads: None,
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: Retention::reclaim(0),
        },
        haematite::RuntimeOptions {
            long_wait_interval: std::time::Duration::from_secs(5),
            reclaim_start: haematite::ReclaimStart::AtOpen,
        },
    )?;
    drop(database);
    if !path.join("durability/config.json").is_file() {
        return Err("full-services fixture creation left no database configuration".into());
    }
    println!("full-services fixture: one empty database, eight shards");
    Ok(())
}

fn main() -> FixtureResult<()> {
    let env_file =
        std::env::var_os("NEXTEST_ENV").ok_or("durability fixture builder requires NEXTEST_ENV")?;
    let run = std::env::var("NEXTEST_RUN_ID")?;
    uuid::Uuid::parse_str(&run)?;
    let executable = std::env::current_exe()?;
    let target = executable
        .parent()
        .ok_or("durability fixture binary has no target parent")?;
    let directory = tempfile::Builder::new()
        .prefix(&format!("durability-fixtures-{run}-"))
        .tempdir_in(target)?;
    ceiling(directory.path(), false)?;
    ceiling(directory.path(), true)?;
    pages(directory.path())?;
    full_services(directory.path())?;
    fs::write(
        directory.path().join("manifest.json"),
        serde_json::to_vec(&serde_json::json!({"settled_claims": SETTLED}))?,
    )?;
    let path = directory
        .path()
        .to_str()
        .ok_or("durability fixture path is not UTF-8")?;
    let mut environment = fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(env_file)?;
    writeln!(environment, "LIMINAL_DURABILITY_FIXTURES={path}")?;
    environment.flush()?;
    let root = directory.keep();
    println!("durability fixtures built once at {}", root.display());
    Ok(())
}