feat(tranquil-store): flaky-device scenario, jemalloc heap-prof

Lewis: May this revision serve well! <lu5a@proton.me>

.gitignore

@@ -1,4 +1,4 @@
-/target
+target/
 .env
 .direnv
 result

crates/tranquil-store/Cargo.toml

@@ -38,10 +38,12 @@ tempfile = { version = "3", optional = true }
 clap = { workspace = true, optional = true }
 toml = { version = "0.8", optional = true }
 tracing-subscriber = { workspace = true, features = ["env-filter"], optional = true }
+libc = "0.2"
 
 [features]
 test-harness = ["dep:tempfile"]
 gauntlet-cli = ["test-harness", "dep:clap", "dep:toml", "dep:tracing-subscriber"]
+gauntlet-jemalloc-prof = []
 
 [[bin]]
 name = "tranquil-gauntlet"
@@ -60,9 +62,8 @@ tranquil-db = { workspace = true }
 sqlx = { workspace = true }
 k256 = { workspace = true }
 rand = { workspace = true }
-tikv-jemallocator = "0.6"
+tikv-jemallocator = { version = "0.6", features = ["profiling", "unprefixed_malloc_on_supported_platforms"] }
 tracing-subscriber = { workspace = true, features = ["env-filter"] }
-libc = "0.2"
 
 [[bench]]
 name = "blockstore"

crates/tranquil-store/src/gauntlet/flaky.rs

