From 4015217a2ebc265d14ba79c70adfeecbddda778c Mon Sep 17 00:00:00 2001 From: Lewis Date: Fri, 29 May 2026 11:08:01 +0300 Subject: [PATCH] feat(store): Clock trait for DST Lewis: May this revision serve well! --- .config/nextest.toml | 20 ++ .containerignore | 12 + .gitignore | 1 + crates/tranquil-pds/src/repo/mod.rs | 5 +- crates/tranquil-pds/src/scheduled.rs | 15 +- crates/tranquil-pds/src/state.rs | 5 +- .../tests/gc_compaction_restart.rs | 7 +- crates/tranquil-store/benches/blockstore.rs | 9 +- crates/tranquil-store/src/backup.rs | 7 +- .../src/blockstore/compaction.rs | 6 +- .../src/blockstore/group_commit.rs | 60 ++-- crates/tranquil-store/src/blockstore/store.rs | 40 ++- crates/tranquil-store/src/clock.rs | 124 +++++++ crates/tranquil-store/src/consistency.rs | 22 +- crates/tranquil-store/src/eventlog/writer.rs | 17 +- .../src/gauntlet/chaos_walker.rs | 13 +- .../tranquil-store/src/gauntlet/invariants.rs | 85 ++--- crates/tranquil-store/src/gauntlet/metrics.rs | 9 +- crates/tranquil-store/src/gauntlet/mod.rs | 8 +- crates/tranquil-store/src/gauntlet/op.rs | 6 + crates/tranquil-store/src/gauntlet/runner.rs | 325 ++++++++++++++---- .../tranquil-store/src/gauntlet/scenarios.rs | 55 ++- crates/tranquil-store/src/gauntlet/shrink.rs | 5 +- crates/tranquil-store/src/gauntlet/soak.rs | 20 +- .../tranquil-store/src/gauntlet/workload.rs | 23 +- crates/tranquil-store/src/lib.rs | 8 +- .../src/metastore/commit_ops.rs | 8 +- .../tranquil-store/src/metastore/handler.rs | 8 +- crates/tranquil-store/src/sim.rs | 27 +- crates/tranquil-store/tests/backup.rs | 9 +- .../tranquil-store/tests/checkpoint_race.rs | 4 +- crates/tranquil-store/tests/common/mod.rs | 19 +- .../tests/compaction_liveness.rs | 9 +- .../tests/compaction_restart.rs | 6 +- .../tests/external_corruption_recovery.rs | 6 +- .../tests/gauntlet_index_backed.rs | 41 ++- crates/tranquil-store/tests/gauntlet_smoke.rs | 16 +- crates/tranquil-store/tests/gc_stress.rs | 11 +- .../tests/rotation_robustness.rs | 12 +- crates/tranquil-store/tests/sim_blockstore.rs | 22 +- frontend/package.json | 5 + frontend/pnpm-lock.yaml | 213 +++++++++++- justfile | 14 +- 43 files changed, 1057 insertions(+), 280 deletions(-) create mode 100644 .containerignore create mode 100644 crates/tranquil-store/src/clock.rs diff --git a/.config/nextest.toml b/.config/nextest.toml index 840556a..9761845 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -106,6 +106,16 @@ slow-timeout = { period = "300s", terminate-after = 8 } filter = "binary(compaction_restart) | binary(mst_refcount_integrity) | binary(gc_compaction_restart)" slow-timeout = { period = "120s", terminate-after = 4 } +[[profile.default.overrides]] +filter = "test(/retention_time_travel_survives_many_seeds/)" +slow-timeout = { period = "300s", terminate-after = 24 } +test-group = "io-heavy-sim" + +[[profile.default.overrides]] +filter = "binary(fd_lifecycle)" +slow-timeout = { period = "300s", terminate-after = 4 } +test-group = "io-heavy-sim" + [[profile.ci.overrides]] filter = "test(/import_with_verification/) | test(/plc_migration/)" test-group = "serial-env-tests" @@ -137,3 +147,13 @@ test-group = "heavy-load-tests" [[profile.ci.overrides]] filter = "binary(repo_lifecycle)" test-group = "heavy-load-tests" + +[[profile.ci.overrides]] +filter = "test(/retention_time_travel_survives_many_seeds/)" +slow-timeout = { period = "300s", terminate-after = 24 } +test-group = "io-heavy-sim" + +[[profile.ci.overrides]] +filter = "binary(fd_lifecycle)" +slow-timeout = { period = "300s", terminate-after = 4 } +test-group = "io-heavy-sim" diff --git a/.containerignore b/.containerignore new file mode 100644 index 0000000..ba30eaf --- /dev/null +++ b/.containerignore @@ -0,0 +1,12 @@ +target/ +.git/ +.jj/ +**/node_modules/ +frontend/dist/ +frontend/coverage/ +frontend/.pnpm-store/ +.direnv/ +result +.env +*.output +reference-pds-bsky/ diff --git a/.gitignore b/.gitignore index ec8e17e..1b04c5c 100644 --- a/.gitignore +++ b/.gitignore @@ -4,5 +4,6 @@ target/ result frontend/node_modules/ frontend/dist/ +frontend/coverage/ frontend/.pnpm-store frontend/.npmrc diff --git a/crates/tranquil-pds/src/repo/mod.rs b/crates/tranquil-pds/src/repo/mod.rs index 894391a..110f081 100644 --- a/crates/tranquil-pds/src/repo/mod.rs +++ b/crates/tranquil-pds/src/repo/mod.rs @@ -8,11 +8,12 @@ use jacquard_repo::error::RepoError; use jacquard_repo::repo::CommitData; use jacquard_repo::storage::BlockStore; use tranquil_store::blockstore::TranquilBlockStore; +use tranquil_store::{RealIO, SystemClock}; #[derive(Clone)] pub enum AnyBlockStore { Postgres(PostgresBlockStore), - TranquilStore(TranquilBlockStore), + TranquilStore(TranquilBlockStore), } impl AnyBlockStore { @@ -23,7 +24,7 @@ impl AnyBlockStore { } } - pub fn as_tranquil_store(&self) -> Option<&TranquilBlockStore> { + pub fn as_tranquil_store(&self) -> Option<&TranquilBlockStore> { match self { Self::TranquilStore(s) => Some(s), Self::Postgres(_) => None, diff --git a/crates/tranquil-pds/src/scheduled.rs b/crates/tranquil-pds/src/scheduled.rs index 1dcab93..e32ccea 100644 --- a/crates/tranquil-pds/src/scheduled.rs +++ b/crates/tranquil-pds/src/scheduled.rs @@ -574,7 +574,10 @@ impl CompactionBlocklist { } fn run_compaction_pass( - store: &tranquil_store::blockstore::TranquilBlockStore, + store: &tranquil_store::blockstore::TranquilBlockStore< + tranquil_store::RealIO, + tranquil_store::SystemClock, + >, liveness_threshold: f64, grace_period_ms: u64, blocklist: &parking_lot::Mutex, @@ -836,7 +839,10 @@ fn cid_to_bytes(cid: &Cid) -> anyhow::Result { } fn walk_repo_dag_sync( - store: &tranquil_store::blockstore::TranquilBlockStore, + store: &tranquil_store::blockstore::TranquilBlockStore< + tranquil_store::RealIO, + tranquil_store::SystemClock, + >, head_cid: &Cid, reachable: &mut std::collections::HashSet, phantom_files: &mut std::collections::HashSet, @@ -960,7 +966,10 @@ fn paginate_repos( } pub fn run_reachability_walk( - store: &tranquil_store::blockstore::TranquilBlockStore, + store: &tranquil_store::blockstore::TranquilBlockStore< + tranquil_store::RealIO, + tranquil_store::SystemClock, + >, repo_repo: &dyn RepoRepository, ) -> anyhow::Result { let rt = tokio::runtime::Handle::current(); diff --git a/crates/tranquil-pds/src/state.rs b/crates/tranquil-pds/src/state.rs index 4569ed8..53a7047 100644 --- a/crates/tranquil-pds/src/state.rs +++ b/crates/tranquil-pds/src/state.rs @@ -471,7 +471,10 @@ impl AppState { } struct TranquilStoreWiring { - blockstore: tranquil_store::blockstore::TranquilBlockStore, + blockstore: tranquil_store::blockstore::TranquilBlockStore< + tranquil_store::RealIO, + tranquil_store::SystemClock, + >, signal_provider: Arc, repos: PostgresRepositories, segments_dir: PathBuf, diff --git a/crates/tranquil-pds/tests/gc_compaction_restart.rs b/crates/tranquil-pds/tests/gc_compaction_restart.rs index c993820..1c65228 100644 --- a/crates/tranquil-pds/tests/gc_compaction_restart.rs +++ b/crates/tranquil-pds/tests/gc_compaction_restart.rs @@ -5,7 +5,12 @@ use common::*; use reqwest::StatusCode; use serde_json::{Value, json}; -fn run_compaction(store: &tranquil_store::blockstore::TranquilBlockStore) { +fn run_compaction( + store: &tranquil_store::blockstore::TranquilBlockStore< + tranquil_store::RealIO, + tranquil_store::SystemClock, + >, +) { let liveness = store.compaction_liveness(0).unwrap(); liveness .iter() diff --git a/crates/tranquil-store/benches/blockstore.rs b/crates/tranquil-store/benches/blockstore.rs index a79cbe8..4a13ec5 100644 --- a/crates/tranquil-store/benches/blockstore.rs +++ b/crates/tranquil-store/benches/blockstore.rs @@ -13,6 +13,7 @@ use sha2::{Digest, Sha256}; use tranquil_store::blockstore::{ BlockStoreConfig, DEFAULT_MAX_FILE_SIZE, GroupCommitConfig, TranquilBlockStore, }; +use tranquil_store::{RealIO, SystemClock}; const DAG_CBOR_CODEC: u64 = 0x71; const SHA2_256_CODE: u64 = 0x12; @@ -60,11 +61,11 @@ fn compute_stats(durations: &mut [Duration]) -> Option { }) } -fn open_store(dir: &Path) -> TranquilBlockStore { +fn open_store(dir: &Path) -> TranquilBlockStore { open_store_sharded(dir, 1) } -fn open_store_sharded(dir: &Path, shard_count: u8) -> TranquilBlockStore { +fn open_store_sharded(dir: &Path, shard_count: u8) -> TranquilBlockStore { TranquilBlockStore::open(BlockStoreConfig { data_dir: dir.join("data"), index_dir: dir.join("index"), @@ -183,7 +184,9 @@ async fn bench_read_throughput(block_count: usize, concurrency: usize) { cids }; - let run_reads = |label: &'static str, store: TranquilBlockStore, cids: Vec| async move { + let run_reads = |label: &'static str, + store: TranquilBlockStore, + cids: Vec| async move { let start = Instant::now(); let handles: Vec<_> = (0..concurrency) diff --git a/crates/tranquil-store/src/backup.rs b/crates/tranquil-store/src/backup.rs index 14046cd..63cdc92 100644 --- a/crates/tranquil-store/src/backup.rs +++ b/crates/tranquil-store/src/backup.rs @@ -8,6 +8,7 @@ use crate::blockstore::{ BlockOffset, BlockstoreSnapshot, CommitEpoch, CommitError, DataFileId, QuiesceGuard, RebuildError, TranquilBlockStore, }; +use crate::clock::SystemClock; use crate::eventlog::{ EventLog, EventLogConfig, EventLogFreezeGuard, EventLogSnapshotState, EventSequence, EventWithMutations, SegmentId, SegmentOffset, @@ -157,14 +158,14 @@ enum BackupLineage<'a> { } pub struct BackupCoordinator<'a, S: StorageIO> { - blockstore: &'a TranquilBlockStore, + blockstore: &'a TranquilBlockStore, eventlog: &'a EventLog, metastore: &'a Metastore, } impl<'a, S: StorageIO + Send + Sync + 'static> BackupCoordinator<'a, S> { pub fn new( - blockstore: &'a TranquilBlockStore, + blockstore: &'a TranquilBlockStore, eventlog: &'a EventLog, metastore: &'a Metastore, ) -> Self { @@ -312,7 +313,7 @@ impl<'a, S: StorageIO + Send + Sync + 'static> BackupCoordinator<'a, S> { BackupManifest { version: BACKUP_FORMAT_VERSION, - created_at_ms: crate::wall_clock_ms().raw(), + created_at_ms: crate::blockstore::WallClockMs::now().raw(), blockstore: { let max_cursor = bs .shard_cursors diff --git a/crates/tranquil-store/src/blockstore/compaction.rs b/crates/tranquil-store/src/blockstore/compaction.rs index 0ebbdbc..1a31de9 100644 --- a/crates/tranquil-store/src/blockstore/compaction.rs +++ b/crates/tranquil-store/src/blockstore/compaction.rs @@ -9,6 +9,7 @@ use super::hint::{HintFileWriter, hint_file_path}; use super::manager::DataFileManager; use super::types::{ BlockLocation, CidBytes, CommitEpoch, CompactionResult, CompactionStats, DataFileId, + WallClockMs, }; #[derive(Debug)] @@ -65,6 +66,7 @@ pub(super) fn compact_on_writer_thread( active_files: &ActiveFileSet, hint_positions: &super::group_commit::ShardHintPositions, epoch: &super::types::EpochCounter, + now: WallClockMs, ) -> Result { if active_files.contains(source_file_id) { return Err(CompactionError::ActiveFileCannotBeCompacted); @@ -89,6 +91,7 @@ pub(super) fn compact_on_writer_thread( new_file_id, current_epoch, grace_period_ms, + now, ); match result { @@ -183,6 +186,7 @@ fn purge_phantom_file( }) } +#[allow(clippy::too_many_arguments)] fn stream_compact( manager: &DataFileManager, index: &BlockIndex, @@ -191,9 +195,9 @@ fn stream_compact( new_file_id: DataFileId, current_epoch: CommitEpoch, grace_period_ms: u64, + now: WallClockMs, ) -> Result<(u64, u64, u64, super::types::HintOffset), CompactionError> { let mut reader = DataFileReader::open(manager.io(), source_fd)?; - let now = crate::wall_clock_ms(); let new_handle = manager.open_for_append(new_file_id)?; let mut writer = DataFileWriter::new(manager.io(), new_handle.fd(), new_file_id)?; diff --git a/crates/tranquil-store/src/blockstore/group_commit.rs b/crates/tranquil-store/src/blockstore/group_commit.rs index 7ade216..2d7759a 100644 --- a/crates/tranquil-store/src/blockstore/group_commit.rs +++ b/crates/tranquil-store/src/blockstore/group_commit.rs @@ -6,6 +6,7 @@ use std::thread; use parking_lot::RwLock; +use crate::clock::{Clock, LogicalNanos}; use crate::fsync_order::PostBlockstoreHook; use super::BlocksSynced; @@ -224,13 +225,14 @@ impl Default for GroupCommitConfig { } } -struct ShardContext { +struct ShardContext { shard_id: ShardId, epoch: EpochCounter, file_ids: Arc, active_files: Arc, hint_positions: Arc, verify_persisted_blocks: bool, + clock: C, } struct ActiveState { @@ -256,8 +258,8 @@ struct SingleShardWriter { } impl SingleShardWriter { - fn spawn( - ctx: ShardContext, + fn spawn( + ctx: ShardContext, manager: DataFileManager, index: Arc, config: GroupCommitConfig, @@ -359,20 +361,22 @@ pub struct GroupCommitWriter { } impl GroupCommitWriter { - pub fn spawn( + pub fn spawn( make_manager: impl Fn() -> DataFileManager, index: Arc, config: GroupCommitConfig, + clock: C, ) -> Result { - Self::spawn_sharded(make_manager, index, config, None, None, 1, None) + Self::spawn_sharded(make_manager, index, config, None, None, 1, None, clock) } - pub fn spawn_with_hook( + pub fn spawn_with_hook( make_manager: impl Fn() -> DataFileManager, index: Arc, config: GroupCommitConfig, post_sync_hook: Option>, initial_epoch: Option, + clock: C, ) -> Result { Self::spawn_sharded( make_manager, @@ -382,10 +386,12 @@ impl GroupCommitWriter { initial_epoch, 1, None, + clock, ) } - pub fn spawn_sharded( + #[allow(clippy::too_many_arguments)] + pub fn spawn_sharded( make_manager: F, index: Arc, config: GroupCommitConfig, @@ -393,9 +399,11 @@ impl GroupCommitWriter { initial_epoch: Option, shard_count: u8, checkpoint_positions: Option<&CheckpointPositions>, + clock: C, ) -> Result where S: StorageIO + 'static, + C: Clock, F: Fn() -> DataFileManager, { let shard_count = shard_count.max(1); @@ -435,6 +443,7 @@ impl GroupCommitWriter { active_files: Arc::clone(&active_files), hint_positions: Arc::clone(&hint_positions), verify_persisted_blocks: config.verify_persisted_blocks, + clock: clock.clone(), }; SingleShardWriter::spawn( ctx, @@ -843,23 +852,24 @@ fn handle_quiesce( let _ = resume.blocking_recv(); } -fn maybe_checkpoint( +fn maybe_checkpoint( index: &BlockIndex, epoch: &EpochCounter, config: &GroupCommitConfig, - last_checkpoint: &mut std::time::Instant, + clock: &C, + last_checkpoint: &mut LogicalNanos, writes_since_checkpoint: &mut u64, hint_positions: &ShardHintPositions, ) { - let interval = std::time::Duration::from_millis(config.checkpoint_interval_ms); - let elapsed = last_checkpoint.elapsed() >= interval; + let interval = LogicalNanos::from_millis(config.checkpoint_interval_ms); + let elapsed = clock.monotonic().saturating_sub(*last_checkpoint) >= interval; let threshold = *writes_since_checkpoint >= config.checkpoint_write_threshold; if !elapsed && !threshold { return; } match index.write_checkpoint(epoch.current(), hint_positions) { Ok(()) => { - *last_checkpoint = std::time::Instant::now(); + *last_checkpoint = clock.monotonic(); *writes_since_checkpoint = 0; tracing::debug!("periodic checkpoint written"); } @@ -880,17 +890,17 @@ fn shutdown_checkpoint( } } -fn commit_loop( +fn commit_loop( manager: &DataFileManager, index: &BlockIndex, receiver: &flume::Receiver, config: &GroupCommitConfig, state: &mut ActiveState, post_sync_hook: Option<&dyn PostBlockstoreHook>, - ctx: &ShardContext, + ctx: &ShardContext, ) { let epoch = &ctx.epoch; - let mut last_checkpoint = std::time::Instant::now(); + let mut last_checkpoint = ctx.clock.monotonic(); let mut writes_since_checkpoint: u64 = 0; loop { @@ -914,6 +924,7 @@ fn commit_loop( &ctx.active_files, &ctx.hint_positions, epoch, + ctx.clock.wall_millis(), ); let _ = response.send(result); continue; @@ -925,7 +936,7 @@ fn commit_loop( let repaired = index.repair_leaked_refcounts( &leaked_cids, epoch.current(), - crate::wall_clock_ms(), + ctx.clock.wall_millis(), ); let _ = response.send(Ok(repaired)); continue; @@ -985,6 +996,7 @@ fn commit_loop( &ctx.active_files, &ctx.hint_positions, epoch, + ctx.clock.wall_millis(), ); let _ = response.send(result); }); @@ -996,7 +1008,7 @@ fn commit_loop( let repaired = index.repair_leaked_refcounts( &leaked_cids, epoch.current(), - crate::wall_clock_ms(), + ctx.clock.wall_millis(), ); let _ = response.send(Ok(repaired)); }); @@ -1005,6 +1017,7 @@ fn commit_loop( index, epoch, config, + &ctx.clock, &mut last_checkpoint, &mut writes_since_checkpoint, &ctx.hint_positions, @@ -1032,13 +1045,13 @@ fn run_post_sync_hook(hook: Option<&dyn PostBlockstoreHook>, proof: &BlocksSynce } } -fn drain_and_process_remaining( +fn drain_and_process_remaining( manager: &DataFileManager, index: &BlockIndex, receiver: &flume::Receiver, state: &mut ActiveState, post_sync_hook: Option<&dyn PostBlockstoreHook>, - ctx: &ShardContext, + ctx: &ShardContext, ) { let epoch = &ctx.epoch; let mut entries: Vec = Vec::new(); @@ -1082,13 +1095,14 @@ fn drain_and_process_remaining( &ctx.active_files, &ctx.hint_positions, epoch, + ctx.clock.wall_millis(), ); let _ = response.send(result); }); repairs.into_iter().for_each(|(leaked_cids, response)| { let repaired = - index.repair_leaked_refcounts(&leaked_cids, epoch.current(), crate::wall_clock_ms()); + index.repair_leaked_refcounts(&leaked_cids, epoch.current(), ctx.clock.wall_millis()); let _ = response.send(Ok(repaired)); }); @@ -1215,12 +1229,12 @@ fn rollback_batch( }); } -fn process_batch( +fn process_batch( manager: &DataFileManager, index: &BlockIndex, batch: &[BatchEntry], state: &mut ActiveState, - ctx: &ShardContext, + ctx: &ShardContext, ) -> Result<(HashMap<[u8; CID_SIZE], BlockLocation>, BlocksSynced), CommitError> { let epoch = &ctx.epoch; let batch_start = std::time::Instant::now(); @@ -1330,7 +1344,7 @@ fn process_batch( let write_nanos = batch_start.elapsed().as_nanos() as u64; let current_epoch = epoch.current(); - let now = crate::wall_clock_ms(); + let now = ctx.clock.wall_millis(); let rollback_on_err = |e: CommitError| -> CommitError { rollback_batch(manager, state, &rotations); diff --git a/crates/tranquil-store/src/blockstore/store.rs b/crates/tranquil-store/src/blockstore/store.rs index 4008f12..c4bc7f4 100644 --- a/crates/tranquil-store/src/blockstore/store.rs +++ b/crates/tranquil-store/src/blockstore/store.rs @@ -11,6 +11,7 @@ use jacquard_repo::error::RepoError; use jacquard_repo::repo::CommitData; use jacquard_repo::storage::BlockStore; +use crate::clock::{Clock, SystemClock}; use crate::fsync_order::PostBlockstoreHook; use crate::io::{OpenOptions, RealIO, StorageIO}; @@ -23,7 +24,7 @@ use super::manager::DataFileManager; use super::reader::{BlockStoreReader, ReadError}; use super::types::{ BlockLocation, BlockOffset, CollectionResult, CompactionResult, DataFileId, EpochCounter, - LivenessInfo, WallClockMs, + LivenessInfo, }; fn cid_to_bytes(cid: &Cid) -> Result<[u8; CID_SIZE], RepoError> { @@ -111,15 +112,16 @@ impl Drop for QuiesceGuard { } } -pub struct TranquilBlockStore { +pub struct TranquilBlockStore { writer: Arc, reader: Arc>, index: Arc, epoch: EpochCounter, data_dir: PathBuf, + clock: C, } -impl Clone for TranquilBlockStore { +impl Clone for TranquilBlockStore { fn clone(&self) -> Self { Self { writer: Arc::clone(&self.writer), @@ -127,6 +129,7 @@ impl Clone for TranquilBlockStore { index: Arc::clone(&self.index), epoch: self.epoch.clone(), data_dir: self.data_dir.clone(), + clock: self.clock.clone(), } } } @@ -170,7 +173,7 @@ impl Default for OpenRetryPolicy { } } -impl TranquilBlockStore { +impl TranquilBlockStore { pub fn open(config: BlockStoreConfig) -> Result { Self::open_with_hook(config, None) } @@ -179,7 +182,7 @@ impl TranquilBlockStore { config: BlockStoreConfig, post_sync_hook: Option>, ) -> Result { - Self::open_with_io_hook(config, RealIO::new, post_sync_hook) + Self::open_with_io_hook(config, RealIO::new, post_sync_hook, SystemClock) } pub fn open_with_retry( @@ -227,18 +230,23 @@ where } } -impl TranquilBlockStore { - pub fn open_with_io(config: BlockStoreConfig, make_io: F) -> Result +impl TranquilBlockStore { + pub fn open_with_io( + config: BlockStoreConfig, + make_io: F, + clock: C, + ) -> Result where F: Fn() -> S + Send + Sync + Clone + 'static, { - Self::open_with_io_hook(config, make_io, None) + Self::open_with_io_hook(config, make_io, None, clock) } pub fn open_with_io_hook( config: BlockStoreConfig, make_io: F, post_sync_hook: Option>, + clock: C, ) -> Result where F: Fn() -> S + Send + Sync + Clone + 'static, @@ -295,6 +303,7 @@ impl TranquilBlockStore { checkpoint_epoch, shard_count, checkpoint_positions, + clock.clone(), ) .map_err(commit_error_to_repo)?; let epoch = writer.epoch().clone(); @@ -317,6 +326,7 @@ impl TranquilBlockStore { index, epoch, data_dir, + clock, }) } @@ -497,9 +507,13 @@ impl TranquilBlockStore { )) } + pub fn clock(&self) -> &C { + &self.clock + } + pub fn collect_dead_blocks(&self, grace_period_ms: u64) -> Result { let current_epoch = self.epoch.current(); - let now = WallClockMs::now(); + let now = self.clock.wall_millis(); Ok(self .index .collect_dead_blocks(current_epoch, now, grace_period_ms)) @@ -536,7 +550,7 @@ impl TranquilBlockStore { grace_period_ms: u64, ) -> Result, RepoError> { let current_epoch = self.epoch.current(); - let now = WallClockMs::now(); + let now = self.clock.wall_millis(); Ok(self .index .liveness_by_file(current_epoch, now, grace_period_ms)) @@ -700,7 +714,7 @@ impl TranquilBlockStore { } } -impl BlockStore for TranquilBlockStore { +impl BlockStore for TranquilBlockStore { async fn get(&self, cid: &Cid) -> Result, RepoError> { let cid_bytes = cid_to_bytes(cid)?; let reader = Arc::clone(&self.reader); @@ -772,7 +786,7 @@ impl BlockStore for TranquilBlockStore } } -impl TranquilBlockStore { +impl TranquilBlockStore { pub async fn decrement_refs(&self, cids: &[Cid]) -> Result<(), RepoError> { if cids.is_empty() { return Ok(()); @@ -920,7 +934,7 @@ mod tests { let index = BlockIndex::open(&index_dir).unwrap(); - let result = TranquilBlockStore::::replay_single_file( + let result = TranquilBlockStore::::replay_single_file( &wrapper, &data_dir, &index, diff --git a/crates/tranquil-store/src/clock.rs b/crates/tranquil-store/src/clock.rs new file mode 100644 index 0000000..2c4e811 --- /dev/null +++ b/crates/tranquil-store/src/clock.rs @@ -0,0 +1,124 @@ +use std::sync::OnceLock; +use std::time::{Duration, Instant}; + +use crate::blockstore::WallClockMs; +use crate::eventlog::TimestampMicros; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] +#[repr(transparent)] +pub struct LogicalNanos(u64); + +impl LogicalNanos { + pub const fn new(nanos: u64) -> Self { + Self(nanos) + } + + pub fn raw(self) -> u64 { + self.0 + } + + pub fn from_millis(ms: u64) -> Self { + Self(ms.saturating_mul(1_000_000)) + } + + pub fn saturating_sub(self, other: Self) -> Self { + Self(self.0.saturating_sub(other.0)) + } +} + +pub trait Clock: Clone + Send + Sync + 'static { + fn unix_micros(&self) -> TimestampMicros; + fn wall_millis(&self) -> WallClockMs; + fn monotonic(&self) -> LogicalNanos; + fn advance(&self, by: Duration); +} + +#[derive(Debug, Clone, Copy, Default)] +pub struct SystemClock; + +static PROCESS_ORIGIN: OnceLock = OnceLock::new(); + +impl Clock for SystemClock { + fn unix_micros(&self) -> TimestampMicros { + TimestampMicros::now() + } + + fn wall_millis(&self) -> WallClockMs { + WallClockMs::now() + } + + fn monotonic(&self) -> LogicalNanos { + let origin = PROCESS_ORIGIN.get_or_init(Instant::now); + LogicalNanos::new(u64::try_from(origin.elapsed().as_nanos()).unwrap_or(u64::MAX)) + } + + fn advance(&self, _by: Duration) {} +} + +#[cfg(any(test, feature = "test-harness"))] +pub use sim_clock::SimClock; + +#[cfg(any(test, feature = "test-harness"))] +mod sim_clock { + use std::sync::Arc; + use std::sync::atomic::{AtomicU64, Ordering}; + use std::time::Duration; + + use super::{Clock, LogicalNanos}; + use crate::blockstore::WallClockMs; + use crate::eventlog::TimestampMicros; + use crate::sim::splitmix64; + + const EPOCH_BASE_MICROS: u64 = 1_700_000_000_000_000; + const ORIGIN_SPREAD_MICROS: u64 = 86_400 * 1_000_000; + + #[derive(Debug)] + struct SimClockState { + logical_nanos: AtomicU64, + unix_origin_micros: u64, + } + + #[derive(Debug, Clone)] + pub struct SimClock { + state: Arc, + } + + impl SimClock { + pub fn new(seed: u64) -> Self { + Self { + state: Arc::new(SimClockState { + logical_nanos: AtomicU64::new(0), + unix_origin_micros: EPOCH_BASE_MICROS + + (splitmix64(seed) % ORIGIN_SPREAD_MICROS), + }), + } + } + } + + impl Clock for SimClock { + fn unix_micros(&self) -> TimestampMicros { + let logical = self.state.logical_nanos.load(Ordering::Acquire); + TimestampMicros::new( + self.state + .unix_origin_micros + .saturating_add(logical / 1_000), + ) + } + + fn wall_millis(&self) -> WallClockMs { + let logical = self.state.logical_nanos.load(Ordering::Acquire); + WallClockMs::new( + (self.state.unix_origin_micros / 1_000).saturating_add(logical / 1_000_000), + ) + } + + fn monotonic(&self) -> LogicalNanos { + LogicalNanos::new(self.state.logical_nanos.load(Ordering::Acquire)) + } + + fn advance(&self, by: Duration) { + let nanos = u64::try_from(by.as_nanos()).unwrap_or(u64::MAX); + self.state.logical_nanos.fetch_add(nanos, Ordering::AcqRel); + } + } +} diff --git a/crates/tranquil-store/src/consistency.rs b/crates/tranquil-store/src/consistency.rs index 9abf945..97b18d8 100644 --- a/crates/tranquil-store/src/consistency.rs +++ b/crates/tranquil-store/src/consistency.rs @@ -5,8 +5,9 @@ use std::path::Path; use crate::blockstore::CID_SIZE; use crate::blockstore::hash_index::BlockIndex; use crate::blockstore::{DataFileId, TranquilBlockStore}; +use crate::clock::{Clock, SystemClock}; use crate::eventlog::{EventLog, EventSequence, SequenceContiguityResult}; -use crate::io::StorageIO; +use crate::io::{RealIO, StorageIO}; use crate::metastore::Metastore; use crate::metastore::encoding::KeyBuilder; use crate::metastore::event_keys::metastore_cursor_key; @@ -245,7 +246,7 @@ impl Default for ConsistencyCheckOptions { } pub fn verify_store_consistency( - blockstore: &TranquilBlockStore, + blockstore: &TranquilBlockStore, metastore: &Metastore, eventlog: &EventLog, ) -> ConsistencyReport { @@ -258,7 +259,7 @@ pub fn verify_store_consistency( } pub fn verify_store_consistency_with_options( - blockstore: &TranquilBlockStore, + blockstore: &TranquilBlockStore, metastore: &Metastore, eventlog: &EventLog, options: ConsistencyCheckOptions, @@ -567,7 +568,7 @@ fn check_cursor_vs_eventlog( } fn check_orphan_data_files( - blockstore: &TranquilBlockStore, + blockstore: &TranquilBlockStore, block_index: &BlockIndex, report: &mut ConsistencyReport, ) { @@ -580,7 +581,7 @@ fn check_orphan_data_files( }; let epoch = blockstore.epoch().current(); - let now = crate::wall_clock_ms(); + let now = blockstore.clock().wall_millis(); let indexed_files = block_index.liveness_by_file(epoch, now, 0); let indexed_file_ids: HashSet = indexed_files.keys().copied().collect(); @@ -600,7 +601,7 @@ fn check_orphan_data_files( } fn check_missing_indexed_files( - blockstore: &TranquilBlockStore, + blockstore: &TranquilBlockStore, block_index: &BlockIndex, report: &mut ConsistencyReport, ) { @@ -613,7 +614,7 @@ fn check_missing_indexed_files( }; let epoch = blockstore.epoch().current(); - let now = crate::wall_clock_ms(); + let now = blockstore.clock().wall_millis(); let indexed_files = block_index.liveness_by_file(epoch, now, 0); indexed_files @@ -622,7 +623,10 @@ fn check_missing_indexed_files( .for_each(|(fid, _)| report.missing_indexed_files.push(*fid)); } -fn check_orphan_hint_files(blockstore: &TranquilBlockStore, report: &mut ConsistencyReport) { +fn check_orphan_hint_files( + blockstore: &TranquilBlockStore, + report: &mut ConsistencyReport, +) { let data_files: HashSet = match blockstore.list_data_files() { Ok(files) => files.into_iter().collect(), Err(e) => { @@ -682,7 +686,7 @@ fn parse_user_hash_from_value(value_bytes: &[u8]) -> Option { } pub fn repair_known_issues( - blockstore: &TranquilBlockStore, + blockstore: &TranquilBlockStore, report: &ConsistencyReport, ) -> RepairResult { let mut result = RepairResult::default(); diff --git a/crates/tranquil-store/src/eventlog/writer.rs b/crates/tranquil-store/src/eventlog/writer.rs index 283dae6..cfb64ed 100644 --- a/crates/tranquil-store/src/eventlog/writer.rs +++ b/crates/tranquil-store/src/eventlog/writer.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use tracing::warn; +use crate::clock::{Clock, SystemClock}; use crate::io::{FileId, StorageIO}; use super::manager::SegmentManager; @@ -12,9 +13,7 @@ use super::segment_file::{ }; use super::segment_index::{DEFAULT_INDEX_INTERVAL, SegmentIndex, rebuild_from_segment}; use super::sidecar::build_sidecar_from_segment; -use super::types::{ - DidHash, EventSequence, EventTypeTag, SegmentId, SegmentOffset, TimestampMicros, -}; +use super::types::{DidHash, EventSequence, EventTypeTag, SegmentId, SegmentOffset}; const VALIDATE_RETRY_ATTEMPTS: u32 = 32; @@ -240,6 +239,16 @@ impl EventLogWriter { did_hash: DidHash, event_type: EventTypeTag, payload: Vec, + ) -> io::Result { + self.append_with_clock(did_hash, event_type, payload, &SystemClock) + } + + pub fn append_with_clock( + &mut self, + did_hash: DidHash, + event_type: EventTypeTag, + payload: Vec, + clock: &C, ) -> io::Result { let payload_len = u32::try_from(payload.len()) .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "payload exceeds u32::MAX"))?; @@ -252,7 +261,7 @@ impl EventLogWriter { } let seq = self.next_seq; - let timestamp = TimestampMicros::now(); + let timestamp = clock.unix_micros(); let event = ValidEvent { seq, diff --git a/crates/tranquil-store/src/gauntlet/chaos_walker.rs b/crates/tranquil-store/src/gauntlet/chaos_walker.rs index 6a9f890..dcf15a9 100644 --- a/crates/tranquil-store/src/gauntlet/chaos_walker.rs +++ b/crates/tranquil-store/src/gauntlet/chaos_walker.rs @@ -6,6 +6,7 @@ use jacquard_repo::mst::NodeData; use super::oracle::{hex_short, try_cid_to_fixed}; use crate::StorageIO; use crate::blockstore::{CidBytes, TranquilBlockStore}; +use crate::clock::Clock; pub enum LookupResult { Found(Cid), @@ -13,8 +14,8 @@ pub enum LookupResult { LostPath, } -pub fn walk_mst_node_cids_tolerant( - store: &TranquilBlockStore, +pub fn walk_mst_node_cids_tolerant( + store: &TranquilBlockStore, root: Cid, lost: &HashSet, ) -> Result, String> { @@ -44,8 +45,8 @@ pub fn walk_mst_node_cids_tolerant( Ok(result) } -pub fn mst_get_tolerant( - store: &TranquilBlockStore, +pub fn mst_get_tolerant( + store: &TranquilBlockStore, root: Cid, target: &str, lost: &HashSet, @@ -76,8 +77,8 @@ pub fn mst_get_tolerant( } } -fn read_node( - store: &TranquilBlockStore, +fn read_node( + store: &TranquilBlockStore, cid_bytes: &CidBytes, ) -> Result { let bytes = match store.get_block_sync(cid_bytes) { diff --git a/crates/tranquil-store/src/gauntlet/invariants.rs b/crates/tranquil-store/src/gauntlet/invariants.rs index a2bcc51..1feeed2 100644 --- a/crates/tranquil-store/src/gauntlet/invariants.rs +++ b/crates/tranquil-store/src/gauntlet/invariants.rs @@ -10,8 +10,9 @@ use super::oracle::{Oracle, hex_short, try_cid_to_fixed}; use crate::blockstore::{ BLOCK_HEADER_SIZE, CidBytes, CompactionError, TranquilBlockStore, hash_to_cid_bytes, }; +use crate::clock::Clock; use crate::eventlog::{EventSequence, SegmentId}; -use crate::io::{RealIO, StorageIO}; +use crate::io::StorageIO; #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct InvariantSet(u32); @@ -100,28 +101,28 @@ pub struct EventLogSnapshot { pub segment_last_ts: Vec<(SegmentId, u64)>, } -pub struct InvariantCtx<'a, S: StorageIO + Send + Sync + 'static = RealIO> { - pub store: &'a Arc>, +pub struct InvariantCtx<'a, S: StorageIO + Send + Sync + 'static, C: Clock> { + pub store: &'a Arc>, pub oracle: &'a Oracle, pub root: Option, pub eventlog: Option<&'a EventLogSnapshot>, } #[async_trait] -pub trait Invariant: Send + Sync { +pub trait Invariant: Send + Sync { fn name(&self) -> &'static str; - async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation>; + async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation>; } pub struct RefcountConservation; #[async_trait] -impl Invariant for RefcountConservation { +impl Invariant for RefcountConservation { fn name(&self) -> &'static str { "RefcountConservation" } - async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation> { let live: Vec<(String, CidBytes)> = ctx.oracle.live_cids_labeled(); let live_set: HashSet = live.iter().map(|(_, c)| *c).collect(); let index: HashMap = ctx @@ -162,12 +163,12 @@ impl Invariant for RefcountConservation pub struct Reachability; #[async_trait] -impl Invariant for Reachability { +impl Invariant for Reachability { fn name(&self) -> &'static str { "Reachability" } - async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation> { let violations: Vec = ctx .oracle .live_cids_labeled() @@ -193,12 +194,12 @@ impl Invariant for Reachability { pub struct AckedWritePersistence; #[async_trait] -impl Invariant for AckedWritePersistence { +impl Invariant for AckedWritePersistence { fn name(&self) -> &'static str { "AckedWritePersistence" } - async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation> { let Some(root) = ctx.root else { if ctx.oracle.live_count() == 0 { return Ok(()); @@ -241,12 +242,12 @@ impl Invariant for AckedWritePersistenc pub struct ReadAfterWrite; #[async_trait] -impl Invariant for ReadAfterWrite { +impl Invariant for ReadAfterWrite { fn name(&self) -> &'static str { "ReadAfterWrite" } - async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation> { let Some(root) = ctx.root else { return Ok(()); }; @@ -295,12 +296,12 @@ impl Invariant for ReadAfterWrite { pub struct CompactionIdempotent; #[async_trait] -impl Invariant for CompactionIdempotent { +impl Invariant for CompactionIdempotent { fn name(&self) -> &'static str { "CompactionIdempotent" } - async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation> { let store_a = ctx.store.clone(); let first = tokio::task::spawn_blocking(move || compact_by_liveness(&store_a)) .await @@ -348,8 +349,8 @@ impl Invariant for CompactionIdempotent } } -fn snapshot( - store: &Arc>, +fn snapshot( + store: &Arc>, ) -> Vec<(CidBytes, u32)> { let mut v: Vec<(CidBytes, u32)> = store .block_index() @@ -363,8 +364,8 @@ fn snapshot( const COMPACT_LIVENESS_CEILING: f64 = 0.99; -fn compact_by_liveness( - store: &TranquilBlockStore, +fn compact_by_liveness( + store: &TranquilBlockStore, ) -> Result<(), String> { let liveness = store .compaction_liveness(0) @@ -386,12 +387,12 @@ fn compact_by_liveness( pub struct HintBackedByData; #[async_trait] -impl Invariant for HintBackedByData { +impl Invariant for HintBackedByData { fn name(&self) -> &'static str { "HintBackedByData" } - async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation> { let store_c = ctx.store.clone(); let result = tokio::task::spawn_blocking(move || { let data: std::collections::HashSet<_> = store_c @@ -435,12 +436,12 @@ impl Invariant for HintBackedByData { pub struct IndexBlocksReadable; #[async_trait] -impl Invariant for IndexBlocksReadable { +impl Invariant for IndexBlocksReadable { fn name(&self) -> &'static str { "IndexBlocksReadable" } - async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation> { let store_c = ctx.store.clone(); let result = tokio::task::spawn_blocking(move || { let entries = store_c.block_index().live_entries_snapshot(); @@ -490,12 +491,12 @@ const INDEX_READABLE_REPORT_CAP: usize = 20; pub struct IndexBackedByDisk; #[async_trait] -impl Invariant for IndexBackedByDisk { +impl Invariant for IndexBackedByDisk { fn name(&self) -> &'static str { "IndexBackedByDisk" } - async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation> { let store_c = ctx.store.clone(); let result = tokio::task::spawn_blocking(move || { let disk: std::collections::HashSet<_> = store_c @@ -544,12 +545,12 @@ impl Invariant for IndexBackedByDisk { pub struct NoOrphanFiles; #[async_trait] -impl Invariant for NoOrphanFiles { +impl Invariant for NoOrphanFiles { fn name(&self) -> &'static str { "NoOrphanFiles" } - async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation> { let store_c = ctx.store.clone(); let result = tokio::task::spawn_blocking(move || { let disk = store_c.list_data_files().map_err(|e| e.to_string())?; @@ -606,12 +607,12 @@ impl Default for ByteBudget { } #[async_trait] -impl Invariant for ByteBudget { +impl Invariant for ByteBudget { fn name(&self) -> &'static str { "ByteBudget" } - async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation> { let store = ctx.store.clone(); let factor = self.overhead_factor; let floor = self.floor_bytes; @@ -643,12 +644,12 @@ impl Invariant for ByteBudget { pub struct ManifestEqualsReality; #[async_trait] -impl Invariant for ManifestEqualsReality { +impl Invariant for ManifestEqualsReality { fn name(&self) -> &'static str { "ManifestEqualsReality" } - async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation> { let store = ctx.store.clone(); tokio::task::spawn_blocking(move || { let listed = store.list_data_files().map_err(|e| e.to_string())?; @@ -715,12 +716,12 @@ impl Invariant for ManifestEqualsRealit pub struct ChecksumCoverage; #[async_trait] -impl Invariant for ChecksumCoverage { +impl Invariant for ChecksumCoverage { fn name(&self) -> &'static str { "ChecksumCoverage" } - async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation> { let violations: Vec = ctx .oracle .live_cids_labeled() @@ -761,12 +762,12 @@ impl Invariant for ChecksumCoverage { pub struct MonotonicSeq; #[async_trait] -impl Invariant for MonotonicSeq { +impl Invariant for MonotonicSeq { fn name(&self) -> &'static str { "MonotonicSeq" } - async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation> { let Some(el) = ctx.eventlog else { return Ok(()); }; @@ -819,12 +820,12 @@ impl Invariant for MonotonicSeq { pub struct FsyncOrdering; #[async_trait] -impl Invariant for FsyncOrdering { +impl Invariant for FsyncOrdering { fn name(&self) -> &'static str { "FsyncOrdering" } - async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation> { let Some(el) = ctx.eventlog else { return Ok(()); }; @@ -873,12 +874,12 @@ impl Invariant for FsyncOrdering { pub struct TombstoneBound; #[async_trait] -impl Invariant for TombstoneBound { +impl Invariant for TombstoneBound { fn name(&self) -> &'static str { "TombstoneBound" } - async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> { + async fn check(&self, ctx: &InvariantCtx<'_, S, C>) -> Result<(), InvariantViolation> { let Some(el) = ctx.eventlog else { return Ok(()); }; @@ -906,15 +907,15 @@ impl Invariant for TombstoneBound { } } -pub fn invariants_for( +pub fn invariants_for( set: InvariantSet, -) -> Vec>> { +) -> Vec>> { let unknown = set.unknown_bits(); assert!( unknown == 0, "invariants_for: unknown InvariantSet bits 0x{unknown:x}; all bits must map to an impl" ); - let candidates: Vec<(InvariantSet, Box>)> = vec![ + let candidates: Vec<(InvariantSet, Box>)> = vec![ ( InvariantSet::REFCOUNT_CONSERVATION, Box::new(RefcountConservation), diff --git a/crates/tranquil-store/src/gauntlet/metrics.rs b/crates/tranquil-store/src/gauntlet/metrics.rs index 4b58317..d776b73 100644 --- a/crates/tranquil-store/src/gauntlet/metrics.rs +++ b/crates/tranquil-store/src/gauntlet/metrics.rs @@ -7,6 +7,7 @@ use tracing::warn; use super::runner::{EventLogState, Harness}; use crate::blockstore::TranquilBlockStore; +use crate::clock::Clock; use crate::io::StorageIO; #[derive(Debug, Clone, Copy, Serialize, Deserialize)] @@ -94,8 +95,8 @@ impl MetricName { } } -pub fn sample_harness( - harness: &Harness, +pub fn sample_harness( + harness: &Harness, elapsed: Duration, ) -> MetricsSample { MetricsSample { @@ -116,8 +117,8 @@ pub fn sample_harness( } } -fn data_file_count( - store: &Arc>, +fn data_file_count( + store: &Arc>, ) -> Option { match store.list_data_files() { Ok(v) => Some(v.len() as u64), diff --git a/crates/tranquil-store/src/gauntlet/mod.rs b/crates/tranquil-store/src/gauntlet/mod.rs index c45c53b..83cc4ce 100644 --- a/crates/tranquil-store/src/gauntlet/mod.rs +++ b/crates/tranquil-store/src/gauntlet/mod.rs @@ -24,8 +24,8 @@ pub use invariants::{ pub use leak::{LeakGateBuildError, LeakGateConfig, LeakViolation, evaluate as evaluate_leak_gate}; pub use metrics::{MetricName, MetricsSample, sample_harness}; pub use op::{ - CollectionName, DidSeed, EventKind, FileChoice, Op, OpStream, PayloadSeed, RecordKey, - RetentionSecs, Seed, ValueSeed, + AdvanceNanos, CollectionName, DidSeed, EventKind, FileChoice, Op, OpStream, PayloadSeed, + RecordKey, RetentionSecs, Seed, ValueSeed, }; pub use oracle::{EventExpectation, Oracle}; pub use overrides::{ConfigOverrides, GroupCommitOverrides, StoreOverrides}; @@ -42,6 +42,6 @@ pub use soak::{ SoakEvent, SoakReport, run_soak, }; pub use workload::{ - ByteRange, DidSpaceSize, KeySpaceSize, OpCount, OpWeights, RetentionMaxSecs, SizeDistribution, - ValueBytes, WorkloadModel, + AdvanceMaxSecs, ByteRange, DidSpaceSize, KeySpaceSize, OpCount, OpWeights, RetentionMaxSecs, + SizeDistribution, ValueBytes, WorkloadModel, }; diff --git a/crates/tranquil-store/src/gauntlet/op.rs b/crates/tranquil-store/src/gauntlet/op.rs index 2a527c1..7fd3062 100644 --- a/crates/tranquil-store/src/gauntlet/op.rs +++ b/crates/tranquil-store/src/gauntlet/op.rs @@ -21,6 +21,9 @@ pub struct PayloadSeed(pub u32); #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] pub struct RetentionSecs(pub u32); +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct AdvanceNanos(pub u64); + #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] pub enum EventKind { Commit, @@ -54,6 +57,9 @@ pub enum Op { RunRetention { max_age_secs: RetentionSecs, }, + AdvanceTime { + by: AdvanceNanos, + }, ReadRecord { collection: CollectionName, rkey: RecordKey, diff --git a/crates/tranquil-store/src/gauntlet/runner.rs b/crates/tranquil-store/src/gauntlet/runner.rs index 00b0437..fe62c7b 100644 --- a/crates/tranquil-store/src/gauntlet/runner.rs +++ b/crates/tranquil-store/src/gauntlet/runner.rs @@ -18,9 +18,10 @@ use crate::blockstore::{ BlockStoreConfig, CidBytes, CompactionError, GroupCommitConfig, TranquilBlockStore, hash_to_cid, hash_to_cid_bytes, }; +use crate::clock::{Clock, SimClock, SystemClock}; use crate::eventlog::{ DEFAULT_INDEX_INTERVAL, DidHash, EventLogWriter, EventTypeTag, MAX_EVENT_PAYLOAD, SegmentId, - SegmentManager, SegmentReader, TimestampMicros, ValidEvent, + SegmentManager, SegmentReader, ValidEvent, }; use crate::io::{RealIO, StorageIO}; use crate::sim::{FaultConfig, PristineGuard, SimulatedIO}; @@ -161,8 +162,8 @@ pub struct EventLogState { pub max_segment_size: u64, } -pub struct Harness { - pub store: Arc>, +pub struct Harness { + pub store: Arc>, pub eventlog: Option>, } @@ -172,8 +173,8 @@ pub struct WriteState { pub eventlog: Option>, } -pub struct SharedState { - pub store: Arc>, +pub struct SharedState { + pub store: Arc>, pub write: tokio::sync::Mutex>, } @@ -381,7 +382,7 @@ async fn run_inner_real_on_root( let segments_dir = segments_subdir(&root); let open = { let segments_dir = segments_dir.clone(); - move |_attempt: usize| -> Result, String> { + move |_attempt: usize| -> Result, String> { let store = TranquilBlockStore::open(cfg.clone()) .map(Arc::new) .map_err(|e| e.to_string())?; @@ -396,7 +397,7 @@ async fn run_inner_real_on_root( } }; if config.writer_concurrency.0 > 1 { - run_inner_generic_concurrent::( + run_inner_generic_concurrent::( config, ops, ops_counter, @@ -406,10 +407,11 @@ async fn run_inner_real_on_root( || {}, tolerate_op_errors, reopen_backoff, + SystemClock, ) .await } else { - run_inner_generic::( + run_inner_generic::( config, ops, ops_counter, @@ -419,6 +421,7 @@ async fn run_inner_real_on_root( || {}, tolerate_op_errors, reopen_backoff, + SystemClock, ) .await } @@ -438,16 +441,22 @@ async fn run_inner_simulated( let eventlog_cfg = config.eventlog; let segments_dir = segments_subdir(dir.path()); let sim: Arc = Arc::new(SimulatedIO::new(config.seed.0, fault)); + let clock = sim.clock(); let sim_for_open = Arc::clone(&sim); + let clock_for_open = clock.clone(); let open = { let segments_dir = segments_dir.clone(); - move |attempt: usize| -> Result>, String> { + move |attempt: usize| -> Result, SimClock>, String> { let _pristine = PristineGuard::new(Arc::clone(&sim_for_open), attempt > 0); let factory_sim = Arc::clone(&sim_for_open); let make_io = move || Arc::clone(&factory_sim); - let store = TranquilBlockStore::>::open_with_io(cfg.clone(), make_io) - .map(Arc::new) - .map_err(|e| e.to_string())?; + let store = TranquilBlockStore::, SimClock>::open_with_io( + cfg.clone(), + make_io, + clock_for_open.clone(), + ) + .map(Arc::new) + .map_err(|e| e.to_string())?; let eventlog = match eventlog_cfg { None => None, Some(elc) => Some( @@ -465,7 +474,7 @@ async fn run_inner_simulated( let sim_for_crash = Arc::clone(&sim); let crash = move || sim_for_crash.crash(); if config.writer_concurrency.0 > 1 { - run_inner_generic_concurrent::, _, _>( + run_inner_generic_concurrent::, SimClock, _, _>( config, ops, ops_counter, @@ -475,10 +484,11 @@ async fn run_inner_simulated( crash, tolerate_errors, Duration::ZERO, + clock, ) .await } else { - run_inner_generic::, _, _>( + run_inner_generic::, SimClock, _, _>( config, ops, ops_counter, @@ -488,6 +498,7 @@ async fn run_inner_simulated( crash, tolerate_errors, Duration::ZERO, + clock, ) .await } @@ -517,7 +528,7 @@ pub(super) fn open_eventlog( } #[allow(clippy::too_many_arguments)] -async fn run_inner_generic( +async fn run_inner_generic( config: GauntletConfig, op_stream: OpStream, ops_counter: Arc, @@ -527,16 +538,18 @@ async fn run_inner_generic( mut crash: Crash, tolerate_op_errors: bool, reopen_backoff: Duration, + clock: C, ) -> GauntletReport where S: StorageIO + Send + Sync + 'static, - Open: FnMut(usize) -> Result, String>, + C: Clock, + Open: FnMut(usize) -> Result, String>, Crash: FnMut(), { let mut oracle = Oracle::new(); let mut violations: Vec = Vec::new(); - let mut harness: Option> = + let mut harness: Option> = match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors, reopen_backoff).await { Ok(h) => Some(h), @@ -569,7 +582,7 @@ where .as_mut() .expect("harness invariant: present when halt_ops is false"); let root_before = root; - match apply_op(live, &mut root, &mut oracle, op, &config.workload).await { + match apply_op(live, &mut root, &mut oracle, op, &config.workload, &clock).await { Ok(()) => {} Err(e) => { if tolerate_op_errors { @@ -750,7 +763,9 @@ pub(super) fn eventlog_snapshot( }) } -fn shutdown_harness(harness: &mut Option>) { +fn shutdown_harness( + harness: &mut Option>, +) { if let Some(h) = harness.as_mut() && let Some(el) = h.eventlog.as_mut() { @@ -762,15 +777,16 @@ fn shutdown_harness(harness: &mut Option( +async fn reopen_with_recovery( open: &mut Open, crash: &mut Crash, tolerate: bool, backoff: Duration, -) -> Result, String> +) -> Result, String> where S: StorageIO + Send + Sync + 'static, - Open: FnMut(usize) -> Result, String>, + C: Clock, + Open: FnMut(usize) -> Result, String>, Crash: FnMut(), { let mut errors: Vec = Vec::new(); @@ -808,8 +824,8 @@ fn sample_distinct(rng: &mut Lcg, n: usize, k: usize) -> Vec { .collect() } -async fn run_quick_check( - store: &Arc>, +async fn run_quick_check( + store: &Arc>, oracle: &Oracle, root: Option, rng: &mut Lcg, @@ -890,21 +906,21 @@ async fn run_quick_check( } } -pub(super) async fn run_invariants( - store: &Arc>, +pub(super) async fn run_invariants( + store: &Arc>, oracle: &Oracle, root: Option, eventlog: Option, set: InvariantSet, ) -> Vec { - let ctx = InvariantCtx:: { + let ctx = InvariantCtx:: { store, oracle, root, eventlog: eventlog.as_ref(), }; let mut out = Vec::new(); - for inv in invariants_for::(set) { + for inv in invariants_for::(set) { if let Err(v) = inv.check(&ctx).await { out.push(v); } @@ -912,8 +928,8 @@ pub(super) async fn run_invariants( out } -fn snapshot_block_index( - store: &TranquilBlockStore, +fn snapshot_block_index( + store: &TranquilBlockStore, ) -> Vec<(CidBytes, u32)> { let mut v: Vec<(CidBytes, u32)> = store .block_index() @@ -972,8 +988,8 @@ fn diff_snapshots(pre: &[(CidBytes, u32)], post: &[(CidBytes, u32)]) -> Option( - store: &Arc>, +pub(super) async fn refresh_oracle_graph( + store: &Arc>, oracle: &mut Oracle, root: Option, ) -> Result<(), String> { @@ -1095,12 +1111,13 @@ fn did_hash_for_seed(seed: DidSeed) -> DidHash { DidHash::from_did(&format!("did:plc:gauntlet{:08x}", seed.0)) } -pub(super) async fn apply_op( - harness: &mut Harness, +pub(super) async fn apply_op( + harness: &mut Harness, root: &mut Option, oracle: &mut Oracle, op: &Op, workload: &WorkloadModel, + clock: &C, ) -> Result<(), OpError> { match op { Op::AddRecord { @@ -1165,8 +1182,8 @@ pub(super) async fn apply_op( let did_hash = did_hash_for_seed(*did_seed); let tag = event_kind_to_tag(*event_kind); let payload = event_payload_bytes(*payload_seed); - let ts_before = TimestampMicros::now().raw(); - match el.writer.append(did_hash, tag, payload) { + let ts_before = clock.unix_micros().raw(); + match el.writer.append_with_clock(did_hash, tag, payload, clock) { Ok(seq) => { oracle.record_event_append(EventExpectation { seq, @@ -1198,7 +1215,11 @@ pub(super) async fn apply_op( let Some(el) = harness.eventlog.as_mut() else { return Ok(()); }; - run_retention(el, oracle, *max_age_secs).map_err(OpError::EventLogRetention) + run_retention(el, oracle, *max_age_secs, clock).map_err(OpError::EventLogRetention) + } + Op::AdvanceTime { by } => { + clock.advance(Duration::from_nanos(by.0)); + Ok(()) } Op::ReadRecord { collection, rkey } => { let Some(r) = *root else { return Ok(()) }; @@ -1228,8 +1249,8 @@ pub(super) async fn apply_op( } } -fn externally_delete_data_file( - store: &std::sync::Arc>, +fn externally_delete_data_file( + store: &std::sync::Arc>, pick: u32, ) -> Result, OpError> { let active = store.block_index().read_write_cursor().map(|c| c.file_id); @@ -1250,26 +1271,26 @@ fn externally_delete_data_file( } } -fn run_retention( +fn run_retention( el: &mut EventLogState, oracle: &mut Oracle, max_age: RetentionSecs, + clock: &C, ) -> Result<(), String> { let sync_result = el.writer.sync().map_err(|e| e.to_string())?; el.manager .io() .sync_dir(el.segments_dir.as_path()) .map_err(|e| e.to_string())?; - let _ = el.writer.rotate_if_needed(); oracle.record_event_sync(sync_result.synced_through); - let active_id = sync_result.segment_id; - let now_us = TimestampMicros::now().raw(); + let now_us = clock.unix_micros().raw(); let max_age_us = u64::from(max_age.0).saturating_mul(1_000_000); let cutoff_us = now_us.saturating_sub(max_age_us); let segments = el.manager.list_segments().map_err(|e| e.to_string())?; + let active_id = segments.last().copied(); segments .iter() - .take_while(|&&id| id != active_id) + .filter(|&&id| Some(id) != active_id) .try_for_each(|&id| -> Result<(), String> { let last_ts = segment_last_timestamp(&el.manager, id).map_err(|e| e.to_string())?; match last_ts { @@ -1293,8 +1314,8 @@ fn segment_last_timestamp( Ok(events.last().map(|e: &ValidEvent| e.timestamp.raw())) } -async fn add_record_atomic( - store: &Arc>, +async fn add_record_atomic( + store: &Arc>, root: Option, collection: &super::op::CollectionName, rkey: &super::op::RecordKey, @@ -1330,8 +1351,8 @@ async fn add_record_atomic( Ok((new_root, true)) } -async fn delete_record_atomic( - store: &Arc>, +async fn delete_record_atomic( + store: &Arc>, old_root: Cid, collection: &super::op::CollectionName, rkey: &super::op::RecordKey, @@ -1387,8 +1408,8 @@ fn diff_obsolete( .map_err(OpError::from) } -async fn commit_atomic( - store: &Arc>, +async fn commit_atomic( + store: &Arc>, blocks: Vec<(CidBytes, Vec)>, obsolete: Vec, ) -> Result<(), OpError> { @@ -1404,8 +1425,8 @@ async fn commit_atomic( const COMPACT_LIVENESS_CEILING: f64 = 0.99; -fn compact_by_liveness( - store: &TranquilBlockStore, +fn compact_by_liveness( + store: &TranquilBlockStore, ) -> Result<(), OpError> { let liveness = store .compaction_liveness(0) @@ -1425,10 +1446,11 @@ fn compact_by_liveness( }) } -async fn apply_op_concurrent( - shared: &Arc>, +async fn apply_op_concurrent( + shared: &Arc>, op: &Op, workload: &WorkloadModel, + clock: &C, ) -> Result<(), OpError> { match op { Op::AddRecord { @@ -1497,12 +1519,12 @@ async fn apply_op_concurrent( let did_hash = did_hash_for_seed(*did_seed); let tag = event_kind_to_tag(*event_kind); let payload = event_payload_bytes(*payload_seed); - let ts_before = TimestampMicros::now().raw(); + let ts_before = clock.unix_micros().raw(); let mut state = shared.write.lock().await; let Some(el) = state.eventlog.as_mut() else { return Ok(()); }; - match el.writer.append(did_hash, tag, payload) { + match el.writer.append_with_clock(did_hash, tag, payload, clock) { Ok(seq) => { let _ = el.writer.rotate_if_needed(); state.oracle.record_event_append(EventExpectation { @@ -1540,7 +1562,11 @@ async fn apply_op_concurrent( let Some(el) = eventlog.as_mut() else { return Ok(()); }; - run_retention(el, oracle, *max_age_secs).map_err(OpError::EventLogRetention) + run_retention(el, oracle, *max_age_secs, clock).map_err(OpError::EventLogRetention) + } + Op::AdvanceTime { by } => { + clock.advance(Duration::from_nanos(by.0)); + Ok(()) } Op::ReadRecord { collection, rkey } => { let r = { shared.write.lock().await.root }; @@ -1575,8 +1601,8 @@ async fn apply_op_concurrent( } #[allow(clippy::too_many_arguments)] -async fn writer_task( - shared: Arc>, +async fn writer_task( + shared: Arc>, ops: Arc>, index: Arc, end: usize, @@ -1584,6 +1610,7 @@ async fn writer_task( ops_counter: Arc, op_errors_counter: Arc, tolerate_op_errors: bool, + clock: C, ) -> Option { loop { let idx = index.fetch_add(1, Ordering::Relaxed); @@ -1591,7 +1618,7 @@ async fn writer_task( return None; } let op = &ops[idx]; - match apply_op_concurrent(&shared, op, &workload).await { + match apply_op_concurrent(&shared, op, &workload, &clock).await { Ok(()) => { ops_counter.fetch_max(idx + 1, Ordering::Relaxed); } @@ -1633,7 +1660,7 @@ fn compute_chunks( } #[allow(clippy::too_many_arguments)] -async fn run_inner_generic_concurrent( +async fn run_inner_generic_concurrent( config: GauntletConfig, op_stream: OpStream, ops_counter: Arc, @@ -1643,10 +1670,12 @@ async fn run_inner_generic_concurrent( mut crash: Crash, tolerate_op_errors: bool, reopen_backoff: Duration, + clock: C, ) -> GauntletReport where S: StorageIO + Send + Sync + 'static, - Open: FnMut(usize) -> Result, String>, + C: Clock, + Open: FnMut(usize) -> Result, String>, Crash: FnMut(), { let ops: Vec = op_stream.into_vec(); @@ -1659,7 +1688,7 @@ where let mut sample_rng = Lcg::new(Seed(config.seed.0 ^ 0x5A5A_5A5A_5A5A_5A5A)); let chunks = compute_chunks(config.restart_policy, total_ops, &mut restart_rng); - let mut harness: Option> = + let mut harness: Option> = match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors, reopen_backoff).await { Ok(h) => Some(h), @@ -1711,6 +1740,7 @@ where Arc::clone(&ops_counter), Arc::clone(&op_errors_counter), tolerate_op_errors, + clock.clone(), ))); } for h in handles.drain(..) { @@ -1916,20 +1946,26 @@ mod tests { attempts: Arc, sim: Arc, store_cfg: BlockStoreConfig, - ) -> impl FnMut(usize) -> Result>, String> + Send + 'static { - move |_attempt: usize| -> Result>, String> { + clock: SimClock, + ) -> impl FnMut(usize) -> Result, SimClock>, String> + Send + 'static + { + move |_attempt: usize| -> Result, SimClock>, String> { let n = attempts.fetch_add(1, Ordering::Relaxed); if n == 0 { return Err("simulated EIO on initial open".to_string()); } let factory_sim = Arc::clone(&sim); let make_io = move || Arc::clone(&factory_sim); - TranquilBlockStore::>::open_with_io(store_cfg.clone(), make_io) - .map(|s| Harness { - store: Arc::new(s), - eventlog: None, - }) - .map_err(|e| e.to_string()) + TranquilBlockStore::, SimClock>::open_with_io( + store_cfg.clone(), + make_io, + clock.clone(), + ) + .map(|s| Harness { + store: Arc::new(s), + eventlog: None, + }) + .map_err(|e| e.to_string()) } } @@ -1939,18 +1975,25 @@ mod tests { let cfg = minimal_config(); let store_cfg = blockstore_config(dir.path(), &cfg.store); let sim: Arc = Arc::new(SimulatedIO::pristine(0)); + let clock = sim.clock(); let attempts = Arc::new(AtomicUsize::new(0)); - let report = run_inner_generic::, _, _>( + let report = run_inner_generic::, SimClock, _, _>( cfg, OpStream::empty(), Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)), - flaky_open(Arc::clone(&attempts), Arc::clone(&sim), store_cfg), + flaky_open( + Arc::clone(&attempts), + Arc::clone(&sim), + store_cfg, + clock.clone(), + ), || {}, true, Duration::ZERO, + clock, ) .await; @@ -1977,18 +2020,25 @@ mod tests { cfg.writer_concurrency = WriterConcurrency(2); let store_cfg = blockstore_config(dir.path(), &cfg.store); let sim: Arc = Arc::new(SimulatedIO::pristine(0)); + let clock = sim.clock(); let attempts = Arc::new(AtomicUsize::new(0)); - let report = run_inner_generic_concurrent::, _, _>( + let report = run_inner_generic_concurrent::, SimClock, _, _>( cfg, OpStream::empty(), Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)), Arc::new(AtomicUsize::new(0)), - flaky_open(Arc::clone(&attempts), Arc::clone(&sim), store_cfg), + flaky_open( + Arc::clone(&attempts), + Arc::clone(&sim), + store_cfg, + clock.clone(), + ), || {}, true, Duration::ZERO, + clock, ) .await; @@ -2007,4 +2057,129 @@ mod tests { "expected at least one retry after first failure, attempts={total}" ); } + + #[tokio::test] + async fn dst_retention_advances_logical_hours_in_real_milliseconds() { + use crate::gauntlet::op::AdvanceNanos; + use futures::stream::StreamExt; + use std::time::Instant; + + const NANOS_PER_HOUR: u64 = 3600 * 1_000_000_000; + + let dir = tempfile::TempDir::new().expect("tempdir"); + let store_cfg = blockstore_config(dir.path(), &minimal_config().store); + let sim: Arc = Arc::new(SimulatedIO::pristine(0xDADA)); + let clock = sim.clock(); + + let store = { + let s = Arc::clone(&sim); + TranquilBlockStore::, SimClock>::open_with_io( + store_cfg, + move || Arc::clone(&s), + clock.clone(), + ) + .map(Arc::new) + .expect("open store") + }; + let eventlog = open_eventlog(Arc::clone(&sim), segments_subdir(dir.path()), 200) + .expect("open eventlog"); + let harness = Harness { + store, + eventlog: Some(eventlog), + }; + + let workload = WorkloadModel::default(); + let appends: Vec = (0..80u32) + .map(|i| Op::AppendEvent { + did_seed: DidSeed(i), + event_kind: EventKind::Commit, + payload_seed: PayloadSeed(i), + }) + .collect(); + + let start = Instant::now(); + let (mut harness, mut root, mut oracle) = futures::stream::iter(appends) + .fold( + (harness, None::, Oracle::new()), + |(mut h, mut r, mut o), op| { + let clk = clock.clone(); + let wl = workload.clone(); + async move { + apply_op(&mut h, &mut r, &mut o, &op, &wl, &clk) + .await + .expect("append op"); + (h, r, o) + } + }, + ) + .await; + + apply_op( + &mut harness, + &mut root, + &mut oracle, + &Op::SyncEventLog, + &workload, + &clock, + ) + .await + .expect("sync"); + let segments_before = harness + .eventlog + .as_ref() + .unwrap() + .manager + .list_segments() + .expect("list before"); + + apply_op( + &mut harness, + &mut root, + &mut oracle, + &Op::AdvanceTime { + by: AdvanceNanos(2 * NANOS_PER_HOUR), + }, + &workload, + &clock, + ) + .await + .expect("advance"); + apply_op( + &mut harness, + &mut root, + &mut oracle, + &Op::RunRetention { + max_age_secs: RetentionSecs(3600), + }, + &workload, + &clock, + ) + .await + .expect("retention"); + let elapsed = start.elapsed(); + + let segments_after = harness + .eventlog + .as_ref() + .unwrap() + .manager + .list_segments() + .expect("list after"); + + assert!( + segments_before.len() > 1, + "expected appends to seal old segments, got {} segment(s)", + segments_before.len() + ); + assert!( + segments_after.len() < segments_before.len(), + "retention after advancing 2 logical hours must delete sealed segments older than the 1h cutoff: before={} after={}", + segments_before.len(), + segments_after.len() + ); + assert!( + elapsed < Duration::from_secs(10), + "advancing 2 logical hours took {elapsed:?} of wall time" + ); + } } diff --git a/crates/tranquil-store/src/gauntlet/scenarios.rs b/crates/tranquil-store/src/gauntlet/scenarios.rs index de094af..e6157d3 100644 --- a/crates/tranquil-store/src/gauntlet/scenarios.rs +++ b/crates/tranquil-store/src/gauntlet/scenarios.rs @@ -6,8 +6,8 @@ use super::runner::{ RestartPolicy, RunLimits, ShardCount, StoreConfig, WallMs, WriterConcurrency, }; use super::workload::{ - ByteRange, DidSpaceSize, KeySpaceSize, OpCount, OpWeights, RetentionMaxSecs, SizeDistribution, - ValueBytes, WorkloadModel, + AdvanceMaxSecs, ByteRange, DidSpaceSize, KeySpaceSize, OpCount, OpWeights, RetentionMaxSecs, + SizeDistribution, ValueBytes, WorkloadModel, }; use crate::blockstore::{GroupCommitConfig, MAX_BLOCK_SIZE}; use crate::sim::FaultConfig; @@ -32,6 +32,7 @@ pub enum Scenario { ContendedWriters, FlakyDevice, ExternalCorruption, + RetentionTimeTravel, } impl Scenario { @@ -55,6 +56,7 @@ impl Scenario { Self::ContendedWriters => "ContendedWriters", Self::FlakyDevice => "FlakyDevice", Self::ExternalCorruption => "ExternalCorruption", + Self::RetentionTimeTravel => "RetentionTimeTravel", } } @@ -78,6 +80,7 @@ impl Scenario { Self::ContendedWriters => "contended-writers", Self::FlakyDevice => "flaky-device", Self::ExternalCorruption => "external-corruption", + Self::RetentionTimeTravel => "retention-time-travel", } } @@ -115,6 +118,9 @@ impl Scenario { Self::ExternalCorruption => { "Rare external data-file deletion mid-workload. Validates phantom-purge self-heal under chaos." } + Self::RetentionTimeTravel => { + "Eventlog retention under logical time travel: random multi-day AdvanceTime jumps interleaved with append/sync/retention. TOMBSTONE_BOUND across fake weeks." + } } } @@ -145,6 +151,7 @@ impl Scenario { Self::ContendedWriters, Self::FlakyDevice, Self::ExternalCorruption, + Self::RetentionTimeTravel, ]; } @@ -219,6 +226,7 @@ pub fn config_for(scenario: Scenario, seed: Seed) -> GauntletConfig { Scenario::ContendedWriters => contended_writers(seed), Scenario::FlakyDevice => flaky_device(seed), Scenario::ExternalCorruption => external_corruption(seed), + Scenario::RetentionTimeTravel => retention_time_travel(seed), } } @@ -251,6 +259,7 @@ fn block_workload( key_space, did_space: DidSpaceSize(32), retention_max_secs: RetentionMaxSecs(3600), + advance_max_secs: AdvanceMaxSecs(7200), } } @@ -666,6 +675,7 @@ fn firehose_fanout(seed: Seed) -> GauntletConfig { key_space: KeySpaceSize(500), did_space: DidSpaceSize(64), retention_max_secs: RetentionMaxSecs(60), + advance_max_secs: AdvanceMaxSecs(7200), }, op_count: OpCount(20_000), invariants: sim_invariants() @@ -706,6 +716,7 @@ fn contended_readers(seed: Seed) -> GauntletConfig { key_space: KeySpaceSize(400), did_space: DidSpaceSize(32), retention_max_secs: RetentionMaxSecs(3600), + advance_max_secs: AdvanceMaxSecs(7200), }, op_count: OpCount(20_000), invariants: sim_invariants(), @@ -773,6 +784,7 @@ fn contended_writers(seed: Seed) -> GauntletConfig { key_space: KeySpaceSize(1_000), did_space: DidSpaceSize(32), retention_max_secs: RetentionMaxSecs(3600), + advance_max_secs: AdvanceMaxSecs(7200), }, op_count: OpCount(20_000), invariants: sim_invariants(), @@ -815,3 +827,42 @@ fn external_corruption(seed: Seed) -> GauntletConfig { tolerate_op_errors: true, } } + +fn retention_time_travel(seed: Seed) -> GauntletConfig { + GauntletConfig { + seed, + io: IoBackend::Simulated { + fault: FaultConfig::none(), + }, + workload: WorkloadModel { + weights: OpWeights { + add: 20, + compact: 3, + checkpoint: 2, + append_event: 30, + sync_event_log: 15, + run_retention: 10, + advance_time: 20, + ..OpWeights::default() + }, + size_distribution: SizeDistribution::Fixed(ValueBytes(96)), + collections: default_collections(), + key_space: KeySpaceSize(400), + did_space: DidSpaceSize(64), + retention_max_secs: RetentionMaxSecs(1_209_600), + advance_max_secs: AdvanceMaxSecs(1_209_600), + }, + op_count: OpCount(1_200), + invariants: sim_invariants() | InvariantSet::MONOTONIC_SEQ | InvariantSet::TOMBSTONE_BOUND, + limits: RunLimits { + max_wall_ms: Some(WallMs(5 * 60_000)), + }, + restart_policy: RestartPolicy::EveryNOps(OpInterval(400)), + store: sim_store(), + eventlog: Some(EventLogConfig { + max_segment_size: MaxSegmentSize(8 * 1024), + }), + writer_concurrency: WriterConcurrency(1), + tolerate_op_errors: false, + } +} diff --git a/crates/tranquil-store/src/gauntlet/shrink.rs b/crates/tranquil-store/src/gauntlet/shrink.rs index 15fbeb8..efbf20c 100644 --- a/crates/tranquil-store/src/gauntlet/shrink.rs +++ b/crates/tranquil-store/src/gauntlet/shrink.rs @@ -112,8 +112,8 @@ mod tests { RestartPolicy, RunLimits, ShardCount, StoreConfig, WriterConcurrency, }; use crate::gauntlet::workload::{ - DidSpaceSize, KeySpaceSize, OpCount, OpWeights, RetentionMaxSecs, SizeDistribution, - ValueBytes, WorkloadModel, + AdvanceMaxSecs, DidSpaceSize, KeySpaceSize, OpCount, OpWeights, RetentionMaxSecs, + SizeDistribution, ValueBytes, WorkloadModel, }; use crate::sim::FaultConfig; @@ -130,6 +130,7 @@ mod tests { key_space: KeySpaceSize(4), did_space: DidSpaceSize(1), retention_max_secs: RetentionMaxSecs(60), + advance_max_secs: AdvanceMaxSecs(7200), }, op_count: OpCount(4), invariants: InvariantSet::EMPTY, diff --git a/crates/tranquil-store/src/gauntlet/soak.rs b/crates/tranquil-store/src/gauntlet/soak.rs index d91ae23..2a480d7 100644 --- a/crates/tranquil-store/src/gauntlet/soak.rs +++ b/crates/tranquil-store/src/gauntlet/soak.rs @@ -18,6 +18,7 @@ use super::runner::{ }; use super::workload::OpCount; use crate::blockstore::TranquilBlockStore; +use crate::clock::Clock; use crate::io::{RealIO, StorageIO}; const OP_ERROR_LOG_THROTTLE: u64 = 1024; @@ -137,7 +138,7 @@ pub async fn run_soak( outcome } -fn shutdown_harness(harness: &mut Harness) { +fn shutdown_harness(harness: &mut Harness) { if let Some(el) = harness.eventlog.as_mut() { if let Err(e) = el.writer.shutdown() { warn!(error = %e, "soak: event log writer shutdown failed"); @@ -146,15 +147,17 @@ fn shutdown_harness(harness: &mut Harness< } } -async fn drive_soak( - harness: &mut Harness, +async fn drive_soak( + harness: &mut Harness, cfg: &SoakConfig, emitter: &mut W, ) -> Result where S: StorageIO + Send + Sync + 'static, + C: Clock, W: Write + Send, { + let clock = harness.store.clock().clone(); let mut oracle = Oracle::new(); let mut root: Option = None; @@ -194,7 +197,16 @@ where if start.elapsed() >= cfg.total_duration { break; } - match apply_op(harness, &mut root, &mut oracle, op, &cfg.gauntlet.workload).await { + match apply_op( + harness, + &mut root, + &mut oracle, + op, + &cfg.gauntlet.workload, + &clock, + ) + .await + { Ok(()) => { ops_executed = ops_executed.saturating_add(1); } diff --git a/crates/tranquil-store/src/gauntlet/workload.rs b/crates/tranquil-store/src/gauntlet/workload.rs index 8563a2e..262fc32 100644 --- a/crates/tranquil-store/src/gauntlet/workload.rs +++ b/crates/tranquil-store/src/gauntlet/workload.rs @@ -1,8 +1,10 @@ use super::op::{ - CollectionName, DidSeed, EventKind, FileChoice, Op, OpStream, PayloadSeed, RecordKey, - RetentionSecs, Seed, ValueSeed, + AdvanceNanos, CollectionName, DidSeed, EventKind, FileChoice, Op, OpStream, PayloadSeed, + RecordKey, RetentionSecs, Seed, ValueSeed, }; +const NANOS_PER_SEC: u64 = 1_000_000_000; + #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub struct ValueBytes(pub u32); @@ -24,6 +26,7 @@ pub struct OpWeights { pub read_record: u32, pub read_block: u32, pub external_delete_data_file: u32, + pub advance_time: u32, } impl OpWeights { @@ -38,6 +41,7 @@ impl OpWeights { + self.read_record + self.read_block + self.external_delete_data_file + + self.advance_time } pub const fn touches_eventlog(&self) -> bool { @@ -82,6 +86,9 @@ pub struct DidSpaceSize(pub u32); #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] pub struct RetentionMaxSecs(pub u32); +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub struct AdvanceMaxSecs(pub u32); + #[derive(Debug, Clone)] pub struct WorkloadModel { pub weights: OpWeights, @@ -90,6 +97,7 @@ pub struct WorkloadModel { pub key_space: KeySpaceSize, pub did_space: DidSpaceSize, pub retention_max_secs: RetentionMaxSecs, + pub advance_max_secs: AdvanceMaxSecs, } impl Default for WorkloadModel { @@ -106,12 +114,14 @@ impl Default for WorkloadModel { read_record: 0, read_block: 0, external_delete_data_file: 0, + advance_time: 0, }, size_distribution: SizeDistribution::Fixed(ValueBytes(64)), collections: vec![CollectionName("app.bsky.feed.post".to_string())], key_space: KeySpaceSize(200), did_space: DidSpaceSize(32), retention_max_secs: RetentionMaxSecs(3600), + advance_max_secs: AdvanceMaxSecs(7200), } } } @@ -142,6 +152,7 @@ impl WorkloadModel { let t7 = t6 + w.run_retention; let t8 = t7 + w.read_record; let t9 = t8 + w.read_block; + let t10 = t9 + w.external_delete_data_file; match bucket { b if b < t1 => Op::AddRecord { @@ -173,9 +184,15 @@ impl WorkloadModel { b if b < t9 => Op::ReadBlock { value_seed: ValueSeed(rng.next_u32()), }, - _ => Op::ExternalDeleteDataFile { + b if b < t10 => Op::ExternalDeleteDataFile { choice: FileChoice(rng.next_u32()), }, + _ => Op::AdvanceTime { + by: AdvanceNanos( + u64::from(rng.next_u32() % self.advance_max_secs.0.max(1)) + * NANOS_PER_SEC, + ), + }, } }) .collect(); diff --git a/crates/tranquil-store/src/lib.rs b/crates/tranquil-store/src/lib.rs index 0610117..564ed48 100644 --- a/crates/tranquil-store/src/lib.rs +++ b/crates/tranquil-store/src/lib.rs @@ -2,6 +2,7 @@ pub mod archival; pub mod backup; pub mod blockstore; pub mod bloom; +pub mod clock; pub mod consistency; pub mod eventlog; pub mod fsync_order; @@ -16,6 +17,9 @@ mod record; mod sim; pub use blockstore::BlocksSynced; +#[cfg(any(test, feature = "test-harness"))] +pub use clock::SimClock; +pub use clock::{Clock, LogicalNanos, SystemClock}; pub use fsync_order::PostBlockstoreHook; #[cfg(any(test, feature = "test-harness"))] pub use harness::{ @@ -31,7 +35,3 @@ pub use sim::{ FaultConfig, LatencyNs, OpRecord, PristineGuard, Probability, SimulatedIO, SyncReorderWindow, sim_proptest_cases, sim_seed_count, sim_seed_range, sim_single_seed, }; - -pub(crate) fn wall_clock_ms() -> blockstore::WallClockMs { - blockstore::WallClockMs::now() -} diff --git a/crates/tranquil-store/src/metastore/commit_ops.rs b/crates/tranquil-store/src/metastore/commit_ops.rs index a095e5c..49b0502 100644 --- a/crates/tranquil-store/src/metastore/commit_ops.rs +++ b/crates/tranquil-store/src/metastore/commit_ops.rs @@ -20,8 +20,9 @@ use super::user_block_ops::UserBlockOps; use super::user_blocks::user_block_user_prefix; use super::user_hash::UserHashMap; use crate::blockstore::TranquilBlockStore; +use crate::clock::SystemClock; use crate::eventlog::EventLogBridge; -use crate::io::StorageIO; +use crate::io::{RealIO, StorageIO}; use tranquil_db_traits::{ ApplyCommitError, ApplyCommitInput, ApplyCommitResult, ImportBlock, ImportRecord, @@ -81,7 +82,7 @@ pub struct CommitOps { user_block_ops: UserBlockOps, backlink_ops: BacklinkOps, event_ops: EventOps, - blockstore: Option, + blockstore: Option>, } impl CommitOps { @@ -110,7 +111,7 @@ impl CommitOps { } } - pub fn with_blockstore(mut self, blockstore: TranquilBlockStore) -> Self { + pub fn with_blockstore(mut self, blockstore: TranquilBlockStore) -> Self { self.blockstore = Some(blockstore); self } @@ -474,7 +475,6 @@ fn parse_user_hash_from_key(key_bytes: &[u8]) -> Option { mod tests { use super::*; use crate::eventlog::{EventLog, EventLogConfig}; - use crate::io::RealIO; use crate::metastore::{Metastore, MetastoreConfig}; use tranquil_db_traits::{CommitEventData, RepoEventType}; use tranquil_types::{Handle, Nsid, Rkey}; diff --git a/crates/tranquil-store/src/metastore/handler.rs b/crates/tranquil-store/src/metastore/handler.rs index dfc2c30..6b373bd 100644 --- a/crates/tranquil-store/src/metastore/handler.rs +++ b/crates/tranquil-store/src/metastore/handler.rs @@ -42,8 +42,9 @@ use super::keys::UserHash; use super::record_ops::ListRecordsQuery; use super::user_hash::UserHashMap; use crate::blockstore::TranquilBlockStore; +use crate::clock::SystemClock; use crate::eventlog::EventLogBridge; -use crate::io::StorageIO; +use crate::io::{RealIO, StorageIO}; use crate::metastore::Metastore; type Tx = oneshot::Sender>; @@ -5956,7 +5957,7 @@ fn dispatch_user(state: &HandlerState, req: UserReque fn handler_loop( metastore: Metastore, bridge: Arc>, - blockstore: Option, + blockstore: Option>, rx: flume::Receiver, thread_index: usize, ) { @@ -6004,7 +6005,7 @@ impl HandlerPool { pub fn spawn( metastore: Metastore, bridge: Arc>, - blockstore: Option, + blockstore: Option>, thread_count: Option, ) -> Self { let count = thread_count @@ -6106,7 +6107,6 @@ impl Drop for HandlerPool { mod tests { use super::*; use crate::eventlog::{EventLog, EventLogConfig}; - use crate::io::RealIO; use crate::metastore::MetastoreConfig; use tranquil_types::{Did, Handle}; diff --git a/crates/tranquil-store/src/sim.rs b/crates/tranquil-store/src/sim.rs index f177822..8c894f1 100644 --- a/crates/tranquil-store/src/sim.rs +++ b/crates/tranquil-store/src/sim.rs @@ -6,11 +6,14 @@ use std::sync::Mutex; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::time::Duration; +use crate::clock::{Clock, SimClock}; use crate::io::{FileId, OpenOptions, StorageIO}; pub const TORN_PAGE_BYTES: usize = 4096; pub const SECTOR_BYTES: usize = 512; +const BASE_IO_SERVICE_NS: u64 = 1_000; + #[derive(Debug, Clone, Copy, PartialEq)] pub struct Probability(f64); @@ -331,6 +334,7 @@ pub struct SimulatedIO { pristine_mode: AtomicBool, rng_seed: u64, latency_counter: AtomicU64, + clock: SimClock, } impl SimulatedIO { @@ -352,9 +356,14 @@ impl SimulatedIO { pristine_mode: AtomicBool::new(false), rng_seed: seed, latency_counter: AtomicU64::new(0), + clock: SimClock::new(seed), } } + pub fn clock(&self) -> SimClock { + self.clock.clone() + } + fn effective_fault_config(&self) -> FaultConfig { if self.pristine_mode.load(Ordering::Relaxed) { FaultConfig::none() @@ -369,13 +378,15 @@ impl SimulatedIO { fn jitter(&self) { let max_ns = self.effective_fault_config().latency_distribution_ns.0; - if max_ns == 0 { - return; - } - let c = self.latency_counter.fetch_add(1, Ordering::Relaxed); - let r = splitmix64(self.rng_seed.wrapping_add(c)); - let ns = r % max_ns; - std::thread::sleep(Duration::from_nanos(ns)); + let extra_ns = match max_ns { + 0 => 0, + max => { + let c = self.latency_counter.fetch_add(1, Ordering::Relaxed); + splitmix64(self.rng_seed.wrapping_add(c)) % max + } + }; + self.clock + .advance(Duration::from_nanos(BASE_IO_SERVICE_NS + extra_ns)); } pub fn pristine(seed: u64) -> Self { @@ -907,7 +918,7 @@ pub fn sim_proptest_cases() -> u32 { u32::try_from(sim_seed_count()).unwrap_or(u32::MAX) } -fn splitmix64(mut x: u64) -> u64 { +pub(crate) fn splitmix64(mut x: u64) -> u64 { x = x.wrapping_add(0x9e3779b97f4a7c15); x = (x ^ (x >> 30)).wrapping_mul(0xbf58476d1ce4e5b9); x = (x ^ (x >> 27)).wrapping_mul(0x94d049bb133111eb); diff --git a/crates/tranquil-store/tests/backup.rs b/crates/tranquil-store/tests/backup.rs index 4438322..3d0727d 100644 --- a/crates/tranquil-store/tests/backup.rs +++ b/crates/tranquil-store/tests/backup.rs @@ -1,7 +1,6 @@ use std::path::Path; use std::sync::Arc; -use tranquil_store::RealIO; use tranquil_store::backup::{ BackupCoordinator, BackupKind, read_manifest, recover_to_sequence, restore_from_backup, restore_from_incremental, verify_backup, @@ -11,6 +10,7 @@ use tranquil_store::blockstore::{ }; use tranquil_store::eventlog::{EventLog, EventLogConfig}; use tranquil_store::metastore::{Metastore, MetastoreConfig}; +use tranquil_store::{RealIO, SystemClock}; fn test_cid(seed: u16) -> CidBytes { let mut cid = [0u8; 36]; @@ -32,7 +32,7 @@ fn block_data(seed: u16) -> Vec { struct TestStore { _dir: tempfile::TempDir, - blockstore: TranquilBlockStore, + blockstore: TranquilBlockStore, eventlog: Arc>, metastore: Metastore, } @@ -163,7 +163,10 @@ fn with_runtime(f: F) { f(); } -fn verify_blocks_readable(store: &TranquilBlockStore, range: std::ops::Range) { +fn verify_blocks_readable( + store: &TranquilBlockStore, + range: std::ops::Range, +) { range.for_each(|i| { let cid = test_cid(i); let data = store.get_block_sync(&cid).unwrap(); diff --git a/crates/tranquil-store/tests/checkpoint_race.rs b/crates/tranquil-store/tests/checkpoint_race.rs index f902d74..95d3d21 100644 --- a/crates/tranquil-store/tests/checkpoint_race.rs +++ b/crates/tranquil-store/tests/checkpoint_race.rs @@ -4,10 +4,10 @@ use std::io; use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use tranquil_store::PostBlockstoreHook; use tranquil_store::blockstore::{ BlockStoreConfig, BlocksSynced, CidBytes, GroupCommitConfig, TranquilBlockStore, }; +use tranquil_store::{PostBlockstoreHook, RealIO, SystemClock}; struct SlowHook; @@ -18,7 +18,7 @@ impl PostBlockstoreHook for SlowHook { } } -fn refcount(store: &TranquilBlockStore, cid: &CidBytes) -> Option { +fn refcount(store: &TranquilBlockStore, cid: &CidBytes) -> Option { store.block_index().get(cid).map(|e| e.refcount.raw()) } diff --git a/crates/tranquil-store/tests/common/mod.rs b/crates/tranquil-store/tests/common/mod.rs index 92443b3..35f8536 100644 --- a/crates/tranquil-store/tests/common/mod.rs +++ b/crates/tranquil-store/tests/common/mod.rs @@ -3,12 +3,12 @@ use std::collections::HashSet; use std::sync::Arc; -use tranquil_store::RealIO; use tranquil_store::blockstore::{ BlockStoreConfig, CidBytes, DEFAULT_MAX_FILE_SIZE, GroupCommitConfig, TranquilBlockStore, }; use tranquil_store::eventlog::{EventLog, EventLogConfig}; use tranquil_store::metastore::{Metastore, MetastoreConfig}; +use tranquil_store::{RealIO, SystemClock}; use tranquil_types::{CidLink, Did, Handle}; use uuid::Uuid; @@ -78,11 +78,11 @@ pub fn with_runtime(f: F) { f(); } -pub fn advance_epoch(store: &TranquilBlockStore) { +pub fn advance_epoch(store: &TranquilBlockStore) { store.apply_commit_blocking(vec![], vec![]).unwrap(); } -pub fn collect_all_dead(store: &TranquilBlockStore) -> HashSet { +pub fn collect_all_dead(store: &TranquilBlockStore) -> HashSet { let result = store.collect_dead_blocks(0).unwrap(); result .candidates @@ -91,7 +91,7 @@ pub fn collect_all_dead(store: &TranquilBlockStore) -> HashSet { .collect() } -pub fn compact_all_sealed(store: &TranquilBlockStore) { +pub fn compact_all_sealed(store: &TranquilBlockStore) { let Ok(files) = store.list_data_files() else { return; }; @@ -118,7 +118,7 @@ pub fn tiny_blockstore_config(dir: &std::path::Path) -> BlockStoreConfig { } } -pub fn compact_by_liveness(store: &TranquilBlockStore) { +pub fn compact_by_liveness(store: &TranquilBlockStore) { let liveness = store.compaction_liveness(0).unwrap(); liveness .iter() @@ -133,7 +133,7 @@ pub fn compact_by_liveness(store: &TranquilBlockStore) { }); } -pub fn compact_lowest_liveness(store: &TranquilBlockStore) { +pub fn compact_lowest_liveness(store: &TranquilBlockStore) { let liveness = store.compaction_liveness(0).unwrap(); let candidate = liveness .iter() @@ -154,7 +154,10 @@ pub fn compact_lowest_liveness(store: &TranquilBlockStore) { } } -pub fn collect_refcounts(store: &TranquilBlockStore, cids: &[CidBytes]) -> Vec<(u32, u32)> { +pub fn collect_refcounts( + store: &TranquilBlockStore, + cids: &[CidBytes], +) -> Vec<(u32, u32)> { cids.iter() .map(|cid| { let seed = u32::from_le_bytes([cid[4], cid[5], cid[6], cid[7]]); @@ -169,7 +172,7 @@ pub fn collect_refcounts(store: &TranquilBlockStore, cids: &[CidBytes]) -> Vec<( } pub struct TestStores { - pub blockstore: TranquilBlockStore, + pub blockstore: TranquilBlockStore, pub eventlog: Arc>, pub metastore: Metastore, } diff --git a/crates/tranquil-store/tests/compaction_liveness.rs b/crates/tranquil-store/tests/compaction_liveness.rs index 289d80c..b43797b 100644 --- a/crates/tranquil-store/tests/compaction_liveness.rs +++ b/crates/tranquil-store/tests/compaction_liveness.rs @@ -5,6 +5,7 @@ use std::collections::HashSet; use tranquil_store::blockstore::{ BlockStoreConfig, CidBytes, GroupCommitConfig, TranquilBlockStore, }; +use tranquil_store::{RealIO, SystemClock}; fn tiny_store_config(dir: &std::path::Path) -> BlockStoreConfig { BlockStoreConfig { @@ -31,7 +32,11 @@ fn make_block(seed: u32, size: usize) -> (CidBytes, Vec) { ) } -fn verify_live_blocks(store: &TranquilBlockStore, live: &HashSet, context: &str) { +fn verify_live_blocks( + store: &TranquilBlockStore, + live: &HashSet, + context: &str, +) { let missing: Vec = live .iter() .copied() @@ -50,7 +55,7 @@ fn verify_live_blocks(store: &TranquilBlockStore, live: &HashSet, context: ); } -fn compact_sealed(store: &TranquilBlockStore) { +fn compact_sealed(store: &TranquilBlockStore) { let files = store.list_data_files().unwrap(); files .iter() diff --git a/crates/tranquil-store/tests/compaction_restart.rs b/crates/tranquil-store/tests/compaction_restart.rs index c160793..cf18d9a 100644 --- a/crates/tranquil-store/tests/compaction_restart.rs +++ b/crates/tranquil-store/tests/compaction_restart.rs @@ -7,15 +7,15 @@ use common::{ collect_refcounts, compact_by_liveness, compact_lowest_liveness, test_cid, tiny_blockstore_config, with_runtime, }; -use tranquil_store::RealIO; use tranquil_store::blockstore::{CidBytes, TranquilBlockStore}; use tranquil_store::eventlog::{EventLog, EventLogBridge, EventLogConfig}; use tranquil_store::metastore::handler::HandlerPool; use tranquil_store::metastore::partitions::Partition; use tranquil_store::metastore::{Metastore, MetastoreConfig}; +use tranquil_store::{RealIO, SystemClock}; struct FullStack { - blockstore: TranquilBlockStore, + blockstore: TranquilBlockStore, _pool: Arc, _event_log: Arc>, } @@ -109,7 +109,7 @@ fn close_full_stack(stack: FullStack, base_dir: &Path) { } fn verify_blocks_and_refcounts( - store: &TranquilBlockStore, + store: &TranquilBlockStore, live_cids: &[CidBytes], expected_refcounts: Option<&[(u32, u32)]>, label: &str, diff --git a/crates/tranquil-store/tests/external_corruption_recovery.rs b/crates/tranquil-store/tests/external_corruption_recovery.rs index a773dd0..ce66ebe 100644 --- a/crates/tranquil-store/tests/external_corruption_recovery.rs +++ b/crates/tranquil-store/tests/external_corruption_recovery.rs @@ -6,12 +6,16 @@ use common::{block_data, test_cid, tiny_blockstore_config, with_runtime}; use tranquil_store::blockstore::{ CompactionResult, DataFileId, TranquilBlockStore, hint_file_path, }; +use tranquil_store::{RealIO, SystemClock}; fn data_file_path(dir: &std::path::Path, file_id: DataFileId) -> std::path::PathBuf { dir.join(format!("{file_id}.tqb")) } -fn populate_with_compaction_history(store: &TranquilBlockStore, live_cids: &[u32]) { +fn populate_with_compaction_history( + store: &TranquilBlockStore, + live_cids: &[u32], +) { live_cids.iter().for_each(|&seed| { store .put_blocks_blocking(vec![(test_cid(seed), block_data(seed))]) diff --git a/crates/tranquil-store/tests/gauntlet_index_backed.rs b/crates/tranquil-store/tests/gauntlet_index_backed.rs index 00d9747..2234dd4 100644 --- a/crates/tranquil-store/tests/gauntlet_index_backed.rs +++ b/crates/tranquil-store/tests/gauntlet_index_backed.rs @@ -5,8 +5,8 @@ use std::sync::Arc; use common::with_runtime; use tranquil_store::blockstore::{BlockStoreConfig, GroupCommitConfig, TranquilBlockStore}; use tranquil_store::gauntlet::{ - Gauntlet, IndexBackedByDisk, Invariant, InvariantCtx, InvariantSet, Oracle, Scenario, Seed, - config_for, + Gauntlet, IndexBackedByDisk, Invariant, InvariantCtx, InvariantSet, Op, OpStream, Oracle, + RetentionSecs, Scenario, Seed, config_for, }; #[test] @@ -101,6 +101,43 @@ async fn external_corruption_scenario_survives_many_seeds() { assert!(failures.is_empty(), "{}", failures.join("\n---\n")); } +#[tokio::test] +async fn retention_time_travel_survives_many_seeds() { + use futures::stream::StreamExt; + + let failures: Vec = futures::stream::iter(tranquil_store::sim_seed_range().map(Seed)) + .map(|seed| async move { + let g = Gauntlet::new(config_for(Scenario::RetentionTimeTravel, seed)) + .expect("build gauntlet"); + let mut ops = g.generate_ops().into_vec(); + ops.push(Op::SyncEventLog); + ops.push(Op::RunRetention { + max_age_secs: RetentionSecs(604_800), + }); + let report = g.run_with_ops(OpStream::from_vec(ops)).await; + (seed, report) + }) + .buffer_unordered(16) + .collect::>() + .await + .into_iter() + .filter(|(_, r)| !r.is_clean()) + .map(|(seed, r)| { + format!( + "seed {}: {} violations\n {}", + seed.0, + r.violations.len(), + r.violations + .iter() + .map(|v| format!("{}: {}", v.invariant, v.detail)) + .collect::>() + .join("\n ") + ) + }) + .collect(); + assert!(failures.is_empty(), "{}", failures.join("\n---\n")); +} + #[tokio::test] #[ignore = "long running, validates iris-class regression over many seeds"] async fn iris_class_regression_30_seeds() { diff --git a/crates/tranquil-store/tests/gauntlet_smoke.rs b/crates/tranquil-store/tests/gauntlet_smoke.rs index db7ff68..b0a7b9d 100644 --- a/crates/tranquil-store/tests/gauntlet_smoke.rs +++ b/crates/tranquil-store/tests/gauntlet_smoke.rs @@ -1,10 +1,10 @@ use tranquil_store::FaultConfig; use tranquil_store::blockstore::GroupCommitConfig; use tranquil_store::gauntlet::{ - CollectionName, ConfigOverrides, DidSpaceSize, Gauntlet, GauntletConfig, GauntletReport, - InvariantSet, IoBackend, KeySpaceSize, MaxFileSize, OpCount, OpInterval, OpWeights, - RegressionRecord, RestartPolicy, RetentionMaxSecs, RunLimits, Scenario, Seed, ShardCount, - SizeDistribution, StoreConfig, StoreOverrides, ValueBytes, WallMs, WorkloadModel, + AdvanceMaxSecs, CollectionName, ConfigOverrides, DidSpaceSize, Gauntlet, GauntletConfig, + GauntletReport, InvariantSet, IoBackend, KeySpaceSize, MaxFileSize, OpCount, OpInterval, + OpWeights, RegressionRecord, RestartPolicy, RetentionMaxSecs, RunLimits, Scenario, Seed, + ShardCount, SizeDistribution, StoreConfig, StoreOverrides, ValueBytes, WallMs, WorkloadModel, WriterConcurrency, config_for, farm, }; @@ -60,6 +60,7 @@ fn fast_sanity_config(seed: Seed) -> GauntletConfig { key_space: KeySpaceSize(100), did_space: DidSpaceSize(32), retention_max_secs: RetentionMaxSecs(3600), + advance_max_secs: AdvanceMaxSecs(7200), }, op_count: OpCount(200), invariants: InvariantSet::REFCOUNT_CONSERVATION @@ -126,6 +127,7 @@ async fn compaction_idempotent_sanity() { key_space: KeySpaceSize(50), did_space: DidSpaceSize(32), retention_max_secs: RetentionMaxSecs(3600), + advance_max_secs: AdvanceMaxSecs(7200), }, op_count: OpCount(300), invariants: InvariantSet::REFCOUNT_CONSERVATION @@ -165,6 +167,7 @@ async fn no_orphan_files_sanity() { key_space: KeySpaceSize(80), did_space: DidSpaceSize(32), retention_max_secs: RetentionMaxSecs(3600), + advance_max_secs: AdvanceMaxSecs(7200), }, op_count: OpCount(200), invariants: InvariantSet::REFCOUNT_CONSERVATION @@ -209,6 +212,7 @@ async fn simulated_pristine_roundtrip() { key_space: KeySpaceSize(80), did_space: DidSpaceSize(32), retention_max_secs: RetentionMaxSecs(3600), + advance_max_secs: AdvanceMaxSecs(7200), }, op_count: OpCount(300), invariants: InvariantSet::REFCOUNT_CONSERVATION @@ -260,6 +264,7 @@ async fn firehose_fanout_pristine_smoke() { key_space: KeySpaceSize(100), did_space: DidSpaceSize(32), retention_max_secs: RetentionMaxSecs(60), + advance_max_secs: AdvanceMaxSecs(7200), }, op_count: OpCount(2_000), invariants: InvariantSet::REFCOUNT_CONSERVATION @@ -312,6 +317,7 @@ async fn contended_readers_pristine_smoke() { key_space: KeySpaceSize(200), did_space: DidSpaceSize(32), retention_max_secs: RetentionMaxSecs(3600), + advance_max_secs: AdvanceMaxSecs(7200), }, op_count: OpCount(1_000), invariants: InvariantSet::REFCOUNT_CONSERVATION @@ -360,6 +366,7 @@ async fn contended_writers_pristine_smoke() { key_space: KeySpaceSize(500), did_space: DidSpaceSize(32), retention_max_secs: RetentionMaxSecs(3600), + advance_max_secs: AdvanceMaxSecs(7200), }, op_count: OpCount(1_000), invariants: InvariantSet::REFCOUNT_CONSERVATION @@ -491,6 +498,7 @@ async fn torn_pages_only_completes_within_budget() { key_space: KeySpaceSize(500), did_space: DidSpaceSize(32), retention_max_secs: RetentionMaxSecs(3600), + advance_max_secs: AdvanceMaxSecs(7200), }, op_count: OpCount(2_000), invariants: InvariantSet::REFCOUNT_CONSERVATION diff --git a/crates/tranquil-store/tests/gc_stress.rs b/crates/tranquil-store/tests/gc_stress.rs index 7825bbe..fcf7bb9 100644 --- a/crates/tranquil-store/tests/gc_stress.rs +++ b/crates/tranquil-store/tests/gc_stress.rs @@ -5,6 +5,7 @@ use tranquil_store::blockstore::{ BlockStoreConfig, CidBytes, DEFAULT_MAX_FILE_SIZE, DataFileId, GroupCommitConfig, TranquilBlockStore, }; +use tranquil_store::{RealIO, SystemClock}; fn test_cid_u32(seed: u32) -> [u8; 36] { let mut cid = [0u8; 36]; @@ -50,7 +51,7 @@ fn with_runtime(f: F) { f(); } -fn collect_all_dead(store: &TranquilBlockStore) -> HashSet { +fn collect_all_dead(store: &TranquilBlockStore) -> HashSet { let result = store.collect_dead_blocks(0).unwrap(); result .candidates @@ -59,7 +60,7 @@ fn collect_all_dead(store: &TranquilBlockStore) -> HashSet { .collect() } -fn compact_all_sealed(store: &TranquilBlockStore) { +fn compact_all_sealed(store: &TranquilBlockStore) { let files = store.list_data_files().unwrap(); let sealed: Vec = files .iter() @@ -72,7 +73,7 @@ fn compact_all_sealed(store: &TranquilBlockStore) { }); } -fn verify_live_readable(store: &TranquilBlockStore, oracle: &GcOracle) { +fn verify_live_readable(store: &TranquilBlockStore, oracle: &GcOracle) { oracle.live_seeds().iter().for_each(|&seed| { let cid = test_cid_u32(seed); let data = store @@ -92,7 +93,7 @@ fn verify_live_readable(store: &TranquilBlockStore, oracle: &GcOracle) { }); } -fn verify_no_live_in_dead(store: &TranquilBlockStore, oracle: &GcOracle) { +fn verify_no_live_in_dead(store: &TranquilBlockStore, oracle: &GcOracle) { let dead = collect_all_dead(store); oracle.live_seeds().iter().for_each(|&seed| { let cid = test_cid_u32(seed); @@ -150,7 +151,7 @@ impl GcOracle { } } -fn advance_epoch(store: &TranquilBlockStore) { +fn advance_epoch(store: &TranquilBlockStore) { store.apply_commit_blocking(vec![], vec![]).unwrap(); } diff --git a/crates/tranquil-store/tests/rotation_robustness.rs b/crates/tranquil-store/tests/rotation_robustness.rs index ed97fa7..65097de 100644 --- a/crates/tranquil-store/tests/rotation_robustness.rs +++ b/crates/tranquil-store/tests/rotation_robustness.rs @@ -10,7 +10,9 @@ use tranquil_store::blockstore::{ BlockStoreConfig, DataFileId, DataFileManager, DataFileWriter, GroupCommitConfig, TranquilBlockStore, }; -use tranquil_store::{FileId, MappedFile, OpenOptions, RealIO, SimulatedIO, StorageIO}; +use tranquil_store::{ + FileId, MappedFile, OpenOptions, RealIO, SimulatedIO, StorageIO, SystemClock, +}; use common::{test_cid, with_runtime}; @@ -149,9 +151,11 @@ fn post_rotation_sync_failure_deletes_new_rotation_files() { shard_count: 1, }; - let store = TranquilBlockStore::::open_with_io(config, move || { - FailingIO::new(Arc::clone(&spec_for_factory)) - }) + let store = TranquilBlockStore::::open_with_io( + config, + move || FailingIO::new(Arc::clone(&spec_for_factory)), + SystemClock, + ) .unwrap(); store diff --git a/crates/tranquil-store/tests/sim_blockstore.rs b/crates/tranquil-store/tests/sim_blockstore.rs index 239689a..efdcf36 100644 --- a/crates/tranquil-store/tests/sim_blockstore.rs +++ b/crates/tranquil-store/tests/sim_blockstore.rs @@ -13,7 +13,7 @@ use tranquil_store::blockstore::{ WallClockMs, WriteCursor, hint_file_path, }; use tranquil_store::{ - FaultConfig, OpenOptions, SimulatedIO, StorageIO, SyncReorderWindow, sim_seed_range, + FaultConfig, OpenOptions, SimClock, SimulatedIO, StorageIO, SyncReorderWindow, sim_seed_range, }; use common::{Rng, advance_epoch, block_data, test_cid, with_runtime}; @@ -717,11 +717,12 @@ fn sim_sync_reorder_loses_first_commit_durability() { { let s = Arc::clone(&sim); - let store = - TranquilBlockStore::>::open_with_io(config.clone(), move || { - Arc::clone(&s) - }) - .unwrap(); + let store = TranquilBlockStore::, SimClock>::open_with_io( + config.clone(), + move || Arc::clone(&s), + sim.clock(), + ) + .unwrap(); store .put_blocks_blocking(vec![(cid, data.clone())]) .unwrap(); @@ -730,9 +731,12 @@ fn sim_sync_reorder_loses_first_commit_durability() { sim.crash(); let s = Arc::clone(&sim); - let store = - TranquilBlockStore::>::open_with_io(config, move || Arc::clone(&s)) - .unwrap(); + let store = TranquilBlockStore::, SimClock>::open_with_io( + config, + move || Arc::clone(&s), + sim.clock(), + ) + .unwrap(); match store.get_block_sync(&cid) { Ok(Some(d)) => assert_eq!(&d[..], &data[..], "block content mismatch after crash"), diff --git a/frontend/package.json b/frontend/package.json index 7f5a42b..beaf91c 100644 --- a/frontend/package.json +++ b/frontend/package.json @@ -7,8 +7,11 @@ "dev": "vite", "build": "vite build", "preview": "vite preview", + "check": "svelte-check --tsconfig ./tsconfig.json", "test": "vitest", "test:run": "vitest run", + "test:watch": "vitest", + "test:ui": "vitest --ui", "test:coverage": "vitest run --coverage" }, "dependencies": { @@ -26,6 +29,8 @@ "@testing-library/jest-dom": "^6.9.1", "@testing-library/svelte": "^5.3.1", "@testing-library/user-event": "^14.6.1", + "@vitest/coverage-v8": "4.1.4", + "@vitest/ui": "4.1.4", "jsdom": "^25.0.1", "svelte": "^5.46.1", "svelte-check": "^4.3.5", diff --git a/frontend/pnpm-lock.yaml b/frontend/pnpm-lock.yaml index 40cc12e..8cd8bb6 100644 --- a/frontend/pnpm-lock.yaml +++ b/frontend/pnpm-lock.yaml @@ -41,10 +41,16 @@ importers: version: 6.9.1 '@testing-library/svelte': specifier: ^5.3.1 - version: 5.3.1(svelte@5.55.3)(vite@7.3.2)(vitest@4.1.4(jsdom@25.0.1)(vite@7.3.2)) + version: 5.3.1(svelte@5.55.3)(vite@7.3.2)(vitest@4.1.4) '@testing-library/user-event': specifier: ^14.6.1 version: 14.6.1(@testing-library/dom@10.4.1) + '@vitest/coverage-v8': + specifier: 4.1.4 + version: 4.1.4(vitest@4.1.4) + '@vitest/ui': + specifier: 4.1.4 + version: 4.1.4(vitest@4.1.4) jsdom: specifier: ^25.0.1 version: 25.0.1 @@ -62,7 +68,7 @@ importers: version: 7.3.2 vitest: specifier: ^4.0.16 - version: 4.1.4(jsdom@25.0.1)(vite@7.3.2) + version: 4.1.4(@vitest/coverage-v8@4.1.4)(@vitest/ui@4.1.4)(jsdom@25.0.1)(vite@7.3.2) packages: @@ -106,18 +112,39 @@ packages: resolution: {integrity: sha512-9NhCeYjq9+3uxgdtp20LSiJXJvN0FeCtNGpJxuMFZ1Kv3cWUNb6DOhJwUvcVCzKGR66cw4njwM6hrJLqgOwbcw==} engines: {node: '>=6.9.0'} + '@babel/helper-string-parser@7.29.7': + resolution: {integrity: sha512-Pb5ijPrZ89GDH8223L4UP8i6QApWxs04RbPQJTeWDV0/keR2E36MeKnyr6LYmUUvqRRI+Iv87SuF1W6ErINzYw==} + engines: {node: '>=6.9.0'} + '@babel/helper-validator-identifier@7.28.5': resolution: {integrity: sha512-qSs4ifwzKJSV39ucNjsvc6WVHs6b7S03sOh2OcHF9UHfVPqWWALUsNUVzhSBiItjRZoLHx7nIarVjqKVusUZ1Q==} engines: {node: '>=6.9.0'} + '@babel/helper-validator-identifier@7.29.7': + resolution: {integrity: sha512-qehxGkRj55h/ff8EMaJ+cYhyaKlHIxqYDn682wQD7RNp9UujOQsHog2uS0r2vzr4pW+sXf90NeeayjcNaX3fFg==} + engines: {node: '>=6.9.0'} + + '@babel/parser@7.29.7': + resolution: {integrity: sha512-hnORnjP/1P/zFEndoeX+n+t1RwWRJiJpM/jO7FW32Kn9r5+sJB2JWOdYo4L6k78j15eCwY3Gm/7364B1EMwtNg==} + engines: {node: '>=6.0.0'} + hasBin: true + '@babel/runtime@7.29.2': resolution: {integrity: sha512-JiDShH45zKHWyGe4ZNVRrCjBz8Nh9TMmZG1kh4QTK8hCBTWBi8Da+i7s1fJw7/lYpM4ccepSNfqzZ/QvABBi5g==} engines: {node: '>=6.9.0'} + '@babel/types@7.29.7': + resolution: {integrity: sha512-4zBIxpPzowiZpusoFkyGVwakdRJUyuH5PxQ/PrqghfdFWWasvnCdPfQXHrenDai+gyLARulZjZowCOj6fjT4pA==} + engines: {node: '>=6.9.0'} + '@badrap/valita@0.4.6': resolution: {integrity: sha512-4kdqcjyxo/8RQ8ayjms47HCWZIF5981oE5nIenbfThKDxWXtEHKipAOWlflpPJzZx9y/JWYQkp18Awr7VuepFg==} engines: {node: '>= 18'} + '@bcoe/v8-coverage@1.0.2': + resolution: {integrity: sha512-6zABk/ECA/QYSCQ1NGiVwwbQerUCZ+TQbp64Q3AgmfNvurHH0j8TtXa1qbShXA6qqkpAj4V5W8pP6mLe1mcMqA==} + engines: {node: '>=18'} + '@csstools/color-helpers@5.1.0': resolution: {integrity: sha512-S11EXWJyy0Mz5SYvRmY8nJYTFFd1LCNV+7cXyAgQtOOuzb4EsgfqDufL+9esx72/eLhsRdGZwaldu/h+E4t4BA==} engines: {node: '>=18'} @@ -474,6 +501,9 @@ packages: '@noble/secp256k1@3.1.0': resolution: {integrity: sha512-+F7iS7tUMaNGXcc9X3PjmjvuQnXEuSjCRNzVVA2xAcKXgCaP0dHYz4SFyt4FKNHef7sOP//xihowcySSS7PK9g==} + '@polka/url@1.0.0-next.29': + resolution: {integrity: sha512-wwQAWhWSuHaag8c4q/KN/vCoeOJYshAIvMQwD4GpSb3OiZklFfvAgmj0VCBBImRpuF/aFgIRzllXlVX93Jevww==} + '@rollup/rollup-android-arm-eabi@4.60.1': resolution: {integrity: sha512-d6FinEBLdIiK+1uACUttJKfgZREXrF0Qc2SmLII7W2AD8FfiZ9Wjd+rD/iRuf5s5dWrr1GgwXCvPqOuDquOowA==} cpu: [arm] @@ -683,6 +713,15 @@ packages: '@types/trusted-types@2.0.7': resolution: {integrity: sha512-ScaPdn1dQczgbl0QFTeTOmVHFULt394XJgOQNoyVhZ6r2vLnMLJfBPd53SB52T/3G36VI1/g2MZaX0cwDuXsfw==} + '@vitest/coverage-v8@4.1.4': + resolution: {integrity: sha512-x7FptB5oDruxNPDNY2+S8tCh0pcq7ymCe1gTHcsp733jYjrJl8V1gMUlVysuCD9Kz46Xz9t1akkv08dPcYDs1w==} + peerDependencies: + '@vitest/browser': 4.1.4 + vitest: 4.1.4 + peerDependenciesMeta: + '@vitest/browser': + optional: true + '@vitest/expect@4.1.4': resolution: {integrity: sha512-iPBpra+VDuXmBFI3FMKHSFXp3Gx5HfmSCE8X67Dn+bwephCnQCaB7qWK2ldHa+8ncN8hJU8VTMcxjPpyMkUjww==} @@ -709,6 +748,11 @@ packages: '@vitest/spy@4.1.4': resolution: {integrity: sha512-XxNdAsKW7C+FLydqFJLb5KhJtl3PGCMmYwFRfhvIgxJvLSXhhVI1zM8f1qD3Zg7RCjTSzDVyct6sghs9UEgBEQ==} + '@vitest/ui@4.1.4': + resolution: {integrity: sha512-EgFR7nlj5iTDYZYCvavjFokNYwr3c3ry0sFiCg+N7B233Nwp+NNx7eoF/XvMWDCKY71xXAG3kFkt97ZHBJVL8A==} + peerDependencies: + vitest: 4.1.4 + '@vitest/utils@4.1.4': resolution: {integrity: sha512-13QMT+eysM5uVGa1rG4kegGYNp6cnQcsTc67ELFbhNLQO+vgsygtYJx2khvdt4gVQqSSpC/KT5FZZxUpP3Oatw==} @@ -744,6 +788,9 @@ packages: resolution: {integrity: sha512-Izi8RQcffqCeNVgFigKli1ssklIbpHnCYc6AknXGYoB6grJqyeby7jv12JUQgmTAnIDnbck1uxksT4dzN3PWBA==} engines: {node: '>=12'} + ast-v8-to-istanbul@1.0.2: + resolution: {integrity: sha512-dKmJxJsGItLmc5CYZKuEjuG6GnBs6PG4gohMhyFOWKaNQoYCuRZJDECaBlHmcG0lv2wc2E0uU8lESmBEumC3DQ==} + asynckit@0.4.0: resolution: {integrity: sha512-Oei9OH4tRh0YqU3GxhX79dM/mwVgvbZJaSNaRk+bshkj0S5cfHcgYakreBjrHwatXKbz+IoIdYLxrKim2MjW0Q==} @@ -917,6 +964,12 @@ packages: picomatch: optional: true + fflate@0.8.3: + resolution: {integrity: sha512-tbZNuJrLwGUp3zshBtdy4W+ORxZuIh8a5ilyIEQDC5rY1f3U20JMry0Ll3WBzU58EZKsEuJFXhb5gwv8CsPvgA==} + + flatted@3.4.2: + resolution: {integrity: sha512-PjDse7RzhcPkIJwy5t7KPWQSZ9cAbzQXcafsetQoD7sOJRQlGikNbx7yZp2OotDnJyrDcbyRq3Ttb18iYOqkxA==} + form-data@4.0.5: resolution: {integrity: sha512-8RipRLol37bNs2bhoV67fiTEvdTrbMUYcFTiy3+wuuOnUog2QBHCZWXDRijWQfAkhBj2Uf5UnVaiWwA5vdd82w==} engines: {node: '>= 6'} @@ -947,6 +1000,10 @@ packages: resolution: {integrity: sha512-ZUKRh6/kUFoAiTAtTYPZJ3hw9wNxx+BIBOijnlG9PnrJsCcSjs1wyyD6vJpaYtgnzDrKYRSqf3OO6Rfa93xsRg==} engines: {node: '>= 0.4'} + has-flag@4.0.0: + resolution: {integrity: sha512-EykJT/Q1KjTWctppgIAgfSO0tKVuZUjhgMr17kqTumMl6Afv3EISleU7qZUzoXDFTAHTDC4NOoG/ZxU3EvlMPQ==} + engines: {node: '>=8'} + has-symbols@1.1.0: resolution: {integrity: sha512-1cDNdwJ2Jaohmb3sg4OmKaMBwuC48sYni5HUw2DvsC8LjGTLK9h+eb1X6RyuOHe4hT0ULCW68iomhjUoKUqlPQ==} engines: {node: '>= 0.4'} @@ -963,6 +1020,9 @@ packages: resolution: {integrity: sha512-Y22oTqIU4uuPgEemfz7NDJz6OeKf12Lsu+QC+s3BVpda64lTiMYCyGwg5ki4vFxkMwQdeZDl2adZoqUgdFuTgQ==} engines: {node: '>=18'} + html-escaper@2.0.2: + resolution: {integrity: sha512-H2iMtd0I4Mt5eYiapRdIDjp+XzelXQ0tFE4JS7YFwFevXXMmOp9myNrUvCg0D6ws8iqkRPBfKHgbwig1SmlLfg==} + http-proxy-agent@7.0.2: resolution: {integrity: sha512-T1gkAiYYDWYx3V5Bmyu7HcfcvL7mUrTWiM6yOfa3PIphViJ/gFPbvidQ+veqSOHci/PxBcDabeUNCzpOODJZig==} engines: {node: '>= 14'} @@ -991,6 +1051,21 @@ packages: is-reference@3.0.3: resolution: {integrity: sha512-ixkJoqQvAP88E6wLydLGGqCJsrFUnqoH6HnaczB8XmDH1oaWU+xxdptvikTgaEhtZ53Ky6YXiBuUI2WXLMCwjw==} + istanbul-lib-coverage@3.2.2: + resolution: {integrity: sha512-O8dpsF+r0WV/8MNRKfnmrtCWhuKjxrq2w+jpzBL5UZKTi2LeVWnWOmWRxFlesJONmc+wLAGvKQZEOanko0LFTg==} + engines: {node: '>=8'} + + istanbul-lib-report@3.0.1: + resolution: {integrity: sha512-GCfE1mtsHGOELCU8e/Z7YWzpmybrx/+dSTfLrvY8qRmaY6zXTKWn6WQIjaAFw069icm6GVMNkgu0NzI4iPZUNw==} + engines: {node: '>=10'} + + istanbul-reports@3.2.0: + resolution: {integrity: sha512-HGYWWS/ehqTV3xN10i23tkPkpH46MLCIMFNCaaKNavAXTF1RkqxawEPtnjnGZ6XKSInBKkiOA5BKS+aZiY3AvA==} + engines: {node: '>=8'} + + js-tokens@10.0.0: + resolution: {integrity: sha512-lM/UBzQmfJRo9ABXbPWemivdCW8V2G8FHaHdypQaIy523snUjog0W71ayWXTjiR+ixeMyVHN2XcpnTd/liPg/Q==} + js-tokens@4.0.0: resolution: {integrity: sha512-RdJUflcE3cUzKiMqQgsCu06FPu9UdIJO0beYbPhHN4k6apgJtifcoCtT9bcxOpYBtpD2kCM6Sbzg4CausW/PKQ==} @@ -1019,6 +1094,13 @@ packages: magic-string@0.30.21: resolution: {integrity: sha512-vd2F4YUyEXKGcLHoq+TEyCjxueSeHnFxyyjNp80yg0XV4vUhnDer/lvvlqM/arB5bXQN5K2/3oinyCRyx8T2CQ==} + magicast@0.5.3: + resolution: {integrity: sha512-pVKE4UdSQ7DvHzivsCIFx2BJn1mHG6KsyrFcaxFx6tONdneEuThrDx0Cj3AMg58KyN4pzYT+LHOotxDQDjNvkw==} + + make-dir@4.0.0: + resolution: {integrity: sha512-hXdUTZYIVOt1Ex//jAQi+wTZZpUpwBj/0QsOzqegb3rGMMeJiSEu5xLHnYfBrRV4RH2+OCSOO95Is/7x1WJ4bw==} + engines: {node: '>=10'} + math-intrinsics@1.1.0: resolution: {integrity: sha512-/IXtbwEk5HTPyEwyKX6hGkYXxM9nbj64B+ilVJnC/R6B0pH5G4V3b0pVbL7DBj4tkhBAppbQUlf6F6Xl9LHu1g==} engines: {node: '>= 0.4'} @@ -1043,6 +1125,10 @@ packages: resolution: {integrity: sha512-tzzskb3bG8LvYGFF/mDTpq3jpI6Q9wc3LEmBaghu+DdCssd1FakN7Bc0hVNmEyGq1bq3RgfkCb3cmQLpNPOroA==} engines: {node: '>=4'} + mrmime@2.0.1: + resolution: {integrity: sha512-Y3wQdFg2Va6etvQ5I82yUhGdsKrcYox6p7FfL1LbK2J4V01F9TGlepTIhnK24t7koZibmg82KGglhA1XK5IsLQ==} + engines: {node: '>=10'} + ms@2.1.3: resolution: {integrity: sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==} @@ -1121,9 +1207,18 @@ packages: resolution: {integrity: sha512-xAg7SOnEhrm5zI3puOOKyy1OMcMlIJZYNJY7xLBwSze0UjhPLnWfj2GF2EpT0jmzaJKIWKHLsaSSajf35bcYnA==} engines: {node: '>=v12.22.7'} + semver@7.8.1: + resolution: {integrity: sha512-rkVq3IXh+4FDGch+KwzX3aV9W3kO54GyEgpvBzSyctDA6Xtd7RJQV1xmXbeQp5v7+VzLOfVqiutSE6GICgPFvg==} + engines: {node: '>=10'} + hasBin: true + siginfo@2.0.0: resolution: {integrity: sha512-ybx0WO1/8bSBLEWXZvEd7gMW3Sn3JFlW3TvX1nREbDLRNQNaeNN8WK0meBwPdAaOI7TtRRRJn/Es1zhrrCHu7g==} + sirv@3.0.2: + resolution: {integrity: sha512-2wcC/oGxHis/BoHkkPwldgiPSYcpZK3JU28WoMVv55yHJgcZ8rlXvuG9iZggz+sU1d4bRgIGASwyWqjxu3FM0g==} + engines: {node: '>=18'} + source-map-js@1.2.1: resolution: {integrity: sha512-UXWMKhLOwVKb728IUtQPXxfYU+usdybtUrK/8uGE8CQMvrhOpwvzDBwj0QhSL7MQc7vIsISBG8VQ8+IDQxpfQA==} engines: {node: '>=0.10.0'} @@ -1138,6 +1233,10 @@ packages: resolution: {integrity: sha512-laJTa3Jb+VQpaC6DseHhF7dXVqHTfJPCRDaEbid/drOhgitgYku/letMUqOXFoWV0zIIUbjpdH2t+tYj4bQMRQ==} engines: {node: '>=8'} + supports-color@7.2.0: + resolution: {integrity: sha512-qpCAvRl9stuOHveKsn7HncJRvv501qIacKzQlO/+Lwxc9+0q2wLyv4Dfvt80/DPn2pqOBsJdDiogXGR9+OvwRw==} + engines: {node: '>=8'} + svelte-check@4.4.6: resolution: {integrity: sha512-kP1zG81EWaFe9ZyTv4ZXv44Csi6Pkdpb7S3oj6m+K2ec/IcDg/a8LsFsnVLqm2nxtkSwsd5xPj/qFkTBgXHXjg==} engines: {node: '>= 18.0.0'} @@ -1189,6 +1288,10 @@ packages: resolution: {integrity: sha512-WMi/OQ2axVTf/ykqCQgXiIct+mSQDFdH2fkwhPwgEwvJ1kSzZRiinb0zF2Xb8u4+OqPChmyI6MEu4EezNJz+FQ==} hasBin: true + totalist@3.0.1: + resolution: {integrity: sha512-sf4i37nQ2LBx4m3wB74y+ubopq6W/dIzXg0FDGjsYnZHVa1Da8FH853wlL2gtUhg+xJXjfk3kUZS3BRoQeoQBQ==} + engines: {node: '>=6'} + tough-cookie@5.1.2: resolution: {integrity: sha512-FVDYdxtnj0G6Qm/DhNPSb8Ju59ULcup3tuJxkFb5K8Bv2pUXILbf0xZWU8PX8Ov19OXljbUyveOFwRMwkXzO+A==} engines: {node: '>=16'} @@ -1311,6 +1414,7 @@ packages: whatwg-encoding@3.1.1: resolution: {integrity: sha512-6qN4hJdMwfYBtE3YBTTHhoeuUrDBPZmbQaxWAqSALV/MeEnR5z1xd8UKud2RAkFoPkmB+hli1TZSnyi84xz1vQ==} engines: {node: '>=18'} + deprecated: Use @exodus/bytes instead for a more spec-conformant and faster implementation whatwg-mimetype@4.0.0: resolution: {integrity: sha512-QaKxh0eNIi2mE9p2vEdzfagOKHCcj1pJ56EEHGQOVxp8r9/iszLUUV7v89x9O1p/T+NlTM5W7jW6+cz4Fq1YVg==} @@ -1423,12 +1527,27 @@ snapshots: js-tokens: 4.0.0 picocolors: 1.1.1 + '@babel/helper-string-parser@7.29.7': {} + '@babel/helper-validator-identifier@7.28.5': {} + '@babel/helper-validator-identifier@7.29.7': {} + + '@babel/parser@7.29.7': + dependencies: + '@babel/types': 7.29.7 + '@babel/runtime@7.29.2': {} + '@babel/types@7.29.7': + dependencies: + '@babel/helper-string-parser': 7.29.7 + '@babel/helper-validator-identifier': 7.29.7 + '@badrap/valita@0.4.6': {} + '@bcoe/v8-coverage@1.0.2': {} + '@csstools/color-helpers@5.1.0': {} '@csstools/css-calc@2.1.4(@csstools/css-parser-algorithms@3.0.5(@csstools/css-tokenizer@3.0.4))(@csstools/css-tokenizer@3.0.4)': @@ -1643,6 +1762,8 @@ snapshots: '@noble/secp256k1@3.1.0': {} + '@polka/url@1.0.0-next.29': {} + '@rollup/rollup-android-arm-eabi@4.60.1': optional: true @@ -1765,14 +1886,14 @@ snapshots: dependencies: svelte: 5.55.3 - '@testing-library/svelte@5.3.1(svelte@5.55.3)(vite@7.3.2)(vitest@4.1.4(jsdom@25.0.1)(vite@7.3.2))': + '@testing-library/svelte@5.3.1(svelte@5.55.3)(vite@7.3.2)(vitest@4.1.4)': dependencies: '@testing-library/dom': 10.4.1 '@testing-library/svelte-core': 1.0.0(svelte@5.55.3) svelte: 5.55.3 optionalDependencies: vite: 7.3.2 - vitest: 4.1.4(jsdom@25.0.1)(vite@7.3.2) + vitest: 4.1.4(@vitest/coverage-v8@4.1.4)(@vitest/ui@4.1.4)(jsdom@25.0.1)(vite@7.3.2) '@testing-library/user-event@14.6.1(@testing-library/dom@10.4.1)': dependencies: @@ -1791,6 +1912,20 @@ snapshots: '@types/trusted-types@2.0.7': {} + '@vitest/coverage-v8@4.1.4(vitest@4.1.4)': + dependencies: + '@bcoe/v8-coverage': 1.0.2 + '@vitest/utils': 4.1.4 + ast-v8-to-istanbul: 1.0.2 + istanbul-lib-coverage: 3.2.2 + istanbul-lib-report: 3.0.1 + istanbul-reports: 3.2.0 + magicast: 0.5.3 + obug: 2.1.1 + std-env: 4.0.0 + tinyrainbow: 3.1.0 + vitest: 4.1.4(@vitest/coverage-v8@4.1.4)(@vitest/ui@4.1.4)(jsdom@25.0.1)(vite@7.3.2) + '@vitest/expect@4.1.4': dependencies: '@standard-schema/spec': 1.1.0 @@ -1826,6 +1961,17 @@ snapshots: '@vitest/spy@4.1.4': {} + '@vitest/ui@4.1.4(vitest@4.1.4)': + dependencies: + '@vitest/utils': 4.1.4 + fflate: 0.8.3 + flatted: 3.4.2 + pathe: 2.0.3 + sirv: 3.0.2 + tinyglobby: 0.2.16 + tinyrainbow: 3.1.0 + vitest: 4.1.4(@vitest/coverage-v8@4.1.4)(@vitest/ui@4.1.4)(jsdom@25.0.1)(vite@7.3.2) + '@vitest/utils@4.1.4': dependencies: '@vitest/pretty-format': 4.1.4 @@ -1850,6 +1996,12 @@ snapshots: assertion-error@2.0.1: {} + ast-v8-to-istanbul@1.0.2: + dependencies: + '@jridgewell/trace-mapping': 0.3.31 + estree-walker: 3.0.3 + js-tokens: 10.0.0 + asynckit@0.4.0: {} axobject-query@4.1.0: {} @@ -2055,6 +2207,10 @@ snapshots: optionalDependencies: picomatch: 4.0.4 + fflate@0.8.3: {} + + flatted@3.4.2: {} + form-data@4.0.5: dependencies: asynckit: 0.4.0 @@ -2092,6 +2248,8 @@ snapshots: gopd@1.2.0: {} + has-flag@4.0.0: {} + has-symbols@1.1.0: {} has-tostringtag@1.0.2: @@ -2106,6 +2264,8 @@ snapshots: dependencies: whatwg-encoding: 3.1.1 + html-escaper@2.0.2: {} + http-proxy-agent@7.0.2: dependencies: agent-base: 7.1.4 @@ -2141,6 +2301,21 @@ snapshots: dependencies: '@types/estree': 1.0.8 + istanbul-lib-coverage@3.2.2: {} + + istanbul-lib-report@3.0.1: + dependencies: + istanbul-lib-coverage: 3.2.2 + make-dir: 4.0.0 + supports-color: 7.2.0 + + istanbul-reports@3.2.0: + dependencies: + html-escaper: 2.0.2 + istanbul-lib-report: 3.0.1 + + js-tokens@10.0.0: {} + js-tokens@4.0.0: {} jsdom@25.0.1: @@ -2185,6 +2360,16 @@ snapshots: dependencies: '@jridgewell/sourcemap-codec': 1.5.5 + magicast@0.5.3: + dependencies: + '@babel/parser': 7.29.7 + '@babel/types': 7.29.7 + source-map-js: 1.2.1 + + make-dir@4.0.0: + dependencies: + semver: 7.8.1 + math-intrinsics@1.1.0: {} memoizee@0.4.17: @@ -2208,6 +2393,8 @@ snapshots: mri@1.2.0: {} + mrmime@2.0.1: {} + ms@2.1.3: {} multiformats@13.4.2: {} @@ -2298,8 +2485,16 @@ snapshots: dependencies: xmlchars: 2.2.0 + semver@7.8.1: {} + siginfo@2.0.0: {} + sirv@3.0.2: + dependencies: + '@polka/url': 1.0.0-next.29 + mrmime: 2.0.1 + totalist: 3.0.1 + source-map-js@1.2.1: {} stackback@0.0.2: {} @@ -2310,6 +2505,10 @@ snapshots: dependencies: min-indent: 1.0.1 + supports-color@7.2.0: + dependencies: + has-flag: 4.0.0 + svelte-check@4.4.6(picomatch@4.0.4)(svelte@5.55.3)(typescript@5.9.3): dependencies: '@jridgewell/trace-mapping': 0.3.31 @@ -2383,6 +2582,8 @@ snapshots: dependencies: tldts-core: 6.1.86 + totalist@3.0.1: {} + tough-cookie@5.1.2: dependencies: tldts: 6.1.86 @@ -2414,7 +2615,7 @@ snapshots: optionalDependencies: vite: 7.3.2 - vitest@4.1.4(jsdom@25.0.1)(vite@7.3.2): + vitest@4.1.4(@vitest/coverage-v8@4.1.4)(@vitest/ui@4.1.4)(jsdom@25.0.1)(vite@7.3.2): dependencies: '@vitest/expect': 4.1.4 '@vitest/mocker': 4.1.4(vite@7.3.2) @@ -2437,6 +2638,8 @@ snapshots: vite: 7.3.2 why-is-node-running: 2.3.0 optionalDependencies: + '@vitest/coverage-v8': 4.1.4(vitest@4.1.4) + '@vitest/ui': 4.1.4(vitest@4.1.4) jsdom: 25.0.1 transitivePeerDependencies: - msw diff --git a/justfile b/justfile index 6a618fd..0cb9050 100644 --- a/justfile +++ b/justfile @@ -164,21 +164,21 @@ container-pull: podman pull atcr.io/tranquil.farm/tranquil-pds:latest frontend-dev: - . ~/.deno/env && cd frontend && deno task dev + cd frontend && pnpm run dev frontend-build: - . ~/.deno/env && cd frontend && deno task build + cd frontend && pnpm run build frontend-check: - . ~/.deno/env && cd frontend && deno task check + cd frontend && pnpm run check frontend-clean: rm -rf frontend/dist frontend/node_modules frontend-test *args: - . ~/.deno/env && cd frontend && VITEST=true deno task test:run {{args}} + cd frontend && VITEST=true pnpm run test:run {{args}} frontend-test-watch: - . ~/.deno/env && cd frontend && VITEST=true deno task test:watch + cd frontend && VITEST=true pnpm run test:watch frontend-test-ui: - . ~/.deno/env && cd frontend && VITEST=true deno task test:ui + cd frontend && VITEST=true pnpm run test:ui frontend-test-coverage: - . ~/.deno/env && cd frontend && VITEST=true deno task test:run --coverage + cd frontend && VITEST=true pnpm run test:run --coverage build-all: frontend-build build