use std::fs;
use std::io::Write;
use std::path::Path;
use std::sync::Arc;
use haematite::{Database, DatabaseConfig, EventStore, NodeCacheBudget, Retention};
use liminal::durability::bridge::block_on;
use liminal::durability::{
ClaimEvidence, CountingStore, DedupCache, DedupDecision, DurabilityError, DurableStore,
HaematiteStore, ProcessingReceipt, 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])
}
fn create(path: &Path) -> FixtureResult<CountingStore<HaematiteStore>> {
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),
})?;
Ok(CountingStore::new(HaematiteStore::new(Arc::new(
EventStore::new(database),
))))
}
fn check_counts(
name: &str,
store: &CountingStore<HaematiteStore>,
batches: u64,
rows: u64,
values: u64,
) -> FixtureResult<()> {
let counts = 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());
}
println!("fixture {name}: {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 store = create(&root.join(name))?;
block_on(seeded.land_on(&store))??;
check_counts(name, &store, 1282, if reclaimed { 3847 } else { 3844 }, 1)?;
drop(store);
Ok(())
}
fn pages(root: &Path) -> FixtureResult<()> {
let store = create(&root.join("paged-read"))?;
let payloads = (0_u64..256)
.map(|sequence| sequence.to_be_bytes().to_vec())
.collect();
block_on(store.append_all("liminal/p0-60/paged-read-shape", payloads, 0))??;
block_on(store.flush())??;
check_counts("paged-read", &store, 1, 256, 0)?;
drop(store);
let store = 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(store.append_all(stream, payloads, 0))??;
}
block_on(store.flush())??;
check_counts("paged-scan", &store, 3, 23, 0)?;
drop(store);
let store = create(&root.join("paged-walk"))?;
for index in 0..48 {
let stream = format!("liminal/scan-bound/walk-{index:03}");
block_on(store.append_all(&stream, vec![format!("{stream}#0").into_bytes()], 0))??;
}
block_on(store.flush())??;
check_counts("paged-walk", &store, 48, 48, 0)?;
drop(store);
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),
})?;
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(())
}