test: xorb reconstruction cache warm-up benchmark and unit test (#24)

## Summary

- Adds a unit test verifying `warm_reconstruction_cache` is triggered on
`open()` and that the watch stays true for subsequent reads on the same
handle (no re-waiting)
- Adds an EC2 integration benchmark (`warm_cache_bench`) measuring TTFB
and sequential throughput for 5 consecutive opens of a 100 MB file

## Background

The streaming path already used the xorb chunk disk cache correctly via
`FileReconstructor.with_chunk_cache()`. This PR adds validation and
numbers to prove it.

## Benchmark results (m6i.2xlarge, /dev/shm cache)

```
Open#  TTFB (ms)  Total (s)  MB/s     cache
-----  ---------  ---------  -------  -----
    1      200.7       0.23    432.0  cold
    2        0.8       0.02   5741.3  warm
    3        0.7       0.01   6789.3  warm
    4        0.6       0.01   7050.2  warm
    5        0.6       0.01   7040.3  warm

cold TTFB 201 ms / 432 MB/s  →  warm TTFB 1 ms / 6.7 GB/s
```

The bench uses `/dev/shm` (tmpfs) on Linux so warm reads hit RAM. On
EBS-backed `/tmp` (m6i default, ~125 MB/s baseline), warm reads would
appear slower than cold S3 reads, making the cache benefit invisible.

## Run

```bash
cargo test --release --test warm_cache_bench -- --nocapture
```

Author: XciD

src/test_mocks.rs

@@ -222,6 +222,10 @@ pub struct MockXet {
     range_fail_count: AtomicU32,
     /// Number of range download calls that should return empty before succeeding.
     range_empty_count: AtomicU32,
+    /// How many times warm_reconstruction_cache was called.
+    pub warm_call_count: AtomicU32,
+    /// Optional delay injected into warm_reconstruction_cache (simulates CAS round-trip).
+    warm_delay_ms: AtomicU64,
 }
 
 impl MockXet {
@@ -235,9 +239,15 @@ impl MockXet {
             writer_fail_after: AtomicU64::new(u64::MAX),
             range_fail_count: AtomicU32::new(0),
             range_empty_count: AtomicU32::new(0),
+            warm_call_count: AtomicU32::new(0),
+            warm_delay_ms: AtomicU64::new(0),
         })
     }
 
+    pub fn set_warm_delay_ms(&self, ms: u64) {
+        self.warm_delay_ms.store(ms, Ordering::SeqCst);
+    }
+
     pub fn add_file(&self, hash: &str, content: &[u8]) {
         self.files.lock().unwrap().insert(hash.to_string(), content.to_vec());
     }
@@ -312,7 +322,13 @@ impl XetOps for MockXet {
         Ok(results)
     }
 
