From 7a3637252756118900acc0752c6f21b67b89dc30 Mon Sep 17 00:00:00 2001 From: Beinan Date: Mon, 19 Oct 2026 21:52:41 +0000 Subject: [PATCH] fix: measure WAL component disk footprint from files --- .github/workflows/rust-test.yml | 1 + crates/lance-context-wal/README.md | 9 ++ crates/lance-context-wal/src/bin/lab.rs | 14 ++- .../src/bin/lab/disk_usage.rs | 90 +++++++++++++++++++ 4 files changed, 112 insertions(+), 2 deletions(-) create mode 100644 crates/lance-context-wal/src/bin/lab/disk_usage.rs diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml index 712a695..faab87e 100644 --- a/.github/workflows/rust-test.yml +++ b/.github/workflows/rust-test.yml @@ -62,6 +62,7 @@ jobs: - name: WAL component correctness smoke tests run: | cargo run -p lance-context-wal --features lab --bin context-wal-lab -- codec --payload-bytes 1024 --rows 8 --iterations 2 + cargo test -p lance-context-wal --features lab --bin context-wal-lab cargo run -p lance-context-wal --features lab --bin context-wal-lab -- disk --directory "$RUNNER_TEMP/wal-disk-smoke" --payload-bytes 1024 --batch 4 --writers 2 --groups-per-writer 2 --rounds 8 - name: WAL placement, admission, record, contract and IPC decoding tests run: | diff --git a/crates/lance-context-wal/README.md b/crates/lance-context-wal/README.md index 6f2ecfd..8166273 100644 --- a/crates/lance-context-wal/README.md +++ b/crates/lance-context-wal/README.md @@ -132,6 +132,15 @@ prepared before timing and reused across rounds. Compression is disabled unless inspection. This measures local durable writes, not replicated ACKs, crashes, power loss or Blob/Lance publication. +Disk footprint is observed after the writers finish and the engine closes, +outside write and reopen timing. `disk_apparent_bytes` sums regular file lengths; +`disk_allocated_bytes` sums their allocated blocks on Unix and is null elsewhere. +`disk_bytes` conservatively sums the larger apparent/allocated size of each file +(apparent size on other platforms). These are directory footprint observations, +not physical bytes written or write amplification. The pinned engine's segment +estimate is reported separately as `engine_estimated_disk_bytes`: large records +can overshoot the target segment size and make that estimate too small. + The pinned raft-engine 0.4.2 has a separate `sync-probe` command to inspect empty sync versus a nonempty synced barrier: diff --git a/crates/lance-context-wal/src/bin/lab.rs b/crates/lance-context-wal/src/bin/lab.rs index 33a1820..0c339d4 100644 --- a/crates/lance-context-wal/src/bin/lab.rs +++ b/crates/lance-context-wal/src/bin/lab.rs @@ -18,6 +18,9 @@ use raft_engine::{Config, Engine, LogBatch, MessageExt, ReadableSize}; use raft_proto::eraftpb::Entry; use serde_json::json; +#[path = "lab/disk_usage.rs"] +mod disk_usage; + struct CountingAllocator; static ALLOCS: AtomicU64 = AtomicU64::new(0); static ALLOC_BYTES: AtomicU64 = AtomicU64::new(0); @@ -314,8 +317,11 @@ fn disk( )?; } let elapsed = writes_started.elapsed(); - let disk_bytes = engine.get_used_size(); + let engine_estimated_disk_bytes = engine.get_used_size(); drop(engine); + // The pinned engine estimates segment count * target size; a large batch + // can overshoot that target. Observe actual files outside write/reopen timing. + let disk_usage = disk_usage::measure(&directory)?; let recovery_started = Instant::now(); let engine = Engine::open(config)?; let reopen = recovery_started.elapsed(); @@ -342,7 +348,11 @@ fn disk( "rounds":rounds, "compression_threshold":compression, "payload_mib": size as f64 * batch as f64 * rounds as f64 * writers as f64 / 1048576.0, "elapsed_seconds":elapsed.as_secs_f64(), "payload_mib_per_second":size as f64 * batch as f64 * rounds as f64 * writers as f64 / 1048576.0 / elapsed.as_secs_f64(), "batch_latency_us_p50":latency.value_at_quantile(0.5),"batch_latency_us_p99":latency.value_at_quantile(0.99), - "disk_bytes":disk_bytes,"reopen_seconds":reopen.as_secs_f64(),"verify_seconds":verify_started.elapsed().as_secs_f64(), + "disk_bytes":disk_usage.accounted_bytes, + "disk_apparent_bytes":disk_usage.apparent_bytes, + "disk_allocated_bytes":disk_usage.allocated_bytes, + "engine_estimated_disk_bytes":engine_estimated_disk_bytes, + "reopen_seconds":reopen.as_secs_f64(),"verify_seconds":verify_started.elapsed().as_secs_f64(), "verified_entries":writers * rounds * batch,"preparation_seconds": writes_started.duration_since(start).as_secs_f64(), "directory":directory}) ); diff --git a/crates/lance-context-wal/src/bin/lab/disk_usage.rs b/crates/lance-context-wal/src/bin/lab/disk_usage.rs new file mode 100644 index 0000000..f81b9fd --- /dev/null +++ b/crates/lance-context-wal/src/bin/lab/disk_usage.rs @@ -0,0 +1,90 @@ +//! Quiescent directory observation for the disk experiment, not a write counter. +use std::path::Path; + +use anyhow::{Context, Result}; + +pub(super) struct DiskUsage { + pub apparent_bytes: u64, + /// Filesystem block allocation is available on Unix only. + pub allocated_bytes: Option, + /// Sum each file's larger apparent/allocated size, matching WAL accounting. + pub accounted_bytes: u64, +} + +pub(super) fn measure(directory: &Path) -> Result { + let mut usage = DiskUsage { + apparent_bytes: 0, + allocated_bytes: cfg!(unix).then_some(0), + accounted_bytes: 0, + }; + for entry in std::fs::read_dir(directory)? { + let metadata = entry?.metadata()?; + if !metadata.is_file() { + continue; + } + let apparent = metadata.len(); + let accounted = apparent; + #[cfg(unix)] + let accounted = { + use std::os::unix::fs::MetadataExt; + let allocated = metadata + .blocks() + .checked_mul(512) + .context("allocated file size overflow")?; + usage.allocated_bytes = Some( + usage + .allocated_bytes + .unwrap() + .checked_add(allocated) + .context("allocated directory size overflow")?, + ); + accounted.max(allocated) + }; + usage.apparent_bytes = usage + .apparent_bytes + .checked_add(apparent) + .context("apparent directory size overflow")?; + usage.accounted_bytes = usage + .accounted_bytes + .checked_add(accounted) + .context("accounted directory size overflow")?; + } + Ok(usage) +} + +#[cfg(test)] +mod tests { + use raft_engine::{Config, Engine, LogBatch, ReadableSize}; + + #[test] + fn oversized_segments_are_measured_after_synced_writes() -> anyhow::Result<()> { + let directory = tempfile::tempdir()?; + let engine = Engine::open(Config { + dir: directory.path().to_string_lossy().into_owned(), + target_file_size: ReadableSize(64 * 1024), + batch_compression_threshold: ReadableSize(0), + ..Default::default() + })?; + let payload = vec![7; 120 * 1024]; + for key in 0u64..8 { + let mut batch = LogBatch::default(); + batch.put(1, key.to_le_bytes().to_vec(), payload.clone())?; + engine.write(&mut batch, true)?; + } + let engine_estimate = engine.get_used_size() as u64; + drop(engine); + + let usage = super::measure(directory.path())?; + assert!(usage.apparent_bytes >= 8 * payload.len() as u64); + assert!(usage.accounted_bytes >= usage.apparent_bytes); + assert!( + engine_estimate < usage.apparent_bytes, + "the pinned engine estimates overshot segments at their target size" + ); + assert_eq!(usage.allocated_bytes.is_some(), cfg!(unix)); + if let Some(allocated) = usage.allocated_bytes { + assert!(usage.accounted_bytes >= allocated); + } + Ok(()) + } +}