diff --git a/AGENTS.md b/AGENTS.md index 0e03277ef..bfe75a7e1 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -146,7 +146,7 @@ The project follows a modular workspace architecture with clear separation of co #### `xtask` (Development Automation) - **Purpose**: Development task automation following cargo-xtask pattern -- **Role**: Build scripts and development utilities, including `cargo xtask archive-bench` — the archive segment benchmarks (store-level presets, node-level paired runs of two `dolos` binaries, dictionary tooling, report); see `xtask/archive-bench/README.md` +- **Role**: Build scripts and development utilities, including `cargo xtask perf` — storage and minibf performance experiments with shared measurement and reporting; see `xtask/perf/README.md` ## Dependency Flow diff --git a/Cargo.lock b/Cargo.lock index feb446e77..dbd7b9cd5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1683,6 +1683,7 @@ dependencies = [ "itertools 0.14.0", "pallas", "rand 0.9.3", + "serde", "tempfile", "tokio", "tokio-stream", @@ -7111,6 +7112,7 @@ name = "xtask" version = "0.1.0" dependencies = [ "anyhow", + "axum 0.8.4", "bech32 0.11.1", "chrono", "clap", @@ -7118,12 +7120,15 @@ dependencies = [ "dolos-cardano", "dolos-core", "dolos-flatfiles", + "dolos-minibf", + "dolos-testing", "hdrhistogram", "hex", "libc", "pallas", "postgres", "postgres-native-tls", + "reqwest 0.12.28", "serde", "serde_json", "sha2 0.10.9", @@ -7131,6 +7136,8 @@ dependencies = [ "tokio", "toml 0.8.20", "tonic 0.12.3", + "tower 0.4.13", + "vergen-gitcl", "xshell", "zstd", ] diff --git a/benches/archive_backends.rs b/benches/archive_backends.rs index 5628ebd77..fe1722308 100644 --- a/benches/archive_backends.rs +++ b/benches/archive_backends.rs @@ -1,10 +1,12 @@ -//! Regression guards for the archive backend's three read-shape families. +//! Regression guards for logs, store-local indexes and compressed bodies. //! //! The store is populated with the synthetic content from //! `dolos_testing::archive` and measured against a fixed key sample: block //! location resolution and full block reads (`get_block_by_slot`), the //! per-account reward point reads (`read_logs` per epoch), and the per-epoch //! temporal-prefix scan (`iter_logs`). +//! The queries module adds populated-domain index lookups, valid compressed +//! bodies, a real Alonzo block and logs-only scale cases. //! //! These are *relative* regression guards on small tempdir populations. //! Absolute numbers say nothing about production behavior — the @@ -16,6 +18,9 @@ use dolos_core::{ }; use dolos_testing::archive::{populate_archive, ArchiveShape}; +#[path = "archive_backends/queries.rs"] +mod queries; + /// The one namespace the log shapes exercise; the name is the real bulk /// namespace so the bench reads like the node. const NS: &str = "account-epochs"; diff --git a/benches/archive_backends/queries.rs b/benches/archive_backends/queries.rs new file mode 100644 index 000000000..8cacc31c5 --- /dev/null +++ b/benches/archive_backends/queries.rs @@ -0,0 +1,182 @@ +use dolos_core::{ArchiveStore, Domain, StateStore}; +use dolos_testing::{ + performance::{ApiFixture, FixtureShape}, + toy_domain::{FjallStores, ToyStores}, +}; +use pallas::ledger::{addresses::Address, traverse::MultiEraBlock}; + +fn fixture(blocks: usize) -> ApiFixture { + ApiFixture::new( + FjallStores::open(), + FixtureShape { + blocks, + ..Default::default() + }, + ) + .unwrap() +} + +#[divan::bench(args = [16, 64], sample_count = 30)] +fn archive_address_tags(bencher: divan::Bencher, blocks: usize) { + let fixture = fixture(blocks); + let key = Address::from_bech32(&fixture.vectors.address) + .unwrap() + .to_vec(); + let expected: Vec<_> = fixture + .vectors + .blocks + .iter() + .take(blocks) + .map(|block| block.slot) + .collect(); + bencher.bench_local(|| { + let actual = fixture + .domain + .archive() + .slots_by_tag("address", &key, 0, u64::MAX) + .unwrap() + .collect::, _>>() + .unwrap(); + assert_eq!(divan::black_box(actual), expected); + }); +} + +#[divan::bench(args = [16, 64], sample_count = 30)] +fn exact_transaction_lookup(bencher: divan::Bencher, blocks: usize) { + let fixture = fixture(blocks); + let keys: Vec<_> = fixture + .vectors + .blocks + .iter() + .take(blocks) + .map(|block| (hex::decode(&block.tx_hashes[0]).unwrap(), block.slot)) + .collect(); + bencher.bench_local(|| { + for (hash, slot) in &keys { + assert_eq!( + fixture + .domain + .archive() + .slot_by_tx_hash(divan::black_box(hash)) + .unwrap(), + Some(*slot) + ); + } + assert_eq!( + fixture + .domain + .archive() + .slot_by_tx_hash(&[0xff; 32]) + .unwrap(), + None + ); + }); +} + +#[divan::bench(args = [16, 64], sample_count = 30)] +fn state_address_tags(bencher: divan::Bencher, blocks: usize) { + let fixture = fixture(blocks); + let key = Address::from_bech32(&fixture.vectors.address) + .unwrap() + .to_vec(); + bencher.bench_local(|| { + let actual = fixture + .domain + .state() + .utxos_by_tag("address", divan::black_box(&key)) + .unwrap(); + assert_eq!(actual.len(), blocks); + }); +} + +#[divan::bench(args = [16, 64], sample_count = 30)] +fn compressed_valid_block_reads(bencher: divan::Bencher, blocks: usize) { + let fixture = fixture(blocks); + bencher.bench_local(|| { + for (index, expected) in fixture.blocks.iter().enumerate().step_by(3) { + let slot = fixture.vectors.blocks[index].slot; + let body = fixture + .domain + .archive() + .get_block_by_slot(&slot) + .unwrap() + .unwrap(); + assert_eq!(divan::black_box(body.as_slice()), expected.as_slice()); + } + }); +} + +#[divan::bench(args = [16, 64], sample_count = 30)] +fn compressed_reverse_block_scan(bencher: divan::Bencher, blocks: usize) { + let fixture = fixture(blocks); + bencher.bench_local(|| { + let mut count = 0; + for (slot, body) in fixture + .domain + .archive() + .get_range(Some(1), Some(blocks as u64 + 1)) + .unwrap() + .rev() + { + let block = MultiEraBlock::decode(&body).unwrap(); + assert_eq!(block.slot(), slot); + divan::black_box(block.header().hash()); + count += 1; + } + assert_eq!(count, blocks); + }); +} + +#[divan::bench(args = [1000, 10000], sample_count = 30)] +fn logs_only_scale(bencher: divan::Bencher, rows: u64) { + use dolos_testing::archive::{populate_archive, ArchiveShape}; + let shape = ArchiveShape { + epochs: 4, + blocks_per_epoch: 0, + log_rows_per_epoch: rows, + slots_per_epoch: 432_000, + seed: 0, + }; + let directory = tempfile::tempdir().unwrap(); + let store = dolos_fjall::archive::ArchiveStore::open( + dolos_cardano::model::build_schema(), + directory.path(), + &Default::default(), + ) + .unwrap(); + populate_archive(&store, "account-epochs", &shape).unwrap(); + assert!(store.get_tip().unwrap().is_none()); + let range = shape.log_key(2, 0)..shape.log_key(3, 0); + bencher.bench_local(|| { + let mut count = 0; + for entry in store.iter_logs("account-epochs", range.clone()).unwrap() { + divan::black_box(entry.unwrap()); + count += 1; + } + assert_eq!(count, rows); + }); +} + +#[divan::bench(sample_count = 50)] +fn compressed_real_alonzo_block(bencher: divan::Bencher) { + use dolos_core::{ArchiveWriter, ChainPoint}; + let body = std::sync::Arc::new( + hex::decode(include_str!("../../crates/cardano/test_data/alonzo27.block").trim()).unwrap(), + ); + let block = MultiEraBlock::decode(&body).unwrap(); + let point = ChainPoint::Specific(block.slot(), block.hash()); + let directory = tempfile::tempdir().unwrap(); + let store = dolos_fjall::archive::ArchiveStore::open( + dolos_cardano::model::build_schema(), + directory.path(), + &Default::default(), + ) + .unwrap(); + let writer = store.start_writer().unwrap(); + writer.apply(&point, &body).unwrap(); + writer.commit().unwrap(); + bencher.bench_local(|| { + let actual = store.get_block_by_slot(&point.slot()).unwrap().unwrap(); + assert_eq!(divan::black_box(actual.as_slice()), body.as_slice()); + }); +} diff --git a/crates/flatfiles/dictionary/README.md b/crates/flatfiles/dictionary/README.md index 83fde89e5..e3445ce94 100644 --- a/crates/flatfiles/dictionary/README.md +++ b/crates/flatfiles/dictionary/README.md @@ -24,7 +24,7 @@ command over the same files yields the same bytes. Three candidates were trained from the same corpus with the same seed and evaluated with the harness's `evaluate` command (then -`dolos-archive-bench evaluate`, now `cargo xtask archive-bench evaluate`) +`dolos-archive-bench evaluate`, now `cargo xtask perf storage evaluate`) on data disjoint from every training segment: mainnet segments 448–455 (156,621 Conway blocks, 877 MiB, the held-out window), one 20,000-block fixture per mainnet era, preprod diff --git a/crates/testing/Cargo.toml b/crates/testing/Cargo.toml index 10616e2d2..2de5467a4 100644 --- a/crates/testing/Cargo.toml +++ b/crates/testing/Cargo.toml @@ -18,3 +18,4 @@ tokio-stream = { workspace = true } futures-core = { workspace = true } futures-util = { workspace = true } itertools = { workspace = true } +serde = { workspace = true, features = ["derive"] } diff --git a/crates/testing/src/archive.rs b/crates/testing/src/archive.rs index 713688dec..3f2049f36 100644 --- a/crates/testing/src/archive.rs +++ b/crates/testing/src/archive.rs @@ -53,7 +53,7 @@ impl ArchiveShape { /// [`Self::block_slots`] and [`populate_archive`] so the slots callers /// sample are exactly the slots the population wrote. pub fn stride(&self) -> u64 { - (self.slots_per_epoch / self.blocks_per_epoch).max(1) + (self.slots_per_epoch / self.blocks_per_epoch.max(1)).max(1) } /// Every block slot the population writes, in ascending order. Callers @@ -97,33 +97,64 @@ fn filler(rng: &mut SplitMix64, len: usize) -> Vec { out } -/// Write `shape` into `store`: one committed writer per epoch carrying that -/// epoch's blocks (~1-2 KiB bodies) and its log rows (~30-60 B values) under -/// `ns`, which must exist in the store's schema. +/// Write `shape` into `store` in batches of at most 1,000 records. Blocks +/// contain filler (~1-2 KiB) and log values contain filler (~30-60 B). +/// Zero blocks per epoch produces a logs-only population. pub fn populate_archive( store: &S, ns: Namespace, shape: &ArchiveShape, ) -> Result<(), ArchiveError> { + populate_archive_batched(store, ns, shape, 1000) +} + +pub fn populate_archive_batched( + store: &S, + ns: Namespace, + shape: &ArchiveShape, + batch_size: usize, +) -> Result<(), ArchiveError> { + if batch_size == 0 + || shape.slots_per_epoch == 0 + || shape.blocks_per_epoch > shape.slots_per_epoch + || shape.epochs.checked_mul(shape.slots_per_epoch).is_none() + { + return Err(ArchiveError::InternalError( + "invalid archive population shape or batch size".into(), + )); + } let mut rng = SplitMix64(shape.seed); let stride = shape.stride(); for epoch in 0..shape.epochs { - let writer = store.start_writer()?; let epoch_start = shape.epoch_start(epoch); - for i in 0..shape.blocks_per_epoch { - let slot = epoch_start + i * stride; - let len = 1024 + (rng.next_u64() % 1024) as usize; - writer.apply(&block_point(slot), &Arc::new(filler(&mut rng, len)))?; + for start in (0..shape.blocks_per_epoch).step_by(batch_size) { + let writer = store.start_writer()?; + for block in start + ..start + .saturating_add(batch_size as u64) + .min(shape.blocks_per_epoch) + { + let slot = epoch_start + block * stride; + let len = 1024 + (rng.next_u64() % 1024) as usize; + writer.apply(&block_point(slot), &Arc::new(filler(&mut rng, len)))?; + } + writer.commit()?; } - for i in 0..shape.log_rows_per_epoch { - let len = 30 + (rng.next_u64() % 31) as usize; - writer.write_log(ns, &shape.log_key(epoch, i), &filler(&mut rng, len))?; + for start in (0..shape.log_rows_per_epoch).step_by(batch_size) { + let writer = store.start_writer()?; + for row in start + ..start + .saturating_add(batch_size as u64) + .min(shape.log_rows_per_epoch) + { + let len = 30 + (rng.next_u64() % 31) as usize; + writer.write_log(ns, &shape.log_key(epoch, row), &filler(&mut rng, len))?; + } + writer.commit()?; } - - writer.commit()?; } Ok(()) diff --git a/crates/testing/src/lib.rs b/crates/testing/src/lib.rs index 23fb31d02..add9526ab 100644 --- a/crates/testing/src/lib.rs +++ b/crates/testing/src/lib.rs @@ -23,6 +23,8 @@ use dolos_core::*; pub mod archive; pub mod blocks; pub mod faults; +pub mod measured; +pub mod performance; pub mod synthetic; pub mod toy_domain; diff --git a/crates/testing/src/measured.rs b/crates/testing/src/measured.rs new file mode 100644 index 000000000..0d05e4b54 --- /dev/null +++ b/crates/testing/src/measured.rs @@ -0,0 +1,376 @@ +use std::{ + ops::Range, + sync::{ + atomic::{AtomicU64, Ordering}, + Arc, + }, +}; + +use dolos_core::{ + archive::Skippable, ArchiveError, ArchiveStore, BlockBody, BlockSlot, ChainPoint, EntityKey, + EntityValue, LogKey, Namespace, StateError, StateStore, TagDimension, TxoRef, UtxoMap, UtxoSet, +}; +use serde::Serialize; + +use crate::toy_domain::ToyStores; + +#[derive(Clone, Default)] +pub struct WorkCounters { + pub log_rows: Arc, + pub log_reads: Arc, + pub block_reads: Arc, + pub decoded_bytes: Arc, + pub exact_lookups: Arc, + pub tag_candidates: Arc, + pub state_entities: Arc, + pub utxo_refs: Arc, + pub utxo_reads: Arc, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)] +pub struct WorkSnapshot { + pub log_rows: u64, + pub log_reads: u64, + pub block_reads: u64, + pub decoded_bytes: u64, + pub exact_lookups: u64, + pub tag_candidates: u64, + pub state_entities: u64, + pub utxo_refs: u64, + pub utxo_reads: u64, +} + +impl WorkCounters { + pub fn snapshot(&self) -> WorkSnapshot { + WorkSnapshot { + log_rows: self.log_rows.load(Ordering::Relaxed), + log_reads: self.log_reads.load(Ordering::Relaxed), + block_reads: self.block_reads.load(Ordering::Relaxed), + decoded_bytes: self.decoded_bytes.load(Ordering::Relaxed), + exact_lookups: self.exact_lookups.load(Ordering::Relaxed), + tag_candidates: self.tag_candidates.load(Ordering::Relaxed), + state_entities: self.state_entities.load(Ordering::Relaxed), + utxo_refs: self.utxo_refs.load(Ordering::Relaxed), + utxo_reads: self.utxo_reads.load(Ordering::Relaxed), + } + } + + pub fn reset(&self) { + self.log_rows.store(0, Ordering::Relaxed); + self.log_reads.store(0, Ordering::Relaxed); + self.block_reads.store(0, Ordering::Relaxed); + self.decoded_bytes.store(0, Ordering::Relaxed); + self.exact_lookups.store(0, Ordering::Relaxed); + self.tag_candidates.store(0, Ordering::Relaxed); + self.state_entities.store(0, Ordering::Relaxed); + self.utxo_refs.store(0, Ordering::Relaxed); + self.utxo_reads.store(0, Ordering::Relaxed); + } + + fn body(&self, body: &[u8]) { + self.block_reads.fetch_add(1, Ordering::Relaxed); + self.decoded_bytes + .fetch_add(body.len() as u64, Ordering::Relaxed); + } +} + +pub struct CountedIter { + inner: Inner, + count: Arc, +} + +impl Iterator for CountedIter { + type Item = Inner::Item; + + fn next(&mut self) -> Option { + let item = self.inner.next()?; + self.count.fetch_add(1, Ordering::Relaxed); + Some(item) + } +} + +impl DoubleEndedIterator for CountedIter { + fn next_back(&mut self) -> Option { + let item = self.inner.next_back()?; + self.count.fetch_add(1, Ordering::Relaxed); + Some(item) + } +} + +pub struct CountedBlocks { + inner: Inner, + counters: WorkCounters, +} + +impl> Iterator for CountedBlocks { + type Item = (BlockSlot, BlockBody); + + fn next(&mut self) -> Option { + let item = self.inner.next()?; + self.counters.body(&item.1); + Some(item) + } +} + +impl> DoubleEndedIterator + for CountedBlocks +{ + fn next_back(&mut self) -> Option { + let item = self.inner.next_back()?; + self.counters.body(&item.1); + Some(item) + } +} + +impl Skippable for CountedBlocks { + fn skip_forward(&mut self, count: usize) { + self.inner.skip_forward(count); + } + + fn skip_backward(&mut self, count: usize) { + self.inner.skip_backward(count); + } +} + +#[derive(Clone)] +pub struct MeasuredArchive { + inner: Inner, + pub counters: WorkCounters, +} + +impl ArchiveStore for MeasuredArchive { + type BlockIter<'a> = CountedBlocks>; + type Writer = Inner::Writer; + type LogIter = CountedIter; + type EntityValueIter = Inner::EntityValueIter; + type SlotIter = CountedIter; + type TagIter = Inner::TagIter; + type ExactIter = Inner::ExactIter; + + fn start_writer(&self) -> Result { + self.inner.start_writer() + } + + fn read_logs( + &self, + ns: Namespace, + keys: &[&LogKey], + ) -> Result>, ArchiveError> { + self.counters + .log_reads + .fetch_add(keys.len() as u64, Ordering::Relaxed); + self.inner.read_logs(ns, keys) + } + + fn iter_logs( + &self, + ns: Namespace, + range: Range, + ) -> Result { + Ok(CountedIter { + inner: self.inner.iter_logs(ns, range)?, + count: self.counters.log_rows.clone(), + }) + } + + fn get_block_by_slot(&self, slot: &BlockSlot) -> Result, ArchiveError> { + let body = self.inner.get_block_by_slot(slot)?; + if let Some(body) = &body { + self.counters.body(body); + } + Ok(body) + } + + fn get_blocks_by_slot(&self, slot: &BlockSlot) -> Result, ArchiveError> { + let bodies = self.inner.get_blocks_by_slot(slot)?; + for body in &bodies { + self.counters.body(body); + } + Ok(bodies) + } + + fn get_range<'a>( + &self, + from: Option, + to: Option, + ) -> Result, ArchiveError> { + Ok(CountedBlocks { + inner: self.inner.get_range(from, to)?, + counters: self.counters.clone(), + }) + } + + fn find_intersect(&self, intersect: &[ChainPoint]) -> Result, ArchiveError> { + self.inner.find_intersect(intersect) + } + + fn get_tip(&self) -> Result, ArchiveError> { + let tip = self.inner.get_tip()?; + if let Some((_, body)) = &tip { + self.counters.body(body); + } + Ok(tip) + } + + fn prune_history(&self, max_slots: u64, max_prune: Option) -> Result { + self.inner.prune_history(max_slots, max_prune) + } + + fn truncate_front(&self, after: &ChainPoint) -> Result<(), ArchiveError> { + self.inner.truncate_front(after) + } + + fn slot_by_block_hash(&self, hash: &[u8]) -> Result, ArchiveError> { + self.counters.exact_lookups.fetch_add(1, Ordering::Relaxed); + self.inner.slot_by_block_hash(hash) + } + + fn slot_by_tx_hash(&self, hash: &[u8]) -> Result, ArchiveError> { + self.counters.exact_lookups.fetch_add(1, Ordering::Relaxed); + self.inner.slot_by_tx_hash(hash) + } + + fn slot_by_block_number(&self, number: u64) -> Result, ArchiveError> { + self.counters.exact_lookups.fetch_add(1, Ordering::Relaxed); + self.inner.slot_by_block_number(number) + } + + fn slots_by_tag( + &self, + dimension: TagDimension, + key: &[u8], + start: BlockSlot, + end: BlockSlot, + ) -> Result { + Ok(CountedIter { + inner: self.inner.slots_by_tag(dimension, key, start, end)?, + count: self.counters.tag_candidates.clone(), + }) + } + + fn iter_archive_tags( + &self, + dimensions: &[TagDimension], + slots: Range, + ) -> Result { + self.inner.iter_archive_tags(dimensions, slots) + } + + fn iter_exact_records(&self, slots: Range) -> Result { + self.inner.iter_exact_records(slots) + } +} + +#[derive(Clone)] +pub struct MeasuredState { + inner: Inner, + pub counters: WorkCounters, +} + +impl StateStore for MeasuredState { + type EntityIter = CountedIter; + type EntityValueIter = CountedIter; + type UtxoIter = Inner::UtxoIter; + type Writer = Inner::Writer; + + fn read_cursor(&self) -> Result, StateError> { + self.inner.read_cursor() + } + + fn start_writer(&self) -> Result { + self.inner.start_writer() + } + + fn read_entities( + &self, + ns: Namespace, + keys: &[&EntityKey], + ) -> Result>, StateError> { + self.counters + .state_entities + .fetch_add(keys.len() as u64, Ordering::Relaxed); + self.inner.read_entities(ns, keys) + } + + fn iter_entities( + &self, + ns: Namespace, + range: Range, + ) -> Result { + Ok(CountedIter { + inner: self.inner.iter_entities(ns, range)?, + count: self.counters.state_entities.clone(), + }) + } + + fn iter_entity_values( + &self, + ns: Namespace, + key: impl AsRef<[u8]>, + ) -> Result { + Ok(CountedIter { + inner: self.inner.iter_entity_values(ns, key)?, + count: self.counters.state_entities.clone(), + }) + } + + fn get_utxos(&self, refs: Vec) -> Result { + self.counters + .utxo_reads + .fetch_add(refs.len() as u64, Ordering::Relaxed); + self.inner.get_utxos(refs) + } + + fn utxos_by_tag(&self, dimension: TagDimension, key: &[u8]) -> Result { + let refs = self.inner.utxos_by_tag(dimension, key)?; + self.counters + .utxo_refs + .fetch_add(refs.len() as u64, Ordering::Relaxed); + Ok(refs) + } + + fn iter_utxos(&self) -> Result { + self.inner.iter_utxos() + } +} + +#[derive(Clone)] +pub struct MeasuredStores { + _inner: Inner, + state: MeasuredState, + archive: MeasuredArchive, +} + +impl MeasuredStores { + pub fn new(inner: Inner) -> Self { + let counters = WorkCounters::default(); + Self { + state: MeasuredState { + inner: inner.state().clone(), + counters: counters.clone(), + }, + archive: MeasuredArchive { + inner: inner.archive().clone(), + counters, + }, + _inner: inner, + } + } +} + +impl ToyStores for MeasuredStores { + type State = MeasuredState; + type Archive = MeasuredArchive; + + fn open() -> Self { + Self::new(Inner::open()) + } + + fn state(&self) -> &Self::State { + &self.state + } + + fn archive(&self) -> &Self::Archive { + &self.archive + } +} diff --git a/crates/testing/src/performance.rs b/crates/testing/src/performance.rs new file mode 100644 index 000000000..ef0e12aa2 --- /dev/null +++ b/crates/testing/src/performance.rs @@ -0,0 +1,170 @@ +use std::sync::Arc; + +use dolos_cardano::{model::AccountEpochLog, rupd::credential_to_key}; +use dolos_core::{ + import::ImportExt, ArchiveStore, ArchiveWriter, ChainError, Domain, LogKey, RawBlock, + TemporalKey, +}; +use pallas::{ + crypto::hash::Hash, + ledger::{ + addresses::{Network, StakeAddress, StakePayload}, + primitives::StakeCredential, + }, +}; +use serde::{Deserialize, Serialize}; + +use crate::{ + measured::MeasuredStores, + synthetic::{build_synthetic_blocks, SyntheticBlockConfig, SyntheticVectors}, + toy_domain::{ToyDomain, ToyStores}, +}; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct FixtureShape { + pub blocks: usize, + pub transactions_per_block: usize, + pub log_rows: usize, + pub pool_stride: usize, + pub page: usize, + pub page_size: usize, + pub seed: u64, +} + +impl Default for FixtureShape { + fn default() -> Self { + Self { + blocks: 16, + transactions_per_block: 3, + log_rows: 256, + pool_stride: 8, + page: 2, + page_size: 2, + seed: 0, + } + } +} + +impl FixtureShape { + pub fn validate(&self) -> Result<(), String> { + if self.blocks < 4 || self.blocks > 10_000 { + return Err("blocks must be between 4 and 10000 (one preview epoch)".into()); + } + if self.transactions_per_block == 0 || self.transactions_per_block > 100 { + return Err("transactions_per_block must be between 1 and 100".into()); + } + if self.log_rows == 0 || self.pool_stride == 0 || self.log_rows < self.pool_stride { + return Err("log_rows must be positive and at least pool_stride > 0".into()); + } + let page_end = self + .page + .checked_mul(self.page_size) + .ok_or("pagination overflow")?; + if self.page == 0 + || self.page_size == 0 + || self.page_size > 100 + || page_end > self.blocks + || page_end > self.log_rows.div_ceil(self.pool_stride) + { + return Err("page must be positive, page_size 1..100, with enough blocks and pool matches to fill the page".into()); + } + Ok(()) + } + + pub fn credential(&self, row: usize) -> StakeCredential { + let mut bytes = [0u8; 28]; + bytes[..8].copy_from_slice(&self.seed.to_be_bytes()); + bytes[20..].copy_from_slice(&(row as u64).to_be_bytes()); + StakeCredential::AddrKeyhash(Hash::from(bytes)) + } + + pub fn stake_address(&self, row: usize) -> String { + let StakeCredential::AddrKeyhash(hash) = self.credential(row) else { + unreachable!() + }; + StakeAddress::new(Network::Testnet, StakePayload::Stake(hash)) + .to_bech32() + .expect("valid stake address") + } +} + +pub struct ApiFixture { + pub domain: ToyDomain>, + pub vectors: SyntheticVectors, + pub blocks: Vec, + pub tail: Vec, + pub shape: FixtureShape, + pub epoch: u64, +} + +impl ApiFixture { + pub fn new( + stores: Stores, + shape: FixtureShape, + ) -> Result> { + shape.validate().map_err(std::io::Error::other)?; + let config = SyntheticBlockConfig { + block_count: shape.blocks * 2, + txs_per_block: shape.transactions_per_block, + slot: 1, + metadata_value: format!("benchmark-{}", shape.seed), + ..Default::default() + }; + let (mut blocks, vectors, chain_config) = build_synthetic_blocks(config); + let tail = blocks.split_off(shape.blocks); + let domain = ToyDomain::with_stores( + Arc::new(dolos_cardano::include::preview::load()), + chain_config, + None, + None, + MeasuredStores::new(stores), + ); + for batch in blocks.chunks(100) { + domain.import_blocks(batch.to_vec())?; + } + let summary = dolos_cardano::eras::load_era_summary::>>( + domain.state(), + )?; + let (epoch, _) = summary.slot_epoch(vectors.blocks[0].slot); + let pool_hash = decode_pool(&vectors.pool_id)?; + for start in (0..shape.log_rows).step_by(1000) { + let writer = domain.archive().start_writer()?; + for row in start..(start + 1000).min(shape.log_rows) { + let key = LogKey::from(( + TemporalKey::from(summary.epoch_start(epoch)), + credential_to_key(&shape.credential(row)), + )); + let log = AccountEpochLog { + active_stake: Some(1_000_000 + row as u64), + pool_id: Some(if row % shape.pool_stride == 0 { + pool_hash + } else { + Hash::from([0x55; 28]) + }), + ..Default::default() + }; + writer.write_log_typed(&key, &log)?; + } + writer.commit()?; + } + domain.archive().counters.reset(); + Ok(Self { + domain, + vectors, + blocks, + tail, + shape, + epoch, + }) + } +} + +pub fn decode_pool(pool: &str) -> Result, ChainError> { + use bech32::FromBase32; + let (_, data, _) = bech32::decode(pool).map_err(|_| ChainError::InvalidPoolParams)?; + let bytes = Vec::::from_base32(&data).map_err(|_| ChainError::InvalidPoolParams)?; + let bytes: [u8; 28] = bytes + .try_into() + .map_err(|_| ChainError::InvalidPoolParams)?; + Ok(Hash::from(bytes)) +} diff --git a/crates/testing/src/toy_domain.rs b/crates/testing/src/toy_domain.rs index e4a72faee..8f0296314 100644 --- a/crates/testing/src/toy_domain.rs +++ b/crates/testing/src/toy_domain.rs @@ -207,6 +207,25 @@ pub struct FjallStores { } impl FjallStores { + pub fn open_in( + parent: &std::path::Path, + state_config: &dolos_core::config::FjallStateConfig, + archive_config: &dolos_core::config::FjallArchiveConfig, + ) -> Result> { + let dir = tempfile::tempdir_in(parent)?; + let state = dolos_fjall::StateStore::open(dir.path().join("state"), state_config)?; + let archive = dolos_fjall::archive::ArchiveStore::open( + dolos_cardano::model::build_schema(), + dir.path().join("archive"), + archive_config, + )?; + Ok(Self { + state, + archive, + _dir: Arc::new(dir), + }) + } + /// Where the stores live: `state/` and `archive/` under it. pub fn path(&self) -> &std::path::Path { self._dir.path() @@ -297,6 +316,18 @@ impl ToyDomain { } impl ToyDomain { + pub fn with_persistent_wal( + mut self, + path: impl AsRef, + ) -> Result { + self.wal = dolos_redb3::wal::RedbWalStore::open( + path, + &dolos_core::config::RedbWalConfig::default(), + )?; + self.bootstrap()?; + Ok(self) + } + /// The general constructor: the backend is named by the caller. /// /// `ToyDomain::new` and friends are this with [`MemoryStores`] filled in. @@ -310,7 +341,16 @@ impl ToyDomain { storage_config: Option, ) -> Self { let stores = B::open(); + Self::with_stores(genesis, config, initial_delta, storage_config, stores) + } + pub fn with_stores( + genesis: Arc, + config: CardanoConfig, + initial_delta: Option, + storage_config: Option, + stores: B, + ) -> Self { let (tip_broadcast, _) = tokio::sync::broadcast::channel(100); let chain = dolos_cardano::CardanoLogic::initialize::( diff --git a/crates/testing/tests/benchmark_fixtures.rs b/crates/testing/tests/benchmark_fixtures.rs new file mode 100644 index 000000000..7eb237f1c --- /dev/null +++ b/crates/testing/tests/benchmark_fixtures.rs @@ -0,0 +1,108 @@ +use dolos_core::{archive::Skippable, ArchiveStore, NamespaceType, StateSchema}; +use dolos_testing::{ + archive::{populate_archive_batched, ArchiveShape}, + measured::MeasuredStores, + toy_domain::{MemoryStores, ToyStores}, +}; + +fn shape(blocks: u64) -> ArchiveShape { + ArchiveShape { + epochs: 2, + blocks_per_epoch: blocks, + log_rows_per_epoch: 17, + slots_per_epoch: 100, + seed: 42, + } +} + +fn schema() -> StateSchema { + let mut schema = StateSchema::default(); + schema.insert("logs", NamespaceType::KeyValue); + schema +} + +#[test] +fn population_is_identical_across_batch_boundaries() { + let first = dolos_core::builtin::MemoryArchiveStore::new(schema()); + let second = dolos_core::builtin::MemoryArchiveStore::new(schema()); + let shape = shape(7); + populate_archive_batched(&first, "logs", &shape, 1).unwrap(); + populate_archive_batched(&second, "logs", &shape, 11).unwrap(); + assert_eq!( + first.get_range(None, None).unwrap().collect::>(), + second.get_range(None, None).unwrap().collect::>(), + ); + let range = shape.log_key(0, 0)..shape.log_key(shape.epochs, 0); + assert_eq!( + first + .iter_logs("logs", range.clone()) + .unwrap() + .collect::, _>>() + .unwrap(), + second + .iter_logs("logs", range) + .unwrap() + .collect::, _>>() + .unwrap(), + ); +} + +#[test] +fn logs_only_population_has_no_bodies_and_rejects_invalid_batch_size() { + let store = dolos_core::builtin::MemoryArchiveStore::new(schema()); + let shape = shape(0); + assert!(populate_archive_batched(&store, "logs", &shape, 0).is_err()); + populate_archive_batched(&store, "logs", &shape, 3).unwrap(); + assert!(store.get_tip().unwrap().is_none()); + assert_eq!( + store + .iter_logs("logs", shape.log_key(0, 0)..shape.log_key(2, 0)) + .unwrap() + .count(), + 34 + ); +} + +#[test] +fn measurement_preserves_reverse_iteration_and_does_not_count_skips_as_reads() { + let stores = MeasuredStores::new(MemoryStores::open()); + let archive = stores.archive(); + populate_archive_batched(archive, "account-epochs", &shape(8), 3).unwrap(); + archive.counters.reset(); + let mut blocks = archive.get_range(Some(0), Some(100)).unwrap(); + blocks.skip_forward(2); + blocks.skip_backward(1); + assert_eq!(archive.counters.snapshot().block_reads, 0); + assert_eq!(blocks.next().unwrap().0, 24); + assert_eq!(blocks.next_back().unwrap().0, 72); + assert_eq!(archive.counters.snapshot().block_reads, 2); + assert!(archive.counters.snapshot().decoded_bytes > 0); + let log_range = shape(8).log_key(0, 0)..shape(8).log_key(1, 0); + let mut logs = archive.iter_logs("account-epochs", log_range).unwrap(); + assert_eq!(archive.counters.snapshot().log_rows, 0); + logs.next().unwrap().unwrap(); + assert_eq!(archive.counters.snapshot().log_rows, 1); +} + +#[test] +fn persistent_wal_is_anchored_at_imported_tip_before_live_replay() { + use dolos_core::{BootstrapExt, Domain, StateStore, SyncExt, WalStore}; + use dolos_testing::performance::{ApiFixture, FixtureShape}; + + let fixture = ApiFixture::new(MemoryStores::open(), FixtureShape::default()).unwrap(); + let imported_tip = fixture.domain.state().read_cursor().unwrap().unwrap(); + let directory = tempfile::tempdir().unwrap(); + let domain = fixture + .domain + .with_persistent_wal(directory.path().join("wal")) + .unwrap(); + assert_eq!(domain.wal().find_tip().unwrap().unwrap().0, imported_tip); + assert!(domain.wal().read_entry(&imported_tip).unwrap().is_some()); + for block in &fixture.tail[..2] { + domain.roll_forward(block.clone()).unwrap(); + } + domain.check_integrity().unwrap(); + domain.rollback(&imported_tip).unwrap(); + assert_eq!(domain.state().read_cursor().unwrap(), Some(imported_tip)); + domain.bootstrap().unwrap(); +} diff --git a/xtask/Cargo.toml b/xtask/Cargo.toml index 425983c6b..675a6b6fc 100644 --- a/xtask/Cargo.toml +++ b/xtask/Cargo.toml @@ -25,6 +25,11 @@ bech32 = { workspace = true } pallas = { workspace = true } chrono = { workspace = true, features = ["clock"] } dolos-flatfiles = { path = "../crates/flatfiles" } +dolos-minibf = { path = "../crates/minibf" } +dolos-testing = { path = "../crates/testing" } +axum = "0.8.4" +tower = { workspace = true, features = ["util"] } +reqwest = { workspace = true } hdrhistogram = "7.5" libc = "0.2" sha2 = "0.10" @@ -34,3 +39,6 @@ zstd = "0.13" [dev-dependencies] tempfile = "3" + +[build-dependencies] +vergen-gitcl = "9.1.0" diff --git a/xtask/README.md b/xtask/README.md index da2e0253a..2dea8c384 100644 --- a/xtask/README.md +++ b/xtask/README.md @@ -132,25 +132,12 @@ Output fields per entity: - **accounts** — `stake,pool,lovelace` - **rewards** — `stake,pool,amount,type,earned_epoch` -### `archive-bench` +### `perf` -Benchmarks for the archive's compressed block segments: store-level -workloads over the production `dolos-flatfiles` store beside a modelled -sink, node-level workloads that drive `dolos` binaries through their import -and API paths so two revisions can be paired, dictionary training and -evaluation, and a report renderer with gate verdicts. - -``` -cargo xtask archive-bench bench --preset all --corpus --segments 448..451 --work --out results.jsonl -cargo xtask archive-bench node --bin baseline=:v3 --bin candidate= --run --immutable --genesis --work --out node.jsonl -cargo xtask archive-bench report results.jsonl node.jsonl -cargo xtask archive-bench train --corpus --sample 440..447=1500 --out cardano.dict -cargo xtask archive-bench evaluate --fixture label=:448..455 --dictionary bundled -``` - -Every option, the gates, the corpus requirements and the committed results -are described in [`archive-bench/README.md`](archive-bench/README.md). The -`smoke` preset runs under `cargo test -p xtask`. +Performance experiments for storage and minibf, with shared load generation, +version comparison and reporting. Start with the [perf overview](perf/README.md), +then choose [storage](perf/storage.md), [minibf](perf/minibf.md) or +[HTTP calibration](perf/http.md). Rust microbenchmarks remain under `cargo bench`. ### `e2e-test` diff --git a/xtask/archive-bench/README.md b/xtask/archive-bench/README.md deleted file mode 100644 index 2a255ec57..000000000 --- a/xtask/archive-bench/README.md +++ /dev/null @@ -1,245 +0,0 @@ -# archive-bench - -Developer benchmarks for the archive's compressed block segments, run as -`cargo xtask archive-bench`. Not part of the `dolos` binary and not a -production code path. Two layers of workloads: - -- **store level** (`bench`): the production `dolos_flatfiles::FlatFileStore` - beside a modelled sink — one zstd frame per block at a physical location, - or the same bodies raw — on the same append, point, page, mixed and - append-under-query workloads, so codec cost and I/O cost can be told apart. - The raw sink is supplemental evidence about the codec; it is not a - production baseline. -- **node level** (`node`): `dolos` binaries driven through their own import - and API paths — `data import-archive` over a node immutable directory, - `data dump-blocks` over the imported history, and the UTxO RPC sync service - under a threaded query driver — so two revisions are measured on the same - host, corpus, durability and concurrency. This is the production - comparison the acceptance gates are judged on. - -`report` renders markdown tables and gate verdicts from any set of result -files; `train` and `evaluate` are the developer-only dictionary tooling. The -`smoke` preset runs every store-level workload on a synthetic corpus in -seconds; `cargo test -p xtask` runs it, so the harness stays green in CI -without a corpus or a binary. `results/` holds the measured runs, one -directory per host and date. - -## Store level - -```sh -cargo xtask archive-bench bench \ - --preset all \ - --corpus /path/to/raw/segments --segments 448..451 \ - --work /fast/disk/archive-bench.noindex \ - --out results.jsonl --repeat 3 --cache warm,evict --evict-from /some/big/dir -cargo xtask archive-bench report results.jsonl -``` - -`bench` appends one JSON record per measurement to `--out`; every record -carries the environment (revision, OS, CPU, memory, zstd version, the -filesystem under `--work`, which counters the host provides), the corpus -(source, block and byte counts, eras, segments), the codec, the repeat and -the cache regime beside the metrics. - -Presets: - -| preset | what runs | -|---|---| -| `write` | the corpus appended in slot order in batches of 1 (live tip), 100 (pull and backfill import) and 500 (mithril bootstrap) blocks, paired across codecs; encode CPU, ingestion throughput, batch commit latency, fsync latency, peak transient heap, process CPU and disk bytes | -| `read` | over a store written once per codec: uniform and local point reads, 100-block pages, a whole scan, at each `--threads` count and each `--cache` regime | -| `mixed` | 90/10, 50/50 and 10/90 point/page mixes at the highest `--threads` | -| `concurrent` | half the corpus written, then the other half appended in batches of 1 and 100 while readers run a 90/10 mix; both sides measured over the same span | -| `all` | all of the above | -| `smoke` | `all` at a size that finishes in seconds; `cargo test -p xtask` runs it on a synthetic corpus | - -Codecs (`--codecs`): `raw` (the reference sink: bodies unframed), `zstd1` -and `zstd3` (one dictionary-free frame per block at that level), -`zstd1-dict` and `zstd3-dict` (the same with `--dictionary`, by default the -one bundled in `dolos-flatfiles`; `zstd3-dict` is the production codec's -parameters) and `store` (the production `FlatFileStore` itself, appending -and reading exactly as the node does: one frame per block with the bundled -dictionary, serial encoding, one `fdatasync` per touched segment per batch). -The store codec is measured end to end — its records carry the whole append -in `write_ms` and the whole read in `decode_us_per_block`, with no encode or -fsync split, and `--encode-threads` and `--no-fsync` do not reach it — and -it skips the `nocache` regime, since it opens its own descriptors. - -Corpora: `--corpus DIR --segments 446..449` walks raw `NNNNNN.segment` files -(the pre-v4 segment layout, bodies concatenated as CBOR items) and decodes -each with pallas for its slot and era; `--immutable DIR` reads a Cardano -node immutable directory through pallas; `--synthetic N` is the seeded -stand-in the smoke test uses. - -Cache regimes (`--cache`): `warm` primes every segment; `evict` drops the -segments' pages and leaves readahead on (`posix_fadvise` on Linux; on macOS -there is no unprivileged drop, so `--evict-from DIR --evict-gib N` streams -that much of other data through the cache instead); `nocache` evicts and -then reads with caching off at the descriptor (`F_NOCACHE`, which also -disables readahead — pessimistic for scans). A regime the host cannot -produce is skipped and said so on stderr. Cold numbers on macOS carry the -stream method in their `cache` field; treat them as approximate. - -Counters: thread CPU is `CLOCK_THREAD_CPUTIME_ID`; process CPU and disk -bytes come from `proc_pid_rusage` on macOS and `getrusage` plus -`/proc/self/io` on Linux; peak transient heap is a counting global -allocator in the `cargo-xtask` binary. A counter the host lacks is `null` -in the record and named under `environment.counters`. - -### Store-level gates - -`report` judges every non-raw codec against raw on the median over paired -repeats: at least 90% of raw ingestion throughput and at most 10% higher -p95 commit latency on every write workload, including the writer under -concurrent query load. These are the provisional codec gates; a failure is -a finding, not a reason to weaken fsync or add a raw mode. - -Only like is compared with like. Records are grouped by run (harness -revision and dirty state, host), by corpus, by workload, and by every -setting of the measurement other than its codec and repeat (batch size, -encoder threads, fsync, reader threads, mix, seed, eviction method); a -candidate keeps its dictionary identity, so two dictionaries are two -candidates. A verdict needs exactly one raw and one candidate sample per -repeat in that group: a candidate without a raw partner, repeats that do -not match, or a repeat recorded twice (a rerun appended to the same file) is -reported `UNPAIRED` with the reason, its ratios shown and its verdict -withheld. An older record that never wrote a setting shows it as `?` and -never matches one that did. - -Read workloads get the same ratios against raw in a table of their own, -without a verdict: against the raw sink a warm point read is a memcpy out -of the page cache, so the ratio says how many microseconds decoding adds, -not whether an API budget holds. The API budget is the node level's. - -## Node level - -```sh -cargo xtask archive-bench node \ - --bin baseline=/path/to/dolos-9165dbd8:v3 \ - --bin candidate=target/release/dolos \ - --run 2026-09-09-m4 \ - --immutable /fast/disk/immutable-window \ - --genesis /path/to/mainnet/genesis \ - --work /fast/disk/archive-bench.noindex/work \ - --out node.jsonl --repeat 3 --threads 1,8 --ops 20000 \ - --cache warm,evict --evict-from /some/big/dir --evict-gib 16 -cargo xtask archive-bench report node.jsonl -``` - -Every `--bin` is `LABEL=PATH[:STORAGE_VERSION]` (the version the binary's -`dolos.toml` declares; `v4` unless given). Every repeat runs each binary in -turn on a fresh, disposable instance under `--work`, so the two sides see -the same host in the same state; `--run` names the paired run and every -record carries it beside the label, the binary's SHA-256 and `--version`. - -Workloads (`--workloads`, default all four): - -| workload | what runs | measured | -|---|---|---| -| `import` | `dolos data import-archive` over the immutable directory in batches of `--chunk-size` (500, the mithril bootstrap shape) | wall, blocks/s, raw MiB/s, ms per batch, the child's CPU and peak RSS from `wait4`, segment bytes, compression ratio, index bytes | -| `live` | the same command with one block per commit over the first `--live-blocks` blocks: the live-tip shape, one `fdatasync` and one index commit per block | the same | -| `read` | `dolos data dump-blocks` over the corpus (the node's own scan of every frame), then `dolos serve` with the UTxO RPC service: `FetchBlock` by slot (uniform and 80%-local), `DumpHistory` pages of 100, at each `--threads` count and each `--cache` regime | ops/s, blocks/s, point and page latency histograms, the server's CPU, disk bytes and peak footprint over the workload | -| `mixed` | 90/10, 50/50 and 10/90 point/page mixes at the highest `--threads` | the same | - -The corpus is what `--immutable` yields through pallas, minus its last -chunk (which the reader treats as not yet immutable). Two things follow -from how the node reads that directory: - -- **It has to begin with chunk `00000`**: pallas refuses a directory whose - first block is not the genesis block, and `import-archive` starts from - the archive's tip, which is empty. A window copied out of a full immutable - directory therefore carries chunks `00000` and `00001` beside the range - under test; the first Byron epoch is small (a few thousand tiny blocks) - and the records name the eras so it can be discounted. -- **The read side needs an era summary.** An archive-only import writes - neither the block-hash index nor the era summary the block mappers read, - so before `serve` the harness seeds the state with `doctor rebuild-state - --stop-epoch 1` — a replay of exactly that first Byron epoch, which is the - other reason chunk `00001` has to be there (the replay stops at the block - that opens epoch 1). Reads then resolve blocks by slot; slot 0 and the - two-block Byron boundary slots are left out of the query set because the - service cannot name them unambiguously. The scan covers the contiguous - tail of the corpus (`dump-blocks` asserts consecutive block numbers). - -Every response is checked for success; `--verify` also checks the returned -header against the corpus. A workload with errors or mismatches fails the -run rather than recording a number. - -Not measured at the node level, and why: **append under concurrent API -load**. `import-archive` and `serve` each take the store's exclusive lock, -and a `daemon` that syncs while serving needs an upstream, which makes the -run neither offline nor reproducible. The store-level `concurrent` preset — -the production `FlatFileStore` under readers — is the evidence for that -shape. **Batch commit latency percentiles** are not observable from -outside the process; the node records carry the mean (`ms_per_batch`) and -the store-level `write` preset carries the p50/p95/p99 of the production -store's append. - -### Node gates - -`report` pairs every label against the one named `baseline` within one run -name, corpus, workload and set of settings, on the median over paired -repeats, and needs one sample per label per repeat: ingestion (`import-500` -and `import-1`) passes at **at least 90% of baseline throughput**; point -reads pass at **at most 10% higher p95 latency**; pages and scans are -reported with their ratios and no verdict. These are the acceptance -batch's production budgets. - -### Production baseline - -The last revision whose store appended raw bodies is dolos `9165dbd8` (the -merge of #1311, the parent of the direct-write cutover #1312). It reads a -`v3` configuration, so its `--bin` spec ends in `:v3`. Build it from a -detached worktree (`git worktree add --detach ../dolos-baseline 9165dbd8 && -cargo build --release --bin dolos`) and pass the binary's path. - -## Dictionary - -```sh -cargo xtask archive-bench train \ - --corpus /path/to/mainnet/segments \ - --sample 8,25,40,60,120,250,330,380=1000 --sample 440..447=1500 \ - --seed 0 --max-size 112640 --out cardano.dict -cargo xtask archive-bench evaluate \ - --fixture mainnet-heldout=/path/to/mainnet/segments:448..455 \ - --fixture preprod=/path/to/preprod/segments:100,200,300 \ - --fixture preview=immutable:/path/to/preview/immutable \ - --dictionary bundled --dictionary cardano.dict --out evaluate.jsonl -``` - -`train` samples a seeded subset of each segment's blocks (`--segments` -with `--samples-per-segment`, or weighted `--sample SEGMENTS=COUNT` groups) -and runs zstd's trainer; `.json` records every input file's SHA-256, -the sample per segment, the seed, the size cap and the zstd version, so the -same command over the same files yields the same bytes. `evaluate` encodes -and decodes every block of every fixture with the dictionary-free codec and -each dictionary given, checks the round trip, and reports ratio, encode -throughput and decode cost per block, per era. - -Dictionary preparation is developer tooling and nothing else: the node has -no training, sealing or dictionary command. The bundled asset lives at -`crates/flatfiles/dictionary/cardano.dict` with its provenance beside it; -`dolos_flatfiles::BUNDLED_DICTIONARY` exposes it and -`crates/flatfiles/tests/bundled_dictionary.rs` pins its hash and the -128 KiB budget. Changing those bytes is a storage-format decision. - -## Results - -`results/` holds the measured runs, one directory per host and date, each -with the raw `*.jsonl` records, the `report` output over them, the runbook -that produced them and a `REPORT.md` with the corpus, host, verdicts and -findings spelled out. Older directories keep the commands as they were run -at the time; where the command has since moved, the runbook says so at the -top rather than rewriting history. - -Before publishing results, replace workstation paths, usernames and -temporary session directories in records, provenance and commands with -consistent relative corpus/work labels. Preserve corpus hashes, revision -metadata and measured values; regenerate reports from the sanitized -records because run IDs include environment metadata. Local harness output -is not automatically sanitized. `xtask/tests/published_artifacts.rs` guards -the committed results and dictionary provenance against common leaks. - -The research this tooling grew out of — the reads-only measurements the -cutover design was first argued from — lives in the txpipe knowledge base -(`solution/dolos/kb/archive-flatfile-compression/`); the committed results -here are the evidence for the implementation as shipped. diff --git a/xtask/archive-bench/results/2026-09-09-m4-apfs-ssd-automatic/REPORT.md b/xtask/archive-bench/results/2026-09-09-m4-apfs-ssd-automatic/REPORT.md index 31f898109..86d00368f 100644 --- a/xtask/archive-bench/results/2026-09-09-m4-apfs-ssd-automatic/REPORT.md +++ b/xtask/archive-bench/results/2026-09-09-m4-apfs-ssd-automatic/REPORT.md @@ -148,7 +148,7 @@ Peak encoded buffers are 4,841,487 bytes at batch 500 and 5,836,179 at batch incompressible body among mixed synthetic bodies stays serial in its own window: peak scratch is 16,842,752 bytes, without an owned copy of that frame. The general bound remains `max(W, B(M)) + (P + 2) * B(M)`; native contexts -are bounded by P + 1. See the [implementation notes](../../OFFLINE-IMPORT.md). +are bounded by P + 1. See the [implementation notes](../../../perf/encoding.md). [verification.json](verification.json) records completed exit statuses: clippy/build exit 0; default tests 1,457 passed / 48 ignored; all-features diff --git a/xtask/archive-bench/results/2026-09-09-m4-apfs-ssd-offline-import/REPORT.md b/xtask/archive-bench/results/2026-09-09-m4-apfs-ssd-offline-import/REPORT.md index f4ae0c0df..8a1966f01 100644 --- a/xtask/archive-bench/results/2026-09-09-m4-apfs-ssd-offline-import/REPORT.md +++ b/xtask/archive-bench/results/2026-09-09-m4-apfs-ssd-offline-import/REPORT.md @@ -68,7 +68,7 @@ the historical arm has no such counter, not a measured zero. The counters include retained serial scratch, extra encoder scratch and owned completed frames. They exclude native zstd contexts and allocator/vector metadata, which are included in process RSS and bounded as documented in the -[implementation notes](../../OFFLINE-IMPORT.md). +[implementation notes](../../../perf/encoding.md). The one-block Byron shape is the first 2,000 inherited blocks. The modern shape contains the required original Byron origin anchor followed by 1,999 diff --git a/xtask/build.rs b/xtask/build.rs new file mode 100644 index 000000000..0365d0b7d --- /dev/null +++ b/xtask/build.rs @@ -0,0 +1,7 @@ +use vergen_gitcl::{Emitter, GitclBuilder}; + +fn main() -> Result<(), Box> { + let git = GitclBuilder::default().sha(false).dirty(true).build()?; + Emitter::default().add_instructions(&git)?.emit()?; + Ok(()) +} diff --git a/xtask/perf/README.md b/xtask/perf/README.md new file mode 100644 index 000000000..e00291e3c --- /dev/null +++ b/xtask/perf/README.md @@ -0,0 +1,59 @@ +# Performance experiments + +`cargo xtask perf` orchestrates experiments; `cargo bench --bench archive_backends` +measures isolated Rust operations. Neither a synthetic run nor a passing relative +comparison establishes mainnet capacity. + +| Guide | Commands | Subject | +|---|---|---| +| [Storage](storage.md) | `perf storage run`, `node`, `train`, `evaluate` | Compression, I/O, import and storage-facing RPC | +| [Minibf](minibf.md) | `perf minibf run`, `compare`, `check` | In-process routes, live replay and version comparisons | +| [HTTP calibration](http.md) | `perf minibf http` | Minibf on prepared, populated nodes | + +## Build and run + +From the repository root: + +```sh +cargo build --release -p xtask +XTASK="$PWD/target/release/cargo-xtask" +"$XTASK" perf --help +``` + +The guides use this release binary. Use each subcommand's `--help` for the full +option list. Put `--work` on the disk being measured; use a new output file for +each experiment because measurements append JSONL records. + +```sh +"$XTASK" perf report results.jsonl +``` + +Reporting accepts storage, minibf or mixed records. It renders findings; +`perf minibf check` additionally exits nonzero when a comparison cannot pass. +Match corpus, host, build settings, cache, durability and load across arms. +Missing counters are unavailable, not zero; simulated cold-cache results are +not equivalent to physical cold reads. + +## Layout and compatibility + +`xtask::perf::{storage,minibf}` share measurement, provenance, load generation +and reporting. Reusable chain fixtures and store counters live in `dolos-testing`. +Harness checks are `perf_cli`, `storage_perf_smoke` and `minibf_perf`; CI runs +correctness checks, not noisy wall-clock thresholds. + +`archive-bench` remains a legacy storage command; `bench` aliases `storage run`. +New scripts should use the commands above. + +## Published evidence + +Historical records and their original commands remain under `archive-bench/results`: +[compression study](../archive-bench/results/2026-09-08-m4-apfs-ssd/REPORT.md) and +[cutover acceptance](../archive-bench/results/2026-09-09-m4-apfs-ssd-acceptance/REPORT.md). +The [encoding notes](encoding.md#evidence) cover subsequent automatic-encoding runs. +Each run separates findings (`REPORT.md`), reproduction (`RUNBOOK.md`), generated +tables and raw JSONL. They are evidence, not current CLI instructions. + +Before publishing, replace local paths and session identifiers with consistent +labels. Preserve hashes, revisions and measurements, then regenerate tables from +the sanitized records. Output is not automatically sanitized; the +`published_artifacts` test guards committed evidence and dictionary provenance. diff --git a/xtask/archive-bench/OFFLINE-IMPORT.md b/xtask/perf/encoding.md similarity index 96% rename from xtask/archive-bench/OFFLINE-IMPORT.md rename to xtask/perf/encoding.md index e30b5b1db..599666bf7 100644 --- a/xtask/archive-bench/OFFLINE-IMPORT.md +++ b/xtask/perf/encoding.md @@ -75,9 +75,9 @@ framing, dictionary, compression level or storage-version claim is made. ## Evidence The initial offline candidate and its exact identities are recorded in -`results/2026-09-09-m4-apfs-ssd-offline-import`; those measurements do not +`../archive-bench/results/2026-09-09-m4-apfs-ssd-offline-import`; those measurements do not describe the automatic candidate, whose evidence is in -`results/2026-09-09-m4-apfs-ssd-automatic`. +`../archive-bench/results/2026-09-09-m4-apfs-ssd-automatic`. Use the checked-in measurement patches on disposable source trees to add actual commit timing, then compare pinned raw production `9165dbd8`, merged serial @@ -88,7 +88,7 @@ commit p95. Also report tiny, modern, bootstrap-sized and concurrent workloads. Historical runbooks mention a Python patch generator that has since been removed. Reproduce the instrumentation directly from the saved patches instead: -set `EVIDENCE` to the absolute path of this directory's `results` directory, +set `EVIDENCE` to the absolute path of `xtask/archive-bench/results`, then run the matching command from each disposable checkout's root. ```sh diff --git a/xtask/perf/http.md b/xtask/perf/http.md new file mode 100644 index 000000000..e2edc89aa --- /dev/null +++ b/xtask/perf/http.md @@ -0,0 +1,57 @@ +# HTTP calibration + +Use the release binary from the [overview](README.md#build-and-run). +Prepare populated baseline/candidate nodes with equivalent logical data and +pause sync. The driver does not restore, start, change or stop nodes. + +## Describe the experiment + +Create a manifest, replacing the placeholders with measured configuration and +independently verified expected responses—not just the candidate's output: + +```json +{ + "schema": 1, + "dataset": "", + "network": "mainnet", + "tip_hash": "<64-hex-digit tip hash>", + "server_environment": {"cpu": "", "memory_bytes": 0, "disk": ""}, + "settings": { + "storage_version": "v4", "dictionary": "", + "max_scan_items": 3000, "cache": "", + "durability": "production-defaults" + }, + "cases": [{ + "name": "pool-blocks", "path": "/epochs//blocks/?count=2", + "fields": [], "expected": ["", ""], + "array": true, "live": false + }] +} +``` + +`fields` lists JSON pointers to compare per item when `array` is true, or once +for an object response otherwise. Empty `fields` compares whole values; nonempty +projections verify only selected fields. Every measured response must also match +the validated warmup response's complete hash. + +## Run both arms + +```sh +"$XTASK" perf minibf http --url http://127.0.0.1:3000 --manifest mainnet.json \ + --server-binary /path/to/baseline/dolos --server-revision BASELINE_COMMIT \ + --run mainnet-01 --label baseline --out mainnet.jsonl --repeat 1 --repeat-start 0 +``` + +Run the candidate with its URL, binary, revision and label. Keep the manifest, +run name and request settings identical. Repeat with increasing `--repeat-start`, +alternating arm order; use at least three pairs before applying +[`perf minibf check`](minibf.md#compare-revisions) to the combined JSONL. +Authentication requires HTTPS. `--project-id-env VARIABLE_NAME` reads a secret without +recording its value in the command. + +## Interpretation + +The manifest tip is checked before and after each workload; changes invalidate +the result. Server binary/revision are operator assertions, not remote attestation. +CPU, I/O and RSS belong to the **client**, not the server; collect server metrics +separately. Stable-tip calibration does not establish live-sync performance. diff --git a/xtask/perf/minibf.md b/xtask/perf/minibf.md new file mode 100644 index 000000000..f21c1439c --- /dev/null +++ b/xtask/perf/minibf.md @@ -0,0 +1,91 @@ +# Minibf experiments + +Use the release binary from the [overview](README.md#build-and-run). +`run` executes actual Axum routes and query facades against fresh Fjall state, +archive and persistent WAL stores, without a network socket. + +```sh +"$XTASK" perf minibf run --work /path/to/scratch --out smoke.jsonl \ + --run smoke --repeat 1 --requests 10 +``` + +This is a harness check, deliberately too small to pass a performance gate. + +## Workloads and scale + +Select workloads with `--cases name,name` (default: all): + +| Cases | Access shape | +|---|---| +| `epoch-latest` | Current state/stake path | +| `epoch-stakes-page`, `epoch-stakes-sparse-pool` | Typed log scans, filtering and pagination | +| `epoch-blocks-pool-page`, `epoch-blocks-pool-no-match` | Compressed-body reads and issuer filtering | +| `epoch-blocks-reverse` | Reverse block traversal | +| `address-transactions-page` | Archive tags joined with blocks | +| `account-utxos-wide` | State tags, UTxOs and archive metadata | +| `transaction-utxos` | Transaction lookup and input/output mapping | + +Scale with `--blocks`, `--transactions-per-block`, `--log-rows`, `--pool-stride`, +`--page` and `--page-size`; keep `--seed`, `--cache-mib` and `--max-scan-items` +fixed when comparing versions. Pages must fit the generated population. + +Fixtures contain synthetic Conway blocks in preview epoch zero and an equally +sized replay tail. Setup is untimed; logs are populated in bounded batches, but +blocks and genesis inputs remain in memory. Limits are 10,000 initial blocks +and 100 transactions per block. This does not reproduce mainnet compression, +LSM depth, epoch transitions, pruning or the production traffic mix. + +## Load and measurement + +`--rates 0` is closed-loop load; positive rates schedule independent arrivals. +`--concurrency` bounds in-flight requests; excess arrivals count as rejected. +Late in-process requests are drained, since cancellation cannot stop their +blocking work. Errors, timeouts and rejections preserve records and fail the run. +The timeout is a completion budget, not a hard process deadline; a stuck blocking +operation requires an external process watchdog. Cancelling only its future +would free a concurrency slot while the storage work was still running. + +Each request validates fixture-derived identities, ordering and values, then +checks the complete normalized response hash. Warmup primes the route: this is +`route-primed`, not cold. Latency includes scheduling, response collection, +validation and hashing; service latency excludes scheduling delay. + +Counters track yielded rows, decoded bodies, lookups and state reads—not internal +LSM visits, physical frame bytes, separate codec CPU or semaphore queue depth. +Process resources include the harness; lifetime peak RSS is not per-request RSS. + +## Live replay + +```sh +"$XTASK" perf minibf run --work /path/to/scratch --out live.jsonl --run live \ + --live --blocks 256 --requests 1000 --rates 25 --timeout-ms 1000 \ + --write-interval-ms 250 --cases account-utxos-wide +``` + +The writer uses normal `roll_forward`, including WAL/state/archive commits. +Bootstrap anchors the WAL at the imported tip; replay can roll back to that +anchor, not into the pre-import history. +The tail must span arrival duration plus timeout; the interval is a minimum +delay, not guaranteed ingestion throughput. Latest/reverse-tip cases are excluded. +Counters include API and writer work; writer latency covers the whole roll-forward. + +## Compare revisions + +Prepare release binaries with the same harness/fixture protocol and equivalent +build settings; the runner does not build or patch historical revisions. + +```sh +"$XTASK" perf minibf compare \ + --bin baseline=/path/to/baseline/cargo-xtask \ + --bin candidate=/path/to/candidate/cargo-xtask -- \ + --work /path/to/scratch --out paired.jsonl --run comparison-01 \ + --requests 1000 --repeat 3 +"$XTASK" perf minibf check paired.jsonl +``` + +Arm order alternates per repeat. Default gates require three paired repeats, +1,000 successful requests each, p95 ratio ≤1.10, p99 ratio ≤1.20 and throughput +ratio ≥0.90. These are provisional experiment budgets, not a production SLO; +use more samples for p99. Optional `--p95-budget-ms`/`--p99-budget-ms` apply to +both arms. Missing, duplicate, incompatible, failed or undersampled evidence cannot +pass. Confirm synthetic findings with [real-node calibration](http.md). diff --git a/xtask/perf/storage.md b/xtask/perf/storage.md new file mode 100644 index 000000000..a288d0869 --- /dev/null +++ b/xtask/perf/storage.md @@ -0,0 +1,84 @@ +# Storage experiments + +Use the release binary from the [overview](README.md#build-and-run). +These experiments do not invoke minibf handlers. + +## Store workloads + +```sh +"$XTASK" perf storage run --synthetic 400 --preset smoke \ + --work /path/to/scratch --out smoke.jsonl --verify +``` + +For real data, replace `--synthetic` with `--immutable DIR` or +`--corpus DIR --segments 448..451`. The latter expects pre-v4 raw segment files, +not current compressed flatfiles. Use `--preset all --repeat 3` for a fuller run. + +| Preset | Work | +|---|---| +| `write` | Append batches of 1, 100 and 500 blocks | +| `read` | Point reads, 100-block pages and full scans | +| `mixed` | 90/10, 50/50 and 10/90 point/page mixes | +| `concurrent` | Append while readers run | +| `all`, `smoke` | All workloads; smoke bounds their size | + +`--codecs` compares `raw`, `zstd1`, `zstd3`, their `-dict` variants and `store`. +Raw/zstd sinks isolate codec cost; `store` uses the production `FlatFileStore`. +Raw is **not** a supported production mode. The store measures whole operations, +not separate codec/fsync time; `--encode-threads` and `--no-fsync` do not affect it. +The production store now selects serial or bounded parallel encoding +automatically; see [encoding policy and evidence](encoding.md). + +Cache choices are `warm`, `evict` and `nocache`. On macOS, eviction requires +`--evict-from DIR --evict-gib N` and is approximate. `nocache` also disables +readahead; the production store skips it. Unsupported regimes are reported and +skipped. Do not weaken durability to obtain a passing comparison. +Cache-setting failures and out-of-range scan latencies fail the experiment; +they are not reported as successful measurements with altered cache or timing. + +Store reports pair codecs against raw with identical settings and repeats. +Provisional write gates require throughput ≥90% of raw and p95 commit latency +≤110%; read ratios are descriptive, not API latency gates. + +## Compare Dolos binaries + +```sh +"$XTASK" perf storage node \ + --bin baseline=/path/to/baseline/dolos --bin candidate=/path/to/candidate/dolos \ + --run storage-01 --immutable /path/to/immutable --genesis /path/to/genesis \ + --work /path/to/scratch --out node.jsonl --repeat 3 --verify +``` + +The runner creates fresh disposable instances. Binary specs are +`LABEL=PATH[:STORAGE_VERSION]`, defaulting to v4. Historical raw revision +`9165dbd8` needs `:v3`; that is a compression baseline, not a redb comparison. + +`import` uses 500-block batches; `live` models one-block commits through offline +import, **not** daemon sync. `read` uses `dump-blocks` and UTxO RPC point/page +queries; `mixed` combines RPC point/page reads. This runner does not append while +serving: use the store's `concurrent` preset or [minibf live replay](minibf.md#live-replay). + +The immutable corpus must include chunks `00000` and `00001`; its final chunk +is excluded. Before serving, the runner replays the first Byron epoch to seed +the era summary. Ambiguous boundary slots are excluded from RPC queries; scans +cover the contiguous tail. An arbitrary copied window is insufficient. + +Reports pair against the `baseline` label: import throughput must be ≥90% and +point-read p95 ≤110% of baseline. Pages/scans have ratios but no verdict. +Uninstrumented nodes report mean batch time; explicitly instrumented records +also gate commit p95 at ≤110% of baseline and do not pair with uninstrumented runs. + +## Dictionary experiments + +```sh +"$XTASK" perf storage train --corpus /path/to/raw/segments \ + --sample 440..447=1500 --seed 0 --out candidate.dict +"$XTASK" perf storage evaluate \ + --fixture heldout=/path/to/raw/segments:448..455 \ + --dictionary bundled --dictionary candidate.dict --out dictionaries.jsonl +``` + +Training writes dictionary provenance beside the asset. Evaluation verifies +round trips and reports size, encoding throughput and decode cost on held-out +blocks. Changing the [bundled dictionary](../../crates/flatfiles/dictionary/README.md) +is a storage-format decision, not an operator tuning option. diff --git a/xtask/src/lib.rs b/xtask/src/lib.rs index 199685d02..5c5409fd0 100644 --- a/xtask/src/lib.rs +++ b/xtask/src/lib.rs @@ -1,3 +1,3 @@ //! Library half of the xtask binary: the pieces its tests drive directly. -pub mod archive_bench; +pub mod perf; diff --git a/xtask/src/main.rs b/xtask/src/main.rs index bbd562a1e..0a40956c5 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -1,8 +1,8 @@ use anyhow::Result; use clap::{Parser, Subcommand}; use xshell::{cmd, Shell}; -use xtask::archive_bench; -use xtask::archive_bench::measure::PeakAlloc; +use xtask::perf; +use xtask::perf::measure::PeakAlloc; mod bootstrap; mod config; @@ -27,9 +27,14 @@ enum Commands { /// Run e2e tests E2eTest, - /// Archive segment benchmarks (bench, node, train, evaluate, report) + /// Storage and minibf performance experiments with shared measurement and + /// reporting #[command(subcommand)] - ArchiveBench(archive_bench::Cmd), + Perf(perf::Cmd), + + /// Compatibility entry point for storage benchmarks; prefer perf storage + #[command(subcommand)] + ArchiveBench(perf::storage::Cmd), /// Bootstrap a local Mithril snapshot into an instance BootstrapMithrilLocal(bootstrap::BootstrapArgs), @@ -55,7 +60,8 @@ fn main() -> Result<()> { println!("Running sync tests..."); cmd!(sh, "cargo test --test sync -- --ignored --nocapture").run()?; } - Commands::ArchiveBench(cmd) => archive_bench::run(cmd)?, + Commands::Perf(cmd) => perf::run(cmd)?, + Commands::ArchiveBench(cmd) => perf::storage::run(cmd)?, Commands::BootstrapMithrilLocal(args) => bootstrap::run(&sh, &args)?, Commands::GroundTruth(cmd) => ground_truth::run(cmd)?, Commands::TestInstance(cmd) => test_instance::run(&sh, cmd)?, diff --git a/xtask/src/archive_bench/dictionary.rs b/xtask/src/perf/dictionary.rs similarity index 100% rename from xtask/src/archive_bench/dictionary.rs rename to xtask/src/perf/dictionary.rs diff --git a/xtask/src/perf/load.rs b/xtask/src/perf/load.rs new file mode 100644 index 000000000..196df1451 --- /dev/null +++ b/xtask/src/perf/load.rs @@ -0,0 +1,132 @@ +use super::measure; +use anyhow::Context; +use serde_json::{json, Value}; +use std::time::{Duration, Instant}; +use tokio::task::JoinSet; + +struct Outcome { + latency: u64, + service: u64, + bytes: usize, + error: Option, +} + +fn collect(outcome: Outcome, stats: &mut LoadStats, timeout: Duration) { + stats.latency.record(outcome.latency.max(1)).unwrap(); + stats.service.record(outcome.service.max(1)).unwrap(); + stats.response_bytes += outcome.bytes; + if let Some(error) = outcome.error { + stats.errors += 1; + if stats.first_error.is_none() { + stats.first_error = Some(error); + } + } else if outcome.latency > timeout.as_nanos() as u64 { + stats.timeouts += 1; + } else { + stats.completed += 1; + } +} + +struct LoadStats { + latency: hdrhistogram::Histogram, + service: hdrhistogram::Histogram, + completed: usize, + errors: usize, + timeouts: usize, + rejected: usize, + response_bytes: usize, + peak_in_flight: usize, + first_error: Option, +} + +impl Default for LoadStats { + fn default() -> Self { + Self { + latency: measure::histogram(), + service: measure::histogram(), + completed: 0, + errors: 0, + timeouts: 0, + rejected: 0, + response_bytes: 0, + peak_in_flight: 0, + first_error: None, + } + } +} + +pub async fn drive_requests( + request: Request, + requests: usize, + concurrency: usize, + rate: u64, + timeout: Duration, +) -> anyhow::Result +where + Request: Fn() -> Response, + Response: std::future::Future> + Send + 'static, +{ + anyhow::ensure!( + requests > 0 && concurrency > 0, + "requests and concurrency must be positive" + ); + let mut tasks = JoinSet::new(); + let mut stats = LoadStats::default(); + let start = Instant::now(); + for index in 0..requests { + let scheduled = if rate == 0 { + Instant::now() + } else { + start + Duration::from_secs_f64(index as f64 / rate as f64) + }; + if rate > 0 { + tokio::time::sleep_until(scheduled.into()).await; + } + while let Some(outcome) = tasks.try_join_next() { + collect(outcome?, &mut stats, timeout); + } + if tasks.len() >= concurrency { + if rate > 0 { + stats.rejected += 1; + continue; + } + collect( + tasks.join_next().await.context("missing request")??, + &mut stats, + timeout, + ); + } + let scheduled = if rate == 0 { Instant::now() } else { scheduled }; + let response = request(); + tasks.spawn(async move { + let service_start = Instant::now(); + let result = response.await; + Outcome { + latency: scheduled.elapsed().as_nanos().min(3_600_000_000_000) as u64, + service: service_start.elapsed().as_nanos().min(3_600_000_000_000) as u64, + bytes: result.as_ref().copied().unwrap_or(0), + error: result.err().map(|error| error.to_string()), + } + }); + stats.peak_in_flight = stats.peak_in_flight.max(tasks.len()); + } + while let Some(outcome) = tasks.join_next().await { + collect(outcome?, &mut stats, timeout); + } + if rate > 0 { + tokio::time::sleep_until( + (start + Duration::from_secs_f64(requests as f64 / rate as f64)).into(), + ) + .await; + } + let elapsed = start.elapsed().as_secs_f64(); + Ok(json!({ + "requests": requests, "completed": stats.completed, "errors": stats.errors, + "timeouts": stats.timeouts, "rejected": stats.rejected, + "first_error": stats.first_error, "elapsed_seconds": elapsed, + "completed_per_second": stats.completed as f64 / elapsed, + "latency": measure::histogram_json(&stats.latency), + "service_latency": measure::histogram_json(&stats.service), + "response_bytes": stats.response_bytes, "peak_in_flight": stats.peak_in_flight, + })) +} diff --git a/xtask/src/archive_bench/measure.rs b/xtask/src/perf/measure.rs similarity index 100% rename from xtask/src/archive_bench/measure.rs rename to xtask/src/perf/measure.rs diff --git a/xtask/src/perf/minibf/cases.rs b/xtask/src/perf/minibf/cases.rs new file mode 100644 index 000000000..66e62053b --- /dev/null +++ b/xtask/src/perf/minibf/cases.rs @@ -0,0 +1,224 @@ +use axum::{ + body::{to_bytes, Body}, + http::Request, + Router, +}; +use dolos_testing::{performance::ApiFixture, toy_domain::ToyStores}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; +use tower::ServiceExt; + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct Case { + pub name: String, + pub path: String, + pub fields: Vec, + pub expected: Value, + pub array: bool, + pub live: bool, +} + +impl Case { + pub fn verify(&self, value: &Value) -> anyhow::Result<()> { + let project = |row: &Value| -> Value { + if self.fields.is_empty() { + row.clone() + } else { + Value::Array( + self.fields + .iter() + .map(|field| row.pointer(field).cloned().unwrap_or(Value::Null)) + .collect(), + ) + } + }; + let actual = if self.array { + let rows = value + .as_array() + .ok_or_else(|| anyhow::anyhow!("{}: expected array", self.name))?; + Value::Array(rows.iter().map(project).collect()) + } else { + project(value) + }; + anyhow::ensure!( + actual == self.expected, + "{}: response differs from fixture: expected {}, got {}", + self.name, + self.expected, + actual + ); + Ok(()) + } +} + +pub fn cases(fixture: &ApiFixture) -> Vec { + let shape = &fixture.shape; + let vectors = &fixture.vectors; + let epoch = fixture.epoch; + let count = shape.page_size; + let page = shape.page; + let skip = (page - 1) * count; + let issuer_hash = pallas::crypto::hash::Hasher::<224>::hash(&[0x10, 0x11]); + let issuer_pool = + bech32::encode::(bech32::Hrp::parse("pool").unwrap(), issuer_hash.as_ref()) + .unwrap(); + let pool_expected: Vec = (0..shape.log_rows) + .step_by(shape.pool_stride) + .skip((page - 1) * count) + .take(count) + .map(|row| { + json!([ + shape.stake_address(row), + (1_000_000 + row as u64).to_string() + ]) + }) + .collect(); + let tx_expected: Vec = vectors + .blocks + .iter() + .skip(skip) + .take(count) + .map(|block| json!([block.tx_hashes[0]])) + .collect(); + let utxo_expected: Vec = vectors + .blocks + .iter() + .take(shape.blocks) + .flat_map(|block| &block.tx_hashes) + .take(4) + .map(|hash| json!([hash, 0])) + .collect(); + vec![ + Case { + name: "epoch-latest".into(), + path: "/epochs/latest".into(), + fields: vec!["/epoch".into(), "/block_count".into()], + expected: json!([epoch, shape.blocks]), + array: false, + live: false, + }, + Case { + name: "epoch-blocks-pool-page".into(), + path: format!("/epochs/{epoch}/blocks/{issuer_pool}?count={count}&page={page}"), + fields: vec![], + expected: json!(vectors + .blocks + .iter() + .skip(skip) + .take(count) + .map(|block| &block.block_hash) + .collect::>()), + array: true, + live: true, + }, + Case { + name: "epoch-stakes-sparse-pool".into(), + path: format!( + "/epochs/{epoch}/stakes/{}?count={count}&page={page}", + vectors.pool_id + ), + fields: vec!["/stake_address".into(), "/amount".into()], + expected: json!(pool_expected), + array: true, + live: true, + }, + Case { + name: "epoch-stakes-page".into(), + path: format!("/epochs/{epoch}/stakes?count={count}&page={page}"), + fields: vec!["/stake_address".into(), "/amount".into()], + expected: json!((skip..skip + count) + .map(|row| json!([ + shape.stake_address(row), + (1_000_000 + row as u64).to_string() + ])) + .collect::>()), + array: true, + live: true, + }, + Case { + name: "epoch-blocks-pool-no-match".into(), + path: format!("/epochs/{epoch}/blocks/{}?count=5", vectors.pool_id), + fields: vec![], + expected: json!([]), + array: true, + live: true, + }, + Case { + name: "epoch-blocks-reverse".into(), + path: format!("/epochs/{epoch}/blocks?count=2&order=desc"), + fields: vec![], + expected: json!(vectors + .blocks + .iter() + .take(shape.blocks) + .rev() + .take(2) + .map(|block| &block.block_hash) + .collect::>()), + array: true, + live: false, + }, + Case { + name: "address-transactions-page".into(), + path: format!( + "/addresses/{}/transactions?count={count}&page={page}&order=asc", + vectors.address + ), + fields: vec!["/tx_hash".into()], + expected: json!(tx_expected), + array: true, + live: true, + }, + Case { + name: "account-utxos-wide".into(), + path: format!( + "/accounts/{}/utxos?count=4&order=asc", + vectors.stake_address + ), + fields: vec!["/tx_hash".into(), "/output_index".into()], + expected: json!(utxo_expected), + array: true, + live: true, + }, + Case { + name: "transaction-utxos".into(), + path: format!("/txs/{}/utxos", vectors.tx_hash), + fields: vec![ + "/hash".into(), + "/outputs/0/output_index".into(), + "/outputs/0/address".into(), + ], + expected: json!([vectors.tx_hash, 0, vectors.address]), + array: false, + live: true, + }, + ] +} + +pub async fn request(router: Router, case: &Case) -> anyhow::Result { + response(router, case).await.map(|(bytes, _)| bytes) +} + +pub fn response_hash(value: &Value) -> String { + format!( + "{:x}", + Sha256::digest(serde_json::to_vec(value).expect("JSON response")) + ) +} + +pub async fn response(router: Router, case: &Case) -> anyhow::Result<(usize, String)> { + let request = Request::builder().uri(&case.path).body(Body::empty())?; + let response = router.oneshot(request).await?; + let status = response.status(); + let body = to_bytes(response.into_body(), 16 * 1024 * 1024).await?; + anyhow::ensure!( + status == axum::http::StatusCode::OK, + "{}: HTTP {status}: {}", + case.name, + String::from_utf8_lossy(&body) + ); + let value: Value = serde_json::from_slice(&body)?; + case.verify(&value)?; + Ok((body.len(), response_hash(&value))) +} diff --git a/xtask/src/perf/minibf/http.rs b/xtask/src/perf/minibf/http.rs new file mode 100644 index 000000000..3c03ef60d --- /dev/null +++ b/xtask/src/perf/minibf/http.rs @@ -0,0 +1,284 @@ +use std::{fs::OpenOptions, io::Write, path::PathBuf, time::Duration}; + +use anyhow::Context; +use reqwest::{Client, Url}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; + +use super::cases::Case; +use crate::perf::load::drive_requests; +use crate::perf::{command_line, measure, parse_list}; + +#[derive(clap::Args)] +pub struct Args { + #[arg(long)] + pub url: Url, + #[arg(long)] + pub manifest: PathBuf, + #[arg(long)] + pub server_binary: PathBuf, + #[arg(long)] + pub server_revision: String, + #[arg(long)] + pub out: PathBuf, + #[arg(long)] + pub run: String, + #[arg(long)] + pub label: String, + #[arg(long, default_value_t = 3)] + pub repeat: usize, + #[arg(long, default_value_t = 0)] + pub repeat_start: usize, + #[arg(long, default_value_t = 1000)] + pub requests: usize, + #[arg(long, default_value_t = 4)] + pub concurrency: usize, + #[arg(long, default_value = "0")] + pub rates: String, + #[arg(long, default_value_t = 15_000)] + pub timeout_ms: u64, + #[arg(long)] + pub project_id_env: Option, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Manifest { + pub schema: u32, + pub dataset: String, + pub network: String, + pub tip_hash: String, + pub server_environment: Value, + pub settings: Value, + pub cases: Vec, +} + +impl Manifest { + pub fn validate(&self) -> anyhow::Result<()> { + anyhow::ensure!(self.schema == 1, "unsupported manifest schema"); + anyhow::ensure!( + !self.dataset.is_empty() && !self.network.is_empty(), + "dataset and network are required" + ); + anyhow::ensure!( + self.tip_hash.len() == 64 && self.tip_hash.bytes().all(|byte| byte.is_ascii_hexdigit()), + "tip_hash must be a block hash" + ); + anyhow::ensure!( + self.server_environment.is_object() && self.settings.is_object(), + "server_environment and settings must be objects" + ); + for field in [ + "storage_version", + "dictionary", + "max_scan_items", + "cache", + "durability", + ] { + anyhow::ensure!( + !self.settings[field].is_null(), + "manifest settings missing {field}" + ); + } + anyhow::ensure!( + !self.cases.is_empty(), + "manifest must declare at least one case" + ); + let mut names = std::collections::BTreeSet::new(); + for case in &self.cases { + anyhow::ensure!( + !case.name.is_empty() && names.insert(&case.name), + "case names must be nonempty and unique" + ); + anyhow::ensure!( + case.path.starts_with('/') + && !case.path.starts_with("//") + && !case.path.contains('#'), + "case path must be an absolute path on the same server" + ); + anyhow::ensure!( + !case.expected.is_null(), + "each case needs an independent response oracle" + ); + } + Ok(()) + } +} + +async fn json_response(client: &Client, url: Url) -> anyhow::Result<(Value, usize)> { + let mut response = client.get(url).send().await?.error_for_status()?; + anyhow::ensure!( + response.status() == reqwest::StatusCode::OK, + "expected HTTP 200, got {}", + response.status() + ); + let mut body = Vec::new(); + while let Some(chunk) = response.chunk().await? { + anyhow::ensure!( + body.len() + chunk.len() <= 16 * 1024 * 1024, + "response exceeds 16 MiB" + ); + body.extend_from_slice(&chunk); + } + Ok((serde_json::from_slice(&body)?, body.len())) +} + +async fn check_tip(client: &Client, base: &Url, manifest: &Manifest) -> anyhow::Result<()> { + let (tip, _) = json_response(client, base.join("/blocks/latest")?).await?; + anyhow::ensure!( + tip["hash"] == manifest.tip_hash, + "node tip differs from the manifest; use a fully populated stable-tip instance" + ); + Ok(()) +} + +pub fn run(args: Args) -> anyhow::Result<()> { + anyhow::ensure!( + args.project_id_env.is_none() || args.url.scheme() == "https", + "--project-id-env requires HTTPS; refusing cleartext credentials" + ); + anyhow::ensure!( + matches!(args.url.scheme(), "http" | "https") && args.url.host_str().is_some(), + "url must use HTTP(S)" + ); + anyhow::ensure!( + args.url.username().is_empty() + && args.url.password().is_none() + && args.url.query().is_none() + && args.url.fragment().is_none(), + "use an origin URL without credentials, query or fragment" + ); + anyhow::ensure!(args.url.path() == "/", "url must be the server origin"); + anyhow::ensure!( + args.requests > 0 && args.concurrency > 0 && args.repeat > 0 && args.timeout_ms > 0, + "request counts, concurrency, repeats and timeout must be positive" + ); + anyhow::ensure!( + !args.run.is_empty() && !args.label.is_empty() && !args.server_revision.is_empty(), + "run, label and server revision are required" + ); + anyhow::ensure!( + args.repeat_start.checked_add(args.repeat).is_some(), + "repeat range overflow" + ); + let rates: Vec = parse_list(&args.rates)?; + let unique: std::collections::BTreeSet<_> = rates.iter().collect(); + anyhow::ensure!( + !rates.is_empty() && rates.len() == unique.len(), + "rates must be nonempty and unique" + ); + let manifest_bytes = std::fs::read(&args.manifest)?; + let manifest: Manifest = serde_json::from_slice(&manifest_bytes)?; + manifest.validate()?; + let timeout = Duration::from_millis(args.timeout_ms); + let mut headers = reqwest::header::HeaderMap::new(); + if let Some(variable) = &args.project_id_env { + let value = + std::env::var(variable).context("project-id environment variable is unavailable")?; + let mut value = reqwest::header::HeaderValue::from_str(&value)?; + value.set_sensitive(true); + headers.insert("project_id", value); + } + let client = Client::builder() + .timeout(timeout) + .redirect(reqwest::redirect::Policy::none()) + .default_headers(headers) + .build()?; + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .enable_all() + .build()?; + let environment = measure::environment(&[]); + let binary_sha256 = format!("{:x}", Sha256::digest(std::fs::read(&args.server_binary)?)); + let fixture = json!({ + "manifest_sha256": format!("{:x}", Sha256::digest(serde_json::to_vec(&manifest)?)), + "dataset": manifest.dataset, "network": manifest.network, "tip_hash": manifest.tip_hash, + "server_environment": manifest.server_environment, + }); + let suite: Vec<_> = manifest.cases.iter().map(|case| &case.name).collect(); + let mut file = OpenOptions::new() + .create(true) + .append(true) + .open(&args.out)?; + let mut failed = false; + for repeat in args.repeat_start..args.repeat_start + args.repeat { + for rate in &rates { + for case in &manifest.cases { + runtime.block_on(check_tip(&client, &args.url, &manifest))?; + let url = args.url.join(&case.path)?; + anyhow::ensure!( + url.origin() == args.url.origin(), + "case URL escaped the configured server origin" + ); + let expected_response = runtime.block_on(async { + let (body, _) = json_response(&client, url.clone()).await?; + case.verify(&body)?; + Ok::<_, anyhow::Error>(super::cases::response_hash(&body)) + })?; + let before = measure::counters(); + let mut metrics = runtime.block_on(drive_requests( + || { + let client = client.clone(); + let url = url.clone(); + let case = case.clone(); + let expected = expected_response.clone(); + async move { + let (body, bytes) = json_response(&client, url).await?; + case.verify(&body)?; + anyhow::ensure!( + super::cases::response_hash(&body) == expected, + "response changed after fixture validation" + ); + Ok(bytes) + } + }, + args.requests, + args.concurrency, + *rate, + timeout, + ))?; + metrics["resources"] = measure::counters().delta(&before).json(); + let tip_valid = runtime + .block_on(check_tip(&client, &args.url, &manifest)) + .is_ok(); + metrics["tip_unchanged"] = json!(tip_valid); + metrics["kind"] = json!("minibf"); + metrics["workload"] = json!(case.name); + metrics["work_scope"] = json!("unavailable-http"); + metrics["resources_scope"] = json!("client-only"); + failed |= !tip_valid + || ["errors", "timeouts", "rejected"] + .iter() + .any(|field| metrics[*field].as_u64().unwrap_or(0) > 0); + let mut case_fixture = fixture.clone(); + case_fixture["case"] = serde_json::to_value(case)?; + case_fixture["response_sha256"] = json!(expected_response); + let record = json!({ + "schema": 1, "run": args.run, "label": args.label, "repeat": repeat, + "suite": suite, "binary_sha256": binary_sha256, "environment": environment, + "build": { + "revision": args.server_revision, "identity_source": "operator-supplied-server-binary", + "harness_revision": env!("VERGEN_GIT_SHA"), "harness_dirty": env!("VERGEN_GIT_DIRTY"), + }, + "fixture": case_fixture, "metrics": metrics, + "settings": { + "transport": "http", "server": manifest.settings, + "requests": args.requests, "concurrency": args.concurrency, "rate": rate, + "timeout_ms": args.timeout_ms, "cache": "route-primed", + "mode": "read-only-stable-tip", "runtime_workers": 4, + }, + "command": command_line(), + }); + serde_json::to_writer(&mut file, &record)?; + file.write_all(b"\n")?; + file.flush()?; + } + } + } + anyhow::ensure!( + !failed, + "HTTP workloads failed, overloaded or changed tip; results retained" + ); + Ok(()) +} diff --git a/xtask/src/perf/minibf/mod.rs b/xtask/src/perf/minibf/mod.rs new file mode 100644 index 000000000..9bb51df4e --- /dev/null +++ b/xtask/src/perf/minibf/mod.rs @@ -0,0 +1,495 @@ +pub mod cases; +pub mod http; +pub mod report; + +use std::{ + fs::OpenOptions, + io::Write, + path::PathBuf, + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }, + time::{Duration, Instant}, +}; + +use super::load::drive_requests; +use anyhow::Context; +use clap::Parser; +use dolos_core::{ + config::{FjallArchiveConfig, FjallStateConfig, MinibfConfig}, + Domain, SyncExt, +}; +use dolos_testing::{ + performance::{ApiFixture, FixtureShape}, + toy_domain::FjallStores, +}; +use serde_json::{json, Value}; +use sha2::{Digest, Sha256}; + +use super::measure; + +#[derive(clap::Subcommand)] +pub enum Cmd { + #[command(about = "Measure verified routes on fresh persistent fixtures")] + Run(Box), + #[command(about = "Pair benchmark-enabled revisions in alternating order")] + Compare(Box), + #[command(about = "Calibrate HTTP workloads against a prepared stable-tip node")] + Http(Box), + #[command(about = "Enforce paired budgets and reject incomplete evidence")] + Check { + #[arg(required = true)] + files: Vec, + #[command(flatten)] + budgets: report::Budgets, + }, +} + +pub fn dispatch(command: Cmd) -> anyhow::Result<()> { + match command { + Cmd::Run(args) => run(*args), + Cmd::Compare(args) => compare(*args), + Cmd::Http(args) => http::run(*args), + Cmd::Check { files, budgets } => { + budgets.validate()?; + let records = super::report::load(&files)?; + let (rendered, passed) = report::assess(&records, &budgets); + print!("{rendered}"); + anyhow::ensure!(passed, "minibf comparison did not pass; see report"); + Ok(()) + } + } +} + +pub const CASE_NAMES: &[&str] = &[ + "epoch-latest", + "epoch-blocks-pool-page", + "epoch-stakes-sparse-pool", + "epoch-stakes-page", + "epoch-blocks-pool-no-match", + "epoch-blocks-reverse", + "address-transactions-page", + "account-utxos-wide", + "transaction-utxos", +]; + +#[derive(Clone, clap::Args)] +pub struct Args { + #[arg(long)] + pub work: PathBuf, + #[arg(long)] + pub out: PathBuf, + #[arg(long)] + pub run: String, + #[arg(long, default_value = "candidate")] + pub label: String, + #[arg(long, default_value_t = 3)] + pub repeat: usize, + #[arg(long, default_value_t = 0)] + pub repeat_start: usize, + #[arg(long, default_value_t = 16)] + pub blocks: usize, + #[arg(long, default_value_t = 3)] + pub transactions_per_block: usize, + #[arg(long, default_value_t = 256)] + pub log_rows: usize, + #[arg(long, default_value_t = 8)] + pub pool_stride: usize, + #[arg(long, default_value_t = 2)] + pub page: usize, + #[arg(long, default_value_t = 2)] + pub page_size: usize, + #[arg(long, default_value_t = 0)] + pub seed: u64, + #[arg(long, default_value_t = 200)] + pub requests: usize, + #[arg(long, default_value_t = 4)] + pub concurrency: usize, + #[arg(long, default_value = "0")] + pub rates: String, + #[arg(long, default_value_t = 15_000)] + pub timeout_ms: u64, + #[arg(long, default_value_t = 3_000)] + pub max_scan_items: u64, + #[arg(long, default_value_t = 64)] + pub cache_mib: usize, + #[arg(long, default_value = "all")] + pub cases: String, + #[arg(long)] + pub live: bool, + #[arg(long, default_value_t = 100)] + pub write_interval_ms: u64, +} + +#[derive(clap::Args)] +pub struct CompareArgs { + #[arg(long = "bin", required = true, num_args = 1)] + pub bins: Vec, + #[arg(last = true, required = true, allow_hyphen_values = true)] + pub args: Vec, +} + +#[derive(Parser)] +struct ChildArgs { + #[command(flatten)] + args: Args, +} + +impl Args { + pub fn shape(&self) -> FixtureShape { + FixtureShape { + blocks: self.blocks, + transactions_per_block: self.transactions_per_block, + log_rows: self.log_rows, + pool_stride: self.pool_stride, + page: self.page, + page_size: self.page_size, + seed: self.seed, + } + } + + fn validate(&self) -> anyhow::Result> { + self.shape().validate().map_err(anyhow::Error::msg)?; + anyhow::ensure!( + !self.run.trim().is_empty() && !self.label.trim().is_empty(), + "run and label must not be empty" + ); + anyhow::ensure!( + self.requests > 0 && self.concurrency > 0 && self.repeat > 0, + "requests, concurrency and repeat must be positive" + ); + anyhow::ensure!( + self.timeout_ms > 0 && self.cache_mib > 0 && self.max_scan_items > 0, + "timeout, cache and scan limit must be positive" + ); + let rates: Vec = super::parse_list(&self.rates)?; + anyhow::ensure!(!rates.is_empty(), "rates must not be empty"); + let mut unique = rates.clone(); + unique.sort_unstable(); + unique.dedup(); + anyhow::ensure!( + rates.len() == unique.len(), + "rates must not contain duplicates" + ); + anyhow::ensure!( + self.repeat_start.checked_add(self.repeat).is_some(), + "repeat range overflow" + ); + if self.live { + anyhow::ensure!( + self.write_interval_ms > 0 && !rates.contains(&0), + "live workloads require a positive offered rate and write interval" + ); + let duration_ms = self.requests as f64 * 1000.0 / *rates.iter().min().unwrap() as f64; + anyhow::ensure!((self.blocks as f64 * self.write_interval_ms as f64) > duration_ms + self.timeout_ms as f64, "tail too short: increase --blocks or --write-interval-ms to span requests plus timeout"); + } + Ok(rates) + } +} + +pub async fn drive( + router: axum::Router, + case: cases::Case, + requests: usize, + concurrency: usize, + rate: u64, + timeout: Duration, +) -> anyhow::Result { + drive_requests( + move || { + let router = router.clone(); + let case = case.clone(); + async move { cases::request(router, &case).await } + }, + requests, + concurrency, + rate, + timeout, + ) + .await +} + +async fn measure_case(args: &Args, name: &str, rate: u64) -> anyhow::Result<(Value, Value)> { + let state_config = FjallStateConfig { + cache: Some(args.cache_mib), + ..Default::default() + }; + let archive_config = FjallArchiveConfig { + cache: Some(args.cache_mib), + ..Default::default() + }; + let stores = FjallStores::open_in(&args.work, &state_config, &archive_config) + .map_err(anyhow::Error::msg)?; + let store_path = stores.path().to_path_buf(); + let mut fixture = ApiFixture::new(stores, args.shape()).map_err(anyhow::Error::msg)?; + fixture.domain = fixture.domain.with_persistent_wal(store_path.join("wal"))?; + let case = cases::cases(&fixture) + .into_iter() + .find(|case| case.name == name) + .context("unknown case")?; + anyhow::ensure!( + !args.live || case.live, + "{name} changes its expected response during live sync" + ); + let config = MinibfConfig::new("127.0.0.1:0".parse()?).with_max_scan_items(args.max_scan_items); + let router = dolos_minibf::build_router(config, fixture.domain.clone()); + let (_, expected_response) = cases::response(router.clone(), &case) + .await + .context("fixture validation / warmup")?; + fixture.domain.archive().counters.reset(); + let mut digest = Sha256::new(); + for block in fixture.blocks.iter().chain(&fixture.tail) { + digest.update((block.len() as u64).to_be_bytes()); + digest.update(block.as_ref()); + } + let fixture_identity = json!({ + "schema": 1, "shape": args.shape(), "chain_sha256": format!("{:x}", digest.finalize()), + "epoch": fixture.epoch, "initial_tip": fixture.vectors.blocks[args.blocks - 1].slot, + "case": case, "network": "synthetic-preview", "body_era": "conway", + "response_sha256": expected_response, + }); + let stop = Arc::new(AtomicBool::new(false)); + let before = measure::counters(); + let writer = if args.live { + let domain = fixture.domain.clone(); + let blocks = fixture.tail.clone(); + let stop = stop.clone(); + let interval = Duration::from_millis(args.write_interval_ms); + Some(std::thread::spawn(move || -> Result { + let start = Instant::now(); + let mut commits = measure::histogram(); + let mut written = 0usize; + for block in blocks { + if stop.load(Ordering::Relaxed) { + break; + } + let commit_start = Instant::now(); + domain + .roll_forward(block) + .map_err(|error| error.to_string())?; + commits + .record(commit_start.elapsed().as_nanos().max(1) as u64) + .map_err(|error| error.to_string())?; + written += 1; + std::thread::park_timeout(interval); + } + Ok(json!({ + "blocks": written, "elapsed_seconds": start.elapsed().as_secs_f64(), + "roll_forward_latency": measure::histogram_json(&commits), + })) + })) + } else { + None + }; + let measured = drive_requests( + move || { + let router = router.clone(); + let case = case.clone(); + let expected = expected_response.clone(); + async move { + let (bytes, actual) = cases::response(router, &case).await?; + anyhow::ensure!( + actual == expected, + "response changed after fixture validation" + ); + Ok(bytes) + } + }, + args.requests, + args.concurrency, + rate, + Duration::from_millis(args.timeout_ms), + ) + .await; + stop.store(true, Ordering::Relaxed); + let writer_metrics = if let Some(writer) = writer { + writer.thread().unpark(); + let result = writer + .join() + .map_err(|_| anyhow::anyhow!("sync writer panicked"))? + .map_err(anyhow::Error::msg)?; + anyhow::ensure!( + result["blocks"].as_u64().unwrap_or(0) > 0, + "writer made no progress" + ); + Some(result) + } else { + None + }; + let counters = measure::counters().delta(&before); + let mut metrics = measured?; + metrics["kind"] = json!("minibf"); + metrics["workload"] = json!(name); + metrics["work"] = serde_json::to_value(fixture.domain.archive().counters.snapshot())?; + metrics["work_scope"] = json!(if args.live { "api-and-writer" } else { "api" }); + metrics["resources"] = counters.json(); + metrics["writer"] = json!(writer_metrics); + metrics["storage_bytes"] = json!({ + "state": directory_bytes(&store_path.join("state"))?, + "archive": directory_bytes(&store_path.join("archive"))?, + "wal": directory_bytes(&store_path.join("wal"))?, + }); + Ok((fixture_identity, metrics)) +} + +fn directory_bytes(path: &std::path::Path) -> std::io::Result { + let metadata = match path.metadata() { + Ok(metadata) => metadata, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(0), + Err(error) => return Err(error), + }; + if metadata.is_file() { + return Ok(metadata.len()); + } + let mut total = 0; + for entry in std::fs::read_dir(path)? { + total += directory_bytes(&entry?.path())?; + } + Ok(total) +} + +pub fn run(args: Args) -> anyhow::Result<()> { + let rates = args.validate()?; + std::fs::create_dir_all(&args.work)?; + let selected: Vec<&str> = if args.cases == "all" { + CASE_NAMES + .iter() + .copied() + .filter(|name| !args.live || !["epoch-blocks-reverse", "epoch-latest"].contains(name)) + .collect() + } else { + args.cases.split(',').collect() + }; + let mut unique = std::collections::BTreeSet::new(); + anyhow::ensure!( + selected + .iter() + .all(|name| CASE_NAMES.contains(name) && unique.insert(*name)), + "unknown or duplicate case; available: {}", + CASE_NAMES.join(",") + ); + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(4) + .enable_all() + .build()?; + let environment = measure::environment(&[&args.work]); + let binary = std::env::current_exe()?; + let binary_sha256 = format!("{:x}", Sha256::digest(std::fs::read(binary)?)); + let mut file = OpenOptions::new() + .create(true) + .append(true) + .open(&args.out)?; + let mut failed = false; + for repeat in args.repeat_start..args.repeat_start + args.repeat { + for rate in &rates { + for name in &selected { + eprintln!("minibf {} repeat {repeat}: {name} rate={rate}", args.label); + let (fixture, metrics) = runtime.block_on(measure_case(&args, name, *rate))?; + failed |= ["errors", "timeouts", "rejected"] + .iter() + .any(|field| metrics[*field].as_u64().unwrap_or(0) > 0); + let record = json!({ + "schema": 1, "run": args.run, "label": args.label, "repeat": repeat, + "suite": selected, + "binary_sha256": binary_sha256, "environment": environment, + "build": { + "revision": env!("VERGEN_GIT_SHA"), + "dirty": env!("VERGEN_GIT_DIRTY"), + "profile": if cfg!(debug_assertions) { "debug" } else { "release" }, + }, + "fixture": fixture, "metrics": metrics, + "settings": { + "requests": args.requests, "concurrency": args.concurrency, "rate": rate, + "timeout_ms": args.timeout_ms, "max_scan_items": args.max_scan_items, + "cache_mib_per_store": args.cache_mib, "cache": "route-primed", + "durability": "production-defaults", "storage_version": "v4", + "dictionary": super::dictionary::Dictionary::bundled().id().to_string(), + "mode": if args.live { "live-sync" } else { "read-only" }, + "write_interval_ms": args.write_interval_ms, "runtime_workers": 4, + }, + "command": super::command_line(), + }); + serde_json::to_writer(&mut file, &record)?; + file.write_all(b"\n")?; + file.flush()?; + } + } + } + anyhow::ensure!( + !failed, + "one or more minibf workloads failed, timed out or overloaded; results retained" + ); + Ok(()) +} + +pub fn compare(args: CompareArgs) -> anyhow::Result<()> { + let parsed = + ChildArgs::try_parse_from(std::iter::once("minibf".to_string()).chain(args.args.clone()))?; + parsed.args.validate()?; + let mut binaries = std::collections::BTreeMap::new(); + for spec in &args.bins { + let (label, path) = spec + .split_once('=') + .context("--bin must be LABEL=PATH to cargo-xtask")?; + anyhow::ensure!( + !label.is_empty() && !path.is_empty(), + "empty binary label or path" + ); + anyhow::ensure!( + binaries + .insert(label.to_string(), PathBuf::from(path).canonicalize()?) + .is_none(), + "duplicate label" + ); + } + anyhow::ensure!( + binaries.contains_key("baseline") && binaries.len() >= 2, + "supply baseline and at least one candidate" + ); + anyhow::ensure!( + !args.args.iter().any(|arg| arg == "--label" + || arg.starts_with("--label=") + || arg == "--repeat-start" + || arg.starts_with("--repeat-start=")), + "comparison owns label and repeat-start" + ); + let mut child_args = args.args.clone(); + let mut position = 0; + while position < child_args.len() { + if child_args[position] == "--repeat" { + child_args.drain(position..position + 2); + } else if child_args[position].starts_with("--repeat=") { + child_args.remove(position); + } else { + position += 1; + } + } + let mut failed = false; + for repeat in 0..parsed.args.repeat { + let mut ordered: Vec<_> = binaries.iter().collect(); + if repeat % 2 == 1 { + ordered.reverse(); + } + for (label, binary) in ordered { + let status = std::process::Command::new(binary) + .arg("perf") + .arg("minibf") + .arg("run") + .args(&child_args) + .arg("--label") + .arg(label) + .arg("--repeat") + .arg("1") + .arg("--repeat-start") + .arg(repeat.to_string()) + .status()?; + failed |= !status.success(); + } + } + anyhow::ensure!( + !failed, + "at least one comparison arm failed; inspect retained records" + ); + Ok(()) +} diff --git a/xtask/src/perf/minibf/report.rs b/xtask/src/perf/minibf/report.rs new file mode 100644 index 000000000..b97216180 --- /dev/null +++ b/xtask/src/perf/minibf/report.rs @@ -0,0 +1,318 @@ +use std::{ + collections::{BTreeMap, BTreeSet}, + fmt::Write as _, +}; + +use serde_json::Value; + +#[derive(Clone, clap::Args)] +pub struct Budgets { + #[arg(long, default_value_t = 1.10)] + pub max_p95_ratio: f64, + #[arg(long, default_value_t = 1.20)] + pub max_p99_ratio: f64, + #[arg(long, default_value_t = 0.90)] + pub min_throughput_ratio: f64, + #[arg(long, default_value_t = 3)] + pub min_repeats: usize, + #[arg(long, default_value_t = 1000)] + pub min_samples: u64, + #[arg(long)] + pub p95_budget_ms: Option, + #[arg(long)] + pub p99_budget_ms: Option, +} + +impl Default for Budgets { + fn default() -> Self { + Self { + max_p95_ratio: 1.10, + max_p99_ratio: 1.20, + min_throughput_ratio: 0.90, + min_repeats: 3, + min_samples: 1000, + p95_budget_ms: None, + p99_budget_ms: None, + } + } +} + +impl Budgets { + pub fn validate(&self) -> anyhow::Result<()> { + let positive = |number: f64| number.is_finite() && number > 0.0; + anyhow::ensure!( + positive(self.max_p95_ratio) + && positive(self.max_p99_ratio) + && positive(self.min_throughput_ratio), + "ratios must be finite and positive" + ); + anyhow::ensure!( + self.min_repeats > 0 && self.min_samples > 0, + "minimum repeats and samples must be positive" + ); + anyhow::ensure!( + self.p95_budget_ms + .into_iter() + .chain(self.p99_budget_ms) + .all(positive), + "latency budgets must be finite and positive" + ); + Ok(()) + } +} + +fn group_key(record: &Value) -> String { + let mut environment = record["environment"].clone(); + if let Some(environment) = environment.as_object_mut() { + environment.remove("revision"); + environment.remove("recorded_at"); + } + serde_json::to_string(&serde_json::json!({ + "schema": record["schema"], "run": record["run"], + "suite": record["suite"], + "environment": environment, "fixture": record["fixture"], "settings": record["settings"], + "build_profile": record["build"]["profile"], + })) + .unwrap() +} + +fn metric(record: &Value, pointer: &str) -> Option { + record + .pointer(pointer)? + .as_f64() + .filter(|number| number.is_finite() && *number > 0.0) +} + +fn valid(record: &Value) -> bool { + let metrics = &record["metrics"]; + let count = metrics["completed"].as_u64(); + let hash = |value: &Value| { + value.as_str().is_some_and(|value| { + value.len() == 64 && value.bytes().all(|byte| byte.is_ascii_hexdigit()) + }) + }; + record["schema"] == 1 + && record["run"] + .as_str() + .is_some_and(|value| !value.is_empty()) + && record["environment"].is_object() + && record["fixture"].is_object() + && record["settings"].is_object() + && record["label"] + .as_str() + .is_some_and(|value| !value.is_empty()) + && hash(&record["binary_sha256"]) + && hash(&record["fixture"]["response_sha256"]) + && count.is_some_and(|count| count > 0) + && (record["settings"]["transport"] != "http" || metrics["tip_unchanged"] == true) + && metrics["requests"].as_u64() == count + && record["settings"]["requests"].as_u64() == count + && metrics["latency"]["count"].as_u64() == count + && ["errors", "timeouts", "rejected"] + .iter() + .all(|field| metrics[*field].as_u64() == Some(0)) + && metric(record, "/metrics/latency/p95_us").is_some() + && metric(record, "/metrics/latency/p99_us").is_some() + && metric(record, "/metrics/completed_per_second").is_some() +} + +fn median(values: &mut [f64]) -> f64 { + values.sort_by(f64::total_cmp); + let middle = values.len() / 2; + if values.len().is_multiple_of(2) { + (values[middle - 1] + values[middle]) / 2.0 + } else { + values[middle] + } +} + +fn map_repeats<'a>(arm: &[&'a Value]) -> Option> { + let mut repeats = BTreeMap::new(); + let mut identities = BTreeSet::new(); + for record in arm { + let repeat = record["repeat"].as_u64()?; + if repeats.insert(repeat, *record).is_some() { + return None; + } + identities.insert(record["binary_sha256"].as_str()?); + } + (identities.len() == 1).then_some(repeats) +} + +pub fn assess(records: &[Value], budgets: &Budgets) -> (String, bool) { + let mut groups: BTreeMap>> = BTreeMap::new(); + for record in records + .iter() + .filter(|record| record["metrics"]["kind"] == "minibf") + { + groups + .entry(group_key(record)) + .or_default() + .entry(record["label"].as_str().unwrap_or("").to_string()) + .or_default() + .push(record); + } + if groups.is_empty() { + return ("No minibf records.\n".into(), false); + } + let mut output = String::from("## Minibf paired gates\n\n"); + writeln!(output, "Budgets: p95 ratio ≤ {:.3}; p99 ratio ≤ {:.3}; completed throughput ratio ≥ {:.3}; at least {} paired repeats and {} successful requests per repeat.\n", budgets.max_p95_ratio, budgets.max_p99_ratio, budgets.min_throughput_ratio, budgets.min_repeats, budgets.min_samples).unwrap(); + output.push_str("| run / workload | candidate | pairs | p95/p99 ratios (p95 min–max) | throughput ratio | baseline p95/p99 ms | candidate p95/p99 ms | verdict |\n|---|---|---:|---:|---:|---:|---:|---|\n"); + let mut passed = true; + let mut suites: BTreeMap, BTreeSet)> = BTreeMap::new(); + for record in records + .iter() + .filter(|record| record["metrics"]["kind"] == "minibf") + { + let mut scope = record.clone(); + if let Some(fixture) = scope["fixture"].as_object_mut() { + fixture.remove("case"); + fixture.remove("response_sha256"); + } + let scope = format!( + "{}:{}:{}", + group_key(&scope), + record["label"], + record["repeat"] + ); + let (expected, actual) = suites.entry(scope).or_default(); + if let Some(names) = record["suite"].as_array() { + expected.extend(names.iter().filter_map(Value::as_str).map(str::to_string)); + } + if let Some(name) = record["metrics"]["workload"].as_str() { + actual.insert(name.to_string()); + } + } + for (expected, actual) in suites.values() { + if expected.is_empty() || expected != actual { + output.push_str("| incomplete suite | — | 0 | — | — | — | — | INVALID: workload records missing or undeclared |\n"); + passed = false; + } + } + for labels in groups.values() { + let example = labels.values().next().unwrap()[0]; + let title = format!( + "{} / {} / {} rps", + example["run"].as_str().unwrap_or("?"), + example["metrics"]["workload"].as_str().unwrap_or("?"), + example["settings"]["rate"] + ); + let Some(baseline) = labels.get("baseline") else { + writeln!( + output, + "| {title} | — | 0 | — | — | — | — | UNPAIRED: no baseline |" + ) + .unwrap(); + passed = false; + continue; + }; + if labels.len() == 1 { + writeln!( + output, + "| {title} | — | 0 | — | — | — | — | UNPAIRED: no candidate |" + ) + .unwrap(); + passed = false; + } + for (label, candidate) in labels + .iter() + .filter(|(label, _)| label.as_str() != "baseline") + { + let pairs = map_repeats(baseline) + .zip(map_repeats(candidate)) + .filter(|(baseline, candidate)| baseline.keys().eq(candidate.keys())); + let Some((baseline, candidate)) = pairs else { + writeln!(output, "| {title} | {label} | 0 | — | — | — | — | UNPAIRED: duplicates, identities or repeats differ |").unwrap(); + passed = false; + continue; + }; + if !baseline + .values() + .chain(candidate.values()) + .all(|record| valid(record)) + { + writeln!(output, "| {title} | {label} | {} | — | — | — | — | INVALID: failed requests or incomplete metrics |", baseline.len()).unwrap(); + passed = false; + continue; + } + let mut latency = Vec::new(); + let mut tail_latency = Vec::new(); + let mut throughput = Vec::new(); + let mut baseline_p95 = Vec::new(); + let mut baseline_p99 = Vec::new(); + let mut candidate_p95 = Vec::new(); + let mut candidate_p99 = Vec::new(); + let mut absolute_pass = true; + for (repeat, baseline_record) in &baseline { + let candidate_record = candidate[repeat]; + let p95 = |record| metric(record, "/metrics/latency/p95_us").unwrap(); + let p99 = |record| metric(record, "/metrics/latency/p99_us").unwrap(); + latency.push(p95(candidate_record) / p95(baseline_record)); + tail_latency.push(p99(candidate_record) / p99(baseline_record)); + throughput.push( + metric(candidate_record, "/metrics/completed_per_second").unwrap() + / metric(baseline_record, "/metrics/completed_per_second").unwrap(), + ); + baseline_p95.push(p95(baseline_record) / 1000.0); + baseline_p99.push(p99(baseline_record) / 1000.0); + candidate_p95.push(p95(candidate_record) / 1000.0); + candidate_p99.push(p99(candidate_record) / 1000.0); + for record in [*baseline_record, candidate_record] { + absolute_pass &= budgets + .p95_budget_ms + .is_none_or(|limit| p95(record) / 1000.0 <= limit); + absolute_pass &= budgets + .p99_budget_ms + .is_none_or(|limit| p99(record) / 1000.0 <= limit); + } + } + let latency_ratio = median(&mut latency); + let tail_ratio = median(&mut tail_latency); + let throughput_ratio = median(&mut throughput); + let enough = baseline.len() >= budgets.min_repeats + && baseline.values().chain(candidate.values()).all(|record| { + record["metrics"]["completed"].as_u64().unwrap() >= budgets.min_samples + }); + let verdict = if !enough { + "INSUFFICIENT" + } else if !absolute_pass { + "FAIL: absolute budget" + } else if latency_ratio > budgets.max_p95_ratio + || tail_ratio > budgets.max_p99_ratio + || throughput_ratio < budgets.min_throughput_ratio + { + "FAIL: regression" + } else { + "PASS" + }; + passed &= verdict == "PASS"; + writeln!(output, "| {title} | {label} | {} | {:.3}/{:.3} ({:.3}–{:.3}) | {:.3} | {:.3}/{:.3} | {:.3}/{:.3} | {verdict} |", + baseline.len(), latency_ratio, tail_ratio, latency[0], latency[latency.len() - 1], throughput_ratio, + median(&mut baseline_p95), median(&mut baseline_p99), median(&mut candidate_p95), median(&mut candidate_p99)).unwrap(); + } + } + output.push_str("\nLatency includes scheduling delay; rejected arrivals and invalid/late responses prevent a pass. Small-fixture passes do not establish mainnet capacity.\n"); + output.push_str("\n## Minibf work and resources\n\n| label / repeat / workload | scope | log rows | tag candidates | block reads | CPU ms | peak RSS bytes | sync blocks |\n|---|---|---:|---:|---:|---:|---:|---:|\n"); + for record in records + .iter() + .filter(|record| record["metrics"]["kind"] == "minibf") + { + let metrics = &record["metrics"]; + writeln!( + output, + "| {} / {} / {} | {} | {} | {} | {} | {} | {} | {} |", + record["label"].as_str().unwrap_or("?"), + record["repeat"], + metrics["workload"].as_str().unwrap_or("?"), + metrics["work_scope"].as_str().unwrap_or("?"), + metrics["work"]["log_rows"], + metrics["work"]["tag_candidates"], + metrics["work"]["block_reads"], + metrics["resources"]["cpu_ms"], + metrics["resources"]["max_rss_bytes"], + metrics["writer"]["blocks"], + ) + .unwrap(); + } + (output, passed) +} diff --git a/xtask/src/perf/mod.rs b/xtask/src/perf/mod.rs new file mode 100644 index 000000000..5bc6f5202 --- /dev/null +++ b/xtask/src/perf/mod.rs @@ -0,0 +1,62 @@ +pub mod dictionary; +pub mod load; +pub mod measure; +pub mod minibf; +pub mod report; +pub mod storage; + +#[derive(clap::Subcommand)] +pub enum Cmd { + #[command( + subcommand, + about = "Storage, codec and import performance experiments" + )] + Storage(storage::Cmd), + #[command(subcommand, about = "Minibf endpoint and HTTP performance experiments")] + Minibf(minibf::Cmd), + #[command(about = "Render shared benchmark records")] + Report { + #[arg(required = true)] + files: Vec, + }, +} + +pub fn run(command: Cmd) -> anyhow::Result<()> { + match command { + Cmd::Storage(command) => storage::run(command), + Cmd::Minibf(command) => minibf::dispatch(command), + Cmd::Report { files } => { + print!("{}", report::render(&report::load(&files)?)); + Ok(()) + } + } +} + +pub fn parse_list(s: &str) -> anyhow::Result> +where + T::Err: std::fmt::Display, +{ + s.split(',') + .map(str::trim) + .filter(|p| !p.is_empty()) + .map(|p| p.parse::().map_err(|e| anyhow::anyhow!("{p}: {e}"))) + .collect() +} + +/// The command line as a shell would need it typed, so a recorded command +/// replays. +pub fn command_line() -> String { + std::env::args() + .map(|arg| shell_word(&arg)) + .collect::>() + .join(" ") +} + +fn shell_word(arg: &str) -> String { + let plain = |c: char| c.is_ascii_alphanumeric() || "-_./=,:@+%".contains(c); + if !arg.is_empty() && arg.chars().all(plain) { + arg.to_string() + } else { + format!("'{}'", arg.replace('\'', "'\\''")) + } +} diff --git a/xtask/src/archive_bench/report.rs b/xtask/src/perf/report.rs similarity index 98% rename from xtask/src/archive_bench/report.rs rename to xtask/src/perf/report.rs index 67040639f..029a6290f 100644 --- a/xtask/src/archive_bench/report.rs +++ b/xtask/src/perf/report.rs @@ -239,6 +239,23 @@ fn ms(us: f64) -> String { /// Render every record kind present as a markdown section. pub fn render(records: &[Value]) -> String { + if records + .iter() + .any(|record| record["metrics"]["kind"] == "minibf") + { + let other: Vec<_> = records + .iter() + .filter(|record| record["metrics"]["kind"] != "minibf") + .cloned() + .collect(); + let mut rendered = if other.is_empty() { + String::new() + } else { + render(&other) + }; + rendered.push_str(&super::minibf::report::assess(records, &Default::default()).0); + return rendered; + } let mut out = String::new(); let runs = runs(records); if !runs.is_empty() { diff --git a/xtask/src/archive_bench/codec.rs b/xtask/src/perf/storage/codec.rs similarity index 96% rename from xtask/src/archive_bench/codec.rs rename to xtask/src/perf/storage/codec.rs index c506889cf..0f6beb9e6 100644 --- a/xtask/src/archive_bench/codec.rs +++ b/xtask/src/perf/storage/codec.rs @@ -531,7 +531,7 @@ impl Reader { if !self.files.contains_key(&segment) { let file = File::open(segment_path(&self.dir, segment))?; if self.nocache { - set_nocache(&file); + set_nocache(&file)?; } self.files.insert(segment, file); } @@ -623,13 +623,32 @@ fn read_exact_at(file: &File, mut buf: &mut [u8], mut offset: u64) -> io::Result } #[cfg(target_os = "macos")] -fn set_nocache(file: &File) { +fn set_nocache(file: &File) -> io::Result<()> { use std::os::unix::io::AsRawFd; - unsafe { libc::fcntl(file.as_raw_fd(), libc::F_NOCACHE, 1) }; + if unsafe { libc::fcntl(file.as_raw_fd(), libc::F_NOCACHE, 1) } == -1 { + return Err(io::Error::last_os_error()); + } + Ok(()) } #[cfg(not(target_os = "macos"))] -fn set_nocache(_file: &File) {} +fn set_nocache(_file: &File) -> io::Result<()> { + Ok(()) +} + +#[cfg(all(test, target_os = "macos"))] +mod nocache_tests { + use super::*; + + #[test] + fn nocache_reports_unsupported_descriptors() { + let file = tempfile::tempfile().unwrap(); + set_nocache(&file).unwrap(); + let (socket, _peer) = std::os::unix::net::UnixStream::pair().unwrap(); + let descriptor: std::os::fd::OwnedFd = socket.into(); + assert!(set_nocache(&File::from(descriptor)).is_err()); + } +} #[cfg(target_os = "linux")] fn drop_cache(file: &File, offset: u64, len: u64) { diff --git a/xtask/src/archive_bench/corpus.rs b/xtask/src/perf/storage/corpus.rs similarity index 100% rename from xtask/src/archive_bench/corpus.rs rename to xtask/src/perf/storage/corpus.rs diff --git a/xtask/src/archive_bench/mod.rs b/xtask/src/perf/storage/mod.rs similarity index 93% rename from xtask/src/archive_bench/mod.rs rename to xtask/src/perf/storage/mod.rs index f421b1019..d8674eb54 100644 --- a/xtask/src/archive_bench/mod.rs +++ b/xtask/src/perf/storage/mod.rs @@ -1,6 +1,6 @@ //! Developer benchmarks for the archive's compressed block segments. //! -//! `cargo xtask archive-bench` measures the archive's write and read paths +//! `cargo xtask perf storage` measures the archive's write and read paths //! on real block corpora and renders paired comparisons from the records. //! The store-level workloads run the production `dolos_flatfiles` store //! beside a modelled sink (one frame per block, or raw bodies) so codec and @@ -12,14 +12,13 @@ pub mod codec; pub mod corpus; -pub mod dictionary; -pub mod measure; pub mod node; pub mod presets; -pub mod report; pub mod train; pub mod workloads; +use super::{command_line, dictionary, measure, parse_list, report}; + use std::io::Write; use std::path::PathBuf; @@ -96,9 +95,9 @@ impl CorpusArgs { } } -/// Options of the `bench` subcommand. +/// Options of the `run` subcommand. #[derive(clap::Args)] -pub struct BenchArgs { +pub struct RunArgs { #[command(flatten)] corpus: CorpusArgs, @@ -174,7 +173,8 @@ pub struct BenchArgs { #[derive(Subcommand)] pub enum Cmd { /// Run a store-level preset and append JSON records to --out - Bench(Box), + #[command(alias = "bench")] + Run(Box), /// Drive a dolos binary through import and API workloads Node(Box), @@ -233,17 +233,6 @@ pub enum Cmd { Report { files: Vec }, } -pub fn parse_list(s: &str) -> anyhow::Result> -where - T::Err: std::fmt::Display, -{ - s.split(',') - .map(str::trim) - .filter(|p| !p.is_empty()) - .map(|p| p.parse::().map_err(|e| anyhow::anyhow!("{p}: {e}"))) - .collect() -} - fn codecs(spec: &str, dictionary: &str) -> anyhow::Result> { let mut out = Vec::new(); for name in spec.split(',').map(str::trim).filter(|p| !p.is_empty()) { @@ -263,28 +252,10 @@ fn codecs(spec: &str, dictionary: &str) -> anyhow::Result> Ok(out) } -/// The command line as a shell would need it typed, so a recorded command -/// replays. -pub fn command_line() -> String { - std::env::args() - .map(|arg| shell_word(&arg)) - .collect::>() - .join(" ") -} - -fn shell_word(arg: &str) -> String { - let plain = |c: char| c.is_ascii_alphanumeric() || "-_./=,:@+%".contains(c); - if !arg.is_empty() && arg.chars().all(plain) { - arg.to_string() - } else { - format!("'{}'", arg.replace('\'', "'\\''")) - } -} - pub fn run(cmd: Cmd) -> anyhow::Result<()> { match cmd { - Cmd::Bench(args) => { - let BenchArgs { + Cmd::Run(args) => { + let RunArgs { corpus, preset, work, diff --git a/xtask/src/archive_bench/node.rs b/xtask/src/perf/storage/node.rs similarity index 98% rename from xtask/src/archive_bench/node.rs rename to xtask/src/perf/storage/node.rs index 64546da1d..314d656c5 100644 --- a/xtask/src/archive_bench/node.rs +++ b/xtask/src/perf/storage/node.rs @@ -449,6 +449,14 @@ fn terminate(child: &mut Child) -> io::Result { wait_child(child) } +fn serving_outcome( + served: anyhow::Result<()>, + cleanup: io::Result, +) -> anyhow::Result { + served?; + Ok(cleanup?) +} + fn dir_bytes(dir: &Path) -> u64 { let Ok(entries) = std::fs::read_dir(dir) else { return 0; @@ -1328,8 +1336,7 @@ pub fn run(args: NodeArgs) -> anyhow::Result<()> { } Ok(()) })(); - let usage = terminate(&mut server)?; - served?; + let usage = serving_outcome(served, terminate(&mut server))?; if usage.status != 0 && usage.status != -(libc_sigterm()) { eprintln!( " warning: {} serve exited with {}", @@ -1385,6 +1392,21 @@ fn libc_sigterm() -> i32 { mod tests { use super::*; + #[test] + fn cleanup_does_not_mask_workload_errors() { + let result = serving_outcome( + Err(anyhow::anyhow!("server exited before listening")), + Err(io::Error::from_raw_os_error(libc::ECHILD)), + ); + assert_eq!( + result.err().unwrap().to_string(), + "server exited before listening" + ); + let result = serving_outcome(Ok(()), Err(io::Error::other("cleanup failed"))); + assert_eq!(result.err().unwrap().to_string(), "cleanup failed"); + assert!(serving_outcome(Ok(()), Ok(ChildUsage::default())).is_ok()); + } + #[test] fn a_binary_spec_names_its_label_path_and_version() { let b = Bin::parse("baseline=/tmp/dolos-old:v3").unwrap(); diff --git a/xtask/src/archive_bench/presets.rs b/xtask/src/perf/storage/presets.rs similarity index 97% rename from xtask/src/archive_bench/presets.rs rename to xtask/src/perf/storage/presets.rs index 1d3e3322e..7ff1c9a3b 100644 --- a/xtask/src/archive_bench/presets.rs +++ b/xtask/src/perf/storage/presets.rs @@ -115,15 +115,6 @@ fn clean(dir: &Path, keep: bool) -> io::Result<()> { Ok(()) } -fn store_files(dir: &Path) -> io::Result> { - let mut files: Vec = std::fs::read_dir(dir)? - .filter_map(|e| e.ok().map(|e| e.path())) - .filter(|p| p.extension().is_some_and(|x| x == "segment")) - .collect(); - files.sort(); - Ok(files) -} - /// One read workload of a run and whether the cache is put back into the /// regime before it runs. pub struct ReadStep { @@ -268,7 +259,7 @@ pub fn run(preset: Preset, corpus: &Corpus, opts: &Options, out: &mut dyn Write) }, )?; let locations = outcome.locations; - let files = store_files(&dir)?; + let files = outcome.files; for ®ime in &opts.regimes { if regime == Regime::NoCache && !codec.reads_without_cache() { eprintln!( diff --git a/xtask/src/archive_bench/train.rs b/xtask/src/perf/storage/train.rs similarity index 97% rename from xtask/src/archive_bench/train.rs rename to xtask/src/perf/storage/train.rs index 17489988e..82d3cb538 100644 --- a/xtask/src/archive_bench/train.rs +++ b/xtask/src/perf/storage/train.rs @@ -103,15 +103,16 @@ pub fn evaluate(fixtures: &[Fixture], codecs: &[(String, Codec)]) -> io::Result< let mut decoder = codec.decoder()?; let mut frames: Vec> = Vec::with_capacity(fixture.corpus.blocks.len()); let mut per_era: std::collections::BTreeMap<&str, (u64, u64)> = Default::default(); - let cpu = thread_cpu_ns(); + let mut encode_ns = 0u64; for block in &fixture.corpus.blocks { + let cpu = thread_cpu_ns(); let frame = encoder.encode(&block.body)?; + encode_ns = encode_ns.saturating_add(thread_cpu_ns().saturating_sub(cpu)); let e = per_era.entry(block.era).or_default(); e.0 += block.body.len() as u64; e.1 += frame.len() as u64; frames.push(frame.to_vec()); } - let encode_ns = thread_cpu_ns().saturating_sub(cpu); let cpu = thread_cpu_ns(); for (block, frame) in fixture.corpus.blocks.iter().zip(&frames) { let body = decoder.decode(frame)?; diff --git a/xtask/src/archive_bench/workloads.rs b/xtask/src/perf/storage/workloads.rs similarity index 95% rename from xtask/src/archive_bench/workloads.rs rename to xtask/src/perf/storage/workloads.rs index 4a3d5ff67..dca24195d 100644 --- a/xtask/src/archive_bench/workloads.rs +++ b/xtask/src/perf/storage/workloads.rs @@ -25,6 +25,7 @@ pub struct WriteParams { pub struct WriteOutcome { pub locations: Vec, + pub files: Vec, pub metrics: Value, } @@ -82,6 +83,7 @@ pub fn write_corpus( let process = counters().delta(&before); let heap_peak = heap_peak().map(|p| p.saturating_sub(heap_base)); let blocks = locations.len() as u64; + let files = sink.files(); let metrics = json!({ "kind": "write", "workload": params.name, @@ -105,7 +107,7 @@ pub fn write_corpus( "batch_latency": histogram_json(&batch_hist), "fsync_latency": histogram_json(&fsync_hist), "segment_crossings": crossings, - "segments": sink.files().len(), + "segments": files.len(), "heap_peak_bytes": heap_peak, "import_buffers": sink.append_stats().map(|stats| json!({ "encoders_peak": stats.parallel_encoders_peak, @@ -115,7 +117,11 @@ pub fn write_corpus( })), "process": process.json(), }); - Ok(WriteOutcome { locations, metrics }) + Ok(WriteOutcome { + locations, + files, + metrics, + }) } /// What the page cache holds when a read workload starts. @@ -428,11 +434,32 @@ impl<'a> Worker<'a> { for i in range { self.read_one(i)?; } - self.acc.page.record(t.elapsed().as_nanos() as u64).unwrap(); + record_scan_latency(&mut self.acc.page, t.elapsed())?; Ok(()) } } +fn record_scan_latency( + histogram: &mut hdrhistogram::Histogram, + elapsed: std::time::Duration, +) -> io::Result<()> { + let nanos = u64::try_from(elapsed.as_nanos()).map_err(io::Error::other)?; + histogram.record(nanos.max(1)).map_err(io::Error::other) +} + +#[cfg(test)] +mod scan_tests { + use super::*; + + #[test] + fn over_range_scans_fail_without_panicking_or_clipping() { + let mut histogram = histogram(); + record_scan_latency(&mut histogram, std::time::Duration::from_secs(1)).unwrap(); + assert!(record_scan_latency(&mut histogram, std::time::Duration::from_secs(7200)).is_err()); + assert_eq!(histogram.len(), 1); + } +} + fn thread_seed(seed: u64, thread: usize) -> u64 { seed ^ (thread as u64 + 1).wrapping_mul(0x9E37_79B9_7F4A_7C15) } diff --git a/xtask/tests/minibf_perf.rs b/xtask/tests/minibf_perf.rs new file mode 100644 index 000000000..fcb97dcf2 --- /dev/null +++ b/xtask/tests/minibf_perf.rs @@ -0,0 +1,371 @@ +use std::time::Duration; + +use axum::{routing::get, Json, Router}; +use dolos_core::{config::MinibfConfig, Domain}; +use dolos_testing::{ + performance::{ApiFixture, FixtureShape}, + toy_domain::{FjallStores, MemoryStores, ToyStores}, +}; +use serde_json::{json, Value}; +use xtask::perf::minibf::{ + cases, drive, + report::{assess, Budgets}, +}; + +#[test] +fn http_credentials_are_rejected_before_loading_inputs_or_secrets() { + let result = std::process::Command::new(env!("CARGO_BIN_EXE_cargo-xtask")) + .args([ + "perf", + "minibf", + "http", + "--url", + "http://127.0.0.1:1", + "--manifest", + "missing-manifest.json", + "--server-binary", + "missing-dolos", + "--server-revision", + "test", + "--out", + "unused.jsonl", + "--run", + "test", + "--label", + "baseline", + "--project-id-env", + "DOLOS_PERF_TEST_ABSENT_SECRET", + ]) + .env_remove("DOLOS_PERF_TEST_ABSENT_SECRET") + .output() + .unwrap(); + assert!(!result.status.success()); + assert!(String::from_utf8_lossy(&result.stderr).contains("--project-id-env requires HTTPS")); +} + +async fn verify_fixture(stores: Stores) { + let fixture = ApiFixture::new(stores, FixtureShape::default()).unwrap(); + let router = dolos_minibf::build_router( + MinibfConfig::new("127.0.0.1:0".parse().unwrap()), + fixture.domain.clone(), + ); + for case in cases::cases(&fixture) { + fixture.domain.archive().counters.reset(); + cases::request(router.clone(), &case).await.unwrap(); + let work = fixture.domain.archive().counters.snapshot(); + match case.name.as_str() { + "epoch-stakes-sparse-pool" => { + assert!(work.log_rows > 5); + assert_eq!(work.block_reads, 0); + } + "epoch-blocks-pool-no-match" => { + assert_eq!(work.block_reads, fixture.shape.blocks as u64); + assert!(work.decoded_bytes > 0); + } + "address-transactions-page" => assert!(work.tag_candidates > 0), + "account-utxos-wide" => { + assert_eq!( + work.utxo_refs, + (fixture.shape.blocks * fixture.shape.transactions_per_block) as u64 + ); + assert_eq!(work.exact_lookups, work.utxo_refs); + } + _ => {} + } + } +} + +#[tokio::test(flavor = "multi_thread")] +async fn route_fixtures_and_work_guards_on_memory() { + verify_fixture(MemoryStores::open()).await; +} + +#[tokio::test(flavor = "multi_thread")] +async fn route_fixtures_and_work_guards_on_fjall() { + verify_fixture(FjallStores::open()).await; +} + +fn scalar_case() -> cases::Case { + cases::Case { + name: "fixture".into(), + path: "/fixture".into(), + fields: vec![], + expected: json!([1]), + array: true, + live: true, + } +} + +#[tokio::test] +async fn overload_keeps_offered_requests_and_bounds_in_flight() { + let router = Router::new().route( + "/fixture", + get(|| async { + tokio::time::sleep(Duration::from_millis(30)).await; + Json(json!([1])) + }), + ); + let result = drive(router, scalar_case(), 20, 1, 10000, Duration::from_secs(1)) + .await + .unwrap(); + assert_eq!(result["peak_in_flight"], 1); + assert!(result["rejected"].as_u64().unwrap() > 0); + assert_eq!( + result["completed"].as_u64().unwrap() + result["rejected"].as_u64().unwrap(), + 20 + ); +} + +#[tokio::test] +async fn wrong_responses_and_deadlines_never_count_as_success() { + let router = Router::new().route("/fixture", get(|| async { Json(json!([])) })); + let result = drive(router, scalar_case(), 3, 1, 0, Duration::from_secs(1)) + .await + .unwrap(); + assert_eq!(result["completed"], 0); + assert_eq!(result["errors"], 3); + let router = Router::new().route( + "/fixture", + get(|| async { + tokio::time::sleep(Duration::from_millis(10)).await; + Json(json!([1])) + }), + ); + let result = drive(router, scalar_case(), 1, 1, 0, Duration::from_millis(1)) + .await + .unwrap(); + assert_eq!(result["completed"], 0); + assert_eq!(result["timeouts"], 1); +} + +fn record(label: &str, repeat: u64, p95: f64) -> Value { + json!({ + "schema": 1, "run": "test", "label": label, "repeat": repeat, + "suite": ["test"], + "binary_sha256": if label == "baseline" { "a".repeat(64) } else { "b".repeat(64) }, + "environment": {"os": "test", "revision": label}, + "fixture": {"schema": 1, "seed": 0, "response_sha256": "c".repeat(64)}, + "settings": {"requests": 1000, "rate": 0}, + "metrics": { + "kind": "minibf", "workload": "test", "requests": 1000, + "completed": 1000, "errors": 0, "timeouts": 0, "rejected": 0, + "latency": {"count": 1000, "p95_us": p95, "p99_us": p95 * 2.0}, + "completed_per_second": 100.0, + } + }) +} + +fn paired() -> Vec { + (0..3) + .flat_map(|repeat| { + [ + record("baseline", repeat, 100.0), + record("candidate", repeat, 105.0), + ] + }) + .collect() +} + +#[test] +fn pairing_refuses_missing_duplicate_incompatible_or_failed_evidence() { + let records = paired(); + assert!(assess(&records, &Budgets::default()).1); + let mut missing = records.clone(); + missing.pop(); + assert!(!assess(&missing, &Budgets::default()).1); + let mut duplicate = records.clone(); + duplicate.push(records[0].clone()); + assert!(!assess(&duplicate, &Budgets::default()).1); + for pointer in ["/settings/requests", "/fixture/seed", "/metrics/errors"] { + let mut changed = records.clone(); + *changed[1].pointer_mut(pointer).unwrap() = json!(1); + assert!(!assess(&changed, &Budgets::default()).1, "{pointer}"); + } + let mut changed_binary = records.clone(); + changed_binary[3]["binary_sha256"] = json!("c".repeat(64)); + assert!(!assess(&changed_binary, &Budgets::default()).1); + let mut absent_metric = records.clone(); + absent_metric[1]["metrics"]["latency"]["p95_us"] = Value::Null; + assert!(!assess(&absent_metric, &Budgets::default()).1); +} + +#[test] +fn gates_detect_regression_and_baseline_absolute_budget_failure() { + let mut records = paired(); + for record in records + .iter_mut() + .filter(|record| record["label"] == "candidate") + { + record["metrics"]["latency"]["p95_us"] = json!(120); + } + assert!(assess(&records, &Budgets::default()) + .0 + .contains("FAIL: regression")); + let budgets = Budgets { + p95_budget_ms: Some(0.09), + ..Default::default() + }; + assert!(assess(&paired(), &budgets) + .0 + .contains("FAIL: absolute budget")); + let budgets = Budgets { + min_samples: 2000, + ..Default::default() + }; + assert!(assess(&paired(), &budgets).0.contains("INSUFFICIENT")); +} + +#[test] +fn live_replay_cli_records_writer_progress() { + let temp = tempfile::tempdir().unwrap(); + let output = temp.path().join("live.jsonl"); + let result = std::process::Command::new(env!("CARGO_BIN_EXE_cargo-xtask")) + .args([ + "perf", + "minibf", + "run", + "--run", + "smoke", + "--repeat", + "1", + "--requests", + "3", + "--rates", + "20", + "--timeout-ms", + "1000", + "--live", + "--write-interval-ms", + "500", + "--cases", + "epoch-blocks-pool-no-match", + ]) + .arg("--work") + .arg(temp.path()) + .arg("--out") + .arg(&output) + .output() + .unwrap(); + assert!( + result.status.success(), + "{}", + String::from_utf8_lossy(&result.stderr) + ); + let record: Value = + serde_json::from_str(std::fs::read_to_string(output).unwrap().trim()).unwrap(); + assert_eq!(record["metrics"]["completed"], 3); + assert!(record["metrics"]["writer"]["blocks"].as_u64().unwrap() > 0); + assert_eq!(record["metrics"]["work_scope"], "api-and-writer"); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn http_calibration_checks_responses_and_refuses_tip_changes() { + use std::sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }; + let temp = tempfile::tempdir().unwrap(); + let changed = Arc::new(AtomicBool::new(false)); + let change_on_read = Arc::new(AtomicBool::new(false)); + let tip_changed = changed.clone(); + let trigger = change_on_read.clone(); + let read_changed = changed.clone(); + let router = Router::new() + .route("/blocks/latest", get(move || { + let changed = tip_changed.clone(); + async move { Json(json!({"hash": if changed.load(Ordering::Relaxed) { "b".repeat(64) } else { "a".repeat(64) }})) } + })) + .route("/fixture", get(move || { + let trigger = trigger.clone(); + let changed = read_changed.clone(); + async move { + if trigger.load(Ordering::Relaxed) { changed.store(true, Ordering::Relaxed); } + Json(json!([1])) + } + })); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { + axum::serve(listener, router).await.unwrap(); + }); + let manifest = temp.path().join("manifest.json"); + std::fs::write(&manifest, serde_json::to_vec(&json!({ + "schema": 1, "dataset": "test", "network": "test", "tip_hash": "a".repeat(64), + "server_environment": {"host": "test"}, + "settings": {"storage_version": "v4", "dictionary": "test", "max_scan_items": 3000, "cache": "test", "durability": "test"}, + "cases": [scalar_case()], + })).unwrap()).unwrap(); + for changes in [false, true] { + change_on_read.store(changes, Ordering::Relaxed); + let output = temp.path().join(format!("http-{changes}.jsonl")); + let manifest = manifest.clone(); + let output_argument = output.clone(); + let result = tokio::task::spawn_blocking(move || { + std::process::Command::new(env!("CARGO_BIN_EXE_cargo-xtask")) + .args([ + "perf", + "minibf", + "http", + "--run", + "smoke", + "--label", + "candidate", + "--server-revision", + "test", + "--requests", + "3", + "--repeat", + "1", + ]) + .arg("--url") + .arg(format!("http://{address}")) + .arg("--server-binary") + .arg(env!("CARGO_BIN_EXE_cargo-xtask")) + .arg("--manifest") + .arg(manifest) + .arg("--out") + .arg(output_argument) + .output() + .unwrap() + }) + .await + .unwrap(); + assert_eq!( + result.status.success(), + !changes, + "{}", + String::from_utf8_lossy(&result.stderr) + ); + let record: Value = + serde_json::from_str(std::fs::read_to_string(output).unwrap().trim()).unwrap(); + assert_eq!(record["metrics"]["tip_unchanged"], !changes); + assert_eq!(record["metrics"]["resources_scope"], "client-only"); + } + server.abort(); +} + +#[test] +fn incomplete_declared_suites_cannot_pass() { + let mut records = paired(); + for record in &mut records { + record["suite"] = json!(["test", "missing-workload"]); + } + assert!(!assess(&records, &Budgets::default()).1); +} + +#[test] +fn p99_regressions_and_response_changes_cannot_pass() { + let mut records = paired(); + for record in records + .iter_mut() + .filter(|record| record["label"] == "candidate") + { + record["metrics"]["latency"]["p99_us"] = json!(300); + } + assert!(assess(&records, &Budgets::default()) + .0 + .contains("FAIL: regression")); + let mut records = paired(); + records[1]["fixture"]["response_sha256"] = json!("different-response"); + assert!(!assess(&records, &Budgets::default()).1); +} diff --git a/xtask/tests/perf_cli.rs b/xtask/tests/perf_cli.rs new file mode 100644 index 000000000..f6e1fa76b --- /dev/null +++ b/xtask/tests/perf_cli.rs @@ -0,0 +1,86 @@ +use std::process::Command; + +fn binary() -> &'static str { + env!("CARGO_BIN_EXE_cargo-xtask") +} + +#[test] +fn storage_and_minibf_have_separate_command_trees() { + for subject in ["storage", "minibf"] { + let output = Command::new(binary()) + .args(["perf", subject, "--help"]) + .output() + .unwrap(); + assert!(output.status.success()); + let help = String::from_utf8(output.stdout).unwrap(); + if subject == "storage" { + assert!(help.contains("train")); + assert!(!help.contains("minibf")); + } else { + assert!(help.contains("compare")); + assert!(!help.contains("train")); + } + } + assert!(Command::new(binary()) + .args(["perf", "storage", "run", "--help"]) + .output() + .unwrap() + .status + .success()); + assert!(Command::new(binary()) + .args(["archive-bench", "bench", "--help"]) + .status() + .unwrap() + .success()); +} + +#[test] +fn paired_runner_uses_minibf_command_and_shared_report() { + let temp = tempfile::tempdir().unwrap(); + let records = temp.path().join("paired.jsonl"); + let result = Command::new(binary()) + .args(["perf", "minibf", "compare", "--bin"]) + .arg(format!("baseline={}", binary())) + .arg("--bin") + .arg(format!("candidate={}", binary())) + .args(["--", "--work"]) + .arg(temp.path()) + .arg("--out") + .arg(&records) + .args([ + "--run", + "taxonomy-smoke", + "--requests", + "2", + "--repeat", + "1", + "--cases", + "epoch-stakes-sparse-pool", + ]) + .output() + .unwrap(); + assert!( + result.status.success(), + "{}", + String::from_utf8_lossy(&result.stderr) + ); + assert_eq!( + std::fs::read_to_string(&records).unwrap().lines().count(), + 2 + ); + let report = Command::new(binary()) + .args(["perf", "report"]) + .arg(&records) + .output() + .unwrap(); + assert!(report.status.success()); + assert!(String::from_utf8(report.stdout) + .unwrap() + .contains("INSUFFICIENT")); + let gate = Command::new(binary()) + .args(["perf", "minibf", "check"]) + .arg(&records) + .output() + .unwrap(); + assert!(!gate.status.success()); +} diff --git a/xtask/tests/archive_bench_smoke.rs b/xtask/tests/storage_perf_smoke.rs similarity index 82% rename from xtask/tests/archive_bench_smoke.rs rename to xtask/tests/storage_perf_smoke.rs index ae0788810..de20e5793 100644 --- a/xtask/tests/archive_bench_smoke.rs +++ b/xtask/tests/storage_perf_smoke.rs @@ -1,13 +1,13 @@ //! The smoke preset over a synthetic corpus: every workload runs, every //! body read back matches, and the records carry what a report needs. -use xtask::archive_bench::codec::Codec; -use xtask::archive_bench::corpus::Corpus; -use xtask::archive_bench::dictionary::Dictionary; -use xtask::archive_bench::presets::{run, Options, Preset}; -use xtask::archive_bench::report; -use xtask::archive_bench::train::{evaluate, Fixture}; -use xtask::archive_bench::workloads::{EvictOptions, Regime}; +use xtask::perf::dictionary::Dictionary; +use xtask::perf::report; +use xtask::perf::storage::codec::Codec; +use xtask::perf::storage::corpus::Corpus; +use xtask::perf::storage::presets::{run, Options, Preset}; +use xtask::perf::storage::train::{evaluate, Fixture}; +use xtask::perf::storage::workloads::{EvictOptions, Regime}; #[test] fn automatic_store_selects_parallel_in_a_two_worker_process() { @@ -16,8 +16,9 @@ fn automatic_store_selects_parallel_in_a_two_worker_process() { let status = std::process::Command::new(env!("CARGO_BIN_EXE_cargo-xtask")) .env("RAYON_NUM_THREADS", "2") .args([ - "archive-bench", - "bench", + "perf", + "storage", + "run", "--preset", "write", "--synthetic", @@ -202,3 +203,32 @@ fn evaluation_round_trips_and_ranks_the_dictionary() { assert!(r["metrics"]["ratio"].as_f64().unwrap() < 1.0); } } + +#[test] +fn write_outcome_carries_the_sink_segment_files() { + use xtask::perf::storage::workloads::{write_corpus, WriteParams}; + + let corpus = Corpus::synthetic(3, 400, 2); + for codec in [Codec::Raw, Codec::Store] { + let directory = tempfile::tempdir().unwrap(); + std::fs::write(directory.path().join("unrelated.txt"), "not a segment").unwrap(); + let outcome = write_corpus( + &corpus, + &codec, + directory.path(), + &WriteParams { + name: "files".into(), + batch: 100, + encode_threads: 1, + fsync: true, + }, + ) + .unwrap(); + assert!(!outcome.files.is_empty()); + assert_eq!(outcome.metrics["segments"], outcome.files.len()); + assert!(outcome.files.iter().all(|file| file.is_file() + && file + .extension() + .is_some_and(|extension| extension == "segment"))); + } +}