@@ -0,0 +1,448 @@
+use std::env;
+use std::num::NonZeroU32;
+use std::path::{Path, PathBuf};
+use std::process::{Command, Output};
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use tempfile::TempDir;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct UpIntervalSecs(pub NonZeroU32);
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct DownIntervalSecs(pub NonZeroU32);
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct BackingMegabytes(pub u32);
+
+const fn nz(n: u32) -> NonZeroU32 {
+    match NonZeroU32::new(n) {
+        Some(v) => v,
+        None => panic!("zero interval not permitted"),
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct FlakyConfig {
+    pub up_interval: UpIntervalSecs,
+    pub down_interval: DownIntervalSecs,
+    pub backing_mb: BackingMegabytes,
+}
+
+impl FlakyConfig {
+    pub const fn default_stress() -> Self {
+        Self {
+            up_interval: UpIntervalSecs(nz(8)),
+            down_interval: DownIntervalSecs(nz(2)),
+            backing_mb: BackingMegabytes(256),
+        }
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum FlakyError {
+    #[error("not running as root, EUID != 0")]
+    NotRoot,
+    #[error("tool missing: {0}")]
+    ToolMissing(&'static str),
+    #[error("kernel target dm-flakey unavailable: {0}")]
+    DmFlakeyMissing(String),
+    #[error("{tool} failed: status={status}, stderr={stderr}")]
+    CommandFailed {
+        tool: &'static str,
+        status: String,
+        stderr: String,
+    },
+    #[error("io: {0}")]
+    Io(#[from] std::io::Error),
+}
+
+impl FlakyError {
+    pub const fn is_env_absent(&self) -> bool {
+        matches!(
+            self,
+            Self::NotRoot | Self::ToolMissing(_) | Self::DmFlakeyMissing(_)
+        )
+    }
+}
+
+static MOUNT_COUNTER: AtomicU64 = AtomicU64::new(0);
+
+pub struct FlakyMount {
+    mount_point: TempDir,
+    mapper_name: String,
+    mapper_path: PathBuf,
+    loop_device: PathBuf,
+    backing_file: PathBuf,
+    _backing_tempdir: TempDir,
+}
+
+impl FlakyMount {
+    pub fn try_new(cfg: &FlakyConfig) -> Result<Self, FlakyError> {
+        ensure_root()?;
+        ["losetup", "dmsetup", "mkfs.ext4", "mount", "umount"]
+            .iter()
+            .copied()
+            .try_for_each(ensure_tool)?;
+        probe_dm_flakey()?;
+        let _ = reap_stale_mounts();
+
+        let backing_dir = TempDir::new()?;
+        let backing_file = backing_dir.path().join("backing.img");
+        allocate_backing(&backing_file, cfg.backing_mb)?;
+
+        let loop_device = attach_loop(&backing_file)?;
+        let sectors = sector_count(&loop_device)?;
+
+        if let Err(e) = mkfs_ext4(&loop_device) {
+            let _ = detach_loop(&loop_device);
+            return Err(e);
+        }
+
+        let mapper_name = format!(
+            "tranquil-flaky-{}-{}",
+            std::process::id(),
+            MOUNT_COUNTER.fetch_add(1, Ordering::Relaxed),
+        );
+        match dm_create(
+            &mapper_name,
+            &loop_device,
+            sectors,
+            cfg.up_interval,
+            cfg.down_interval,
+        ) {
+            Ok(()) => {}
+            Err(e) => {
+                let _ = detach_loop(&loop_device);
+                return Err(e);
+            }
+        }
+        let mapper_path = PathBuf::from(format!("/dev/mapper/{mapper_name}"));
+
+        let mount_point = TempDir::new()?;
+        if let Err(e) = mount_ext4(&mapper_path, mount_point.path()) {
+            let _ = dm_remove(&mapper_name);
+            let _ = detach_loop(&loop_device);
+            return Err(e);
+        }
+
+        Ok(Self {
+            mount_point,
+            mapper_name,
+            mapper_path,
+            loop_device,
+            backing_file,
+            _backing_tempdir: backing_dir,
+        })
+    }
+
+    pub fn path(&self) -> &Path {
+        self.mount_point.path()
+    }
+
+    pub fn mapper_name(&self) -> &str {
+        &self.mapper_name
+    }
+
+    pub fn mapper_path(&self) -> &Path {
+        &self.mapper_path
+    }
+
+    pub fn loop_device(&self) -> &Path {
+        &self.loop_device
+    }
+
+    pub fn backing_file(&self) -> &Path {
+        &self.backing_file
+    }
+}
+
+impl Drop for FlakyMount {
+    fn drop(&mut self) {
+        if let Err(e) = umount(self.mount_point.path()) {
+            tracing::warn!(
+                mount = %self.mount_point.path().display(),
+                error = %e,
+                "flaky umount failed, trying lazy unmount",
+            );
+            if let Err(e2) = umount_lazy(self.mount_point.path()) {
+                tracing::warn!(
+                    mount = %self.mount_point.path().display(),
+                    error = %e2,
+                    "flaky lazy unmount also failed, device may leak",
+                );
+            }
+        }
+        if let Err(e) = dm_remove(&self.mapper_name) {
+            tracing::warn!(
+                name = %self.mapper_name,
+                error = %e,
+                "flaky dm remove failed, mapper device may leak",
+            );
+        }
+        if let Err(e) = detach_loop(&self.loop_device) {
+            tracing::warn!(
+                device = %self.loop_device.display(),
+                error = %e,
+                "flaky loop detach failed, loop device may leak",
+            );
+        }
+    }
+}
+
+#[cfg(unix)]
+fn ensure_root() -> Result<(), FlakyError> {
+    if unsafe { libc::geteuid() } == 0 {
+        Ok(())
+    } else {
+        Err(FlakyError::NotRoot)
+    }
+}
+
+#[cfg(not(unix))]
+fn ensure_root() -> Result<(), FlakyError> {
+    Err(FlakyError::NotRoot)
+}
+
+fn ensure_tool(tool: &'static str) -> Result<(), FlakyError> {
+    match find_in_path(tool) {
+        Some(_) => Ok(()),
+        None => Err(FlakyError::ToolMissing(tool)),
+    }
+}
+
+fn find_in_path(tool: &str) -> Option<PathBuf> {
+    let path = env::var_os("PATH")?;
+    env::split_paths(&path).find_map(|dir| {
+        let candidate = dir.join(tool);
+        is_executable_file(&candidate).then_some(candidate)
+    })
+}
+
+#[cfg(unix)]
+fn is_executable_file(p: &Path) -> bool {
+    use std::os::unix::fs::PermissionsExt;
+    std::fs::metadata(p)
+        .map(|m| m.is_file() && (m.permissions().mode() & 0o111) != 0)
+        .unwrap_or(false)
+}
+
+#[cfg(not(unix))]
+fn is_executable_file(p: &Path) -> bool {
+    p.is_file()
+}
+
+fn probe_dm_flakey() -> Result<(), FlakyError> {
+    let out = Command::new("dmsetup").arg("targets").output()?;
+    if !out.status.success() {
+        return Err(FlakyError::DmFlakeyMissing(stringify_output(&out)));
+    }
+    let stdout = String::from_utf8_lossy(&out.stdout);
+    if !stdout.lines().any(|l| l.starts_with("flakey")) {
+        return Err(FlakyError::DmFlakeyMissing(stdout.into_owned()));
+    }
+    Ok(())
+}
+
+fn reap_stale_mounts() -> Result<(), FlakyError> {
+    let out = Command::new("dmsetup")
+        .arg("ls")
+        .arg("--target")
+        .arg("flakey")
+        .output()?;
+    if !out.status.success() {
+        return Ok(());
+    }
+    String::from_utf8_lossy(&out.stdout)
+        .lines()
+        .filter_map(parse_flaky_entry)
+        .filter(|(_, pid)| !pid_alive(*pid))
+        .for_each(|(name, _)| {
+            let loop_device = mapper_backing_loop(&name);
+            if let Err(e) = dm_remove(&name) {
+                tracing::warn!(name, error = %e, "reap: dm_remove stale mapper failed");
+                return;
+            }
+            if let Some(loop_dev) = loop_device
+                && let Err(e) = detach_loop(&loop_dev)
+            {
+                tracing::warn!(
+                    device = %loop_dev.display(),
+                    error = %e,
+                    "reap: detach_loop stale device failed",
+                );
+            }
+        });
+    Ok(())
+}
+
+fn parse_flaky_entry(line: &str) -> Option<(String, u32)> {
+    let name = line.split_whitespace().next()?;
+    let suffix = name.strip_prefix("tranquil-flaky-")?;
+    let pid_str = suffix.split('-').next()?;
+    let pid = pid_str.parse::<u32>().ok()?;
+    Some((name.to_string(), pid))
+}
+
+fn pid_alive(pid: u32) -> bool {
+    Path::new(&format!("/proc/{pid}")).exists()
+}
+
+fn mapper_backing_loop(name: &str) -> Option<PathBuf> {
+    let out = Command::new("dmsetup")
+        .arg("deps")
+        .arg("-o")
+        .arg("devname")
+        .arg(name)
+        .output()
+        .ok()?;
+    if !out.status.success() {
+        return None;
+    }
+    let text = String::from_utf8_lossy(&out.stdout);
+    let inner = text.split('(').nth(1)?;
+    let dev = inner.split(')').next()?.trim();
+    if dev.is_empty() {
+        None
+    } else {
+        Some(PathBuf::from(format!("/dev/{dev}")))
+    }
+}
+
+fn allocate_backing(path: &Path, size: BackingMegabytes) -> Result<(), FlakyError> {
+    let out = Command::new("truncate")
+        .arg("-s")
+        .arg(format!("{}M", size.0))
+        .arg(path)
+        .output()?;
+    check_status("truncate", &out)
+}
+
+fn attach_loop(backing: &Path) -> Result<PathBuf, FlakyError> {
+    let out = Command::new("losetup")
+        .arg("--find")
+        .arg("--show")
+        .arg(backing)
+        .output()?;
+    check_status("losetup", &out)?;
+    let raw = String::from_utf8_lossy(&out.stdout).trim().to_string();
+    if raw.is_empty() {
+        return Err(FlakyError::CommandFailed {
+            tool: "losetup",
+            status: "exit 0".to_string(),
+            stderr: "no device path on stdout".to_string(),
+        });
+    }
+    Ok(PathBuf::from(raw))
+}
+
+fn detach_loop(device: &Path) -> Result<(), FlakyError> {
+    let out = Command::new("losetup").arg("-d").arg(device).output()?;
+    check_status("losetup -d", &out)
+}
+
+fn sector_count(device: &Path) -> Result<u64, FlakyError> {
+    let out = Command::new("blockdev")
+        .arg("--getsz")
+        .arg(device)
+        .output()?;
+    check_status("blockdev", &out)?;
+    let raw = String::from_utf8_lossy(&out.stdout).trim().to_string();
+    raw.parse::<u64>().map_err(|_| FlakyError::CommandFailed {
+        tool: "blockdev",
+        status: "exit 0".to_string(),
+        stderr: format!("could not parse sector count: {raw:?}"),
+    })
+}
+
+fn dm_create(
+    name: &str,
+    loop_device: &Path,
+    sectors: u64,
+    up: UpIntervalSecs,
+    down: DownIntervalSecs,
+) -> Result<(), FlakyError> {
+    let table = format!(
+        "0 {sectors} flakey {} 0 {} {}",
+        loop_device.display(),
+        up.0.get(),
+        down.0.get(),
+    );
+    let mut child = Command::new("dmsetup")
+        .arg("create")
+        .arg(name)
+        .stdin(std::process::Stdio::piped())
+        .stdout(std::process::Stdio::piped())
+        .stderr(std::process::Stdio::piped())
+        .spawn()?;
+    if let Some(stdin) = child.stdin.as_mut() {
+        use std::io::Write;
+        stdin.write_all(table.as_bytes())?;
+    }
+    let out = child.wait_with_output()?;
+    check_status("dmsetup create", &out)
+}
+
+fn dm_remove(name: &str) -> Result<(), FlakyError> {
+    let out = Command::new("dmsetup")
+        .arg("remove")
+        .arg("--retry")
+        .arg(name)
+        .output()?;
+    check_status("dmsetup remove", &out)
+}
+
+fn mkfs_ext4(device: &Path) -> Result<(), FlakyError> {
+    let out = Command::new("mkfs.ext4")
+        .arg("-q")
+        .arg("-F")
+        .arg(device)
+        .output()?;
+    check_status("mkfs.ext4", &out)
+}
+
+fn mount_ext4(device: &Path, target: &Path) -> Result<(), FlakyError> {
+    let out = Command::new("mount")
+        .arg("-t")
+        .arg("ext4")
+        .arg(device)
+        .arg(target)
+        .output()?;
+    check_status("mount", &out)
+}
+
+fn umount(target: &Path) -> Result<(), FlakyError> {
+    let out = Command::new("umount").arg(target).output()?;
+    check_status("umount", &out)
+}
+
+fn umount_lazy(target: &Path) -> Result<(), FlakyError> {
+    let out = Command::new("umount").arg("-l").arg(target).output()?;
+    check_status("umount -l", &out)
+}
+
+fn check_status(tool: &'static str, out: &Output) -> Result<(), FlakyError> {
+    if out.status.success() {
+        Ok(())
+    } else {
+        Err(FlakyError::CommandFailed {
+            tool,
+            status: format!("{}", out.status),
+            stderr: String::from_utf8_lossy(&out.stderr).into_owned(),
+        })
+    }
+}
+
+fn stringify_output(out: &Output) -> String {
+    let mut s = String::new();
+    if !out.stdout.is_empty() {
+        s.push_str(&String::from_utf8_lossy(&out.stdout));
+    }
+    if !out.stderr.is_empty() {
+        if !s.is_empty() {
+            s.push('\n');
+        }
+        s.push_str(&String::from_utf8_lossy(&out.stderr));
+    }
+    s
+}

crates/tranquil-store/src/gauntlet/invariants.rs

@@ -351,7 +351,7 @@ fn snapshot<S: StorageIO + Send + Sync + 'static>(
         .into_iter()
         .map(|(c, r)| (c, r.raw()))
         .collect();
-    v.sort_unstable_by(|a, b| a.0.cmp(&b.0));
+    v.sort_unstable_by_key(|a| a.0);
     v
 }
 

crates/tranquil-store/src/gauntlet/mod.rs

@@ -1,4 +1,5 @@
 pub mod farm;
+pub mod flaky;
 pub mod invariants;
 pub mod leak;
 pub mod metrics;
@@ -12,6 +13,9 @@ pub mod shrink;
 pub mod soak;
 pub mod workload;
 
+pub use flaky::{
+    BackingMegabytes, DownIntervalSecs, FlakyConfig, FlakyError, FlakyMount, UpIntervalSecs,
+};
 pub use invariants::{
     EventLogSnapshot, Invariant, InvariantSet, InvariantViolation, SnapshotEvent, invariants_for,
 };

crates/tranquil-store/src/gauntlet/runner.rs

@@ -7,6 +7,7 @@ use std::time::Duration;
 use cid::Cid;
 use jacquard_repo::mst::Mst;
 
+use super::flaky::{FlakyConfig, FlakyMount};
 use super::invariants::{
     EventLogSnapshot, InvariantCtx, InvariantSet, InvariantViolation, SnapshotEvent, invariants_for,
 };
@@ -27,6 +28,7 @@ use crate::sim::{FaultConfig, SimulatedIO};
 #[derive(Debug, Clone, Copy)]
 pub enum IoBackend {
     Real,
+    RealWithFlaky { flaky: FlakyConfig },
     Simulated { fault: FaultConfig },
 }
 
@@ -218,6 +220,14 @@ impl Gauntlet {
                     op_errors_counter.clone(),
                     restarts_counter.clone(),
                 )),
+                IoBackend::RealWithFlaky { flaky } => Box::pin(run_inner_real_with_flaky(
+                    self.config,
+                    flaky,
+                    ops,
+                    ops_counter.clone(),
+                    op_errors_counter.clone(),
+                    restarts_counter.clone(),
+                )),
                 IoBackend::Simulated { fault } => Box::pin(run_inner_simulated(
                     self.config,
                     fault,
@@ -261,9 +271,97 @@ async fn run_inner_real(
     restarts_counter: Arc<AtomicUsize>,
 ) -> GauntletReport {
     let dir = tempfile::TempDir::new().expect("tempdir");
-    let cfg = blockstore_config(dir.path(), &config.store);
+    let root = dir.path().to_path_buf();
+    let report = run_inner_real_on_root(
+        config,
+        root,
+        ops,
+        ops_counter,
+        op_errors_counter,
+        restarts_counter,
+        false,
+        Duration::ZERO,
+    )
+    .await;
+    drop(dir);
+    report
+}
+
+async fn run_inner_real_with_flaky(
+    config: GauntletConfig,
+    flaky_cfg: FlakyConfig,
+    ops: OpStream,
+    ops_counter: Arc<AtomicUsize>,
+    op_errors_counter: Arc<AtomicUsize>,
+    restarts_counter: Arc<AtomicUsize>,
+) -> GauntletReport {
+    let mount = match FlakyMount::try_new(&flaky_cfg) {
+        Ok(m) => m,
+        Err(e) => {
+            let invariant = if e.is_env_absent() {
+                "FlakyEnvironment"
+            } else {
+                "FlakyOperational"
+            };
+            return GauntletReport {
+                seed: config.seed,
+                ops_executed: OpsExecuted(0),
+                op_errors: OpErrorCount(0),
+                restarts: RestartCount(0),
+                violations: vec![InvariantViolation {
+                    invariant,
+                    detail: format!("flaky mount: {e}"),
+                }],
+                ops: OpStream::empty(),
+            };
+        }
+    };
+    let root = mount.path().join("store");
+    if let Err(e) = std::fs::create_dir_all(&root) {
+        return GauntletReport {
+            seed: config.seed,
+            ops_executed: OpsExecuted(0),
+            op_errors: OpErrorCount(0),
+            restarts: RestartCount(0),
+            violations: vec![InvariantViolation {
+                invariant: "FlakyOperational",
+                detail: format!("create_dir_all {}: {e}", root.display()),
+            }],
+            ops: OpStream::empty(),
+        };
+    }
+    let down_ms = u64::from(flaky_cfg.down_interval.0.get())
+        .saturating_mul(1_000)
+        .saturating_add(500);
+    let report = run_inner_real_on_root(
+        config,
+        root,
+        ops,
+        ops_counter,
+        op_errors_counter,
+        restarts_counter,
+        true,
+        Duration::from_millis(down_ms),
+    )
+    .await;
+    drop(mount);
+    report
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn run_inner_real_on_root(
+    config: GauntletConfig,
+    root: PathBuf,
+    ops: OpStream,
+    ops_counter: Arc<AtomicUsize>,
+    op_errors_counter: Arc<AtomicUsize>,
+    restarts_counter: Arc<AtomicUsize>,
+    tolerate_op_errors: bool,
+    reopen_backoff: Duration,
+) -> GauntletReport {
+    let cfg = blockstore_config(&root, &config.store);
     let eventlog_cfg = config.eventlog;
-    let segments_dir = segments_subdir(dir.path());
+    let segments_dir = segments_subdir(&root);
     let open = {
         let segments_dir = segments_dir.clone();
         move || -> Result<Harness<RealIO>, String> {
@@ -289,7 +387,8 @@ async fn run_inner_real(
             restarts_counter,
             open,
             || {},
-            false,
+            tolerate_op_errors,
+            reopen_backoff,
         )
         .await
     } else {
@@ -301,7 +400,8 @@ async fn run_inner_real(
             restarts_counter,
             open,
             || {},
-            false,
+            tolerate_op_errors,
+            reopen_backoff,
         )
         .await
     }
@@ -356,6 +456,7 @@ async fn run_inner_simulated(
             open,
             crash,
             tolerate_errors,
+            Duration::ZERO,
         )
         .await
     } else {
@@ -368,6 +469,7 @@ async fn run_inner_simulated(
             open,
             crash,
             tolerate_errors,
+            Duration::ZERO,
         )
         .await
     }
@@ -406,6 +508,7 @@ async fn run_inner_generic<S, Open, Crash>(
     mut open: Open,
     mut crash: Crash,
     tolerate_op_errors: bool,
+    reopen_backoff: Duration,
 ) -> GauntletReport
 where
     S: StorageIO + Send + Sync + 'static,
@@ -474,7 +577,7 @@ where
             oracle.record_crash();
         }
         shutdown_harness(&mut harness);
-        match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors) {
+        match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors, reopen_backoff).await {
             Ok(reopened) => {
                 harness = Some(reopened);
                 let n = restarts_counter.fetch_add(1, Ordering::Relaxed) + 1;
@@ -510,7 +613,7 @@ where
         crash();
         oracle.record_crash();
         shutdown_harness(&mut harness);
-        match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors) {
+        match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors, reopen_backoff).await {
             Ok(reopened) => harness = Some(reopened),
             Err(detail) => {
                 violations.push(InvariantViolation {
@@ -551,7 +654,7 @@ where
     {
         let pre_snapshot = snapshot_block_index(&live.store);
         shutdown_harness(&mut harness);
-        match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors) {
+        match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors, reopen_backoff).await {
             Ok(reopened) => {
                 let post_snapshot = snapshot_block_index(&reopened.store);
                 if let Some(detail) = diff_snapshots(&pre_snapshot, &post_snapshot) {
@@ -636,10 +739,11 @@ fn shutdown_harness<S: StorageIO + Send + Sync + 'static>(harness: &mut Option<H
 
 const MAX_REOPEN_ATTEMPTS: usize = 5;
 
-fn reopen_with_recovery<S, Open, Crash>(
+async fn reopen_with_recovery<S, Open, Crash>(
     open: &mut Open,
     crash: &mut Crash,
     tolerate: bool,
+    backoff: Duration,
 ) -> Result<Harness<S>, String>
 where
     S: StorageIO + Send + Sync + 'static,
@@ -647,19 +751,22 @@ where
     Crash: FnMut(),
 {
     let mut errors: Vec<String> = Vec::new();
-    (0..MAX_REOPEN_ATTEMPTS)
-        .find_map(|attempt| match open() {
-            Ok(h) => Some(Ok(h)),
+    for attempt in 0..MAX_REOPEN_ATTEMPTS {
+        if attempt > 0 && !backoff.is_zero() {
+            tokio::time::sleep(backoff).await;
+        }
+        match open() {
+            Ok(h) => return Ok(h),
             Err(e) => {
                 errors.push(format!("attempt {attempt}: {e}"));
                 if !tolerate {
-                    return Some(Err(errors.join(" | ")));
+                    return Err(errors.join(" | "));
                 }
                 crash();
-                None
             }
-        })
-        .unwrap_or_else(|| Err(errors.join(" | ")))
+        }
+    }
+    Err(errors.join(" | "))
 }
 
 const QUICK_SAMPLE_SIZE: usize = 32;
@@ -777,7 +884,7 @@ fn snapshot_block_index<S: StorageIO + Send + Sync + 'static>(
         .into_iter()
         .map(|(c, r)| (c, r.raw()))
         .collect();
-    v.sort_unstable_by(|a, b| a.0.cmp(&b.0));
+    v.sort_unstable_by_key(|a| a.0);
     v
 }
 
@@ -1450,6 +1557,7 @@ async fn run_inner_generic_concurrent<S, Open, Crash>(
     mut open: Open,
     mut crash: Crash,
     tolerate_op_errors: bool,
+    reopen_backoff: Duration,
 ) -> GauntletReport
 where
     S: StorageIO + Send + Sync + 'static,
@@ -1568,7 +1676,7 @@ where
                     oracle.record_crash();
                 }
                 shutdown_harness(&mut harness);
-                match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors) {
+                match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors, reopen_backoff).await {
                     Ok(reopened) => {
                         harness = Some(reopened);
                         let n = restarts_counter.fetch_add(1, Ordering::Relaxed) + 1;
@@ -1605,7 +1713,7 @@ where
         crash();
         oracle.record_crash();
         shutdown_harness(&mut harness);
-        match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors) {
+        match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors, reopen_backoff).await {
             Ok(reopened) => harness = Some(reopened),
             Err(detail) => {
                 violations.push(InvariantViolation {
@@ -1647,7 +1755,7 @@ where
     {
         let pre_snapshot = snapshot_block_index(&live.store);
         shutdown_harness(&mut harness);
-        match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors) {
+        match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors, reopen_backoff).await {
             Ok(reopened) => {
                 let post_snapshot = snapshot_block_index(&reopened.store);
                 if let Some(detail) = diff_snapshots(&pre_snapshot, &post_snapshot) {

crates/tranquil-store/src/gauntlet/scenarios.rs

@@ -1,3 +1,4 @@
+use super::flaky::FlakyConfig;
 use super::invariants::InvariantSet;
 use super::op::{CollectionName, Seed};
 use super::runner::{
@@ -29,6 +30,7 @@ pub enum Scenario {
     FirehoseFanout,
     ContendedReaders,
     ContendedWriters,
+    FlakyDevice,
 }
 
 impl Scenario {
@@ -50,6 +52,7 @@ impl Scenario {
             Self::FirehoseFanout => "FirehoseFanout",
             Self::ContendedReaders => "ContendedReaders",
             Self::ContendedWriters => "ContendedWriters",
+            Self::FlakyDevice => "FlakyDevice",
         }
     }
 
@@ -71,6 +74,7 @@ impl Scenario {
             Self::FirehoseFanout => "firehose-fanout",
             Self::ContendedReaders => "contended-readers",
             Self::ContendedWriters => "contended-writers",
+            Self::FlakyDevice => "flaky-device",
         }
     }
 
@@ -102,6 +106,9 @@ impl Scenario {
             Self::ContendedWriters => {
                 "Add/delete heavy, 32 writer tasks, simulated moderate faults."
             }
+            Self::FlakyDevice => {
+                "Real IO on ext4 atop dm-flakey. Requires root with dm-flakey available, skips otherwise."
+            }
         }
     }
 
@@ -130,6 +137,7 @@ impl Scenario {
         Self::FirehoseFanout,
         Self::ContendedReaders,
         Self::ContendedWriters,
+        Self::FlakyDevice,
     ];
 }
 
@@ -202,6 +210,7 @@ pub fn config_for(scenario: Scenario, seed: Seed) -> GauntletConfig {
         Scenario::FirehoseFanout => firehose_fanout(seed),
         Scenario::ContendedReaders => contended_readers(seed),
         Scenario::ContendedWriters => contended_writers(seed),
+        Scenario::FlakyDevice => flaky_device(seed),
     }
 }
 
@@ -682,6 +691,37 @@ fn contended_readers(seed: Seed) -> GauntletConfig {
     }
 }
 
+fn flaky_device(seed: Seed) -> GauntletConfig {
+    GauntletConfig {
+        seed,
+        io: IoBackend::RealWithFlaky {
+            flaky: FlakyConfig::default_stress(),
+        },
+        workload: block_workload(
+            block_weights(80, 5, 10, 5),
+            SizeDistribution::Fixed(ValueBytes(128)),
+            KeySpaceSize(500),
+        ),
+        op_count: OpCount(20_000),
+        invariants: InvariantSet::REFCOUNT_CONSERVATION
+            | InvariantSet::REACHABILITY
+            | InvariantSet::ACKED_WRITE_PERSISTENCE
+            | InvariantSet::READ_AFTER_WRITE
+            | InvariantSet::RESTART_IDEMPOTENT
+            | InvariantSet::NO_ORPHAN_FILES
+            | InvariantSet::MANIFEST_EQUALS_REALITY
+            | InvariantSet::BYTE_BUDGET
+            | InvariantSet::CHECKSUM_COVERAGE,
+        limits: RunLimits {
+            max_wall_ms: Some(WallMs(30 * 60_000)),
+        },
+        restart_policy: RestartPolicy::EveryNOps(OpInterval(1_000)),
+        store: tiny_store(),
+        eventlog: None,
+        writer_concurrency: WriterConcurrency(1),
+    }
+}
+
 fn contended_writers(seed: Seed) -> GauntletConfig {
     GauntletConfig {
         seed,

crates/tranquil-store/tests/gauntlet_flaky.rs

@@ -0,0 +1,70 @@
+use std::num::NonZeroU32;
+
+use tranquil_store::gauntlet::{
+    BackingMegabytes, DownIntervalSecs, FlakyConfig, FlakyMount, Gauntlet, Scenario, Seed,
+    UpIntervalSecs, config_for,
+};
+
+#[tokio::test]
+#[ignore = "requires root + dm-flakey; run under a privileged container"]
+async fn flaky_device_scenario_sanity() {
+    let cfg = config_for(Scenario::FlakyDevice, Seed(1));
+    let op_count = cfg.op_count.0;
+    let report = Gauntlet::new(cfg).expect("build gauntlet").run().await;
+    let env_skip = report
+        .violations
+        .iter()
+        .any(|v| v.invariant == "FlakyEnvironment");
+    if env_skip {
+        eprintln!(
+            "flaky environment unavailable: {}",
+            report
+                .violations
+                .iter()
+                .map(|v| format!("{}: {}", v.invariant, v.detail))
+                .collect::<Vec<_>>()
+                .join(", ")
+        );
+        return;
+    }
+    let failures: Vec<String> = report
+        .violations
+        .iter()
+        .map(|v| format!("{}: {}", v.invariant, v.detail))
+        .collect();
+    assert!(failures.is_empty(), "violations: {failures:?}");
+    let floor = op_count / 2;
+    assert!(
+        report.ops_executed.0 >= floor,
+        "flaky ops_executed {} below floor {} of {op_count}: {} op_errors, {} restarts",
+        report.ops_executed.0,
+        floor,
+        report.op_errors.0,
+        report.restarts.0,
+    );
+}
+
+#[test]
+#[ignore = "requires root + dm-flakey; exercises mount/teardown without running Gauntlet"]
+fn flaky_mount_setup_teardown() {
+    let cfg = FlakyConfig {
+        up_interval: UpIntervalSecs(NonZeroU32::new(5).unwrap()),
+        down_interval: DownIntervalSecs(NonZeroU32::new(1).unwrap()),
+        backing_mb: BackingMegabytes(64),
+    };
+    let mount = match FlakyMount::try_new(&cfg) {
+        Ok(m) => m,
+        Err(e) if e.is_env_absent() => {
+            eprintln!("flaky environment unavailable: {e}");
+            return;
+        }
+        Err(e) => panic!("flaky mount setup failed: {e}"),
+    };
+    assert!(mount.path().exists(), "mount path should exist");
+    assert!(mount.mapper_path().exists(), "mapper device should exist");
+    let marker = mount.path().join("marker");
+    std::fs::write(&marker, b"ok").expect("write through flaky mount");
+    let back = std::fs::read(&marker).expect("read back");
+    assert_eq!(back, b"ok");
+    drop(mount);
+}

crates/tranquil-store/tests/gauntlet_soak.rs

@@ -5,6 +5,10 @@ use tranquil_store::gauntlet::{
     LeakGateConfig, Scenario, Seed, SoakConfig, SoakReport, config_for, run_soak,
 };
 
+#[cfg(feature = "gauntlet-jemalloc-prof")]
+#[global_allocator]
+static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
+
 fn soak_hours() -> Option<f64> {
     std::env::var("GAUNTLET_SOAK_HOURS")
         .ok()

crates/tranquil-store/tests/mst_refcount_integrity.rs

@@ -41,8 +41,8 @@ async fn compute_obsolete_from_diff<S: BlockStore + Sync + Send + 'static>(
 ) -> Vec<Cid> {
     let diff = old_mst.diff(new_mst).await.unwrap();
     std::iter::once(old_commit_cid)
-        .chain(diff.removed_mst_blocks.into_iter())
-        .chain(diff.removed_cids.into_iter())
+        .chain(diff.removed_mst_blocks)
+        .chain(diff.removed_cids)
         .collect()
 }
 

crates/tranquil-sync/src/repo.rs

@@ -238,7 +238,7 @@ async fn get_repo_since(state: &AppState, did: &Did, head_cid: &Cid, since: &str
 
         chunk
             .iter()
-            .zip(blocks.into_iter())
+            .zip(blocks)
             .filter_map(|(cid, block_opt)| block_opt.map(|block| (*cid, block)))
             .for_each(|(cid, block)| car_bytes.extend_from_slice(&encode_car_block(&cid, &block)));
     }

justfile

@@ -49,6 +49,45 @@ gauntlet-sweep CONFIG SEEDS="8" DUMP="proptest-regressions":
 gauntlet-soak HOURS="24" OUTPUT="":
     SQLX_OFFLINE=true GAUNTLET_SOAK_HOURS={{HOURS}} GAUNTLET_SOAK_OUTPUT={{OUTPUT}} cargo nextest run -p tranquil-store --features tranquil-store/test-harness --profile gauntlet-soak --test gauntlet_soak --run-ignored all -- soak_long_leak_gate
 
+gauntlet-soak-heapprof HOURS="24" OUTPUT="" PREFIX="jeprof.gauntlet":
+    SQLX_OFFLINE=true \
+    GAUNTLET_SOAK_HOURS={{HOURS}} \
+    GAUNTLET_SOAK_OUTPUT={{OUTPUT}} \
+    MALLOC_CONF="prof:true,prof_active:true,prof_final:true,lg_prof_sample:19,prof_prefix:{{PREFIX}}" \
+    cargo nextest run -p tranquil-store \
+        --features tranquil-store/test-harness,tranquil-store/gauntlet-jemalloc-prof \
+        --profile gauntlet-soak --test gauntlet_soak --run-ignored all -- soak_long_leak_gate
+
+gauntlet-flaky SEED="1":
+    SQLX_OFFLINE=true cargo nextest run -p tranquil-store --features tranquil-store/test-harness --test gauntlet_flaky --run-ignored all
+
+fuzz-target TARGET SECONDS="60" SANITIZER="address":
+    cd crates/tranquil-store/fuzz && cargo +nightly fuzz run --sanitizer {{SANITIZER}} {{TARGET}} -- -max_total_time={{SECONDS}}
+
+fuzz-pr SECONDS="60":
+    cd crates/tranquil-store/fuzz && cargo +nightly fuzz run --sanitizer address decode_block_record -- -max_total_time={{SECONDS}}
+    cd crates/tranquil-store/fuzz && cargo +nightly fuzz run --sanitizer address decode_hint_record -- -max_total_time={{SECONDS}}
+    cd crates/tranquil-store/fuzz && cargo +nightly fuzz run --sanitizer address segment_scan -- -max_total_time={{SECONDS}}
+    cd crates/tranquil-store/fuzz && cargo +nightly fuzz run --sanitizer address metastore_key_codec -- -max_total_time={{SECONDS}}
+    cd crates/tranquil-store/fuzz && cargo +nightly fuzz run --sanitizer address gauntlet_micro -- -max_total_time={{SECONDS}}
+
+fuzz-nightly SECONDS="21600":
+    cd crates/tranquil-store/fuzz && cargo +nightly fuzz run --sanitizer address decode_block_record -- -max_total_time={{SECONDS}}
+    cd crates/tranquil-store/fuzz && cargo +nightly fuzz run --sanitizer address decode_hint_record -- -max_total_time={{SECONDS}}
+    cd crates/tranquil-store/fuzz && cargo +nightly fuzz run --sanitizer address segment_scan -- -max_total_time={{SECONDS}}
+    cd crates/tranquil-store/fuzz && cargo +nightly fuzz run --sanitizer address metastore_key_codec -- -max_total_time={{SECONDS}}
+    cd crates/tranquil-store/fuzz && cargo +nightly fuzz run --sanitizer address gauntlet_micro -- -max_total_time={{SECONDS}}
+
+fuzz-ubsan TARGET SECONDS="60":
+    cd crates/tranquil-store/fuzz && cargo +nightly fuzz run --sanitizer undefined {{TARGET}} -- -max_total_time={{SECONDS}}
+
+test-store-asan:
+    SQLX_OFFLINE=true \
+        ASAN_OPTIONS="halt_on_error=1:abort_on_error=1:detect_leaks=1" \
+        RUSTFLAGS="-Zsanitizer=address" \
+        RUSTDOCFLAGS="-Zsanitizer=address" \
+        cargo +nightly nextest run -p tranquil-store --features tranquil-store/test-harness --target x86_64-unknown-linux-gnu
+
 test-unit:
     SQLX_OFFLINE=true cargo test --test dpop_unit --test validation_edge_cases --test scope_edge_cases