ddsha441981 · VedantMadane · Aug 24, 2026
diff --git a/Cargo.toml b/Cargo.toml
@@ -69,6 +69,9 @@ required-features = ["std"]
 [[example]]
 name = "real_world_latency_spikes"
 required-features = ["std"]
+[[example]]
+name = "soak_test"
+required-features = ["std"]
 
 [profile.release]
 opt-level = 3

diff --git a/examples/soak_test.rs b/examples/soak_test.rs
@@ -0,0 +1,386 @@
+// Copyright (c) 2026 Deendayal Kumawat. All rights reserved.
+// Licensed under the MIT OR Apache-2.0 license.
+
+//! 24-hour soak / endurance test for `ShardedPulseMap`.
+//!
+//! Verifies stability under sustained concurrent load: no panic, no deadlock,
+//! monotonic eviction counts, bounded `len()`, RSS growth within tolerance,
+//! and sentinel key integrity before TTL expiry.
+//!
+//! ```bash
+//! # Full 24-hour soak
+//! cargo run --release --example soak_test --features std
+//!
+//! # Quick smoke (default CI-friendly duration: 5s unless --duration is set)
+//! cargo run --release --example soak_test --features std -- --duration 5
+//!
+//! # 1-hour CI smoke
+//! cargo run --release --example soak_test --features std -- --duration 3600
+//! ```
+
+use pulse_map::ShardedPulseMap;
+use std::env;
+use std::process;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::Arc;
+use std::thread;
+use std::time::{Duration, Instant};
+
+/// Default duration: 24 hours (per issue). Override with `--duration <secs>`.
+const DEFAULT_DURATION_SECS: u64 = 24 * 3600;
+/// Minimum duration accepted by CLI (issue asks for min 1h for CI; we allow
+/// shorter values so local/PR smoke can finish quickly).
+const MIN_DURATION_SECS: u64 = 1;
+/// Stats + integrity check interval.
+const STATS_INTERVAL: Duration = Duration::from_secs(60);
+/// When total duration is under 2 minutes, print stats more often.
+const SHORT_STATS_INTERVAL: Duration = Duration::from_secs(1);
+/// Writer / reader thread counts from the issue.
+const NUM_WRITERS: usize = 8;
+const NUM_READERS: usize = 4;
+/// Buckets per shard (issue: 4096).
+const BUCKETS_PER_SHARD: usize = 4096;
+/// Global TTL in insertion epochs (issue: 10_000).
+const TTL_EPOCHS: u64 = 10_000;
+/// Sentinel keys live this many epochs (short enough to re-check often, long
+/// enough that a concurrent writer wave does not expire them mid-check).
+const SENTINEL_TTL: u64 = 50_000;
+const SENTINEL_COUNT: u64 = 32;
+/// Fail if RSS grows more than this fraction from the first sample.
+const RSS_GROWTH_LIMIT: f64 = 0.10;
+
+fn parse_duration_secs(args: &[String]) -> u64 {
+    let mut i = 0;
+    while i < args.len() {
+        if args[i] == "--duration" {
+            if let Some(v) = args.get(i + 1) {
+                match v.parse::<u64>() {
+                    Ok(n) if n >= MIN_DURATION_SECS => return n,
+                    Ok(_) => {
+                        eprintln!(
+                            "error: --duration must be >= {MIN_DURATION_SECS} seconds"
+                        );
+                        process::exit(2);
+                    }
+                    Err(_) => {
+                        eprintln!("error: invalid --duration value: {v}");
+                        process::exit(2);
+                    }
+                }
+            }
+            eprintln!("error: --duration requires a value in seconds");
+            process::exit(2);
+        }
+        if let Some(rest) = args[i].strip_prefix("--duration=") {
+            match rest.parse::<u64>() {
+                Ok(n) if n >= MIN_DURATION_SECS => return n,
+                Ok(_) => {
+                    eprintln!("error: --duration must be >= {MIN_DURATION_SECS} seconds");
+                    process::exit(2);
+                }
+                Err(_) => {
+                    eprintln!("error: invalid --duration value: {rest}");
+                    process::exit(2);
+                }
+            }
+        }
+        i += 1;
+    }
+    DEFAULT_DURATION_SECS
+}
+
+/// Best-effort RSS in bytes. Linux: `/proc/self/status` VmRSS.
+/// Other platforms: returns `None` (RSS check skipped).
+fn read_rss_bytes() -> Option<u64> {
+    #[cfg(target_os = "linux")]
+    {
+        let status = std::fs::read_to_string("/proc/self/status").ok()?;
+        for line in status.lines() {
+            if let Some(rest) = line.strip_prefix("VmRSS:") {
+                let kb: u64 = rest
+                    .split_whitespace()
+                    .next()?
+                    .parse()
+                    .ok()?;
+                return Some(kb.saturating_mul(1024));
+            }
+        }
+        None
+    }
+    #[cfg(not(target_os = "linux"))]
+    {
+        None
+    }
+}
+
+fn fmt_bytes(n: u64) -> String {
+    const MB: f64 = 1024.0 * 1024.0;
+    format!("{:.1}MB", n as f64 / MB)
+}
+
+fn fmt_elapsed(d: Duration) -> String {
+    let total = d.as_secs();
+    let h = total / 3600;
+    let m = (total % 3600) / 60;
+    let s = total % 60;
+    format!("{h:02}:{m:02}:{s:02}")
+}
+
+fn fmt_ops(n: u64) -> String {
+    if n >= 1_000_000_000 {
+        format!("{:.2}B", n as f64 / 1e9)
+    } else if n >= 1_000_000 {
+        format!("{:.1}M", n as f64 / 1e6)
+    } else if n >= 1_000 {
+        format!("{:.1}K", n as f64 / 1e3)
+    } else {
+        format!("{n}")
+    }
+}
+
+fn main() {
+    let args: Vec<String> = env::args().skip(1).collect();
+    let duration_secs = parse_duration_secs(&args);
+    let duration = Duration::from_secs(duration_secs);
+    let stats_every = if duration_secs < 120 {
+        SHORT_STATS_INTERVAL
+    } else {
+        STATS_INTERVAL
+    };
+
+    println!("═══════════════════════════════════════════");
+    println!("  ShardedPulseMap soak test");
+    println!("  duration={}s writers={NUM_WRITERS} readers={NUM_READERS}", duration_secs);
+    println!("  buckets/shard={BUCKETS_PER_SHARD} ttl={TTL_EPOCHS}");
+    println!("═══════════════════════════════════════════\n");
+
+    let map = Arc::new(ShardedPulseMap::<u64, u64>::new(BUCKETS_PER_SHARD));
+    map.set_ttl(TTL_EPOCHS);
+
+    let stop = Arc::new(AtomicBool::new(false));
+    let ops = Arc::new(AtomicU64::new(0));
+    let fail = Arc::new(AtomicBool::new(false));
+    let fail_reason = Arc::new(std::sync::Mutex::new(String::new()));
+
+    let set_fail = |reason: String| {
+        let mut g = fail_reason.lock().unwrap();
+        if g.is_empty() {
+            *g = reason;
+        }
+        fail.store(true, Ordering::SeqCst);
+        stop.store(true, Ordering::SeqCst);
+    };
+    let _ = &set_fail; // keep type-check happy when unused path
+
+    // ── Writers ──
+    let mut handles = Vec::new();
+    for tid in 0..NUM_WRITERS {
+        let map = Arc::clone(&map);
+        let stop = Arc::clone(&stop);
+        let ops = Arc::clone(&ops);
+        handles.push(thread::Builder::new()
+            .name(format!("writer-{tid}"))
+            .spawn(move || {
+                let mut i: u64 = tid as u64;
+                while !stop.load(Ordering::Relaxed) {
+                    // Spread keys so shards fill under TTL churn.
+                    let key = i.wrapping_mul(0x9E37_79B9_7F4A_7C15) ^ (tid as u64) << 48;
+                    map.insert(key, key ^ 0xDEAD_BEEF);
+                    ops.fetch_add(1, Ordering::Relaxed);
+                    i = i.wrapping_add(NUM_WRITERS as u64);
+                }
+            })
+            .expect("spawn writer"));
+    }
+
+    // ── Readers ──
+    for tid in 0..NUM_READERS {
+        let map = Arc::clone(&map);
+        let stop = Arc::clone(&stop);
+        let ops = Arc::clone(&ops);
+        handles.push(thread::Builder::new()
+            .name(format!("reader-{tid}"))
+            .spawn(move || {
+                let mut i: u64 = tid as u64;
+                while !stop.load(Ordering::Relaxed) {
+                    let key = i.wrapping_mul(0x9E37_79B9_7F4A_7C15) ^ ((tid as u64) % NUM_WRITERS as u64) << 48;
+                    let _ = map.get(&key);
+                    ops.fetch_add(1, Ordering::Relaxed);
+                    i = i.wrapping_add(NUM_READERS as u64);
+                }
+            })
+            .expect("spawn reader"));
+    }
+
+    let start = Instant::now();
+    let mut last_stats = Instant::now();
+    let mut last_evictions: usize = 0;
+    let mut initial_rss: Option<u64> = None;
+    let mut last_rss: Option<u64> = None;
+    let mut tick: u64 = 0;
+
+    // Prime a few inserts so capacity/len are meaningful before first sample.
+    for s in 0..SENTINEL_COUNT {
+        map.insert_ttl(u64::MAX - s, s.wrapping_mul(3) + 1, SENTINEL_TTL);
+    }
+    thread::sleep(Duration::from_millis(50));
+
+    while start.elapsed() < duration && !fail.load(Ordering::SeqCst) {
+        let sleep_for = stats_every
+            .checked_sub(last_stats.elapsed())
+            .unwrap_or(Duration::from_millis(0));
+        if !sleep_for.is_zero() {
+            thread::sleep(sleep_for.min(Duration::from_millis(200)));
+        }
+        if last_stats.elapsed() < stats_every && start.elapsed() < duration {
+            continue;
+        }
+        last_stats = Instant::now();
+        tick += 1;
+
+        let len = map.len();
+        let cap = map.capacity();
+        let load = if cap == 0 {
+            0.0
+        } else {
+            (len as f64 / cap as f64) * 100.0
+        };
+        let evictions = map.eviction_count();
+        let total_ops = ops.load(Ordering::Relaxed);
+        let rss = read_rss_bytes();
+        if initial_rss.is_none() {
+            initial_rss = rss;
+        }
+        last_rss = rss;
+
+        let rss_s = rss.map(fmt_bytes).unwrap_or_else(|| "n/a".into());
+        println!(
+            "[{}] len={len} cap={cap} load={load:.1}% evictions={evictions} rss={rss_s} ops={}",
+            fmt_elapsed(start.elapsed()),
+            fmt_ops(total_ops)
+        );
+
+        // Failure: len exceeds capacity
+        if len > cap {
+            let msg = format!("len ({len}) exceeded capacity ({cap})");
+            eprintln!("FAIL: {msg}");
+            let mut g = fail_reason.lock().unwrap();
+            if g.is_empty() {
+                *g = msg;
+            }
+            fail.store(true, Ordering::SeqCst);
+            break;
+        }
+
+        // Failure: eviction_count must be monotonic (never wraps/resets downward)
+        if evictions < last_evictions {
+            let msg = format!(
+                "eviction_count decreased: {last_evictions} -> {evictions}"
+            );
+            eprintln!("FAIL: {msg}");
+            let mut g = fail_reason.lock().unwrap();
+            if g.is_empty() {
+                *g = msg;
+            }
+            fail.store(true, Ordering::SeqCst);
+            break;
+        }
+        last_evictions = evictions;
+
+        // Failure: RSS growth > 10% from initial (Linux only)
+        if let (Some(init), Some(now)) = (initial_rss, rss) {
+            if init > 0 {
+                let growth = (now as f64 - init as f64) / init as f64;
+                if growth > RSS_GROWTH_LIMIT {
+                    let msg = format!(
+                        "RSS grew {:.1}% ({} -> {}), limit {:.0}%",
+                        growth * 100.0,
+                        fmt_bytes(init),
+                        fmt_bytes(now),
+                        RSS_GROWTH_LIMIT * 100.0
+                    );
+                    eprintln!("FAIL: {msg}");
+                    let mut g = fail_reason.lock().unwrap();
+                    if g.is_empty() {
+                        *g = msg;
+                    }
+                    fail.store(true, Ordering::SeqCst);
+                    break;
+                }
+            }
+        }
+
+        // Sentinel integrity: re-insert then verify immediately
+        for s in 0..SENTINEL_COUNT {
+            let k = u64::MAX - s;
+            let v = s.wrapping_mul(3) + 1 + tick; // rotate expected value
+            map.insert_ttl(k, v, SENTINEL_TTL);
+            match map.get(&k) {
+                Some(got) if got == v => {}
+                Some(got) => {
+                    let msg = format!("sentinel {k} wrong value: got {got}, want {v}");
+                    eprintln!("FAIL: {msg}");
+                    let mut g = fail_reason.lock().unwrap();
+                    if g.is_empty() {
+                        *g = msg;
+                    }
+                    fail.store(true, Ordering::SeqCst);
+                    break;
+                }
+                None => {
+                    let msg = format!("sentinel {k} missing after insert");
+                    eprintln!("FAIL: {msg}");
+                    let mut g = fail_reason.lock().unwrap();
+                    if g.is_empty() {
+                        *g = msg;
+                    }
+                    fail.store(true, Ordering::SeqCst);
+                    break;
+                }
+            }
+        }
+        if fail.load(Ordering::SeqCst) {
+            break;
+        }
+    }
+
+    stop.store(true, Ordering::SeqCst);
+    for h in handles {
+        // Writers/readers exit promptly after stop; join to surface panics.
+        if let Err(e) = h.join() {
+            eprintln!("FAIL: worker thread panicked: {e:?}");
+            fail.store(true, Ordering::SeqCst);
+            let mut g = fail_reason.lock().unwrap();
+            if g.is_empty() {
+                *g = format!("worker panic: {e:?}");
+            }
+        }
+    }
+
+    let elapsed = start.elapsed();
+    let total_ops = ops.load(Ordering::Relaxed);
+    let rss_drift = match (initial_rss, last_rss) {
+        (Some(a), Some(b)) => {
+            let delta = b as i64 - a as i64;
+            let sign = if delta >= 0 { "+" } else { "-" };
+            format!("{sign}{}", fmt_bytes(delta.unsigned_abs()))
+        }
+        _ => "n/a".into(),
+    };
+
+    if fail.load(Ordering::SeqCst) {
+        let reason = fail_reason.lock().unwrap().clone();
+        eprintln!(
+            "[{}] FAILED — {reason}. Total ops: {}, RSS drift: {rss_drift}",
+            fmt_elapsed(elapsed),
+            fmt_ops(total_ops)
+        );
+        process::exit(1);
+    }
+
+    println!(
+        "[{}] PASSED — soak complete. Total ops: {}, RSS drift: {rss_drift}",
+        fmt_elapsed(elapsed),
+        fmt_ops(total_ops)
+    );
+}