feat(test): test that websockets get closed

.config/nextest.toml

@@ -68,10 +68,6 @@ test-group = "serial-env-tests"
 filter = "package(tranquil-signal)"
 test-group = "serial-env-tests"
 
-[[profile.default.overrides]]
-filter = "package(tranquil-config)"
-test-group = "serial-env-tests"
-
 [[profile.default.overrides]]
 filter = "binary(whole_story)"
 test-group = "heavy-load-tests"
@@ -122,10 +118,6 @@ test-group = "serial-env-tests"
 filter = "package(tranquil-signal)"
 test-group = "serial-env-tests"
 
-[[profile.ci.overrides]]
-filter = "package(tranquil-config)"
-test-group = "serial-env-tests"
-
 [[profile.ci.overrides]]
 filter = "binary(whole_story)"
 test-group = "heavy-load-tests"

Cargo.toml

@@ -26,7 +26,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.6.0"
+version = "0.5.7"
 edition = "2024"
 license = "AGPL-3.0-or-later"
 
@@ -93,7 +93,6 @@ ipld-core = "0.4"
 iroh-car = "0.5"
 jacquard-common = { version = "0.9", features = ["crypto-k256"] }
 jacquard-repo = "0.9"
-lettre = { version = "0.11", default-features = false, features = ["builder", "smtp-transport", "tokio1", "tokio1-rustls-tls", "pool", "dkim", "tracing"] }
 jsonwebtoken = { version = "10.2", features = ["rust_crypto"] }
 k256 = { version = "0.13", features = ["ecdsa", "pem", "pkcs8"] }
 metrics = "0.24"
@@ -106,8 +105,6 @@ p384 = { version = "0.13", features = ["ecdsa"] }
 rand = "0.8"
 redis = { version = "1.0", features = ["tokio-comp", "connection-manager"] }
 regex = "1"
-rsa = "0.9"
-secrecy = { version = "0.10", features = ["serde"] }
 reqwest = { version = "0.12", default-features = false, features = ["json", "rustls-tls-webpki-roots", "http2", "charset", "macos-system-configuration"] }
 serde = { version = "1.0", features = ["derive"] }
 serde_bytes = "0.11"

crates/tranquil-comms/Cargo.toml

@@ -10,14 +10,7 @@ tranquil-signal = { workspace = true }
 
 async-trait = { workspace = true }
 base64 = { workspace = true }
-ed25519-dalek = { workspace = true }
-futures = { workspace = true }
-hickory-resolver = { workspace = true }
-lettre = { workspace = true }
-rand = { workspace = true }
 reqwest = { workspace = true }
-rsa = { workspace = true }
-secrecy = { workspace = true }
 serde_json = { workspace = true }
 sqlx = { workspace = true }
 thiserror = { workspace = true }
@@ -25,7 +18,3 @@ tokio = { workspace = true }
 tracing = { workspace = true }
 tranquil-db-traits = { workspace = true }
 uuid = { workspace = true }
-
-[dev-dependencies]
-chrono = { workspace = true }
-tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time", "io-util", "net"] }

crates/tranquil-comms/src/email/types.rs