-    async fn warm_reconstruction_cache(&self, _xet_hash: &str) {}
+    async fn warm_reconstruction_cache(&self, _xet_hash: &str) {
+        self.warm_call_count.fetch_add(1, Ordering::SeqCst);
+        let delay = self.warm_delay_ms.load(Ordering::SeqCst);
+        if delay > 0 {
+            tokio::time::sleep(Duration::from_millis(delay)).await;
+        }
+    }
 
     fn download_stream_boxed(&self, file_info: &XetFileInfo, offset: u64) -> Result<Box<dyn DownloadStreamOps>> {
         let prev_fail = self.range_fail_count.load(Ordering::SeqCst);

src/virtual_fs/tests.rs

@@ -2230,3 +2230,86 @@ fn shutdown_flushes_dirty() {
     let logs = hub.take_batch_log();
     assert!(!logs.is_empty());
 }
+
+// ── Reconstruction cache warm-up tests ──────────────────────────────
+
+/// `open()` must trigger exactly one `warm_reconstruction_cache` call per file handle,
+/// and that call must complete before the first `read()` is served.
+///
+/// We inject a 50 ms delay into the warm call to ensure the warm task is still
+/// running when `read()` is called.  If `read()` did NOT wait for the watch, the
+/// first read would race with the warm task and the assertion on `warm_call_count`
+/// at read time might pass trivially — so we also verify timing:
+///   - with the warm delay the first read takes ≥ 50 ms (waited for warm)
+///   - subsequent reads on the same handle take < 10 ms (watch already true)
+///   - a second `open()` on the same inode triggers a second warm call.
+#[test]
+fn warm_cache_triggered_on_open_sequential_read() {
+    const WARM_DELAY_MS: u64 = 50;
+    const FILE_SIZE: u64 = 1024 * 1024; // 1 MiB — small so data fetch is trivial
+
+    let hub = MockHub::new();
+    hub.add_file("file.bin", FILE_SIZE, Some("hash_abc"), None);
+    let xet = MockXet::new();
+    let data: Vec<u8> = (0..FILE_SIZE as usize).map(|i| (i % 251) as u8).collect();
+    xet.add_file("hash_abc", &data);
+    xet.set_warm_delay_ms(WARM_DELAY_MS);
+
+    let (rt, vfs) = vfs_readonly(&hub, &xet);
+
+    rt.block_on(async {
+        let attr = vfs.lookup(ROOT_INODE, "file.bin").await.unwrap();
+
+        // ── open #1 ──────────────────────────────────────────────────
+        // Warm task starts in background with 50 ms delay.
+        // We yield briefly so the task is definitely running before read().
+        let fh1 = vfs.open(attr.ino, false, false, None).await.unwrap();
+        tokio::task::yield_now().await;
+
+        // First read must wait for warm to complete — expect ≥ WARM_DELAY_MS.
+        let t0 = std::time::Instant::now();
+        let (chunk, _) = vfs.read(fh1, 0, 4096).await.unwrap();
+        let elapsed_first = t0.elapsed();
+
+        assert_eq!(chunk.len(), 4096);
+        assert_eq!(chunk[0], 0u8); // pattern byte at offset 0
+        assert!(
+            elapsed_first.as_millis() >= WARM_DELAY_MS as u128,
+            "first read finished in {}ms — should have waited for warm ({} ms)",
+            elapsed_first.as_millis(),
+            WARM_DELAY_MS,
+        );
+
+        // Warm was called exactly once so far.
+        assert_eq!(xet.warm_call_count.load(std::sync::atomic::Ordering::SeqCst), 1);
+
+        // Subsequent reads on the same handle are immediate (watch already true).
+        let t1 = std::time::Instant::now();
+        let (chunk2, _) = vfs.read(fh1, 4096, 4096).await.unwrap();
+        let elapsed_second = t1.elapsed();
+        assert_eq!(chunk2.len(), 4096);
+        assert_eq!(chunk2[0], (4096 % 251) as u8);
+        assert!(
+            elapsed_second.as_millis() < 10,
+            "second read took {}ms — should be near-instant (no warm wait)",
+            elapsed_second.as_millis(),
+        );
+
+        vfs.release(fh1).await;
+
+        // ── open #2 ──────────────────────────────────────────────────
+        // Each open() is an independent handle with its own warm task.
+        let fh2 = vfs.open(attr.ino, false, false, None).await.unwrap();
+        tokio::task::yield_now().await;
+
+        vfs.read(fh2, 0, 4096).await.unwrap();
+
+        assert_eq!(
+            xet.warm_call_count.load(std::sync::atomic::Ordering::SeqCst),
+            2,
+            "second open should trigger a second warm call"
+        );
+
+        vfs.release(fh2).await;
+    });
+}

tests/warm_cache_bench.rs

@@ -0,0 +1,148 @@
+/// Benchmark: xorb reconstruction cache warm-up benefit on sequential reads.
+///
+/// Each open() reads the full file and reports:
+///   TTFB  — time to first byte (blocked on warm_reconstruction_cache)
+///   MB/s  — end-to-end sequential throughput including TTFB
+///
+/// Open #1 = cold: FUSE process fetches the reconstruction plan from CAS.
+/// Open #2+ = warm: in-process CachedXetClient cache hit, plan already stored.
+///
+/// Run on EC2:
+///   cargo test --release --test warm_cache_bench -- --nocapture
+mod common;
+
+use std::io::Read;
+use std::time::Instant;
+
+const FILE_SIZE: usize = 100 * 1024 * 1024; // 100 MB
+const N_OPENS: usize = 5;
+const READ_CHUNK: usize = 4 * 1024 * 1024; // 4 MB per read syscall
+
+#[tokio::test]
+async fn bench_xorb_reconstruction_cache() {
+    let (token, bucket_id, hub) = match common::setup_bucket("warm-cache-bench").await {
+        Some(cfg) => cfg,
+        None => return,
+    };
+
+    let filename = "seq_100mb.bin";
+    let data = common::generate_pattern(FILE_SIZE);
+    let write_config = common::build_write_config(&hub).await;
+
+    let tmp = std::env::temp_dir().join(format!("hf-warm-bench-{}", std::process::id()));
+    std::fs::create_dir_all(&tmp).ok();
+    let staging = tmp.join(filename);
+    std::fs::write(&staging, &data).expect("write staging");
+
+    let file_info = common::upload_file(write_config, &staging).await;
+    let xet_hash = file_info.hash().to_string();
+    eprintln!(
+        "Uploaded {} ({} MB), xet_hash={}",
+        filename,
+        FILE_SIZE / (1024 * 1024),
+        xet_hash
+    );
+    std::fs::remove_dir_all(&tmp).ok();
+
+    hub.batch_operations(&[hf_mount::hub_api::BatchOp::AddFile {
+        path: filename.to_string(),
+        xet_hash,
+        mtime: std::time::SystemTime::now()
+            .duration_since(std::time::UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_millis() as u64,
+        content_type: None,
+    }])
+    .await
+    .expect("batch add");
+
+    let pid = std::process::id();
+    let mount = format!("/tmp/hf-warm-bench-{}", pid);
+    // Use /dev/shm (tmpfs) on Linux for RAM-speed cache reads, falling back to /tmp.
+    let shm_dir = format!("/dev/shm/hf-warm-cache-{}", pid);
+    let cache = if cfg!(target_os = "linux") && std::fs::create_dir_all(&shm_dir).is_ok() {
+        shm_dir
+    } else {
+        format!("/tmp/hf-warm-cache-{}", pid)
+    };
+
+    eprintln!("Cache dir: {}", cache);
+
+    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+        let child = common::mount_bucket(&bucket_id, &mount, &cache, &["--read-only"]);
+        let file_path = format!("{}/{}", mount, filename);
+        let size_mb = FILE_SIZE as f64 / (1024.0 * 1024.0);
+
+        eprintln!(
+            "\n=== Xorb reconstruction cache: sequential read ({} MB) ===",
+            FILE_SIZE / (1024 * 1024)
+        );
+        eprintln!(
+            "  {:>6}  {:>10}  {:>10}  {:>10}  {}",
+            "Open#", "TTFB (ms)", "Total (s)", "MB/s", "cache"
+        );
+        eprintln!("  {:-<6}  {:-<10}  {:-<10}  {:-<10}  {:-<5}", "", "", "", "", "");
+
+        let mut results: Vec<(f64, f64, f64)> = Vec::new(); // (ttfb_ms, total_s, mbps)
+        let mut buf = vec![0u8; READ_CHUNK];
+
+        for i in 0..N_OPENS {
+            let t0 = Instant::now();
+            let mut f = std::fs::File::open(&file_path).expect("open");
+
+            // First read: blocks until warm_reconstruction_cache completes (cold = CAS round-trip)
+            let n = f.read(&mut buf).expect("first read");
+            let ttfb = t0.elapsed();
+            assert!(n > 0, "empty first read");
+            assert_eq!(buf[0], 0u8, "data mismatch at open {}", i + 1);
+
+            // Drain the rest
+            let mut total = n;
+            loop {
+                let n = f.read(&mut buf).expect("seq read");
+                if n == 0 {
+                    break;
+                }
+                total += n;
+            }
+            let elapsed = t0.elapsed();
+            assert_eq!(total, FILE_SIZE, "size mismatch at open {}", i + 1);
+
+            let ttfb_ms = ttfb.as_secs_f64() * 1000.0;
+            let total_s = elapsed.as_secs_f64();
+            let mbps = size_mb / total_s;
+            let label = if i == 0 { "cold" } else { "warm" };
+
+            eprintln!(
+                "  {:>6}  {:>10.1}  {:>10.2}  {:>10.1}  {}",
+                i + 1,
+                ttfb_ms,
+                total_s,
+                mbps,
+                label,
+            );
+            results.push((ttfb_ms, total_s, mbps));
+        }
+
+        common::unmount(&mount, child, 10);
+        results
+    }));
+
+    std::fs::remove_dir_all(&mount).ok();
+    std::fs::remove_dir_all(&cache).ok();
+    common::delete_bucket(&common::endpoint(), &token, &bucket_id).await;
+
+    let results = match result {
+        Ok(r) => r,
+        Err(e) => std::panic::resume_unwind(e),
+    };
+
+    let (cold_ttfb, _, cold_mbps) = results[0];
+    let warm_ttfb_avg = results[1..].iter().map(|(t, _, _)| t).sum::<f64>() / (results.len() - 1) as f64;
+    let warm_mbps_avg = results[1..].iter().map(|(_, _, m)| m).sum::<f64>() / (results.len() - 1) as f64;
+
+    eprintln!(
+        "\nSummary: cold TTFB {:.0} ms / {:.1} MB/s  →  warm TTFB {:.0} ms / {:.1} MB/s",
+        cold_ttfb, cold_mbps, warm_ttfb_avg, warm_mbps_avg,
+    );
+}