fix: measure WAL component disk footprint from files

.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: |

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:
 

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})
     );

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<u64>,
+    /// Sum each file's larger apparent/allocated size, matching WAL accounting.
+    pub accounted_bytes: u64,
+}
+
+pub(super) fn measure(directory: &Path) -> Result<DiskUsage> {
+    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(())
+    }
+}