@@ -1,291 +0,0 @@
-use std::path::PathBuf;
-
-#[derive(Debug, thiserror::Error)]
-pub enum ParseError {
-    #[error("empty value")]
-    Empty,
-    #[error("invalid character {0:?}")]
-    InvalidChar(char),
-    #[error("zero {0}")]
-    Zero(&'static str),
-    #[error("invalid TLS mode {0:?}")]
-    InvalidTlsMode(String),
-}
-
-fn parse_token(raw: &str, lowercase: bool, strip_trailing_dot: bool) -> Result<String, ParseError> {
-    let mut s = raw.trim();
-    if strip_trailing_dot {
-        s = s.trim_end_matches('.');
-    }
-    match s {
-        "" => Err(ParseError::Empty),
-        _ if s.chars().any(char::is_whitespace) => Err(ParseError::InvalidChar(' ')),
-        _ => Ok(match lowercase {
-            true => s.to_lowercase(),
-            false => s.to_string(),
-        }),
-    }
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
-pub struct SmtpHost(String);
-
-impl SmtpHost {
-    pub fn parse(raw: &str) -> Result<Self, ParseError> {
-        parse_token(raw, true, false).map(Self)
-    }
-
-    pub fn as_str(&self) -> &str {
-        &self.0
-    }
-}
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
-pub struct SmtpPort(u16);
-
-impl SmtpPort {
-    pub fn parse(raw: u16) -> Result<Self, ParseError> {
-        match raw {
-            0 => Err(ParseError::Zero("smtp port")),
-            n => Ok(Self(n)),
-        }
-    }
-
-    pub fn as_u16(self) -> u16 {
-        self.0
-    }
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
-pub struct HeloName(String);
-
-impl HeloName {
-    pub fn parse(raw: &str) -> Result<Self, ParseError> {
-        parse_token(raw, false, false).map(Self)
-    }
-
-    pub fn as_str(&self) -> &str {
-        &self.0
-    }
-
-    pub fn into_inner(self) -> String {
-        self.0
-    }
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
-pub struct EmailDomain(String);
-
-impl EmailDomain {
-    pub fn parse(raw: &str) -> Result<Self, ParseError> {
-        parse_token(raw, true, true).map(Self)
-    }
-
-    pub fn as_str(&self) -> &str {
-        &self.0
-    }
-
-    pub fn into_inner(self) -> String {
-        self.0
-    }
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
-pub struct MxHost(String);
-
-impl MxHost {
-    pub fn parse(raw: &str) -> Result<Self, ParseError> {
-        parse_token(raw, true, true).map(Self)
-    }
-
-    pub fn as_str(&self) -> &str {
-        &self.0
-    }
-}
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
-pub struct MxPriority(u16);
-
-impl MxPriority {
-    pub fn new(value: u16) -> Self {
-        Self(value)
-    }
-
-    pub fn as_u16(self) -> u16 {
-        self.0
-    }
-}
-
-#[derive(Debug, Clone)]
-pub struct MxRecord {
-    pub priority: MxPriority,
-    pub host: MxHost,
-}
-
-#[derive(Debug, Clone, PartialEq, Eq, Hash)]
-pub struct DkimSelector(String);
-
-impl DkimSelector {
-    pub fn parse(raw: &str) -> Result<Self, ParseError> {
-        let trimmed = raw.trim();
-        let valid = !trimmed.is_empty() && trimmed.split('.').all(valid_subdomain);
-        match valid {
-            true => Ok(Self(trimmed.to_string())),
-            false => Err(ParseError::InvalidChar('?')),
-        }
-    }
-
-    pub fn into_inner(self) -> String {
-        self.0
-    }
-}
-
-fn valid_subdomain(seg: &str) -> bool {
-    let starts_alnum = seg
-        .chars()
-        .next()
-        .is_some_and(|c| c.is_ascii_alphanumeric());
-    let ends_alnum = seg
-        .chars()
-        .next_back()
-        .is_some_and(|c| c.is_ascii_alphanumeric());
-    let body_ok = seg.chars().all(|c| c.is_ascii_alphanumeric() || c == '-');
-    starts_alnum && ends_alnum && body_ok
-}
-
-#[derive(Debug, Clone)]
-pub struct DkimKeyPath(PathBuf);
-
-impl DkimKeyPath {
-    pub fn parse(raw: &str) -> Result<Self, ParseError> {
-        let trimmed = raw.trim();
-        match trimmed.is_empty() {
-            true => Err(ParseError::Empty),
-            false => Ok(Self(PathBuf::from(trimmed))),
-        }
-    }
-
-    pub fn as_path(&self) -> &std::path::Path {
-        &self.0
-    }
-}
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct SmtpUsername(String);
-
-impl SmtpUsername {
-    pub fn parse(raw: &str) -> Result<Self, ParseError> {
-        match raw.is_empty() {
-            true => Err(ParseError::Empty),
-            false => Ok(Self(raw.to_string())),
-        }
-    }
-
-    pub fn into_inner(self) -> String {
-        self.0
-    }
-}
-
-#[derive(Clone)]
-pub struct SmtpPassword(secrecy::SecretString);
-
-impl SmtpPassword {
-    pub fn parse(raw: &str) -> Result<Self, ParseError> {
-        match raw.is_empty() {
-            true => Err(ParseError::Empty),
-            false => Ok(Self(secrecy::SecretString::from(raw.to_string()))),
-        }
-    }
-
-    pub fn expose(&self) -> &str {
-        use secrecy::ExposeSecret;
-        self.0.expose_secret()
-    }
-}
-
-impl std::fmt::Debug for SmtpPassword {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.write_str("SmtpPassword(***)")
-    }
-}
-
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
-pub enum TlsMode {
-    Implicit,
-    Starttls,
-    None,
-}
-
-impl TlsMode {
-    pub fn parse(raw: &str) -> Result<Self, ParseError> {
-        match raw.to_ascii_lowercase().as_str() {
-            "implicit" => Ok(Self::Implicit),
-            "starttls" => Ok(Self::Starttls),
-            "none" => Ok(Self::None),
-            other => Err(ParseError::InvalidTlsMode(other.to_string())),
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn smtp_host_lowercases_and_trims() {
-        let h = SmtpHost::parse("  SMTP.NEL.PET  ").unwrap();
-        assert_eq!(h.as_str(), "smtp.nel.pet");
-    }
-
-    #[test]
-    fn smtp_host_rejects_whitespace() {
-        assert!(SmtpHost::parse("a b").is_err());
-    }
-
-    #[test]
-    fn smtp_host_rejects_empty() {
-        assert!(SmtpHost::parse("").is_err());
-        assert!(SmtpHost::parse("   ").is_err());
-    }
-
-    #[test]
-    fn smtp_port_rejects_zero() {
-        assert!(SmtpPort::parse(0).is_err());
-        assert_eq!(SmtpPort::parse(587).unwrap().as_u16(), 587);
-    }
-
-    #[test]
-    fn email_domain_strips_trailing_dot() {
-        assert_eq!(EmailDomain::parse("Nel.pet.").unwrap().as_str(), "nel.pet");
-    }
-
-    #[test]
-    fn dkim_selector_validates() {
-        assert!(DkimSelector::parse("default").is_ok());
-        assert!(DkimSelector::parse("s1.nel.pet").is_ok());
-        assert!(DkimSelector::parse("s2024-q1").is_ok());
-        assert!(DkimSelector::parse("mailo-2024.nel.pet").is_ok());
-        assert!(DkimSelector::parse("a-b").is_ok());
-        assert!(DkimSelector::parse("").is_err());
-        assert!(DkimSelector::parse("a..b").is_err());
-        assert!(DkimSelector::parse("-leading").is_err());
-        assert!(DkimSelector::parse("trailing-").is_err());
-        assert!(DkimSelector::parse("s_under").is_err());
-    }
-
-    #[test]
-    fn tls_mode_parses_known_modes() {
-        assert_eq!(TlsMode::parse("STARTTLS").unwrap(), TlsMode::Starttls);
-        assert_eq!(TlsMode::parse("implicit").unwrap(), TlsMode::Implicit);
-        assert_eq!(TlsMode::parse("none").unwrap(), TlsMode::None);
-        assert!(TlsMode::parse("garbage").is_err());
-    }
-
-    #[test]
-    fn smtp_password_redacts_in_debug() {
-        let p = SmtpPassword::parse("hunter2").unwrap();
-        let dbg = format!("{:?}", p);
-        assert_eq!(dbg, "SmtpPassword(***)");
-        assert!(!dbg.contains("hunter2"));
-    }
-}

crates/tranquil-config/src/lib.rs

@@ -5,14 +5,6 @@ use std::sync::OnceLock;
 
 static CONFIG: OnceLock<TranquilConfig> = OnceLock::new();
 
-const REMOVED_ENV_VARS: &[(&str, &str)] = &[(
-    "SENDMAIL_PATH",
-    "the sendmail-binary transport was replaced with native SMTP. \
-     Configure MAIL_SMARTHOST_HOST for relay delivery, or leave it unset to \
-     deliver directly via recipient MX records. See example.toml for the full \
-     MAIL_* surface.",
-)];
-
 /// Errors discovered during configuration validation.
 #[derive(Debug)]
 pub struct ConfigError {
@@ -170,14 +162,6 @@ impl TranquilConfig {
     pub fn validate(&self, ignore_secrets: bool) -> Result<(), ConfigError> {
         let mut errors = Vec::new();
 
-        // -- removed config ---------------------------------------------------
-        errors.extend(
-            REMOVED_ENV_VARS
-                .iter()
-                .filter(|(var, _)| std::env::var_os(var).is_some())
-                .map(|(var, guidance)| format!("{var} is no longer supported: {guidance}")),
-        );
-
         // -- secrets ----------------------------------------------------------
         if !ignore_secrets && !self.secrets.allow_insecure && !cfg!(test) {
             if let Some(ref s) = self.secrets.jwt_secret {
@@ -226,85 +210,6 @@ impl TranquilConfig {
             }
         }
 
-        // -- email smarthost --------------------------------------------------
-        match self.email.smarthost.tls.to_ascii_lowercase().as_str() {
-            "implicit" | "starttls" => {}
-            "none" => {
-                if self.email.smarthost.password.is_some() {
-                    errors.push(
-                        "email.smarthost.tls = \"none\" with email.smarthost.password set \
-                         would transmit credentials in plaintext; use \"starttls\" or \"implicit\""
-                            .to_string(),
-                    );
-                }
-            }
-            other => errors.push(format!(
-                "email.smarthost.tls must be \"implicit\", \"starttls\", or \"none\", got \"{other}\""
-            )),
-        }
-
-        let smarthost_host_set = self
-            .email
-            .smarthost
-            .host
-            .as_deref()
-            .is_some_and(|h| !h.is_empty());
-        let username_set = self.email.smarthost.username.is_some();
-        let password_set = self.email.smarthost.password.is_some();
-        if !smarthost_host_set && (username_set || password_set) {
-            errors.push(
-                "email.smarthost.username or email.smarthost.password is set but \
-                 email.smarthost.host is empty; credentials would be silently ignored"
-                    .to_string(),
-            );
-        }
-        if smarthost_host_set && username_set != password_set {
-            errors.push(
-                "email.smarthost.username and email.smarthost.password must both be set or \
-                 both unset; otherwise authentication would silently degrade to anonymous"
-                    .to_string(),
-            );
-        }
-
-        if self.email.smarthost.command_timeout_secs == 0 {
-            errors.push("email.smarthost.command_timeout_secs must be at least 1".to_string());
-        }
-        if self.email.smarthost.total_timeout_secs == 0 {
-            errors.push("email.smarthost.total_timeout_secs must be at least 1".to_string());
-        }
-        if self.email.smarthost.pool_size == 0 {
-            errors.push("email.smarthost.pool_size must be at least 1".to_string());
-        }
-
-        if self.email.direct_mx.max_concurrent_sends == 0 {
-            errors.push("email.direct_mx.max_concurrent_sends must be at least 1".to_string());
-        }
-        if self.email.direct_mx.command_timeout_secs == 0 {
-            errors.push("email.direct_mx.command_timeout_secs must be at least 1".to_string());
-        }
-        if self.email.direct_mx.total_timeout_secs == 0 {
-            errors.push("email.direct_mx.total_timeout_secs must be at least 1".to_string());
-        }
-
-        let dkim_set = self.email.dkim.selector.is_some()
-            || self.email.dkim.domain.is_some()
-            || self.email.dkim.private_key_path.is_some();
-        if dkim_set {
-            if self.email.dkim.selector.is_none() {
-                errors
-                    .push("email.dkim.selector is required when any DKIM field is set".to_string());
-            }
-            if self.email.dkim.domain.is_none() {
-                errors.push("email.dkim.domain is required when any DKIM field is set".to_string());
-            }
-            if self.email.dkim.private_key_path.is_none() {
-                errors.push(
-                    "email.dkim.private_key_path is required when any DKIM field is set"
-                        .to_string(),
-                );
-            }
-        }
-
         // -- telegram ---------------------------------------------------------
         if self.telegram.bot_token.is_some() && self.telegram.webhook_secret.is_none() {
             errors.push(
@@ -849,98 +754,9 @@ pub struct EmailConfig {
     #[config(env = "MAIL_FROM_NAME", default = "Tranquil PDS")]
     pub from_name: String,
 
-    /// HELO/EHLO name announced to remote SMTP servers. Applies to both
-    /// smarthost and direct-MX modes. Defaults to the server hostname.
-    #[config(env = "MAIL_HELO_NAME")]
-    pub helo_name: Option<String>,
-
-    #[config(nested)]
-    pub smarthost: SmarthostConfig,
-
-    #[config(nested)]
-    pub direct_mx: DirectMxConfig,
-
-    #[config(nested)]
-    pub dkim: DkimConfig,
-}
-
-#[derive(Debug, Config)]
-pub struct SmarthostConfig {
-    /// SMTP relay host. When set, mail is delivered through this host
-    /// instead of resolving recipient MX records directly.
-    #[config(env = "MAIL_SMARTHOST_HOST")]
-    pub host: Option<String>,
-
-    /// SMTP relay port.
-    #[config(env = "MAIL_SMARTHOST_PORT", default = 587)]
-    pub port: u16,
-
-    /// SMTP authentication username.
-    #[config(env = "MAIL_SMARTHOST_USERNAME")]
-    pub username: Option<String>,
-
-    /// SMTP authentication password.
-    #[config(env = "MAIL_SMARTHOST_PASSWORD")]
-    pub password: Option<String>,
-
-    /// TLS mode. Valid values: "implicit", "starttls", "none". Setting "none"
-    /// alongside a password is rejected at startup to prevent transmitting
-    /// credentials in plaintext.
-    #[config(env = "MAIL_SMARTHOST_TLS", default = "starttls")]
-    pub tls: String,
-
-    /// Max size of the connection pool.
-    #[config(env = "MAIL_SMARTHOST_POOL_SIZE", default = 4)]
-    pub pool_size: u32,
-
-    /// Per-command SMTP timeout in seconds. Bounds the security handshake.
-    #[config(env = "MAIL_SMARTHOST_COMMAND_TIMEOUT_SECS", default = 30)]
-    pub command_timeout_secs: u64,
-
-    /// Total per-message timeout in seconds. Wraps the entire send so a
-    /// stuck relay cannot stall the comms queue.
-    #[config(env = "MAIL_SMARTHOST_TOTAL_TIMEOUT_SECS", default = 60)]
-    pub total_timeout_secs: u64,
-}
-
-#[derive(Debug, Config)]
-pub struct DirectMxConfig {
-    /// Per-command SMTP timeout in seconds.
-    #[config(env = "MAIL_COMMAND_TIMEOUT_SECS", default = 30)]
-    pub command_timeout_secs: u64,
-
-    /// Total per-message timeout across all MX attempts in seconds.
-    #[config(env = "MAIL_TOTAL_TIMEOUT_SECS", default = 60)]
-    pub total_timeout_secs: u64,
-
-    /// Max number of concurrent direct-MX sends. Limits the load placed
-    /// on any single recipient MX during a backlog drain.
-    #[config(env = "MAIL_MAX_CONCURRENT_SENDS", default = 8)]
-    pub max_concurrent_sends: usize,
-
-    /// Require STARTTLS on every MX hop. When false, TLS is
-    /// attempted opportunistically and the session falls back to plaintext
-    /// if the remote does not advertise STARTTLS. Set true to refuse
-    /// plaintext delivery, at the cost of failing sends to MX hosts that
-    /// do not support TLS.
-    #[config(env = "MAIL_REQUIRE_TLS", default = false)]
-    pub require_tls: bool,
-}
-
-#[derive(Debug, Config)]
-pub struct DkimConfig {
-    /// DKIM selector. When unset, outgoing mail is not signed.
-    #[config(env = "MAIL_DKIM_SELECTOR")]
-    pub selector: Option<String>,
-
-    /// DKIM signing domain.
-    #[config(env = "MAIL_DKIM_DOMAIN")]
-    pub domain: Option<String>,
-
-    /// Path to the DKIM private key in PEM format. Supports RSA and
-    /// Ed25519 keys.
-    #[config(env = "MAIL_DKIM_KEY_PATH")]
-    pub private_key_path: Option<String>,
+    /// Path to the `sendmail` binary.
+    #[config(env = "SENDMAIL_PATH", default = "/usr/sbin/sendmail")]
+    pub sendmail_path: String,
 }
 
 #[derive(Debug, Config)]
@@ -1380,64 +1196,3 @@ pub struct TranquilStoreConfig {
 pub fn template() -> String {
     confique::toml::template::<TranquilConfig>(confique::toml::FormatOptions::default())
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    fn seed_required_env() {
-        let required = [
-            ("PDS_HOSTNAME", "test.local"),
-            ("DATABASE_URL", "postgres://localhost/test"),
-            ("TRANQUIL_PDS_ALLOW_INSECURE_SECRETS", "1"),
-            ("INVITE_CODE_REQUIRED", "false"),
-            ("ENABLE_PDS_HOSTED_DID_WEB", "true"),
-            ("TRANQUIL_LEXICON_OFFLINE", "1"),
-        ];
-        required
-            .iter()
-            .filter(|(k, _)| std::env::var_os(k).is_none())
-            .for_each(|(k, v)| unsafe { std::env::set_var(k, v) });
-    }
-
-    #[test]
-    fn serial_validate_rejects_legacy_sendmail_path() {
-        seed_required_env();
-        unsafe { std::env::set_var("SENDMAIL_PATH", "/usr/sbin/sendmail") };
-        let config = TranquilConfig::builder()
-            .env()
-            .load()
-            .expect("load fresh config");
-        let result = config.validate(true);
-        unsafe { std::env::remove_var("SENDMAIL_PATH") };
-
-        let err = result.expect_err("validate must reject SENDMAIL_PATH");
-        let mentions_sendmail = err.errors.iter().any(|e| e.contains("SENDMAIL_PATH"));
-        assert!(
-            mentions_sendmail,
-            "errors did not mention SENDMAIL_PATH: {:?}",
-            err.errors
-        );
-    }
-
-    #[test]
-    fn serial_validate_passes_when_no_legacy_env_set() {
-        seed_required_env();
-        unsafe { std::env::remove_var("SENDMAIL_PATH") };
-        let config = TranquilConfig::builder()
-            .env()
-            .load()
-            .expect("load fresh config");
-        let result = config.validate(true);
-        let leaked_legacy = result
-            .as_ref()
-            .err()
-            .map(|e| e.errors.iter().any(|s| s.contains("SENDMAIL_PATH")))
-            .unwrap_or(false);
-        assert!(
-            !leaked_legacy,
-            "validate spuriously flagged SENDMAIL_PATH when unset: {:?}",
-            result
-        );
-    }
-}

crates/tranquil-pds/src/state.rs

@@ -523,16 +523,13 @@ fn wire_tranquil_store(
     let metastore =
         Metastore::open(&metastore_dir, metastore_config).expect("failed to open metastore");
 
-    let blockstore = TranquilBlockStore::open_with_retry(
-        BlockStoreConfig {
-            data_dir: blockstore_data_dir,
-            index_dir: blockstore_index_dir,
-            max_file_size: store_cfg.max_blockstore_file_size,
-            group_commit: Default::default(),
-            shard_count: tranquil_store::blockstore::DEFAULT_SHARD_COUNT,
-        },
-        tranquil_store::blockstore::OpenRetryPolicy::default(),
-    )
+    let blockstore = TranquilBlockStore::open(BlockStoreConfig {
+        data_dir: blockstore_data_dir,
+        index_dir: blockstore_index_dir,
+        max_file_size: store_cfg.max_blockstore_file_size,
+        group_commit: Default::default(),
+        shard_count: tranquil_store::blockstore::DEFAULT_SHARD_COUNT,
+    })
     .expect("failed to open blockstore");
 
     let event_log = EventLog::open(

crates/tranquil-store/src/bin/tranquil_gauntlet.rs

@@ -64,16 +64,6 @@ enum Cmd {
         #[arg(long)]
         config: Option<PathBuf>,
 
-        /// Tempdir parent for `IoBackend::Real` seeds only - ignored for
-        /// flaky-mount and simulated backends. Repeatable; each rayon worker
-        /// thread is pinned to one root so concurrent seeds on different
-        /// threads land on different mounts. Default `/tmp`. Also reads
-        /// colon-separated paths from `GAUNTLET_SCRATCH_ROOTS`. Set
-        /// `RAYON_NUM_THREADS=N` to cap workers; for full distribution pass
-        /// one root per worker.
-        #[arg(long)]
-        scratch_root: Vec<PathBuf>,
-
         /// Skip shrinking when dumping regressions.
         #[arg(long)]
         no_shrink: bool,
@@ -105,13 +95,6 @@ enum Cmd {
         #[arg(long)]
         dump_regressions: Option<PathBuf>,
 
-        /// Same as `farm --scratch-root`: tempdir parent for
-        /// `IoBackend::Real` seeds only, pinned per worker thread. Ignored
-        /// for flaky-mount and simulated backends. Repeatable; reads
-        /// colon-separated paths from `GAUNTLET_SCRATCH_ROOTS`.
-        #[arg(long)]
-        scratch_root: Vec<PathBuf>,
-
         /// Skip shrinking when dumping regressions.
         #[arg(long)]
         no_shrink: bool,
@@ -176,8 +159,6 @@ struct ConfigFile {
     #[serde(default)]
     dump_regressions: Option<PathBuf>,
     #[serde(default)]
-    scratch_roots: Vec<PathBuf>,
-    #[serde(default)]
     overrides: ConfigOverrides,
 }
 
@@ -197,8 +178,6 @@ struct SweepConfigFile {
     #[serde(default)]
     dump_regressions: Option<PathBuf>,
     #[serde(default)]
-    scratch_roots: Vec<PathBuf>,
-    #[serde(default)]
     base_overrides: ConfigOverrides,
     #[serde(default)]
     axes: SweepAxes,
@@ -426,7 +405,6 @@ struct FarmPlan {
     seeds: u64,
     hours: Option<f64>,
     dump_regressions: Option<PathBuf>,
-    scratch_roots: Vec<PathBuf>,
     overrides: ConfigOverrides,
     shrink: bool,
     shrink_budget: usize,
@@ -440,7 +418,6 @@ fn resolve_farm(
     hours: Option<f64>,
     dump_regressions: Option<PathBuf>,
     config: Option<PathBuf>,
-    scratch_root: Vec<PathBuf>,
     shrink: bool,
     shrink_budget: usize,
 ) -> Result<FarmPlan, String> {
@@ -466,11 +443,6 @@ fn resolve_farm(
     }
     let dump_regressions =
         dump_regressions.or_else(|| file.as_ref().and_then(|f| f.dump_regressions.clone()));
-    let file_scratch_roots = file
-        .as_ref()
-        .map(|f| f.scratch_roots.clone())
-        .unwrap_or_default();
-    let scratch_roots = resolve_scratch_roots(scratch_root, file_scratch_roots)?;
     let overrides = file.map(|f| f.overrides).unwrap_or_default();
     Ok(FarmPlan {
         scenario,
@@ -478,50 +450,12 @@ fn resolve_farm(
         seeds,
         hours,
         dump_regressions,
-        scratch_roots,
         overrides,
         shrink,
         shrink_budget,
     })
 }
 
-const SCRATCH_ROOTS_ENV: &str = "GAUNTLET_SCRATCH_ROOTS";
-
-fn resolve_scratch_roots(
-    cli: Vec<PathBuf>,
-    config_file: Vec<PathBuf>,
-) -> Result<Vec<PathBuf>, String> {
-    let env_roots: Vec<PathBuf> = std::env::var(SCRATCH_ROOTS_ENV)
-        .ok()
-        .filter(|s| !s.is_empty())
-        .map(|s| s.split(':').map(PathBuf::from).collect())
-        .unwrap_or_default();
-    let candidate: Vec<PathBuf> = if !cli.is_empty() {
-        cli
-    } else if !config_file.is_empty() {
-        config_file
-    } else {
-        env_roots
-    };
-    candidate
-        .into_iter()
-        .map(|p| validate_scratch_root(&p).map(|_| p))
-        .collect()
-}
-
-fn validate_scratch_root(path: &Path) -> Result<(), String> {
-    match path.metadata() {
-        Ok(m) if m.is_dir() => {}
-        Ok(_) => return Err(format!("scratch root not a directory: {}", path.display())),
-        Err(e) => return Err(format!("scratch root {}: {e}", path.display())),
-    }
-    tempfile::Builder::new()
-        .prefix(".tranquil-gauntlet-probe-")
-        .tempfile_in(path)
-        .map(|_| ())
-        .map_err(|e| format!("scratch root {} not writable: {e}", path.display()))
-}
-
 fn validate_hours(h: f64) -> Result<(), String> {
     if !h.is_finite() || h <= 0.0 {
         return Err(format!("invalid --hours={h}: must be positive and finite"));
@@ -630,7 +564,6 @@ fn main() -> ExitCode {
             hours,
             dump_regressions,
             config,
-            scratch_root,
             no_shrink,
             shrink_budget,
         } => {
@@ -641,7 +574,6 @@ fn main() -> ExitCode {
                 hours,
                 dump_regressions,
                 config,
-                scratch_root,
                 !no_shrink,
                 shrink_budget,
             ) {
@@ -693,7 +625,6 @@ fn main() -> ExitCode {
             seed_start,
             seeds,
             dump_regressions,
-            scratch_root,
             no_shrink,
             shrink_budget,
             max_runs,
@@ -703,7 +634,6 @@ fn main() -> ExitCode {
                 seed_start,
                 seeds,
                 dump_regressions,
-                scratch_root,
                 !no_shrink,
                 shrink_budget,
                 max_runs,
@@ -729,20 +659,17 @@ struct SweepPlan {
     seed_start: u64,
     seeds: u64,
     dump_regressions: Option<PathBuf>,
-    scratch_roots: Vec<PathBuf>,
     shrink: bool,
     shrink_budget: usize,
     base_overrides: ConfigOverrides,
     axes: Vec<SweepAxisValues>,
 }
 
-#[allow(clippy::too_many_arguments)]
 fn resolve_sweep(
     config: PathBuf,
     seed_start: Option<u64>,
     seeds: Option<u64>,
     dump_regressions: Option<PathBuf>,
-    scratch_root: Vec<PathBuf>,
     shrink: bool,
     shrink_budget: usize,
     max_runs: u64,
@@ -760,7 +687,6 @@ fn resolve_sweep(
         return Err("--shrink-budget must be greater than zero".to_string());
     }
     let dump_regressions = dump_regressions.or(file.dump_regressions.clone());
-    let scratch_roots = resolve_scratch_roots(scratch_root, file.scratch_roots.clone())?;
     let axes = file.axes.axis_values();
     if axes.is_empty() {
         return Err("sweep produced no combinations".to_string());
@@ -779,7 +705,6 @@ fn resolve_sweep(
         seed_start,
         seeds,
         dump_regressions,
-        scratch_roots,
         shrink,
         shrink_budget,
         base_overrides: file.base_overrides,
@@ -824,7 +749,6 @@ fn run_sweep(plan: SweepPlan, rt: &Runtime, interrupt: Arc<AtomicBool>) -> ExitC
         seed_start,
         seeds,
         dump_regressions,
-        scratch_roots,
         shrink,
         shrink_budget,
         base_overrides,
@@ -858,13 +782,12 @@ fn run_sweep(plan: SweepPlan, rt: &Runtime, interrupt: Arc<AtomicBool>) -> ExitC
         axis_values.apply_to(&mut overrides);
         let combo_start = Instant::now();
         let overrides_for_farm = overrides.clone();
-        let reports = farm::run_many_timed_with_scratch_roots(
+        let reports = farm::run_many_timed(
             move |s| {
                 let mut cfg = config_for(scenario, s);
                 overrides_for_farm.apply_to(&mut cfg);
                 cfg
             },
-            &scratch_roots,
             (seed_start..end).map(Seed),
         );
         let combo_wall = combo_start.elapsed();
@@ -917,7 +840,6 @@ fn run_farm(plan: FarmPlan, rt: &Runtime, interrupt: Arc<AtomicBool>) -> ExitCod
         seeds,
         hours,
         dump_regressions,
-        scratch_roots,
         overrides,
         shrink,
         shrink_budget,
@@ -949,13 +871,12 @@ fn run_farm(plan: FarmPlan, rt: &Runtime, interrupt: Arc<AtomicBool>) -> ExitCod
         };
         let overrides_ref = &overrides;
         let batch_start = Instant::now();
-        let reports = farm::run_many_timed_with_scratch_roots(
+        let reports = farm::run_many_timed(
             |s| {
                 let mut cfg = config_for(scenario, s);
                 overrides_ref.apply_to(&mut cfg);
                 cfg
             },
-            &scratch_roots,
             (next_seed..end).map(Seed),
         );
         let batch_wall = batch_start.elapsed();

crates/tranquil-store/src/blockstore/compaction.rs

@@ -222,8 +222,7 @@ fn stream_compact<S: StorageIO>(
                 .io()
                 .sync_dir(manager.data_dir())
                 .map_err(CompactionError::from)
-        })
-        .and_then(|()| manager.io().barrier().map_err(CompactionError::from));
+        });
 
     let _ = manager.io().close(hint_fd);
 

crates/tranquil-store/src/blockstore/group_commit.rs

@@ -1343,10 +1343,6 @@ fn process_batch<S: StorageIO>(
         )
         .map_err(|e| rollback_on_err(CommitError::from(e)))?;
     hint_writer.sync().map_err(|e| rollback_on_err(e.into()))?;
-    manager
-        .io()
-        .barrier()
-        .map_err(|e| rollback_on_err(e.into()))?;
     let sync_nanos = t.elapsed().as_nanos() as u64;
 
     if !rotations.is_empty() {

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

@@ -27,7 +27,7 @@ pub use hint::{
 pub use manager::{CachedHandle, DEFAULT_MAX_FILE_SIZE, DataFileManager};
 pub use reader::{BlockStoreReader, ReadError};
 pub use store::QuiesceGuard;
-pub use store::{BlockStoreConfig, DEFAULT_SHARD_COUNT, OpenRetryPolicy, TranquilBlockStore};
+pub use store::{BlockStoreConfig, DEFAULT_SHARD_COUNT, TranquilBlockStore};
 pub use types::{
     BlockLength, BlockLocation, BlockOffset, BlockstoreSnapshot, CidBytes, CollectionResult,
     CommitEpoch, CompactionResult, DataFileId, EpochCounter, HintOffset, IndexEntry, LivenessInfo,

crates/tranquil-store/src/blockstore/store.rs

@@ -1,9 +1,7 @@
 use std::collections::HashMap;
 use std::io;
-use std::num::NonZeroU8;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
-use std::time::Duration;
 
 use bytes::Bytes;
 use cid::Cid;
@@ -152,24 +150,6 @@ impl Drop for WriterHandle {
     }
 }
 
-#[derive(Clone, Copy, Debug)]
-pub struct OpenRetryPolicy {
-    pub max_attempts: NonZeroU8,
-    pub initial_backoff: Duration,
-    pub max_backoff: Duration,
-}
-
-impl Default for OpenRetryPolicy {
-    fn default() -> Self {
-        const DEFAULT_MAX_ATTEMPTS: NonZeroU8 = NonZeroU8::new(5).unwrap();
-        Self {
-            max_attempts: DEFAULT_MAX_ATTEMPTS,
-            initial_backoff: Duration::from_millis(100),
-            max_backoff: Duration::from_secs(2),
-        }
-    }
-}
-
 impl TranquilBlockStore<RealIO> {
     pub fn open(config: BlockStoreConfig) -> Result<Self, RepoError> {
         Self::open_with_hook(config, None)
@@ -181,50 +161,6 @@ impl TranquilBlockStore<RealIO> {
     ) -> Result<Self, RepoError> {
         Self::open_with_io_hook(config, RealIO::new, post_sync_hook)
     }
-
-    pub fn open_with_retry(
-        config: BlockStoreConfig,
-        policy: OpenRetryPolicy,
-    ) -> Result<Self, RepoError> {
-        retry_with_backoff(policy, &mut |_| Self::open(config.clone()))
-    }
-}
-
-fn retry_with_backoff<T, F>(policy: OpenRetryPolicy, op: &mut F) -> Result<T, RepoError>
-where
-    F: FnMut(u8) -> Result<T, RepoError>,
-{
-    retry_attempt(policy, op, 0, policy.initial_backoff)
-}
-
-fn retry_attempt<T, F>(
-    policy: OpenRetryPolicy,
-    op: &mut F,
-    attempt: u8,
-    backoff: Duration,
-) -> Result<T, RepoError>
-where
-    F: FnMut(u8) -> Result<T, RepoError>,
-{
-    match op(attempt) {
-        Ok(t) => Ok(t),
-        Err(e) if attempt + 1 >= policy.max_attempts.get() => Err(e),
-        Err(e) => {
-            tracing::warn!(
-                attempt,
-                error = %e,
-                backoff_ms = u64::try_from(backoff.as_millis()).unwrap_or(u64::MAX),
-                "blockstore open failed, retrying"
-            );
-            std::thread::sleep(backoff);
-            retry_attempt(
-                policy,
-                op,
-                attempt + 1,
-                (backoff * 2).min(policy.max_backoff),
-            )
-        }
-    }
 }
 
 impl<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
@@ -395,7 +331,15 @@ impl<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
         let scan_pos = &mut { start_offset };
         let (scanned_entries, last_valid_end) = std::iter::from_fn(|| {
             match super::data_file::decode_block_record(io, fd, *scan_pos, file_size) {
-                Err(e) => Some(Err(e)),
+                Err(e) => {
+                    tracing::warn!(
+                        file_id = %file_id,
+                        offset = scan_pos.raw(),
+                        error = %e,
+                        "IO error during recovery scan, stopping"
+                    );
+                    None
+                }
                 Ok(None) => None,
                 Ok(Some(ReadBlockRecord::Valid {
                     offset,
@@ -410,7 +354,7 @@ impl<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
                     let record_size = BLOCK_RECORD_OVERHEAD as u64 + u64::from(raw_len);
                     let new_end = offset.advance(record_size);
                     *scan_pos = new_end;
-                    Some(Ok((
+                    Some((
                         cid_bytes,
                         BlockLocation {
                             file_id,
@@ -418,30 +362,20 @@ impl<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
                             length,
                         },
                         new_end,
-                    )))
+                    ))
                 }
                 Ok(Some(ReadBlockRecord::Corrupted { .. } | ReadBlockRecord::Truncated { .. })) => {
                     None
                 }
             }
         })
-        .try_fold(
+        .fold(
             (Vec::new(), start_offset),
-            |(mut entries, _), item: io::Result<_>| {
-                let (cid, loc, new_end) = item?;
+            |(mut entries, _), (cid, loc, new_end)| {
                 entries.push((cid, loc));
-                Ok::<_, io::Error>((entries, new_end))
+                (entries, new_end)
             },
-        )
-        .map_err(|e| {
-            tracing::warn!(
-                file_id = %file_id,
-                offset = scan_pos.raw(),
-                error = %e,
-                "IO error during recovery scan, aborting to preserve durable tail"
-            );
-            RepoError::storage(e)
-        })?;
+        );
 
         if file_size > last_valid_end.raw() {
             tracing::info!(
@@ -779,213 +713,3 @@ impl<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
         Ok(self.index.get(&cid_bytes).map(|entry| entry.refcount.raw()))
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    use std::collections::HashMap;
-    use std::sync::Mutex;
-    use std::sync::atomic::{AtomicBool, Ordering};
-
-    use crate::blockstore::data_file::{
-        BLOCK_FORMAT_VERSION, BLOCK_HEADER_SIZE, BLOCK_MAGIC, encode_block_record,
-    };
-    use crate::blockstore::manager::DATA_FILE_EXTENSION;
-    use crate::io::FileId;
-
-    struct EioOnReadAtRange {
-        inner: RealIO,
-        target_path: PathBuf,
-        target_min: u64,
-        target_max: u64,
-        fired: AtomicBool,
-        fd_paths: Mutex<HashMap<FileId, PathBuf>>,
-    }
-
-    impl StorageIO for EioOnReadAtRange {
-        fn open(&self, path: &Path, opts: OpenOptions) -> io::Result<FileId> {
-            let fd = self.inner.open(path, opts)?;
-            self.fd_paths
-                .lock()
-                .unwrap()
-                .insert(fd, path.to_path_buf());
-            Ok(fd)
-        }
-
-        fn close(&self, fd: FileId) -> io::Result<()> {
-            self.fd_paths.lock().unwrap().remove(&fd);
-            self.inner.close(fd)
-        }
-
-        fn read_at(&self, fd: FileId, offset: u64, buf: &mut [u8]) -> io::Result<usize> {
-            let path_match = self.fd_paths.lock().unwrap().get(&fd).cloned();
-            let in_target_range = path_match.as_ref() == Some(&self.target_path)
-                && offset >= self.target_min
-                && offset <= self.target_max;
-            if in_target_range && !self.fired.swap(true, Ordering::SeqCst) {
-                return Err(io::Error::other("simulated EIO on read"));
-            }
-            self.inner.read_at(fd, offset, buf)
-        }
-
-        fn write_at(&self, fd: FileId, offset: u64, buf: &[u8]) -> io::Result<usize> {
-            self.inner.write_at(fd, offset, buf)
-        }
-
-        fn sync(&self, fd: FileId) -> io::Result<()> {
-            self.inner.sync(fd)
-        }
-
-        fn file_size(&self, fd: FileId) -> io::Result<u64> {
-            self.inner.file_size(fd)
-        }
-
-        fn truncate(&self, fd: FileId, size: u64) -> io::Result<()> {
-            self.inner.truncate(fd, size)
-        }
-
-        fn rename(&self, from: &Path, to: &Path) -> io::Result<()> {
-            self.inner.rename(from, to)
-        }
-
-        fn delete(&self, path: &Path) -> io::Result<()> {
-            self.inner.delete(path)
-        }
-
-        fn mkdir(&self, path: &Path) -> io::Result<()> {
-            self.inner.mkdir(path)
-        }
-
-        fn sync_dir(&self, path: &Path) -> io::Result<()> {
-            self.inner.sync_dir(path)
-        }
-
-        fn list_dir(&self, path: &Path) -> io::Result<Vec<PathBuf>> {
-            self.inner.list_dir(path)
-        }
-    }
-
-    #[test]
-    fn scan_and_index_does_not_truncate_acked_block_on_transient_eio() {
-        let tmp = tempfile::TempDir::new().unwrap();
-        let data_dir = tmp.path().join("data");
-        let index_dir = tmp.path().join("index");
-        std::fs::create_dir_all(&data_dir).unwrap();
-        std::fs::create_dir_all(&index_dir).unwrap();
-
-        let file_id = DataFileId::new(0);
-        let file_path = data_dir.join(format!("{file_id}.{DATA_FILE_EXTENSION}"));
-
-        let setup = RealIO::new();
-        let fd = setup.open(&file_path, OpenOptions::read_write()).unwrap();
-        let mut header = [0u8; BLOCK_HEADER_SIZE];
-        header[..4].copy_from_slice(&BLOCK_MAGIC);
-        header[4] = BLOCK_FORMAT_VERSION;
-        setup.write_all_at(fd, 0, &header).unwrap();
-
-        let cid_a = [0xAAu8; CID_SIZE];
-        let data_a = vec![1u8; 64];
-        let block_a_offset = BlockOffset::new(BLOCK_HEADER_SIZE as u64);
-        let len_a =
-            encode_block_record(&setup, fd, block_a_offset, &cid_a, &data_a).unwrap();
-
-        let block_b_offset_raw = BLOCK_HEADER_SIZE as u64 + len_a;
-        let block_b_offset = BlockOffset::new(block_b_offset_raw);
-        let cid_b = [0xBBu8; CID_SIZE];
-        let data_b = vec![2u8; 64];
-        let len_b = encode_block_record(&setup, fd, block_b_offset, &cid_b, &data_b).unwrap();
-
-        setup.sync(fd).unwrap();
-        setup.close(fd).unwrap();
-        drop(setup);
-
-        let total_size = block_b_offset_raw + len_b;
-        assert_eq!(std::fs::metadata(&file_path).unwrap().len(), total_size);
-
-        let wrapper = EioOnReadAtRange {
-            inner: RealIO::new(),
-            target_path: file_path.clone(),
-            target_min: block_b_offset_raw,
-            target_max: block_b_offset_raw + (BLOCK_RECORD_OVERHEAD as u64) - 1,
-            fired: AtomicBool::new(false),
-            fd_paths: Mutex::new(HashMap::new()),
-        };
-
-        let index = BlockIndex::open(&index_dir).unwrap();
-
-        let result = TranquilBlockStore::<EioOnReadAtRange>::replay_single_file(
-            &wrapper,
-            &data_dir,
-            &index,
-            file_id,
-            BlockOffset::new(BLOCK_HEADER_SIZE as u64),
-        );
-
-        assert!(
-            result.is_err(),
-            "replay must surface transient EIO instead of silently truncating"
-        );
-
-        let post_size = std::fs::metadata(&file_path).unwrap().len();
-        assert_eq!(
-            post_size, total_size,
-            "scan truncated durable acked block past EIO point: expected {total_size} bytes, got {post_size}"
-        );
-    }
-
-    fn instant_policy(max_attempts: u8) -> OpenRetryPolicy {
-        OpenRetryPolicy {
-            max_attempts: NonZeroU8::new(max_attempts).expect("max_attempts must be nonzero"),
-            initial_backoff: Duration::ZERO,
-            max_backoff: Duration::ZERO,
-        }
-    }
-
-    #[test]
-    fn retry_with_backoff_succeeds_on_first_attempt() {
-        let calls = std::sync::atomic::AtomicUsize::new(0);
-        let result = retry_with_backoff(instant_policy(5), &mut |_| {
-            calls.fetch_add(1, Ordering::Relaxed);
-            Ok::<u8, RepoError>(42)
-        });
-        assert_eq!(result.expect("ok"), 42);
-        assert_eq!(calls.load(Ordering::Relaxed), 1);
-    }
-
-    #[test]
-    fn retry_with_backoff_recovers_after_transient_failures() {
-        let calls = std::sync::atomic::AtomicUsize::new(0);
-        let result = retry_with_backoff(instant_policy(5), &mut |_| {
-            let n = calls.fetch_add(1, Ordering::Relaxed);
-            if n >= 2 {
-                Ok::<u8, RepoError>(7)
-            } else {
-                Err(RepoError::storage(io::Error::other("transient EIO")))
-            }
-        });
-        assert_eq!(result.expect("ok"), 7);
-        assert_eq!(calls.load(Ordering::Relaxed), 3);
-    }
-
-    #[test]
-    fn retry_with_backoff_gives_up_after_max_attempts() {
-        let calls = std::sync::atomic::AtomicUsize::new(0);
-        let result: Result<u8, RepoError> = retry_with_backoff(instant_policy(3), &mut |_| {
-            calls.fetch_add(1, Ordering::Relaxed);
-            Err(RepoError::storage(io::Error::other("permanent EIO")))
-        });
-        assert!(result.is_err(), "expected exhaustion error");
-        assert_eq!(calls.load(Ordering::Relaxed), 3);
-    }
-
-    #[test]
-    fn retry_with_backoff_passes_attempt_index_to_op() {
-        let observed = std::sync::Mutex::new(Vec::<u8>::new());
-        let _result: Result<(), RepoError> = retry_with_backoff(instant_policy(4), &mut |attempt| {
-            observed.lock().unwrap().push(attempt);
-            Err(RepoError::storage(io::Error::other("EIO")))
-        });
-        assert_eq!(*observed.lock().unwrap(), vec![0, 1, 2, 3]);
-    }
-}

crates/tranquil-store/src/eventlog/writer.rs

@@ -3,27 +3,16 @@ use std::sync::Arc;
 
 use tracing::warn;
 
-use crate::io::{FileId, StorageIO};
+use crate::io::StorageIO;
 
 use super::manager::SegmentManager;
-use super::segment_file::{
-    SEGMENT_HEADER_SIZE, SEGMENT_MAGIC, SegmentWriter, ValidEvent, ValidateEventRecord,
-    validate_event_record,
-};
+use super::segment_file::{SEGMENT_HEADER_SIZE, SegmentWriter, ValidEvent};
 use super::segment_index::{DEFAULT_INDEX_INTERVAL, SegmentIndex, rebuild_from_segment};
 use super::sidecar::build_sidecar_from_segment;
 use super::types::{
     DidHash, EventSequence, EventTypeTag, SegmentId, SegmentOffset, TimestampMicros,
 };
 
-const VALIDATE_RETRY_ATTEMPTS: u32 = 32;
-
-#[derive(Debug, Clone)]
-struct PendingAppend {
-    event: ValidEvent,
-    offset: SegmentOffset,
-}
-
 #[derive(Debug)]
 pub struct SyncResult {
     pub synced_through: EventSequence,
@@ -42,8 +31,7 @@ pub struct EventLogWriter<S: StorageIO> {
     max_payload: u32,
     event_count_in_segment: usize,
     last_event_offset: Option<SegmentOffset>,
-    pending: Vec<PendingAppend>,
-    poisoned: bool,
+    pending_events: Vec<ValidEvent>,
 }
 
 impl<S: StorageIO> EventLogWriter<S> {
@@ -95,25 +83,10 @@ impl<S: StorageIO> EventLogWriter<S> {
             max_payload,
             event_count_in_segment: 0,
             last_event_offset: None,
-            pending: Vec::new(),
-            poisoned: false,
+            pending_events: Vec::new(),
         })
     }
 
-    fn truncate_and_init_fresh(
-        manager: Arc<SegmentManager<S>>,
-        fd: FileId,
-        active_id: SegmentId,
-        prev_segments: &[SegmentId],
-        index_interval: usize,
-        max_payload: u32,
-    ) -> io::Result<Self> {
-        manager.io().truncate(fd, 0)?;
-        let next_seq = find_last_seq_from_segments(&manager, prev_segments, max_payload)?
-            .map_or(EventSequence::new(1), |s| s.next());
-        Self::init_fresh(manager, active_id, next_seq, index_interval, max_payload)
-    }
-
     fn recover_active(
         manager: Arc<SegmentManager<S>>,
         segments: &[SegmentId],
@@ -124,19 +97,6 @@ impl<S: StorageIO> EventLogWriter<S> {
         let handle = manager.open_for_append(active_id)?;
         let fd = handle.fd();
 
-        let prev_segments = &segments[..segments.len().saturating_sub(1)];
-
-        if highest_segment_has_torn_header(manager.io(), fd)? {
-            return Self::truncate_and_init_fresh(
-                Arc::clone(&manager),
-                fd,
-                active_id,
-                prev_segments,
-                index_interval,
-                max_payload,
-            );
-        }
-
         let (index, last_seq_in_active) = match rebuild_from_segment(
             manager.io(),
             fd,
@@ -147,11 +107,15 @@ impl<S: StorageIO> EventLogWriter<S> {
             Err(rebuild_err) => {
                 let file_size = manager.io().file_size(fd)?;
                 if file_size <= SEGMENT_HEADER_SIZE as u64 {
-                    return Self::truncate_and_init_fresh(
+                    manager.io().truncate(fd, 0)?;
+                    let prev_segments = &segments[..segments.len().saturating_sub(1)];
+                    let next_seq =
+                        find_last_seq_from_segments(&manager, prev_segments, max_payload)?
+                            .map_or(EventSequence::new(1), |s| s.next());
+                    return Self::init_fresh(
                         Arc::clone(&manager),
-                        fd,
                         active_id,
-                        prev_segments,
+                        next_seq,
                         index_interval,
                         max_payload,
                     );
@@ -167,6 +131,8 @@ impl<S: StorageIO> EventLogWriter<S> {
 
         let position = SegmentOffset::new(manager.io().file_size(fd)?);
 
+        let prev_segments = &segments[..segments.len().saturating_sub(1)];
+
         let next_seq = match last_seq_in_active {
             Some(seq) => {
                 if let Some(sealed_last) =
@@ -230,8 +196,7 @@ impl<S: StorageIO> EventLogWriter<S> {
             max_payload,
             event_count_in_segment,
             last_event_offset,
-            pending: Vec::new(),
-            poisoned: false,
+            pending_events: Vec::new(),
         })
     }
 
@@ -262,20 +227,28 @@ impl<S: StorageIO> EventLogWriter<S> {
             payload,
         };
 
-        self.append_inner(event).map(|_| seq)
+        let offset = self.active_writer.append_event(self.manager.io(), &event)?;
+
+        let should_index = self.event_count_in_segment == 0
+            || self
+                .event_count_in_segment
+                .is_multiple_of(self.index_interval);
+        if should_index {
+            self.active_index.record(seq, offset);
+        }
+
+        self.event_count_in_segment = self
+            .event_count_in_segment
+            .checked_add(1)
+            .expect("event_count_in_segment overflow");
+        self.last_event_offset = Some(offset);
+        self.next_seq = seq.next();
+        self.pending_events.push(event);
+
+        Ok(seq)
     }
 
     pub fn append_valid_event(&mut self, event: ValidEvent) -> io::Result<()> {
-        self.append_inner(event)
-    }
-
-    fn append_inner(&mut self, event: ValidEvent) -> io::Result<()> {
-        if self.poisoned {
-            return Err(io::Error::other(
-                "writer poisoned by partial-valid sync; reopen required",
-            ));
-        }
-
         let offset = self.active_writer.append_event(self.manager.io(), &event)?;
 
         let should_index = self.event_count_in_segment == 0
@@ -292,52 +265,21 @@ impl<S: StorageIO> EventLogWriter<S> {
             .expect("event_count_in_segment overflow");
         self.last_event_offset = Some(offset);
         self.next_seq = event.seq.next();
-        self.pending.push(PendingAppend { event, offset });
+        self.pending_events.push(event);
 
         Ok(())
     }
 
+    pub fn peek_pending_event(&self, seq: EventSequence) -> Option<&ValidEvent> {
+        self.pending_events.iter().find(|e| e.seq == seq)
+    }
+
     pub fn sync(&mut self) -> io::Result<SyncResult> {
-        if self.poisoned {
-            return Err(io::Error::other(
-                "writer poisoned by partial-valid sync; reopen required",
-            ));
-        }
-
-        if !self.pending.is_empty() {
+        if !self.pending_events.is_empty() {
             self.active_writer.sync(self.manager.io())?;
-            self.manager.io().barrier()?;
         }
 
-        let pending = std::mem::take(&mut self.pending);
-
-        let fd = self.active_writer.fd();
-        let file_size = self.manager.io().file_size(fd)?;
-
-        let valid_count = pending
-            .iter()
-            .take_while(|p| {
-                validate_with_retry(
-                    self.manager.io(),
-                    fd,
-                    p.offset,
-                    file_size,
-                    self.max_payload,
-                    p.event.seq,
-                )
-            })
-            .count();
-
-        if valid_count < pending.len() {
-            self.poisoned = true;
-        }
-
-        let flushed: Vec<ValidEvent> = pending
-            .into_iter()
-            .take(valid_count)
-            .map(|p| p.event)
-            .collect();
-
+        let flushed = std::mem::take(&mut self.pending_events);
         self.synced_seq = flushed.last().map(|e| e.seq).unwrap_or(self.synced_seq);
 
         Ok(SyncResult {
@@ -348,22 +290,12 @@ impl<S: StorageIO> EventLogWriter<S> {
         })
     }
 
-    pub fn is_poisoned(&self) -> bool {
-        self.poisoned
-    }
-
     pub fn rotate_if_needed(&mut self) -> io::Result<Option<SegmentId>> {
-        if self.poisoned {
-            return Err(io::Error::other(
-                "writer poisoned by partial-valid sync; reopen required",
-            ));
-        }
-
         if !self.manager.should_rotate(self.active_writer.position()) {
             return Ok(None);
         }
 
-        if !self.pending.is_empty() {
+        if !self.pending_events.is_empty() {
             return Ok(None);
         }
 
@@ -454,40 +386,6 @@ impl<S: StorageIO> EventLogWriter<S> {
     }
 }
 
-fn validate_with_retry<S: StorageIO>(
-    io: &S,
-    fd: FileId,
-    offset: SegmentOffset,
-    file_size: u64,
-    max_payload: u32,
-    expected_seq: EventSequence,
-) -> bool {
-    (0..VALIDATE_RETRY_ATTEMPTS).any(|_| {
-        matches!(
-            validate_event_record(io, fd, offset, file_size, max_payload),
-            Ok(Some(ValidateEventRecord::Valid { seq, .. })) if seq == expected_seq
-        )
-    })
-}
-
-fn highest_segment_has_torn_header<S: StorageIO>(io: &S, fd: FileId) -> io::Result<bool> {
-    let file_size = io.file_size(fd)?;
-    if file_size < SEGMENT_HEADER_SIZE as u64 {
-        return Ok(true);
-    }
-    let outcomes: Vec<bool> = (0..VALIDATE_RETRY_ATTEMPTS)
-        .filter_map(|_| {
-            let mut header = [0u8; SEGMENT_MAGIC.len()];
-            io.read_exact_at(fd, 0, &mut header)
-                .ok()
-                .map(|()| header == SEGMENT_MAGIC)
-        })
-        .collect();
-    let saw_match = outcomes.iter().any(|&ok| ok);
-    let saw_mismatch = outcomes.iter().any(|&ok| !ok);
-    Ok(!saw_match && saw_mismatch)
-}
-
 fn find_last_seq_from_segments<S: StorageIO>(
     manager: &SegmentManager<S>,
     segments: &[SegmentId],
@@ -1196,62 +1094,4 @@ mod tests {
 
         assert!(writer.rotate_if_needed().unwrap().is_none());
     }
-
-    #[test]
-    fn sync_must_not_certify_durability_when_io_sync_silently_drops() {
-        use crate::sim::{FaultConfig, Probability};
-
-        let sim = Arc::new(SimulatedIO::new(
-            0,
-            FaultConfig {
-                sync_failure_probability: Probability::new(1.0),
-                ..FaultConfig::none()
-            },
-        ));
-        sim.set_pristine_mode(true);
-
-        let mgr = Arc::new(
-            SegmentManager::new(Arc::clone(&sim), PathBuf::from("/segments"), 64 * 1024).unwrap(),
-        );
-
-        let mut writer =
-            EventLogWriter::open(Arc::clone(&mgr), DEFAULT_INDEX_INTERVAL, MAX_EVENT_PAYLOAD)
-                .unwrap();
-
-        sim.set_pristine_mode(false);
-
-        writer
-            .append(
-                DidHash::from_did("did:plc:bug2"),
-                EventTypeTag::COMMIT,
-                b"bug2-payload".to_vec(),
-            )
-            .unwrap();
-
-        assert!(
-            writer.sync().is_err(),
-            "sync must surface dropped fsync as an error"
-        );
-        let claimed_synced = writer.synced_seq();
-        assert_eq!(
-            claimed_synced.raw(),
-            0,
-            "synced_seq must not advance past a failed sync"
-        );
-        drop(writer);
-
-        mgr.shutdown();
-        sim.crash();
-        sim.set_pristine_mode(true);
-
-        let reopened =
-            EventLogWriter::open(Arc::clone(&mgr), DEFAULT_INDEX_INTERVAL, MAX_EVENT_PAYLOAD)
-                .unwrap();
-        let actually_durable = reopened.current_seq();
-
-        assert!(
-            actually_durable >= claimed_synced,
-            "writer claimed sync through {claimed_synced} but post-crash recovery only reaches {actually_durable}"
-        );
-    }
 }

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

@@ -1,6 +1,5 @@
 use std::cell::RefCell;
 use std::panic::{AssertUnwindSafe, catch_unwind};
-use std::path::PathBuf;
 use std::time::{Duration, Instant};
 
 use rayon::prelude::*;
@@ -45,17 +44,6 @@ pub fn run_many_timed<F>(
     make_config: F,
     seeds: impl IntoIterator<Item = Seed>,
 ) -> Vec<(GauntletReport, Duration)>
-where
-    F: Fn(Seed) -> GauntletConfig + Sync + Send,
-{
-    run_many_timed_with_scratch_roots(make_config, &[], seeds)
-}
-
-pub fn run_many_timed_with_scratch_roots<F>(
-    make_config: F,
-    scratch_roots: &[PathBuf],
-    seeds: impl IntoIterator<Item = Seed>,
-) -> Vec<(GauntletReport, Duration)>
 where
     F: Fn(Seed) -> GauntletConfig + Sync + Send,
 {
@@ -63,14 +51,10 @@ where
     seeds
         .into_par_iter()
         .map(|s| {
-            let scratch = scratch_for_thread(scratch_roots, rayon::current_thread_index());
             let start = Instant::now();
             let outcome = catch_unwind(AssertUnwindSafe(|| {
                 let cfg = make_config(s);
-                let mut gauntlet = Gauntlet::new(cfg).expect("build gauntlet");
-                if let Some(root) = scratch {
-                    gauntlet = gauntlet.with_scratch_root(root);
-                }
+                let gauntlet = Gauntlet::new(cfg).expect("build gauntlet");
                 with_runtime(|rt| rt.block_on(gauntlet.run()))
             }));
             let report = outcome.unwrap_or_else(|payload| {
@@ -82,14 +66,6 @@ where
         .collect()
 }
 
-fn scratch_for_thread(roots: &[PathBuf], thread_idx: Option<usize>) -> Option<PathBuf> {
-    if roots.is_empty() {
-        None
-    } else {
-        Some(roots[thread_idx.unwrap_or(0) % roots.len()].clone())
-    }
-}
-
 fn panic_report(seed: Seed, payload: Box<dyn std::any::Any + Send>) -> GauntletReport {
     let msg = payload
         .downcast_ref::<&'static str>()
@@ -108,62 +84,3 @@ fn panic_report(seed: Seed, payload: Box<dyn std::any::Any + Send>) -> GauntletR
         ops: OpStream::empty(),
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn scratch_for_thread_returns_none_when_roots_empty() {
-        assert!(scratch_for_thread(&[], Some(0)).is_none());
-        assert!(scratch_for_thread(&[], Some(7)).is_none());
-        assert!(scratch_for_thread(&[], None).is_none());
-    }
-
-    #[test]
-    fn scratch_for_thread_round_robins_across_roots() {
-        let roots = vec![
-            PathBuf::from("/scratch/a"),
-            PathBuf::from("/scratch/b"),
-            PathBuf::from("/scratch/c"),
-        ];
-        let assigned: Vec<PathBuf> = (0..7)
-            .map(|i| scratch_for_thread(&roots, Some(i)).expect("scratch path"))
-            .collect();
-        assert_eq!(
-            assigned,
-            vec![
-                PathBuf::from("/scratch/a"),
-                PathBuf::from("/scratch/b"),
-                PathBuf::from("/scratch/c"),
-                PathBuf::from("/scratch/a"),
-                PathBuf::from("/scratch/b"),
-                PathBuf::from("/scratch/c"),
-                PathBuf::from("/scratch/a"),
-            ]
-        );
-    }
-
-    #[test]
-    fn scratch_for_thread_with_single_root_returns_same_path() {
-        let roots = vec![PathBuf::from("/scratch/only")];
-        (0..5).for_each(|i| {
-            assert_eq!(
-                scratch_for_thread(&roots, Some(i)),
-                Some(PathBuf::from("/scratch/only"))
-            );
-        });
-    }
-
-    #[test]
-    fn scratch_for_thread_falls_back_to_root_zero_outside_pool() {
-        let roots = vec![
-            PathBuf::from("/scratch/a"),
-            PathBuf::from("/scratch/b"),
-        ];
-        assert_eq!(
-            scratch_for_thread(&roots, None),
-            Some(PathBuf::from("/scratch/a"))
-        );
-    }
-}

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

@@ -405,8 +405,6 @@ fn mount_ext4(device: &Path, target: &Path) -> Result<(), FlakyError> {
     let out = Command::new("mount")
         .arg("-t")
         .arg("ext4")
-        .arg("-o")
-        .arg("errors=continue")
         .arg(device)
         .arg(target)
         .output()?;

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

@@ -23,7 +23,7 @@ use crate::eventlog::{
     SegmentManager, SegmentReader, TimestampMicros, ValidEvent,
 };
 use crate::io::{RealIO, StorageIO};
-use crate::sim::{FaultConfig, PristineGuard, SimulatedIO};
+use crate::sim::{FaultConfig, SimulatedIO};
 
 #[derive(Debug, Clone, Copy)]
 pub enum IoBackend {
@@ -178,7 +178,6 @@ pub struct SharedState<S: StorageIO + Send + Sync + 'static> {
 
 pub struct Gauntlet {
     config: GauntletConfig,
-    scratch_root: Option<PathBuf>,
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -186,15 +185,7 @@ pub enum GauntletBuildError {}
 
 impl Gauntlet {
     pub fn new(config: GauntletConfig) -> Result<Self, GauntletBuildError> {
-        Ok(Self {
-            config,
-            scratch_root: None,
-        })
-    }
-
-    pub fn with_scratch_root(mut self, root: PathBuf) -> Self {
-        self.scratch_root = Some(root);
-        self
+        Ok(Self { config })
     }
 
     pub fn generate_ops(&self) -> OpStream {
@@ -220,7 +211,6 @@ impl Gauntlet {
         let ops_counter = Arc::new(AtomicUsize::new(0));
         let op_errors_counter = Arc::new(AtomicUsize::new(0));
         let restarts_counter = Arc::new(AtomicUsize::new(0));
-        let scratch_root = self.scratch_root;
         let fut: std::pin::Pin<Box<dyn std::future::Future<Output = GauntletReport> + Send>> =
             match self.config.io {
                 IoBackend::Real => Box::pin(run_inner_real(
@@ -229,7 +219,6 @@ impl Gauntlet {
                     ops_counter.clone(),
                     op_errors_counter.clone(),
                     restarts_counter.clone(),
-                    scratch_root,
                 )),
                 IoBackend::RealWithFlaky { flaky } => Box::pin(run_inner_real_with_flaky(
                     self.config,
@@ -280,12 +269,8 @@ async fn run_inner_real(
     ops_counter: Arc<AtomicUsize>,
     op_errors_counter: Arc<AtomicUsize>,
     restarts_counter: Arc<AtomicUsize>,
-    scratch_root: Option<PathBuf>,
 ) -> GauntletReport {
-    let dir = match scratch_root.as_deref() {
-        Some(parent) => tempfile::TempDir::new_in(parent).expect("tempdir in scratch root"),
-        None => tempfile::TempDir::new().expect("tempdir"),
-    };
+    let dir = tempfile::TempDir::new().expect("tempdir");
     let root = dir.path().to_path_buf();
     let report = run_inner_real_on_root(
         config,
@@ -379,7 +364,7 @@ async fn run_inner_real_on_root(
     let segments_dir = segments_subdir(&root);
     let open = {
         let segments_dir = segments_dir.clone();
-        move |_attempt: usize| -> Result<Harness<RealIO>, String> {
+        move || -> Result<Harness<RealIO>, String> {
             let store = TranquilBlockStore::open(cfg.clone())
                 .map(Arc::new)
                 .map_err(|e| e.to_string())?;
@@ -439,8 +424,7 @@ async fn run_inner_simulated(
     let sim_for_open = Arc::clone(&sim);
     let open = {
         let segments_dir = segments_dir.clone();
-        move |attempt: usize| -> Result<Harness<Arc<SimulatedIO>>, String> {
-            let _pristine = PristineGuard::new(Arc::clone(&sim_for_open), attempt > 0);
+        move || -> Result<Harness<Arc<SimulatedIO>>, String> {
             let factory_sim = Arc::clone(&sim_for_open);
             let make_io = move || Arc::clone(&factory_sim);
             let store = TranquilBlockStore::<Arc<SimulatedIO>>::open_with_io(cfg.clone(), make_io)
@@ -528,30 +512,28 @@ async fn run_inner_generic<S, Open, Crash>(
 ) -> GauntletReport
 where
     S: StorageIO + Send + Sync + 'static,
-    Open: FnMut(usize) -> Result<Harness<S>, String>,
+    Open: FnMut() -> Result<Harness<S>, String>,
     Crash: FnMut(),
 {
     let mut oracle = Oracle::new();
     let mut violations: Vec<InvariantViolation> = Vec::new();
 
-    let mut harness: Option<Harness<S>> =
-        match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors, reopen_backoff).await
-        {
-            Ok(h) => Some(h),
-            Err(e) => {
-                return GauntletReport {
-                    seed: config.seed,
-                    ops_executed: OpsExecuted(0),
-                    op_errors: OpErrorCount(op_errors_counter.load(Ordering::Relaxed)),
-                    restarts: RestartCount(0),
-                    violations: vec![InvariantViolation {
-                        invariant: "OpenStore",
-                        detail: format!("initial open: {e}"),
-                    }],
-                    ops: OpStream::empty(),
-                };
-            }
-        };
+    let mut harness: Option<Harness<S>> = match open() {
+        Ok(h) => Some(h),
+        Err(e) => {
+            return GauntletReport {
+                seed: config.seed,
+                ops_executed: OpsExecuted(0),
+                op_errors: OpErrorCount(op_errors_counter.load(Ordering::Relaxed)),
+                restarts: RestartCount(0),
+                violations: vec![InvariantViolation {
+                    invariant: "OpenStore",
+                    detail: format!("initial open: {e}"),
+                }],
+                ops: OpStream::empty(),
+            };
+        }
+    };
     let mut root: Option<Cid> = None;
     let mut restart_rng = Lcg::new(Seed(config.seed.0 ^ 0xA5A5_A5A5_A5A5_A5A5));
     let mut sample_rng = Lcg::new(Seed(config.seed.0 ^ 0x5A5A_5A5A_5A5A_5A5A));
@@ -768,7 +750,7 @@ async fn reopen_with_recovery<S, Open, Crash>(
 ) -> Result<Harness<S>, String>
 where
     S: StorageIO + Send + Sync + 'static,
-    Open: FnMut(usize) -> Result<Harness<S>, String>,
+    Open: FnMut() -> Result<Harness<S>, String>,
     Crash: FnMut(),
 {
     let mut errors: Vec<String> = Vec::new();
@@ -776,7 +758,7 @@ where
         if attempt > 0 && !backoff.is_zero() {
             tokio::time::sleep(backoff).await;
         }
-        match open(attempt) {
+        match open() {
             Ok(h) => return Ok(h),
             Err(e) => {
                 errors.push(format!("attempt {attempt}: {e}"));
@@ -1209,10 +1191,6 @@ fn run_retention<S: StorageIO + Send + Sync + 'static>(
     max_age: RetentionSecs,
 ) -> Result<(), String> {
     let sync_result = el.writer.sync().map_err(|e| e.to_string())?;
-    el.manager
-        .io()
-        .sync_dir(el.segments_dir.as_path())
-        .map_err(|e| e.to_string())?;
     let _ = el.writer.rotate_if_needed();
     oracle.record_event_sync(sync_result.synced_through);
     let active_id = sync_result.segment_id;
@@ -1586,7 +1564,7 @@ async fn run_inner_generic_concurrent<S, Open, Crash>(
 ) -> GauntletReport
 where
     S: StorageIO + Send + Sync + 'static,
-    Open: FnMut(usize) -> Result<Harness<S>, String>,
+    Open: FnMut() -> Result<Harness<S>, String>,
     Crash: FnMut(),
 {
     let ops: Vec<Op> = op_stream.into_vec();
@@ -1599,24 +1577,22 @@ where
     let mut sample_rng = Lcg::new(Seed(config.seed.0 ^ 0x5A5A_5A5A_5A5A_5A5A));
     let chunks = compute_chunks(config.restart_policy, total_ops, &mut restart_rng);
 
-    let mut harness: Option<Harness<S>> =
-        match reopen_with_recovery(&mut open, &mut crash, tolerate_op_errors, reopen_backoff).await
-        {
-            Ok(h) => Some(h),
-            Err(e) => {
-                return GauntletReport {
-                    seed: config.seed,
-                    ops_executed: OpsExecuted(0),
-                    op_errors: OpErrorCount(op_errors_counter.load(Ordering::Relaxed)),
-                    restarts: RestartCount(0),
-                    violations: vec![InvariantViolation {
-                        invariant: "OpenStore",
-                        detail: format!("initial open: {e}"),
-                    }],
-                    ops: OpStream::empty(),
-                };
-            }
-        };
+    let mut harness: Option<Harness<S>> = match open() {
+        Ok(h) => Some(h),
+        Err(e) => {
+            return GauntletReport {
+                seed: config.seed,
+                ops_executed: OpsExecuted(0),
+                op_errors: OpErrorCount(op_errors_counter.load(Ordering::Relaxed)),
+                restarts: RestartCount(0),
+                violations: vec![InvariantViolation {
+                    invariant: "OpenStore",
+                    detail: format!("initial open: {e}"),
+                }],
+                ops: OpStream::empty(),
+            };
+        }
+    };
     let mut root: Option<Cid> = None;
     let mut oracle = Oracle::new();
     let mut halt_ops = false;
@@ -1825,125 +1801,3 @@ where
         ops: OpStream::empty(),
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    fn minimal_config() -> GauntletConfig {
-        GauntletConfig {
-            seed: Seed(0),
-            io: IoBackend::Real,
-            workload: WorkloadModel::default(),
-            op_count: OpCount(0),
-            invariants: InvariantSet::EMPTY,
-            limits: RunLimits {
-                max_wall_ms: Some(WallMs(30_000)),
-            },
-            restart_policy: RestartPolicy::Never,
-            store: StoreConfig {
-                max_file_size: MaxFileSize(8 * 1024),
-                group_commit: GroupCommitConfig::default(),
-                shard_count: ShardCount(1),
-            },
-            eventlog: None,
-            writer_concurrency: WriterConcurrency(1),
-        }
-    }
-
-    fn flaky_open(
-        attempts: Arc<AtomicUsize>,
-        sim: Arc<SimulatedIO>,
-        store_cfg: BlockStoreConfig,
-    ) -> impl FnMut(usize) -> Result<Harness<Arc<SimulatedIO>>, String> + Send + 'static {
-        move |_attempt: usize| -> Result<Harness<Arc<SimulatedIO>>, String> {
-            let n = attempts.fetch_add(1, Ordering::Relaxed);
-            if n == 0 {
-                return Err("simulated EIO on initial open".to_string());
-            }
-            let factory_sim = Arc::clone(&sim);
-            let make_io = move || Arc::clone(&factory_sim);
-            TranquilBlockStore::<Arc<SimulatedIO>>::open_with_io(store_cfg.clone(), make_io)
-                .map(|s| Harness {
-                    store: Arc::new(s),
-                    eventlog: None,
-                })
-                .map_err(|e| e.to_string())
-        }
-    }
-
-    #[tokio::test]
-    async fn run_inner_generic_retries_initial_open_on_transient_io_error() {
-        let dir = tempfile::TempDir::new().expect("tempdir");
-        let cfg = minimal_config();
-        let store_cfg = blockstore_config(dir.path(), &cfg.store);
-        let sim: Arc<SimulatedIO> = Arc::new(SimulatedIO::pristine(0));
-        let attempts = Arc::new(AtomicUsize::new(0));
-
-        let report = run_inner_generic::<Arc<SimulatedIO>, _, _>(
-            cfg,
-            OpStream::empty(),
-            Arc::new(AtomicUsize::new(0)),
-            Arc::new(AtomicUsize::new(0)),
-            Arc::new(AtomicUsize::new(0)),
-            flaky_open(Arc::clone(&attempts), Arc::clone(&sim), store_cfg),
-            || {},
-            true,
-            Duration::ZERO,
-        )
-        .await;
-
-        let opens: Vec<&InvariantViolation> = report
-            .violations
-            .iter()
-            .filter(|v| v.invariant == "OpenStore")
-            .collect();
-        assert!(
-            opens.is_empty(),
-            "expected initial open to retry, got OpenStore violations: {opens:?}"
-        );
-        let total = attempts.load(Ordering::Relaxed);
-        assert!(
-            total >= 2,
-            "expected at least one retry after first failure, attempts={total}"
-        );
-    }
-
-    #[tokio::test]
-    async fn run_inner_generic_concurrent_retries_initial_open_on_transient_io_error() {
-        let dir = tempfile::TempDir::new().expect("tempdir");
-        let mut cfg = minimal_config();
-        cfg.writer_concurrency = WriterConcurrency(2);
-        let store_cfg = blockstore_config(dir.path(), &cfg.store);
-        let sim: Arc<SimulatedIO> = Arc::new(SimulatedIO::pristine(0));
-        let attempts = Arc::new(AtomicUsize::new(0));
-
-        let report = run_inner_generic_concurrent::<Arc<SimulatedIO>, _, _>(
-            cfg,
-            OpStream::empty(),
-            Arc::new(AtomicUsize::new(0)),
-            Arc::new(AtomicUsize::new(0)),
-            Arc::new(AtomicUsize::new(0)),
-            flaky_open(Arc::clone(&attempts), Arc::clone(&sim), store_cfg),
-            || {},
-            true,
-            Duration::ZERO,
-        )
-        .await;
-
-        let opens: Vec<&InvariantViolation> = report
-            .violations
-            .iter()
-            .filter(|v| v.invariant == "OpenStore")
-            .collect();
-        assert!(
-            opens.is_empty(),
-            "expected initial open to retry, got OpenStore violations: {opens:?}"
-        );
-        let total = attempts.load(Ordering::Relaxed);
-        assert!(
-            total >= 2,
-            "expected at least one retry after first failure, attempts={total}"
-        );
-    }
-}

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

@@ -9,7 +9,7 @@ use super::workload::{
     ByteRange, DidSpaceSize, KeySpaceSize, OpCount, OpWeights, RetentionMaxSecs, SizeDistribution,
     ValueBytes, WorkloadModel,
 };
-use crate::blockstore::{GroupCommitConfig, MAX_BLOCK_SIZE};
+use crate::blockstore::GroupCommitConfig;
 use crate::sim::FaultConfig;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -402,7 +402,7 @@ fn huge_values(seed: Seed) -> GauntletConfig {
         workload: block_workload(
             block_weights(85, 5, 8, 2),
             SizeDistribution::HeavyTail(
-                ByteRange::new(ValueBytes(256), ValueBytes(MAX_BLOCK_SIZE))
+                ByteRange::new(ValueBytes(256), ValueBytes(16 * 1024 * 1024))
                     .expect("huge_values ByteRange"),
             ),
             KeySpaceSize(64),

crates/tranquil-store/src/io.rs

@@ -104,10 +104,6 @@ pub trait StorageIO: Send + Sync {
     fn sync_dir(&self, path: &Path) -> io::Result<()>;
     fn list_dir(&self, path: &Path) -> io::Result<Vec<PathBuf>>;
 
-    fn barrier(&self) -> io::Result<()> {
-        Ok(())
-    }
-
     fn write_all_at(&self, fd: FileId, offset: u64, buf: &[u8]) -> io::Result<()> {
         let written = Cell::new(0usize);
         std::iter::from_fn(|| (written.get() < buf.len()).then_some(()))
@@ -194,9 +190,6 @@ impl<S: StorageIO> StorageIO for Arc<S> {
     fn list_dir(&self, path: &Path) -> io::Result<Vec<PathBuf>> {
         (**self).list_dir(path)
     }
-    fn barrier(&self) -> io::Result<()> {
-        (**self).barrier()
-    }
     fn mmap_file(&self, fd: FileId) -> io::Result<MappedFile> {
         (**self).mmap_file(fd)
     }

crates/tranquil-store/src/lib.rs

@@ -28,7 +28,7 @@ pub use record::{
 };
 #[cfg(any(test, feature = "test-harness"))]
 pub use sim::{
-    FaultConfig, LatencyNs, OpRecord, Probability, PristineGuard, SimulatedIO, SyncReorderWindow,
+    FaultConfig, LatencyNs, OpRecord, Probability, SimulatedIO, SyncReorderWindow,
     sim_proptest_cases, sim_seed_count, sim_seed_range, sim_single_seed,
 };
 

crates/tranquil-store/src/sim.rs

@@ -2,8 +2,7 @@ use std::collections::{HashMap, HashSet, VecDeque};
 use std::io;
 use std::path::{Path, PathBuf};
 use std::sync::Mutex;
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
-use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::Duration;
 
 use crate::io::{FileId, OpenOptions, StorageIO};
@@ -227,7 +226,6 @@ pub enum OpRecord {
     SyncDir {
         path: PathBuf,
     },
-    Barrier,
 }
 
 struct PendingSync {
@@ -328,7 +326,6 @@ impl SimState {
 pub struct SimulatedIO {
     state: Mutex<SimState>,
     fault_config: FaultConfig,
-    pristine_mode: AtomicBool,
     rng_seed: u64,
     latency_counter: AtomicU64,
 }
@@ -349,26 +346,13 @@ impl SimulatedIO {
                 pending_deletes: Vec::new(),
             }),
             fault_config,
-            pristine_mode: AtomicBool::new(false),
             rng_seed: seed,
             latency_counter: AtomicU64::new(0),
         }
     }
 
-    fn effective_fault_config(&self) -> FaultConfig {
-        if self.pristine_mode.load(Ordering::Relaxed) {
-            FaultConfig::none()
-        } else {
-            self.fault_config
-        }
-    }
-
-    pub fn set_pristine_mode(&self, on: bool) {
-        self.pristine_mode.store(on, Ordering::Relaxed);
-    }
-
     fn jitter(&self) {
-        let max_ns = self.effective_fault_config().latency_distribution_ns.0;
+        let max_ns = self.fault_config.latency_distribution_ns.0;
         if max_ns == 0 {
             return;
         }
@@ -445,30 +429,12 @@ impl SimulatedIO {
     }
 }
 
-pub struct PristineGuard {
-    sim: Arc<SimulatedIO>,
-}
-
-impl PristineGuard {
-    pub fn new(sim: Arc<SimulatedIO>, on: bool) -> Self {
-        sim.set_pristine_mode(on);
-        Self { sim }
-    }
-}
-
-impl Drop for PristineGuard {
-    fn drop(&mut self) {
-        self.sim.set_pristine_mode(false);
-    }
-}
-
 impl StorageIO for SimulatedIO {
     fn open(&self, path: &Path, opts: OpenOptions) -> io::Result<FileId> {
-        let fault = self.effective_fault_config();
         let mut state = self.state.lock().unwrap();
         let seed = self.rng_seed;
 
-        if state.should_fault(seed, fault.io_error_probability) {
+        if state.should_fault(seed, self.fault_config.io_error_probability) {
             return Err(io::Error::other("simulated EIO on open"));
         }
 
@@ -548,7 +514,6 @@ impl StorageIO for SimulatedIO {
 
     fn read_at(&self, id: FileId, offset: u64, buf: &mut [u8]) -> io::Result<usize> {
         self.jitter();
-        let fault = self.effective_fault_config();
         let mut state = self.state.lock().unwrap();
         let sid = state.require_readable(id)?;
         let seed = self.rng_seed;
@@ -557,12 +522,12 @@ impl StorageIO for SimulatedIO {
             return Err(io::Error::other("simulated EIO after delayed sync fault"));
         }
 
-        if state.should_fault(seed, fault.io_error_probability) {
+        if state.should_fault(seed, self.fault_config.io_error_probability) {
             return Err(io::Error::other("simulated EIO on read"));
         }
 
         let read_offset =
-            if state.should_fault(seed, fault.misdirected_read_probability) {
+            if state.should_fault(seed, self.fault_config.misdirected_read_probability) {
                 let drift_sectors = state.next_random_usize(seed, 8) + 1;
                 let drift = (drift_sectors * SECTOR_BYTES) as u64;
                 if state.next_random(seed) < 0.5 {
@@ -591,7 +556,7 @@ impl StorageIO for SimulatedIO {
         let to_read = buf.len().min(available);
         buf[..to_read].copy_from_slice(&storage.buffered[off..off + to_read]);
 
-        if state.should_fault(seed, fault.bit_flip_on_read_probability) && to_read > 0 {
+        if state.should_fault(seed, self.fault_config.bit_flip_on_read_probability) && to_read > 0 {
             let flip_pos = state.next_random_usize(seed, to_read);
             let flip_bit = state.next_random_usize(seed, 8);
             buf[flip_pos] ^= 1 << flip_bit;
@@ -607,7 +572,6 @@ impl StorageIO for SimulatedIO {
 
     fn write_at(&self, id: FileId, offset: u64, buf: &[u8]) -> io::Result<usize> {
         self.jitter();
-        let fault = self.effective_fault_config();
         let mut state = self.state.lock().unwrap();
         let sid = state.require_writable(id)?;
         let seed = self.rng_seed;
@@ -616,12 +580,12 @@ impl StorageIO for SimulatedIO {
             return Err(io::Error::other("simulated EIO after delayed sync fault"));
         }
 
-        if state.should_fault(seed, fault.io_error_probability) {
+        if state.should_fault(seed, self.fault_config.io_error_probability) {
             return Err(io::Error::other("simulated EIO on write"));
         }
 
         let torn_len =
-            if buf.len() > 1 && state.should_fault(seed, fault.torn_page_probability) {
+            if buf.len() > 1 && state.should_fault(seed, self.fault_config.torn_page_probability) {
                 let page_base = (offset as usize) - ((offset as usize) % TORN_PAGE_BYTES);
                 let page_end = page_base + TORN_PAGE_BYTES;
                 let cap = page_end.saturating_sub(offset as usize).min(buf.len());
@@ -637,7 +601,7 @@ impl StorageIO for SimulatedIO {
         let actual_len = match torn_len {
             Some(n) => n,
             None if buf.len() > 1
-                && state.should_fault(seed, fault.partial_write_probability) =>
+                && state.should_fault(seed, self.fault_config.partial_write_probability) =>
             {
                 let partial = state.next_random_usize(seed, buf.len());
                 partial.max(1)
@@ -645,7 +609,7 @@ impl StorageIO for SimulatedIO {
             None => buf.len(),
         };
 
-        let misdirected = state.should_fault(seed, fault.misdirected_write_probability);
+        let misdirected = state.should_fault(seed, self.fault_config.misdirected_write_probability);
         let write_offset = if misdirected {
             let drift_sectors = state.next_random_usize(seed, 8) + 1;
             let drift = (drift_sectors * SECTOR_BYTES) as u64;
@@ -679,7 +643,6 @@ impl StorageIO for SimulatedIO {
 
     fn sync(&self, id: FileId) -> io::Result<()> {
         self.jitter();
-        let fault = self.effective_fault_config();
         let mut state = self.state.lock().unwrap();
         let sid = state.require_open(id)?;
         let seed = self.rng_seed;
@@ -688,22 +651,16 @@ impl StorageIO for SimulatedIO {
             return Err(io::Error::other("simulated EIO after delayed sync fault"));
         }
 
-        if state.should_fault(seed, fault.io_error_probability) {
+        if state.should_fault(seed, self.fault_config.io_error_probability) {
             return Err(io::Error::other("simulated EIO on sync"));
         }
 
-        if state.should_fault(seed, fault.sync_failure_probability) {
-            state.op_log.push(OpRecord::Sync {
-                fd: id,
-                succeeded: false,
-            });
-            return Err(io::Error::other("simulated dropped fsync"));
-        }
+        let sync_succeeded = !state.should_fault(seed, self.fault_config.sync_failure_probability);
+        let poison_after = sync_succeeded
+            && state.should_fault(seed, self.fault_config.delayed_io_error_probability);
+        let reorder_window = self.fault_config.sync_reorder_window.0 as usize;
 
-        let poison_after = state.should_fault(seed, fault.delayed_io_error_probability);
-        let reorder_window = fault.sync_reorder_window.0 as usize;
-
-        let evicted = if reorder_window > 0 {
+        let evicted = if sync_succeeded && reorder_window > 0 {
             let snapshot = state.storage.get(&sid).unwrap().buffered.clone();
             state.pending_syncs.push_back(PendingSync {
                 storage_id: sid,
@@ -729,7 +686,7 @@ impl StorageIO for SimulatedIO {
 
         let storage = state.storage.get_mut(&sid).unwrap();
 
-        if reorder_window == 0 {
+        if sync_succeeded && reorder_window == 0 {
             storage.durable = storage.buffered.clone();
         }
         if poison_after {
@@ -738,7 +695,7 @@ impl StorageIO for SimulatedIO {
 
         state.op_log.push(OpRecord::Sync {
             fd: id,
-            succeeded: true,
+            succeeded: sync_succeeded,
         });
         Ok(())
     }
@@ -817,31 +774,17 @@ impl StorageIO for SimulatedIO {
         Ok(())
     }
 
-    fn barrier(&self) -> io::Result<()> {
-        self.jitter();
-        let mut state = self.state.lock().unwrap();
-        let drained: Vec<PendingSync> = state.pending_syncs.drain(..).collect();
-        drained.into_iter().for_each(|p| {
-            if let Some(storage) = state.storage.get_mut(&p.storage_id) {
-                storage.durable = p.snapshot;
-            }
-        });
-        state.op_log.push(OpRecord::Barrier);
-        Ok(())
-    }
-
     fn sync_dir(&self, path: &Path) -> io::Result<()> {
-        let fault = self.effective_fault_config();
         let mut state = self.state.lock().unwrap();
         let seed = self.rng_seed;
 
-        if state.should_fault(seed, fault.io_error_probability) {
+        if state.should_fault(seed, self.fault_config.io_error_probability) {
             return Err(io::Error::other("simulated EIO on sync_dir"));
         }
 
         let dir_path = path.to_path_buf();
         let actually_persisted =
-            !state.should_fault(seed, fault.dir_sync_failure_probability);
+            !state.should_fault(seed, self.fault_config.dir_sync_failure_probability);
 
         if actually_persisted {
             state.dirs_durable.insert(dir_path.clone());

crates/tranquil-store/tests/gauntlet_smoke.rs

@@ -460,115 +460,3 @@ async fn mst_restart_churn_single_seed() {
     assert_clean(&report);
     assert!(report.restarts.0 >= 1);
 }
-
-#[tokio::test]
-async fn torn_pages_only_completes_within_budget() {
-    let cfg = GauntletConfig {
-        seed: Seed(0),
-        io: IoBackend::Simulated {
-            fault: FaultConfig::torn_pages_only(),
-        },
-        workload: WorkloadModel {
-            weights: OpWeights {
-                add: 80,
-                delete: 10,
-                compact: 5,
-                checkpoint: 5,
-                ..OpWeights::default()
-            },
-            size_distribution: SizeDistribution::Fixed(ValueBytes(128)),
-            collections: vec![
-                CollectionName("app.bsky.feed.post".to_string()),
-                CollectionName("app.bsky.feed.like".to_string()),
-            ],
-            key_space: KeySpaceSize(500),
-            did_space: DidSpaceSize(32),
-            retention_max_secs: RetentionMaxSecs(3600),
-        },
-        op_count: OpCount(2_000),
-        invariants: InvariantSet::REFCOUNT_CONSERVATION
-            | InvariantSet::REACHABILITY
-            | InvariantSet::ACKED_WRITE_PERSISTENCE
-            | InvariantSet::READ_AFTER_WRITE
-            | InvariantSet::RESTART_IDEMPOTENT,
-        limits: RunLimits {
-            max_wall_ms: Some(WallMs(60_000)),
-        },
-        restart_policy: RestartPolicy::CrashAtSyscall(OpInterval(500)),
-        store: StoreConfig {
-            max_file_size: MaxFileSize(16 * 1024),
-            group_commit: GroupCommitConfig {
-                verify_persisted_blocks: true,
-                ..GroupCommitConfig::default()
-            },
-            shard_count: ShardCount(1),
-        },
-        eventlog: None,
-        writer_concurrency: WriterConcurrency(1),
-    };
-    let report = Gauntlet::new(cfg).expect("build gauntlet").run().await;
-    let budget_violations: Vec<&str> = report
-        .violations
-        .iter()
-        .filter(|v| v.invariant == "WallClockBudget")
-        .map(|v| v.detail.as_str())
-        .collect();
-    assert!(
-        budget_violations.is_empty(),
-        "torn-pages exceeded budget: {budget_violations:?}; ops_executed={}",
-        report.ops_executed.0
-    );
-    assert_eq!(
-        report.ops_executed.0, 2_000,
-        "expected all ops to execute under torn-pages-only faults"
-    );
-}
-
-#[tokio::test]
-async fn real_io_gauntlet_uses_scratch_root_for_tempdir() {
-    let scratch = tempfile::TempDir::new().expect("scratch dir");
-    let scratch_path = scratch.path().to_path_buf();
-    let cfg = fast_sanity_config(Seed(11));
-    let report = Gauntlet::new(cfg)
-        .expect("build gauntlet")
-        .with_scratch_root(scratch_path.clone())
-        .run()
-        .await;
-    assert_clean(&report);
-    let entries: Vec<std::path::PathBuf> = std::fs::read_dir(&scratch_path)
-        .expect("read scratch")
-        .filter_map(|e| e.ok().map(|e| e.path()))
-        .collect();
-    assert!(
-        entries.is_empty(),
-        "scratch root must be empty after gauntlet drop, found: {entries:?}"
-    );
-}
-
-#[test]
-fn farm_run_many_timed_with_scratch_roots_honors_assignment() {
-    let scratch = tempfile::TempDir::new().expect("scratch dir");
-    let root_a = scratch.path().join("a");
-    let root_b = scratch.path().join("b");
-    std::fs::create_dir_all(&root_a).expect("mkdir a");
-    std::fs::create_dir_all(&root_b).expect("mkdir b");
-    let roots = vec![root_a.clone(), root_b.clone()];
-    let reports = farm::run_many_timed_with_scratch_roots(
-        |seed| fast_sanity_config(seed),
-        &roots,
-        (0..2).map(Seed),
-    );
-    assert_eq!(reports.len(), 2);
-    reports.iter().for_each(|(r, _)| assert_clean(r));
-    [&root_a, &root_b].iter().for_each(|root| {
-        let leftover: Vec<std::path::PathBuf> = std::fs::read_dir(root)
-            .expect("read scratch root")
-            .filter_map(|e| e.ok().map(|e| e.path()))
-            .collect();
-        assert!(
-            leftover.is_empty(),
-            "scratch root {} must be empty after farm completes, found: {leftover:?}",
-            root.display()
-        );
-    });
-}

crates/tranquil-store/tests/sim_blockstore.rs

@@ -12,9 +12,7 @@ use tranquil_store::blockstore::{
     GroupCommitConfig, HINT_RECORD_SIZE, HintFileWriter, HintOffset, TranquilBlockStore,
     WallClockMs, WriteCursor, hint_file_path,
 };
-use tranquil_store::{
-    FaultConfig, OpenOptions, SimulatedIO, StorageIO, SyncReorderWindow, sim_seed_range,
-};
+use tranquil_store::{FaultConfig, OpenOptions, SimulatedIO, StorageIO, sim_seed_range};
 
 use common::{Rng, advance_epoch, block_data, test_cid, with_runtime};
 
@@ -693,60 +691,3 @@ fn sim_multi_file_rotation_crash_recovery() {
         });
     });
 }
-
-#[test]
-fn sim_sync_reorder_loses_first_commit_durability() {
-    with_runtime(|| {
-        let dir = tempfile::TempDir::new().unwrap();
-        let config = BlockStoreConfig {
-            data_dir: dir.path().join("data"),
-            index_dir: dir.path().join("index"),
-            max_file_size: DEFAULT_MAX_FILE_SIZE,
-            group_commit: GroupCommitConfig::default(),
-            shard_count: 1,
-        };
-
-        let fault = FaultConfig {
-            sync_reorder_window: SyncReorderWindow(4),
-            ..FaultConfig::none()
-        };
-        let sim: Arc<SimulatedIO> = Arc::new(SimulatedIO::new(706, fault));
-
-        let cid = test_cid(0);
-        let data = block_data(0);
-
-        {
-            let s = Arc::clone(&sim);
-            let store = TranquilBlockStore::<Arc<SimulatedIO>>::open_with_io(
-                config.clone(),
-                move || Arc::clone(&s),
-            )
-            .unwrap();
-            store
-                .put_blocks_blocking(vec![(cid, data.clone())])
-                .unwrap();
-        }
-
-        sim.crash();
-
-        let s = Arc::clone(&sim);
-        let store = TranquilBlockStore::<Arc<SimulatedIO>>::open_with_io(config, move || {
-            Arc::clone(&s)
-        })
-        .unwrap();
-
-        match store.get_block_sync(&cid) {
-            Ok(Some(d)) => assert_eq!(
-                &d[..],
-                &data[..],
-                "block content mismatch after crash"
-            ),
-            Ok(None) => panic!(
-                "durability bug: put_blocks_blocking returned Ok but block missing after crash"
-            ),
-            Err(e) => panic!(
-                "durability bug: block read failed after crash: {e}"
-            ),
-        }
-    });
-}

crates/tranquil-store/tests/sim_eventlog.rs

@@ -5,12 +5,10 @@ use std::sync::Arc;
 
 use rayon::prelude::*;
 use tranquil_store::eventlog::{
-    DidHash, EVENT_HEADER_SIZE, EVENT_RECORD_OVERHEAD, EventLogWriter, EventSequence, EventTypeTag,
-    MAX_EVENT_PAYLOAD, SEGMENT_HEADER_SIZE, SegmentId, SegmentManager, SegmentReader, ValidEvent,
-};
-use tranquil_store::{
-    FaultConfig, OpenOptions, Probability, SimulatedIO, StorageIO, sim_seed_range,
+    DidHash, EVENT_RECORD_OVERHEAD, EventLogWriter, EventSequence, EventTypeTag, MAX_EVENT_PAYLOAD,
+    SEGMENT_HEADER_SIZE, SegmentId, SegmentManager, SegmentReader, ValidEvent,
 };
+use tranquil_store::{FaultConfig, Probability, SimulatedIO, StorageIO, sim_seed_range};
 
 use common::Rng;
 
@@ -206,13 +204,13 @@ fn crash_mid_rotation_with_faults() {
             EventLogWriter::open(Arc::clone(&mgr), 256, MAX_EVENT_PAYLOAD)
         }));
 
-        if let Ok(Ok(writer)) = recovery {
-            let recovered = writer.synced_seq().raw();
+        if let Ok(Ok(writer)) = recovery
+            && let Ok(synced_before) = write_result
+        {
             assert!(
-                recovered <= events_per_seg as u64,
-                "seed {seed}: recovered {recovered} > written {events_per_seg}"
+                writer.synced_seq().raw() <= synced_before,
+                "seed {seed}: recovered more events than were synced"
             );
-            let _ = write_result;
         }
     });
 }
@@ -1022,169 +1020,3 @@ fn aggressive_faults_group_sync_recovery() {
         );
     });
 }
-
-#[test]
-fn sync_synced_seq_must_match_durable_valid_prefix() {
-    let asserted = std::sync::atomic::AtomicU64::new(0);
-    let range = sim_seed_range();
-    let total = range.end - range.start;
-    range.into_par_iter().for_each(|seed| {
-        let fault_config = FaultConfig {
-            partial_write_probability: Probability::new(0.05),
-            torn_page_probability: Probability::new(0.01),
-            misdirected_write_probability: Probability::new(0.01),
-            sync_failure_probability: Probability::new(0.03),
-            sync_reorder_window: tranquil_store::SyncReorderWindow(4),
-            ..FaultConfig::none()
-        };
-        let sim = SimulatedIO::new(seed, fault_config);
-        let mgr = setup_manager(sim, 64 * 1024);
-
-        let Ok(mut writer) = EventLogWriter::open(Arc::clone(&mgr), 256, MAX_EVENT_PAYLOAD) else {
-            return;
-        };
-
-        let event_count = 10u64;
-        (1..=event_count).for_each(|i| {
-            let _ = append_test_event(&mut writer, i, seed);
-        });
-
-        let synced_through = match writer.sync() {
-            Ok(r) => r.synced_through.raw(),
-            Err(_) => return,
-        };
-        let _ = mgr.io().sync_dir(Path::new(SEGMENTS_DIR));
-
-        if synced_through == 0 {
-            return;
-        }
-
-        let Ok(handle) = mgr.open_for_read(SegmentId::new(1)) else {
-            return;
-        };
-        let Ok(reader) = SegmentReader::open(mgr.io(), handle.fd(), MAX_EVENT_PAYLOAD) else {
-            return;
-        };
-        let Ok(valid) = reader.valid_prefix() else {
-            return;
-        };
-
-        let durable_max = valid.last().map(|e| e.seq.raw()).unwrap_or(0);
-
-        asserted.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
-        assert!(
-            synced_through <= durable_max,
-            "seed {seed}: sync acked seq {synced_through} but durable valid prefix only reaches {durable_max}, events written: {event_count}, valid_prefix.len()={}",
-            valid.len()
-        );
-    });
-
-    let asserted = asserted.load(std::sync::atomic::Ordering::Relaxed);
-    if total >= 50 {
-        assert!(
-            asserted * 2 >= total,
-            "fewer than half of {total} seeds reached the durability assertion: {asserted}"
-        );
-    }
-}
-
-#[test]
-fn reopen_recovers_from_torn_segment_header() {
-    let sim = SimulatedIO::pristine(0);
-    let mgr = setup_manager(sim, 64 * 1024);
-
-    {
-        let mut writer = EventLogWriter::open(Arc::clone(&mgr), 256, MAX_EVENT_PAYLOAD).unwrap();
-        (1..=3).for_each(|i| {
-            let _ = append_test_event(&mut writer, i, 0);
-        });
-        writer.sync().unwrap();
-    }
-    mgr.shutdown();
-
-    let path = mgr.segment_path(SegmentId::new(1));
-    let fd = mgr
-        .io()
-        .open(&path, OpenOptions::read_write_existing())
-        .unwrap();
-    mgr.io().write_all_at(fd, 0, &[0u8; 4]).unwrap();
-    mgr.io().sync(fd).unwrap();
-    mgr.io().sync_dir(Path::new(SEGMENTS_DIR)).unwrap();
-    mgr.io().close(fd).unwrap();
-
-    let writer = EventLogWriter::open(Arc::clone(&mgr), 256, MAX_EVENT_PAYLOAD)
-        .expect("reopen with torn header on highest-numbered segment must succeed");
-    assert_eq!(writer.active_segment_id(), SegmentId::new(1));
-}
-
-#[test]
-fn partial_valid_sync_poisons_writer_and_acks_only_valid_prefix() {
-    let sim = SimulatedIO::pristine(0);
-    let mgr = setup_manager(sim, 64 * 1024);
-    let mut writer = EventLogWriter::open(Arc::clone(&mgr), 256, MAX_EVENT_PAYLOAD).unwrap();
-
-    let payload = b"payload-x".to_vec();
-    let payload_size = payload.len();
-    let record_size = EVENT_RECORD_OVERHEAD + payload_size;
-
-    (1..=5u64).for_each(|i| {
-        writer
-            .append(
-                DidHash::from_did(&format!("did:plc:user{i}")),
-                EventTypeTag::COMMIT,
-                payload.clone(),
-            )
-            .unwrap();
-    });
-
-    let event_3_start = SEGMENT_HEADER_SIZE + 2 * record_size;
-    let event_3_checksum_offset = event_3_start + EVENT_HEADER_SIZE + payload_size;
-
-    let segment_path = mgr.segment_path(SegmentId::new(1));
-    let corrupt_fd = mgr
-        .io()
-        .open(&segment_path, OpenOptions::read_write_existing())
-        .unwrap();
-    mgr.io()
-        .write_all_at(corrupt_fd, event_3_checksum_offset as u64, &[0xFFu8; 4])
-        .unwrap();
-    mgr.io().close(corrupt_fd).unwrap();
-
-    let result = writer.sync().unwrap();
-    assert_eq!(
-        result.synced_through,
-        EventSequence::new(2),
-        "sync must ack only events 1..=2 with corrupt event 3"
-    );
-    assert_eq!(result.flushed_events.len(), 2);
-    assert!(writer.is_poisoned(), "writer must be poisoned after partial sync");
-
-    let append_after_poison = writer.append(
-        DidHash::from_did("did:plc:after"),
-        EventTypeTag::COMMIT,
-        payload.clone(),
-    );
-    assert!(
-        append_after_poison.is_err(),
-        "append must fail on poisoned writer"
-    );
-
-    let sync_after_poison = writer.sync();
-    assert!(
-        sync_after_poison.is_err(),
-        "sync must fail on poisoned writer"
-    );
-
-    drop(writer);
-    let recovered = EventLogWriter::open(Arc::clone(&mgr), 256, MAX_EVENT_PAYLOAD).unwrap();
-    assert_eq!(
-        recovered.synced_seq(),
-        EventSequence::new(2),
-        "reopen must observe synced_seq matching disk's valid prefix"
-    );
-
-    let valid = read_all_events(&mgr, 0);
-    assert_eq!(valid.len(), 2);
-    assert_eq!(valid[0].seq, EventSequence::new(1));
-    assert_eq!(valid[1].seq, EventSequence::new(2));
-}

crates/tranquil-sync/Cargo.toml

@@ -22,3 +22,9 @@ serde = { workspace = true }
 serde_ipld_dagcbor = { workspace = true }
 tokio = { workspace = true }
 tracing = { workspace = true }
+
+[dev-dependencies]
+axum-test = { version = "19.1.1", features = [ "ws" ] }
+sqlx = { workspace = true }
+tokio-util = { workspace = true }
+tracing-subscriber.workspace = true

crates/tranquil-sync/src/subscribe_repos.rs

@@ -302,6 +302,8 @@ async fn handle_socket_inner(
                         break;
                     };
 
+                    info!("{msg:?}");
+
                     if let Message::Close(_) = msg {
                         info!("Client closed connection");
                         break;
@@ -312,3 +314,44 @@ async fn handle_socket_inner(
     }
     Ok(())
 }
+
+#[cfg(test)]
+mod test {
+    use std::net::SocketAddr;
+    use std::time::Duration;
+
+    use super::super::sync_routes;
+    use super::*;
+    use axum_test::TestServer;
+    use tokio_util::sync::CancellationToken;
+
+    #[tokio::test]
+    async fn test_websockets_closing() {
+        // tracing_subscriber::fmt().init();
+        tranquil_config::ensure_test_defaults();
+        let state = AppState::new(CancellationToken::new()).await.unwrap();
+        let app = sync_routes()
+            .with_state(state)
+            .into_make_service_with_connect_info::<SocketAddr>();
+        let server = TestServer::builder().http_transport().build(app);
+
+        const CONNECTIONS: usize = 100;
+        let mut open_sockets = Vec::with_capacity(CONNECTIONS);
+
+        for _ in 0..CONNECTIONS {
+            let socket = server
+                .get_websocket("/com.atproto.sync.subscribeRepos")
+                .await
+                .into_websocket()
+                .await;
+            open_sockets.push(socket);
+        }
+        assert_eq!(SUBSCRIBER_COUNT.load(Ordering::SeqCst), CONNECTIONS);
+
+        drop(open_sockets);
+        // disgusting awful hack to give tokio time to poll the server futures enough times to actually drop all the
+        // websockets on the other end as well
+        tokio::time::sleep(Duration::from_millis(8)).await;
+        assert_eq!(SUBSCRIBER_COUNT.load(Ordering::SeqCst), 0);
+    }
+}

example.toml

@@ -373,117 +373,12 @@
 # Default value: "Tranquil PDS"
 #from_name = "Tranquil PDS"
 
-# HELO/EHLO name announced to remote SMTP servers. Applies to both
-# smarthost and direct-MX modes. Defaults to the server hostname.
+# Path to the `sendmail` binary.
 #
-# Can also be specified via environment variable `MAIL_HELO_NAME`.
-#helo_name =
-
-[email.smarthost]
-# SMTP relay host. When set, mail is delivered through this host
-# instead of resolving recipient MX records directly.
+# Can also be specified via environment variable `SENDMAIL_PATH`.
 #
-# Can also be specified via environment variable `MAIL_SMARTHOST_HOST`.
-#host =
-
-# SMTP relay port.
-#
-# Can also be specified via environment variable `MAIL_SMARTHOST_PORT`.
-#
-# Default value: 587
-#port = 587
-
-# SMTP authentication username.
-#
-# Can also be specified via environment variable `MAIL_SMARTHOST_USERNAME`.
-#username =
-
-# SMTP authentication password.
-#
-# Can also be specified via environment variable `MAIL_SMARTHOST_PASSWORD`.
-#password =
-
-# TLS mode. Valid values: "implicit", "starttls", "none". Setting "none"
-# alongside a password is rejected at startup to prevent transmitting
-# credentials in plaintext.
-#
-# Can also be specified via environment variable `MAIL_SMARTHOST_TLS`.
-#
-# Default value: "starttls"
-#tls = "starttls"
-
-# Max size of the connection pool.
-#
-# Can also be specified via environment variable `MAIL_SMARTHOST_POOL_SIZE`.
-#
-# Default value: 4
-#pool_size = 4
-
-# Per-command SMTP timeout in seconds. Bounds the security handshake.
-#
-# Can also be specified via environment variable `MAIL_SMARTHOST_COMMAND_TIMEOUT_SECS`.
-#
-# Default value: 30
-#command_timeout_secs = 30
-
-# Total per-message timeout in seconds. Wraps the entire send so a
-# stuck relay cannot stall the comms queue.
-#
-# Can also be specified via environment variable `MAIL_SMARTHOST_TOTAL_TIMEOUT_SECS`.
-#
-# Default value: 60
-#total_timeout_secs = 60
-
-[email.direct_mx]
-# Per-command SMTP timeout in seconds.
-#
-# Can also be specified via environment variable `MAIL_COMMAND_TIMEOUT_SECS`.
-#
-# Default value: 30
-#command_timeout_secs = 30
-
-# Total per-message timeout across all MX attempts in seconds.
-#
-# Can also be specified via environment variable `MAIL_TOTAL_TIMEOUT_SECS`.
-#
-# Default value: 60
-#total_timeout_secs = 60
-
-# Max number of concurrent direct-MX sends. Limits the load placed
-# on any single recipient MX during a backlog drain.
-#
-# Can also be specified via environment variable `MAIL_MAX_CONCURRENT_SENDS`.
-#
-# Default value: 8
-#max_concurrent_sends = 8
-
-# Require STARTTLS on every MX hop. When false, TLS is
-# attempted opportunistically and the session falls back to plaintext
-# if the remote does not advertise STARTTLS. Set true to refuse
-# plaintext delivery, at the cost of failing sends to MX hosts that
-# do not support TLS.
-#
-# Can also be specified via environment variable `MAIL_REQUIRE_TLS`.
-#
-# Default value: false
-#require_tls = false
-
-[email.dkim]
-# DKIM selector. When unset, outgoing mail is not signed.
-#
-# Can also be specified via environment variable `MAIL_DKIM_SELECTOR`.
-#selector =
-
-# DKIM signing domain.
-#
-# Can also be specified via environment variable `MAIL_DKIM_DOMAIN`.
-#domain =
-
-# Path to the DKIM private key in PEM format. Supports RSA and
-# Ed25519 keys.
-#
-# Can also be specified via environment variable `MAIL_DKIM_KEY_PATH`.
-#private_key_path =
+# Default value: "/usr/sbin/sendmail"
+#sendmail_path = "/usr/sbin/sendmail"
 
 [discord]
 # Discord bot token. When unset, Discord integration is disabled.