build: bump workspace to 0.6.2

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

.config/nextest.toml

@@ -68,6 +68,10 @@ 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"
@@ -118,6 +122,10 @@ 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"

.sqlx/query-5bee4ed5296667e4ca7e1a97aec28d30a470b8aee7b378ec9ca4e34de4faf349.json

@@ -0,0 +1,15 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "UPDATE comms_queue\n               SET status = 'failed'::comms_status,\n                   attempts = max_attempts,\n                   last_error = $2,\n                   updated_at = NOW()\n               WHERE id = $1",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "5bee4ed5296667e4ca7e1a97aec28d30a470b8aee7b378ec9ca4e34de4faf349"
+}

.sqlx/query-890aa92acdcb0fe2a3bf04d87e1f16a801d271da7cedc32fc42c2ef5b100faae.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "UPDATE comms_queue\n               SET status = 'processing', updated_at = NOW()\n               WHERE id IN (\n                   SELECT id FROM comms_queue\n                   WHERE status = 'pending'\n                     AND scheduled_for <= $1\n                     AND attempts < max_attempts\n                   ORDER BY scheduled_for ASC\n                   LIMIT $2\n                   FOR UPDATE SKIP LOCKED\n               )\n               RETURNING\n                   id, user_id,\n                   channel as \"channel: CommsChannel\",\n                   comms_type as \"comms_type: CommsType\",\n                   status as \"status: CommsStatus\",\n                   recipient, subject, body, metadata,\n                   attempts, max_attempts, last_error,\n                   created_at, updated_at, scheduled_for, processed_at",
+  "query": "UPDATE comms_queue\n               SET status = 'processing', updated_at = NOW()\n               WHERE id IN (\n                   SELECT id FROM comms_queue\n                   WHERE attempts < max_attempts\n                     AND scheduled_for <= $1\n                     AND (\n                         status = 'pending'\n                         OR (status = 'processing'\n                             AND updated_at < $1 - INTERVAL '10 minutes')\n                     )\n                   ORDER BY scheduled_for ASC\n                   LIMIT $2\n                   FOR UPDATE SKIP LOCKED\n               )\n               RETURNING\n                   id, user_id,\n                   channel as \"channel: CommsChannel\",\n                   comms_type as \"comms_type: CommsType\",\n                   status as \"status: CommsStatus\",\n                   recipient, subject, body, metadata,\n                   attempts, max_attempts, last_error,\n                   created_at, updated_at, scheduled_for, processed_at",
   "describe": {
     "columns": [
       {
@@ -154,5 +154,5 @@
       true
     ]
   },
-  "hash": "8047fda41bd94f819213decb8b3e0aba49a8dbdb10217eefd77e3567f8c9694a"
+  "hash": "890aa92acdcb0fe2a3bf04d87e1f16a801d271da7cedc32fc42c2ef5b100faae"
 }

.sqlx/query-8f4357f7a18ddcf6b686a4555f244d37c35917364b8f917ca6ee2d4030ace742.json

@@ -1,6 +1,6 @@
 {
   "db_name": "PostgreSQL",
-  "query": "\n            DELETE FROM oauth_token\n            WHERE id IN (\n                SELECT id FROM oauth_token\n                WHERE did = $1\n                ORDER BY updated_at ASC\n                OFFSET $2\n            )\n            ",
+  "query": "\n            DELETE FROM oauth_token\n            WHERE id IN (\n                SELECT id FROM oauth_token\n                WHERE did = $1\n                ORDER BY created_at DESC\n                OFFSET $2\n            )\n            ",
   "describe": {
     "columns": [],
     "parameters": {
@@ -11,5 +11,5 @@
     },
     "nullable": []
   },
-  "hash": "56cd24903171eddc2ededd9079ffe10937c34e99b0305f25c980ca754da44625"
+  "hash": "8f4357f7a18ddcf6b686a4555f244d37c35917364b8f917ca6ee2d4030ace742"
 }

Cargo.toml

@@ -26,7 +26,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.5.7"
+version = "0.6.2"
 edition = "2024"
 license = "AGPL-3.0-or-later"
 
@@ -93,6 +93,7 @@ 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"
@@ -105,6 +106,8 @@ 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"

Dockerfile

@@ -1,7 +1,7 @@
 FROM node:24-alpine AS frontend
 RUN corepack enable && corepack prepare pnpm@latest --activate
 WORKDIR /app
-COPY frontend/package.json frontend/pnpm-lock.yaml ./
+COPY frontend/package.json frontend/pnpm-lock.yaml frontend/pnpm-workspace.yaml ./
 RUN pnpm install --frozen-lockfile
 COPY frontend/ ./
 RUN pnpm build
@@ -46,8 +46,7 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry \
     cp target/release/tranquil-server /tmp/tranquil-pds
 
 FROM alpine:3.23
-RUN apk add --no-cache msmtp ca-certificates \
-    && ln -sf /usr/bin/msmtp /usr/sbin/sendmail
+RUN apk add --no-cache ca-certificates
 COPY --from=builder /tmp/tranquil-pds /usr/local/bin/tranquil-pds
 COPY --from=frontend /app/dist /var/lib/tranquil-pds/frontend
 WORKDIR /app

crates/tranquil-comms/Cargo.toml

@@ -10,7 +10,14 @@ 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 }
@@ -18,3 +25,7 @@ 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/dkim.rs

@@ -0,0 +1,227 @@
+use std::fs;
+
+use base64::Engine;
+use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
+use ed25519_dalek::pkcs8::DecodePrivateKey as _;
+use lettre::Message;
+use lettre::message::dkim::{
+    DkimCanonicalization, DkimCanonicalizationType, DkimConfig as LettreDkimConfig,
+    DkimSigningAlgorithm, DkimSigningKey,
+};
+use lettre::message::header::HeaderName;
+use rsa::pkcs1::EncodeRsaPrivateKey;
+use rsa::pkcs8::LineEnding;
+
+use super::types::{DkimKeyPath, DkimSelector, EmailDomain};
+use crate::sender::SendError;
+
+const SIGNED_HEADERS: &[&str] = &[
+    "From",
+    "Sender",
+    "Reply-To",
+    "To",
+    "Cc",
+    "Subject",
+    "Date",
+    "In-Reply-To",
+    "References",
+    "MIME-Version",
+    "Content-Type",
+    "Content-Transfer-Encoding",
+];
+
+pub struct DkimSigner {
+    config: LettreDkimConfig,
+}
+
+impl DkimSigner {
+    pub fn load(
+        selector: DkimSelector,
+        domain: EmailDomain,
+        path: DkimKeyPath,
+    ) -> Result<Self, SendError> {
+        let pem = fs::read_to_string(path.as_path()).map_err(|e| {
+            SendError::DkimSign(format!("read DKIM key {}: {e}", path.as_path().display()))
+        })?;
+        Self::from_pem(selector, domain, &pem)
+    }
+
+    pub fn from_pem(
+        selector: DkimSelector,
+        domain: EmailDomain,
+        pem: &str,
+    ) -> Result<Self, SendError> {
+        let key = parse_key(pem)?;
+        let canonicalization = DkimCanonicalization {
+            header: DkimCanonicalizationType::Relaxed,
+            body: DkimCanonicalizationType::Relaxed,
+        };
+        let headers = SIGNED_HEADERS
+            .iter()
+            .copied()
+            .map(HeaderName::new_from_ascii_str)
+            .collect();
+        let config = LettreDkimConfig::new(
+            selector.into_inner(),
+            domain.into_inner(),
+            key,
+            headers,
+            canonicalization,
+        );
+        Ok(Self { config })
+    }
+
+    pub fn sign(&self, message: &mut Message) {
+        message.sign(&self.config);
+    }
+}
+
+impl std::fmt::Debug for DkimSigner {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("DkimSigner")
+    }
+}
+
+fn parse_key(input: &str) -> Result<DkimSigningKey, SendError> {
+    let trimmed = input.trim_start();
+    match trimmed {
+        s if s.starts_with("-----BEGIN RSA PRIVATE KEY-----") => {
+            DkimSigningKey::new(input, DkimSigningAlgorithm::Rsa)
+                .map_err(|e| SendError::DkimSign(format!("RSA PKCS#1 PEM rejected: {e}")))
+        }
+        s if s.starts_with("-----BEGIN PRIVATE KEY-----") => parse_pkcs8(input),
+        s if s.starts_with("-----BEGIN") => Err(SendError::DkimSign(
+            "unrecognized PEM type; expected an RSA or Ed25519 private key".to_string(),
+        )),
+        _ => DkimSigningKey::new(input.trim(), DkimSigningAlgorithm::Ed25519).map_err(|e| {
+            SendError::DkimSign(format!(
+                "expected base64-encoded 32-byte Ed25519 seed or a PEM-wrapped key: {e}"
+            ))
+        }),
+    }
+}
+
+fn parse_pkcs8(pem: &str) -> Result<DkimSigningKey, SendError> {
+    let ed25519_err = match ed25519_dalek::SigningKey::from_pkcs8_pem(pem) {
+        Ok(key) => {
+            let seed = BASE64_STANDARD.encode(key.to_bytes());
+            return DkimSigningKey::new(&seed, DkimSigningAlgorithm::Ed25519)
+                .map_err(|e| SendError::DkimSign(format!("re-import Ed25519 seed: {e}")));
+        }
+        Err(e) => e,
+    };
+
+    let rsa_err = match rsa::RsaPrivateKey::from_pkcs8_pem(pem) {
+        Ok(key) => {
+            let pkcs1 = key
+                .to_pkcs1_pem(LineEnding::LF)
+                .map_err(|e| SendError::DkimSign(format!("re-encode RSA PKCS#8 as PKCS#1: {e}")))?;
+            return DkimSigningKey::new(pkcs1.as_str(), DkimSigningAlgorithm::Rsa)
+                .map_err(|e| SendError::DkimSign(format!("re-import RSA PKCS#1: {e}")));
+        }
+        Err(e) => e,
+    };
+
+    Err(SendError::DkimSign(format!(
+        "PKCS#8 PEM rejected by both parsers; ed25519: {ed25519_err}; rsa: {rsa_err}"
+    )))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use ed25519_dalek::pkcs8::EncodePrivateKey as _;
+    use lettre::message::Mailbox;
+    use lettre::message::header::ContentType;
+    use rsa::pkcs1::DecodeRsaPrivateKey as _;
+
+    const ED25519_RAW_SEED_B64: &str = "QkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkJCQkI=";
+
+    const RSA_PKCS1_PEM: &str = include_str!("test_fixtures/rsa2048-priv-pkcs1.pem");
+
+    fn ed25519_pkcs8_pem() -> String {
+        let key = ed25519_dalek::SigningKey::from_bytes(&[7u8; 32]);
+        key.to_pkcs8_pem(LineEnding::LF).unwrap().to_string()
+    }
+
+    fn rsa_pkcs8_pem() -> String {
+        let key = rsa::RsaPrivateKey::from_pkcs1_pem(RSA_PKCS1_PEM).unwrap();
+        key.to_pkcs8_pem(LineEnding::LF).unwrap().to_string()
+    }
+
+    fn signer(pem: &str) -> DkimSigner {
+        DkimSigner::from_pem(
+            DkimSelector::parse("default").unwrap(),
+            EmailDomain::parse("nel.pet").unwrap(),
+            pem,
+        )
+        .expect("key should load")
+    }
+
+    fn signed_headers(signer: &DkimSigner) -> String {
+        let from: Mailbox = "sender@nel.pet".parse().unwrap();
+        let to: Mailbox = "recipient@nel.pet".parse().unwrap();
+        let mut message = Message::builder()
+            .from(from)
+            .to(to)
+            .subject("Roundtrip")
+            .header(ContentType::TEXT_PLAIN)
+            .body("Body".to_string())
+            .unwrap();
+        signer.sign(&mut message);
+        String::from_utf8(message.formatted()).unwrap()
+    }
+
+    #[test]
+    fn rejects_garbage() {
+        assert!(matches!(
+            parse_key("not a key"),
+            Err(SendError::DkimSign(_))
+        ));
+    }
+
+    #[test]
+    fn rejects_unknown_pem_type() {
+        let pem = "-----BEGIN OPENSSH PRIVATE KEY-----\nx\n-----END OPENSSH PRIVATE KEY-----\n";
+        match parse_key(pem) {
+            Err(SendError::DkimSign(msg)) => assert!(msg.contains("unrecognized"), "msg: {msg}"),
+            other => panic!("expected unrecognized PEM error, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn ed25519_raw_seed_signs() {
+        let raw = signed_headers(&signer(ED25519_RAW_SEED_B64));
+        assert_signed_with(&raw, "a=ed25519-sha256");
+    }
+
+    #[test]
+    fn ed25519_pkcs8_pem_signs() {
+        let raw = signed_headers(&signer(&ed25519_pkcs8_pem()));
+        assert_signed_with(&raw, "a=ed25519-sha256");
+    }
+
+    #[test]
+    fn rsa_pkcs1_pem_signs() {
+        let raw = signed_headers(&signer(RSA_PKCS1_PEM));
+        assert_signed_with(&raw, "a=rsa-sha256");
+    }
+
+    #[test]
+    fn rsa_pkcs8_pem_signs() {
+        let raw = signed_headers(&signer(&rsa_pkcs8_pem()));
+        assert_signed_with(&raw, "a=rsa-sha256");
+    }
+
+    fn assert_signed_with(raw: &str, algorithm: &str) {
+        assert!(
+            raw.contains("DKIM-Signature:"),
+            "no signature header: {raw}"
+        );
+        assert!(raw.contains(algorithm), "missing {algorithm}: {raw}");
+        assert!(
+            raw.contains("c=relaxed/relaxed"),
+            "expected relaxed/relaxed canonicalization: {raw}"
+        );
+    }
+}

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

@@ -0,0 +1,151 @@
+use lettre::Message;
+use lettre::message::Mailbox;
+use lettre::message::header::ContentType;
+use uuid::Uuid;
+
+use super::types::EmailDomain;
+use crate::sender::SendError;
+use crate::types::QueuedComms;
+
+pub(super) fn build(from: &Mailbox, qc: &QueuedComms) -> Result<Message, SendError> {
+    let to: Mailbox = qc
+        .recipient
+        .parse()
+        .map_err(|e: lettre::address::AddressError| SendError::InvalidRecipient(e.to_string()))?;
+    let subject = qc.subject.as_deref().unwrap_or("Notification");
+    let message_id = format!("<{}@{}>", Uuid::new_v4(), from.email.domain());
+    Message::builder()
+        .from(from.clone())
+        .to(to)
+        .subject(subject)
+        .message_id(Some(message_id))
+        .header(ContentType::TEXT_PLAIN)
+        .body(qc.body.clone())
+        .map_err(|e| SendError::MessageBuild(e.to_string()))
+}
+
+pub(super) fn recipient_domain(message: &Message) -> Result<EmailDomain, SendError> {
+    let envelope = message.envelope();
+    let first = envelope
+        .to()
+        .first()
+        .ok_or_else(|| SendError::MessageBuild("envelope has no recipients".to_string()))?;
+    EmailDomain::parse(first.domain())
+        .map_err(|e| SendError::InvalidRecipient(format!("invalid recipient domain: {e}")))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::types::{CommsChannel, CommsStatus, CommsType};
+    use chrono::Utc;
+    use uuid::Uuid;
+
+    fn from_mailbox() -> Mailbox {
+        "Test Sender <noreply@nel.pet>".parse().unwrap()
+    }
+
+    fn fixture(recipient: &str, subject: Option<&str>, body: &str) -> QueuedComms {
+        QueuedComms {
+            id: Uuid::new_v4(),
+            user_id: None,
+            channel: CommsChannel::Email,
+            comms_type: CommsType::Welcome,
+            status: CommsStatus::Pending,
+            recipient: recipient.to_string(),
+            subject: subject.map(String::from),
+            body: body.to_string(),
+            metadata: None,
+            attempts: 0,
+            max_attempts: 3,
+            last_error: None,
+            created_at: Utc::now(),
+            updated_at: Utc::now(),
+            scheduled_for: Utc::now(),
+            processed_at: None,
+        }
+    }
+
+    #[test]
+    fn build_basic_message() {
+        let msg = build(
+            &from_mailbox(),
+            &fixture("user@nel.pet", Some("Welcome"), "Hello world."),
+        )
+        .unwrap();
+        let raw = String::from_utf8(msg.formatted()).unwrap();
+        let lower = raw.to_lowercase();
+        assert!(raw.contains("From: \"Test Sender\" <noreply@nel.pet>"));
+        assert!(raw.contains("To: user@nel.pet"));
+        assert!(raw.contains("Subject: Welcome"));
+        assert!(lower.contains("content-type: text/plain"));
+        assert!(raw.contains("Hello world."));
+    }
+
+    #[test]
+    fn utf8_subject_is_encoded() {
+        let msg = build(
+            &from_mailbox(),
+            &fixture("user@nel.pet", Some("héllo wörld"), "Body"),
+        )
+        .unwrap();
+        let raw = String::from_utf8(msg.formatted()).unwrap();
+        assert!(raw.contains("=?utf-8?"));
+        assert!(!raw.contains("héllo"));
+    }
+
+    #[test]
+    fn header_injection_rejected() {
+        let result = build(
+            &from_mailbox(),
+            &fixture("x@nel.pet\r\nBcc: evil@x", Some("s"), "b"),
+        );
+        assert!(matches!(result, Err(SendError::InvalidRecipient(_))));
+    }
+
+    #[test]
+    fn subject_crlf_does_not_inject_headers() {
+        let msg = build(
+            &from_mailbox(),
+            &fixture("user@nel.pet", Some("hi\r\nBcc: evil@nel.pet"), "body"),
+        )
+        .expect("subject CRLF should be encoded, not rejected");
+        let raw = String::from_utf8(msg.formatted()).unwrap();
+        assert!(
+            !raw.contains("Bcc:"),
+            "CRLF in subject must not produce a Bcc header: {raw}"
+        );
+        assert!(
+            raw.contains("Subject: ="),
+            "subject with non-printable chars should be RFC 2047 encoded: {raw}"
+        );
+    }
+
+    #[test]
+    fn message_id_uses_from_domain() {
+        let msg = build(&from_mailbox(), &fixture("user@nel.pet", Some("s"), "b")).unwrap();
+        let raw = String::from_utf8(msg.formatted()).unwrap();
+        let line = raw
+            .lines()
+            .find(|l| l.starts_with("Message-ID:") || l.starts_with("Message-Id:"))
+            .expect("message-id header present");
+        assert!(
+            line.contains("@nel.pet>"),
+            "message-id should use From domain: {line}"
+        );
+    }
+
+    #[test]
+    fn missing_subject_uses_default() {
+        let msg = build(&from_mailbox(), &fixture("user@nel.pet", None, "Body")).unwrap();
+        let raw = String::from_utf8(msg.formatted()).unwrap();
+        assert!(raw.contains("Subject: Notification"));
+    }
+
+    #[test]
+    fn recipient_domain_extracted() {
+        let msg = build(&from_mailbox(), &fixture("user@Nel.PET", Some("s"), "b")).unwrap();
+        let d = recipient_domain(&msg).unwrap();
+        assert_eq!(d.as_str(), "nel.pet");
+    }
+}

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

@@ -0,0 +1,194 @@
+pub mod dkim;
+pub mod message;
+mod mx;
+pub mod transport;
+pub mod types;
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use async_trait::async_trait;
+use hickory_resolver::TokioAsyncResolver;
+use lettre::message::Mailbox;
+use lettre::transport::smtp::AsyncSmtpTransport;
+use lettre::transport::smtp::PoolConfig;
+use lettre::transport::smtp::authentication::Credentials;
+use lettre::transport::smtp::extension::ClientId;
+use tokio::sync::Semaphore;
+use tracing::{info, warn};
+
+pub use self::dkim::DkimSigner;
+pub use self::transport::SendMode;
+use self::types::{
+    DkimKeyPath, DkimSelector, EmailDomain, HeloName, SmtpHost, SmtpPassword, SmtpPort,
+    SmtpUsername, TlsMode,
+};
+use crate::sender::{CommsSender, SendError};
+use crate::types::{CommsChannel, QueuedComms};
+
+pub struct EmailSender {
+    from: Mailbox,
+    mode: SendMode,
+    dkim: Option<DkimSigner>,
+}
+
+impl EmailSender {
+    pub fn new(from: Mailbox, mode: SendMode, dkim: Option<DkimSigner>) -> Self {
+        Self { from, mode, dkim }
+    }
+
+    pub fn from_config(cfg: &tranquil_config::TranquilConfig) -> Result<Option<Self>, SendError> {
+        let Some(from_address) = cfg.email.from_address.as_deref().filter(|s| !s.is_empty()) else {
+            info!("Email sender disabled: MAIL_FROM_ADDRESS unset");
+            return Ok(None);
+        };
+        let from = build_from(&cfg.email.from_name, from_address)?;
+        let dkim = build_dkim(&cfg.email.dkim)?;
+        let mode = match cfg
+            .email
+            .smarthost
+            .host
+            .as_deref()
+            .filter(|h| !h.is_empty())
+        {
+            Some(host) => build_smarthost(cfg, host)?,
+            None => build_direct_mx(cfg)?,
+        };
+        info!(?mode, dkim = dkim.is_some(), "Email sender initialized");
+        Ok(Some(Self { from, mode, dkim }))
+    }
+}
+
+fn config_invalid(field: &str, error: impl std::fmt::Display) -> SendError {
+    SendError::ConfigInvalid(format!("{field}: {error}"))
+}
+
+fn build_from(from_name: &str, from_address: &str) -> Result<Mailbox, SendError> {
+    let raw = match from_name.is_empty() {
+        true => from_address.to_string(),
+        false => format!("\"{}\" <{}>", from_name.replace('"', "'"), from_address),
+    };
+    raw.parse::<Mailbox>()
+        .map_err(|e| config_invalid("MAIL_FROM_ADDRESS / MAIL_FROM_NAME", e))
+}
+
+fn build_smarthost(
+    cfg: &tranquil_config::TranquilConfig,
+    host_raw: &str,
+) -> Result<SendMode, SendError> {
+    let host = SmtpHost::parse(host_raw).map_err(|e| config_invalid("MAIL_SMARTHOST_HOST", e))?;
+    let port = SmtpPort::parse(cfg.email.smarthost.port)
+        .map_err(|e| config_invalid("MAIL_SMARTHOST_PORT", e))?;
+    let tls = TlsMode::parse(&cfg.email.smarthost.tls)
+        .map_err(|e| config_invalid("MAIL_SMARTHOST_TLS", e))?;
+    let helo = resolve_helo(cfg)?;
+    let pool = PoolConfig::new()
+        .max_size(cfg.email.smarthost.pool_size)
+        .idle_timeout(Duration::from_secs(60));
+    let command_timeout = Duration::from_secs(cfg.email.smarthost.command_timeout_secs);
+    let total_timeout = Duration::from_secs(cfg.email.smarthost.total_timeout_secs);
+
+    let builder = match tls {
+        TlsMode::Implicit => AsyncSmtpTransport::<lettre::Tokio1Executor>::relay(host.as_str())
+            .map_err(|e| config_invalid("smarthost TLS setup", e))?,
+        TlsMode::Starttls => {
+            AsyncSmtpTransport::<lettre::Tokio1Executor>::starttls_relay(host.as_str())
+                .map_err(|e| config_invalid("smarthost TLS setup", e))?
+        }
+        TlsMode::None => {
+            AsyncSmtpTransport::<lettre::Tokio1Executor>::builder_dangerous(host.as_str())
+        }
+    };
+    let builder = builder
+        .port(port.as_u16())
+        .hello_name(ClientId::Domain(helo.into_inner()))
+        .timeout(Some(command_timeout))
+        .pool_config(pool);
+    let builder = match (
+        cfg.email.smarthost.username.as_deref(),
+        cfg.email.smarthost.password.as_deref(),
+    ) {
+        (Some(u), Some(p)) => {
+            let username =
+                SmtpUsername::parse(u).map_err(|e| config_invalid("MAIL_SMARTHOST_USERNAME", e))?;
+            let password =
+                SmtpPassword::parse(p).map_err(|e| config_invalid("MAIL_SMARTHOST_PASSWORD", e))?;
+            builder.credentials(Credentials::new(
+                username.into_inner(),
+                password.expose().to_string(),
+            ))
+        }
+        _ => builder,
+    };
+    Ok(SendMode::Smarthost {
+        transport: Box::new(builder.build()),
+        total_timeout,
+    })
+}
+
+fn build_direct_mx(cfg: &tranquil_config::TranquilConfig) -> Result<SendMode, SendError> {
+    let helo = resolve_helo(cfg)?;
+    let resolver = TokioAsyncResolver::tokio_from_system_conf()
+        .map(Arc::new)
+        .map_err(|e| config_invalid("system DNS configuration", e))?;
+    let max_concurrent = cfg.email.direct_mx.max_concurrent_sends.max(1);
+    Ok(SendMode::DirectMx {
+        resolver,
+        helo,
+        command_timeout: Duration::from_secs(cfg.email.direct_mx.command_timeout_secs),
+        total_timeout: Duration::from_secs(cfg.email.direct_mx.total_timeout_secs),
+        require_tls: cfg.email.direct_mx.require_tls,
+        inflight: Arc::new(Semaphore::new(max_concurrent)),
+    })
+}
+
+fn resolve_helo(cfg: &tranquil_config::TranquilConfig) -> Result<HeloName, SendError> {
+    let raw = cfg
+        .email
+        .helo_name
+        .clone()
+        .unwrap_or_else(|| cfg.server.hostname_without_port().to_string());
+    HeloName::parse(&raw).map_err(|e| config_invalid(&format!("HELO name {raw:?}"), e))
+}
+
+fn build_dkim(cfg: &tranquil_config::DkimConfig) -> Result<Option<DkimSigner>, SendError> {
+    let selector = match cfg.selector.as_deref() {
+        Some(s) => s,
+        None => return Ok(None),
+    };
+    let domain = cfg
+        .domain
+        .as_deref()
+        .ok_or_else(|| SendError::DkimSign("MAIL_DKIM_DOMAIN required when selector set".into()))?;
+    let key_path = cfg.private_key_path.as_deref().ok_or_else(|| {
+        SendError::DkimSign("MAIL_DKIM_KEY_PATH required when selector set".into())
+    })?;
+    let selector = DkimSelector::parse(selector)
+        .map_err(|e| SendError::DkimSign(format!("invalid DKIM selector: {e}")))?;
+    let domain = EmailDomain::parse(domain)
+        .map_err(|e| SendError::DkimSign(format!("invalid DKIM domain: {e}")))?;
+    let path = DkimKeyPath::parse(key_path)
+        .map_err(|e| SendError::DkimSign(format!("DKIM key path invalid: {e}")))?;
+    DkimSigner::load(selector, domain, path).map(Some)
+}
+
+#[async_trait]
+impl CommsSender for EmailSender {
+    fn channel(&self) -> CommsChannel {
+        CommsChannel::Email
+    }
+
+    async fn send(&self, notification: &QueuedComms) -> Result<(), SendError> {
+        let mut message = message::build(&self.from, notification)?;
+        if let Some(signer) = &self.dkim {
+            signer.sign(&mut message);
+        }
+        match transport::dispatch(&self.mode, message).await {
+            Ok(()) => Ok(()),
+            Err(e) => {
+                warn!(comms_id = %notification.id, error = %e, "SMTP send failed");
+                Err(e)
+            }
+        }
+    }
+}

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

@@ -0,0 +1,250 @@
+use hickory_resolver::TokioAsyncResolver;
+use hickory_resolver::error::{ResolveError, ResolveErrorKind};
+use hickory_resolver::proto::op::ResponseCode;
+use rand::seq::SliceRandom;
+
+use super::types::{EmailDomain, MxHost, MxPriority, MxRecord};
+use crate::sender::SendError;
+
+pub async fn resolve(
+    resolver: &TokioAsyncResolver,
+    domain: &EmailDomain,
+) -> Result<Vec<MxRecord>, SendError> {
+    match resolver.mx_lookup(domain.as_str()).await {
+        Ok(lookup) => interpret_lookup(
+            lookup
+                .iter()
+                .map(|mx| (mx.preference(), mx.exchange().clone())),
+            domain,
+        ),
+        Err(e) => classify_lookup_error(e, domain),
+    }
+}
+
+fn interpret_lookup(
+    items: impl IntoIterator<Item = (u16, hickory_resolver::Name)>,
+    domain: &EmailDomain,
+) -> Result<Vec<MxRecord>, SendError> {
+    let entries: Vec<_> = items.into_iter().collect();
+    match entries.iter().any(|(_, name)| name.is_root()) {
+        true => Err(SendError::DnsPermanent(format!(
+            "null MX record at {}: domain refuses mail",
+            domain.as_str()
+        ))),
+        false => {
+            let records: Vec<MxRecord> = entries
+                .into_iter()
+                .filter_map(|(prio, name)| {
+                    MxHost::parse(&name.to_utf8()).ok().map(|host| MxRecord {
+                        priority: MxPriority::new(prio),
+                        host,
+                    })
+                })
+                .collect();
+            match records.is_empty() {
+                true => implicit_mx(domain),
+                false => Ok(prioritize(records)),
+            }
+        }
+    }
+}
+
+fn prioritize(mut records: Vec<MxRecord>) -> Vec<MxRecord> {
+    records.shuffle(&mut rand::thread_rng());
+    records.sort_by_key(|r| r.priority);
+    records
+}
+
+fn classify_lookup_error(
+    e: ResolveError,
+    domain: &EmailDomain,
+) -> Result<Vec<MxRecord>, SendError> {
+    match e.kind() {
+        ResolveErrorKind::NoRecordsFound { response_code, .. } => match *response_code {
+            ResponseCode::NoError => implicit_mx(domain),
+            ResponseCode::NXDomain => Err(SendError::DnsPermanent(format!(
+                "domain {} does not exist",
+                domain.as_str()
+            ))),
+            other => Err(SendError::DnsTransient(format!(
+                "MX lookup for {} failed with {other}",
+                domain.as_str()
+            ))),
+        },
+        _ => Err(SendError::DnsTransient(e.to_string())),
+    }
+}
+
+fn implicit_mx(domain: &EmailDomain) -> Result<Vec<MxRecord>, SendError> {
+    MxHost::parse(domain.as_str())
+        .map(|host| {
+            vec![MxRecord {
+                priority: MxPriority::new(0),
+                host,
+            }]
+        })
+        .map_err(|e| SendError::DnsPermanent(format!("invalid recipient domain: {e}")))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn record(prio: u16, host: &str) -> MxRecord {
+        MxRecord {
+            priority: MxPriority::new(prio),
+            host: MxHost::parse(host).unwrap(),
+        }
+    }
+
+    #[test]
+    fn prioritize_sorts_by_priority_ascending() {
+        let result = prioritize(vec![
+            record(20, "mx2.nel.pet"),
+            record(10, "mx1.nel.pet"),
+            record(10, "mx1b.nel.pet"),
+        ]);
+        assert_eq!(result[0].priority.as_u16(), 10);
+        assert_eq!(result[1].priority.as_u16(), 10);
+        assert_eq!(result[2].priority.as_u16(), 20);
+    }
+
+    #[test]
+    fn prioritize_randomizes_equal_priority_order() {
+        let attempts: Vec<Vec<String>> = (0..200)
+            .map(|_| {
+                prioritize(vec![
+                    record(10, "a.nel.pet"),
+                    record(10, "b.nel.pet"),
+                    record(10, "c.nel.pet"),
+                    record(10, "d.nel.pet"),
+                ])
+                .into_iter()
+                .map(|r| r.host.as_str().to_string())
+                .collect()
+            })
+            .collect();
+        let distinct: std::collections::HashSet<_> = attempts.iter().cloned().collect();
+        assert!(
+            distinct.len() > 1,
+            "equal-priority MX order should vary across calls; got only {}",
+            distinct.len()
+        );
+    }
+
+    #[test]
+    fn implicit_mx_uses_domain_as_host() {
+        let d = EmailDomain::parse("nel.pet").unwrap();
+        let result = implicit_mx(&d).unwrap();
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[0].priority.as_u16(), 0);
+        assert_eq!(result[0].host.as_str(), "nel.pet");
+    }
+
+    #[test]
+    fn no_error_response_yields_implicit_mx() {
+        let d = EmailDomain::parse("nel.pet").unwrap();
+        let err = ResolveError::from(ResolveErrorKind::NoRecordsFound {
+            query: Box::new(hickory_resolver::proto::op::Query::default()),
+            soa: None,
+            negative_ttl: None,
+            response_code: ResponseCode::NoError,
+            trusted: false,
+        });
+        let result = classify_lookup_error(err, &d).unwrap();
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[0].host.as_str(), "nel.pet");
+    }
+
+    #[test]
+    fn nxdomain_response_is_permanent() {
+        let d = EmailDomain::parse("does-not-exist.invalid").unwrap();
+        let err = ResolveError::from(ResolveErrorKind::NoRecordsFound {
+            query: Box::new(hickory_resolver::proto::op::Query::default()),
+            soa: None,
+            negative_ttl: None,
+            response_code: ResponseCode::NXDomain,
+            trusted: true,
+        });
+        match classify_lookup_error(err, &d) {
+            Err(SendError::DnsPermanent(_)) => {}
+            other => panic!("expected DnsPermanent, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn servfail_response_is_transient() {
+        let d = EmailDomain::parse("nel.pet").unwrap();
+        let err = ResolveError::from(ResolveErrorKind::NoRecordsFound {
+            query: Box::new(hickory_resolver::proto::op::Query::default()),
+            soa: None,
+            negative_ttl: None,
+            response_code: ResponseCode::ServFail,
+            trusted: false,
+        });
+        match classify_lookup_error(err, &d) {
+            Err(SendError::DnsTransient(_)) => {}
+            other => panic!("expected DnsTransient, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn timeout_is_transient() {
+        let d = EmailDomain::parse("nel.pet").unwrap();
+        let err = ResolveError::from(ResolveErrorKind::Timeout);
+        match classify_lookup_error(err, &d) {
+            Err(SendError::DnsTransient(_)) => {}
+            other => panic!("expected DnsTransient, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn message_variant_is_transient() {
+        let d = EmailDomain::parse("nel.pet").unwrap();
+        let err = ResolveError::from(ResolveErrorKind::Message("transient resolver glitch"));
+        match classify_lookup_error(err, &d) {
+            Err(SendError::DnsTransient(_)) => {}
+            other => panic!("expected DnsTransient default, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn null_mx_is_permanent() {
+        let d = EmailDomain::parse("nomail.nel.pet").unwrap();
+        let result = interpret_lookup(vec![(0, hickory_resolver::Name::root())], &d);
+        match result {
+            Err(SendError::DnsPermanent(msg)) => {
+                assert!(msg.contains("null MX"), "msg: {msg}")
+            }
+            other => panic!("expected DnsPermanent, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn null_mx_alongside_real_records_still_permanent() {
+        let d = EmailDomain::parse("mixed.nel.pet").unwrap();
+        let real = hickory_resolver::Name::from_ascii("mx1.nel.pet.").unwrap();
+        let result = interpret_lookup(vec![(10, real), (0, hickory_resolver::Name::root())], &d);
+        assert!(matches!(result, Err(SendError::DnsPermanent(_))));
+    }
+
+    #[test]
+    fn empty_lookup_uses_implicit_mx() {
+        let d = EmailDomain::parse("nel.pet").unwrap();
+        let result = interpret_lookup(Vec::<(u16, hickory_resolver::Name)>::new(), &d).unwrap();
+        assert_eq!(result.len(), 1);
+        assert_eq!(result[0].host.as_str(), "nel.pet");
+    }
+
+    #[test]
+    fn valid_records_pass_through_with_priority_sort() {
+        let d = EmailDomain::parse("nel.pet").unwrap();
+        let mx1 = hickory_resolver::Name::from_ascii("mx1.nel.pet.").unwrap();
+        let mx2 = hickory_resolver::Name::from_ascii("mx2.nel.pet.").unwrap();
+        let result = interpret_lookup(vec![(20, mx2), (10, mx1)], &d).unwrap();
+        assert_eq!(result.len(), 2);
+        assert_eq!(result[0].priority.as_u16(), 10);
+        assert_eq!(result[0].host.as_str(), "mx1.nel.pet");
+        assert_eq!(result[1].priority.as_u16(), 20);
+    }
+}

crates/tranquil-comms/src/email/test_fixtures/rsa2048-priv-pkcs1.pem

@@ -0,0 +1,27 @@
+-----BEGIN RSA PRIVATE KEY-----
+MIIEowIBAAKCAQEAtsQsUV8QpqrygsY+2+JCQ6Fw8/omM71IM2N/R8pPbzbgOl0p
+78MZGsgPOQ2HSznjD0FPzsH8oO2B5Uftws04LHb2HJAYlz25+lN5cqfHAfa3fgmC
+38FfwBkn7l582UtPWZ/wcBOnyCgb3yLcvJrXyrt8QxHJgvWO23ITrUVYszImbXQ6
+7YGS0YhMrbixRzmo2tpm3JcIBtnHrEUMsT0NfFdfsZhTT8YbxBvA8FdODgEwx7u/
+vf3J9qbi4+Kv8cvqyJuleIRSjVXPsIMnoejIn04APPKIjpMyQdnWlby7rNyQtE4+
+CV+jcFjqJbE/Xilcvqxt6DirjFCvYeKYl1uHLwIDAQABAoIBAH7Mg2LA7bB0EWQh
+XiL3SrnZG6BpAHAM9jaQ5RFNjua9z7suP5YUaSpnegg/FopeUuWWjmQHudl8bg5A
+ZPgtoLdYoU8XubfUH19I4o1lUXBPVuaeeqn6Yw/HZCjAbSXkVdz8VbesK092ZD/e
+0/4V/3irsn5lrMSq0L322yfvYKaRDFxKCF7UMnWrGcHZl6Msbv/OffLRk19uYB7t
+4WGhK1zCfKIfgdLJnD0eoI6Q4wU6sJvvpyTe8NDDo8HpdAwNn3YSahSewKp9gHgg
+VIQlTZUdsHxM+R+2RUwJZYj9WSTbq+s1nKICUmjQBPnWbrPW963BE5utQPFt3mOe
+EWRzdsECgYEA3MBhJC1Okq+u5yrFE8plufdwNvm9fg5uYUYafvdlQiXsFTx+XDGm
+FXpuWhP/bheOh1jByzPZ1rvjF57xiZjkIuzcvtePTs/b5fT82K7CydDchkc8qb0W
+2dI40h+13e++sUPKYdC9aqjZHzOgl3kOlkDbyRCF3F8mNDujE49rLWcCgYEA0/MU
+dX5A6VSDb5K+JCNq8vDaBKNGU8GAr2fpYAhtk/3mXLI+/Z0JN0di9ZgeNhhJr2jN
+11OU/2pOButpsgnkIo2y36cOQPf5dQpSgXZke3iNDld3osuLIuPNJn/3C087AtOq
++w4YxZClZLAxiLCqX8SBVrB2IiFCQ70SJ++n8vkCgYEAzmi3rBsNEA1jblVIh1PF
+wJhD/bOQ4nBd92iUV8m9jZdl4wl4YX4u/IBI9MMkIG24YIe2VOl7s9Rk5+4/jNg/
+4QQ2998Y6aljxOZJEdZ+3jQELy4m49OhrTRq2ta5t/Z3CMsJTmLe6f9NXWZpr5iK
+8iVdHOjtMXxqfYaR2jVNEtsCgYAl9uWUQiAoa037v0I1wO5YQ9IZgJGJUSDWynsg
+C4JtPs5zji4ASY+sCipsqWnH8MPKGrC8QClxMr51ONe+30yw78a5jvfbpU9Wqpmq
+vOU0xJwnlH1GeMUcY8eMfOFocjG0yOtYeubvBIDLr0/AFzz9WHp+Z69RX7m53nUR
+GDlyKQKBgDGZVAbUBiB8rerqNbONBAxfipoa4IJ+ntBrFT2DtoIZNbSzaoK+nVbH
+kbWMJycaV5PVOh1lfAiZeWCxQz5RcZh/RS8USnxyMG1j4dP/wLcbdasI8uRaSC6Y
+hFHL5HjhLrIo0HRWySS2b2ztBI2FP1M+MaaGFPHDzm2OyZg85yr3
+-----END RSA PRIVATE KEY-----

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

@@ -0,0 +1,164 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use futures::StreamExt;
+use hickory_resolver::TokioAsyncResolver;
+use lettre::transport::smtp::AsyncSmtpTransport;
+use lettre::transport::smtp::Error as SmtpError;
+use lettre::transport::smtp::client::{Tls, TlsParameters};
+use lettre::transport::smtp::extension::ClientId;
+use lettre::{AsyncTransport, Message, Tokio1Executor};
+use tokio::sync::Semaphore;
+
+use super::message::recipient_domain;
+use super::mx;
+use super::types::{HeloName, MxRecord};
+use crate::sender::SendError;
+
+pub enum SendMode {
+    Smarthost {
+        transport: Box<AsyncSmtpTransport<Tokio1Executor>>,
+        total_timeout: Duration,
+    },
+    DirectMx {
+        resolver: Arc<TokioAsyncResolver>,
+        helo: HeloName,
+        command_timeout: Duration,
+        total_timeout: Duration,
+        require_tls: bool,
+        inflight: Arc<Semaphore>,
+    },
+}
+
+impl std::fmt::Debug for SendMode {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Smarthost { total_timeout, .. } => {
+                write!(f, "SendMode::Smarthost(total_timeout={total_timeout:?})")
+            }
+            Self::DirectMx {
+                helo, require_tls, ..
+            } => write!(
+                f,
+                "SendMode::DirectMx({}, require_tls={require_tls})",
+                helo.as_str()
+            ),
+        }
+    }
+}
+
+pub async fn dispatch(mode: &SendMode, message: Message) -> Result<(), SendError> {
+    match mode {
+        SendMode::Smarthost {
+            transport,
+            total_timeout,
+        } => with_total_timeout(*total_timeout, run_send(transport, message)).await,
+        SendMode::DirectMx {
+            resolver,
+            helo,
+            command_timeout,
+            total_timeout,
+            require_tls,
+            inflight,
+        } => {
+            with_total_timeout(*total_timeout, async {
+                let _permit =
+                    inflight.clone().acquire_owned().await.map_err(|_| {
+                        SendError::SmtpTransient("send semaphore closed".to_string())
+                    })?;
+                send_direct(
+                    resolver.as_ref(),
+                    helo,
+                    *command_timeout,
+                    *require_tls,
+                    message,
+                )
+                .await
+            })
+            .await
+        }
+    }
+}
+
+async fn with_total_timeout<F: std::future::Future<Output = Result<(), SendError>>>(
+    total: Duration,
+    fut: F,
+) -> Result<(), SendError> {
+    tokio::time::timeout(total, fut)
+        .await
+        .unwrap_or(Err(SendError::Timeout))
+}
+
+async fn run_send(
+    transport: &AsyncSmtpTransport<Tokio1Executor>,
+    message: Message,
+) -> Result<(), SendError> {
+    transport
+        .send(message)
+        .await
+        .map(|_| ())
+        .map_err(classify_smtp_error)
+}
+
+async fn send_direct(
+    resolver: &TokioAsyncResolver,
+    helo: &HeloName,
+    command_timeout: Duration,
+    require_tls: bool,
+    message: Message,
+) -> Result<(), SendError> {
+    let domain = recipient_domain(&message)?;
+    let mxs = mx::resolve(resolver, &domain).await?;
+    let outcome = futures::stream::iter(mxs)
+        .fold(None::<Result<(), SendError>>, |acc, mx_record| {
+            let message = message.clone();
+            async move {
+                match &acc {
+                    Some(Ok(())) | Some(Err(SendError::SmtpPermanent(_))) => acc,
+                    _ => Some(
+                        attempt_one_host(mx_record, helo, command_timeout, require_tls, message)
+                            .await,
+                    ),
+                }
+            }
+        })
+        .await;
+    outcome.unwrap_or_else(|| {
+        Err(SendError::SmtpTransient(format!(
+            "no MX records returned for {}",
+            domain.as_str()
+        )))
+    })
+}
+
+async fn attempt_one_host(
+    mx_record: MxRecord,
+    helo: &HeloName,
+    command_timeout: Duration,
+    require_tls: bool,
+    message: Message,
+) -> Result<(), SendError> {
+    let host = mx_record.host.as_str().to_string();
+    let tls_params = TlsParameters::new(host.clone())
+        .map_err(|e| SendError::SmtpTransient(format!("TLS params for {host}: {e}")))?;
+    let tls = match require_tls {
+        true => Tls::Required(tls_params),
+        false => Tls::Opportunistic(tls_params),
+    };
+    let transport: AsyncSmtpTransport<Tokio1Executor> =
+        AsyncSmtpTransport::<Tokio1Executor>::builder_dangerous(&host)
+            .port(25)
+            .tls(tls)
+            .hello_name(ClientId::Domain(helo.as_str().to_string()))
+            .timeout(Some(command_timeout))
+            .build();
+    run_send(&transport, message).await
+}
+
+fn classify_smtp_error(e: SmtpError) -> SendError {
+    match () {
+        _ if e.is_permanent() => SendError::SmtpPermanent(e.to_string()),
+        _ if e.is_timeout() => SendError::Timeout,
+        _ => SendError::SmtpTransient(e.to_string()),
+    }
+}

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

@@ -0,0 +1,291 @@
+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-comms/src/lib.rs

@@ -1,13 +1,15 @@
+pub mod email;
 mod locale;
 mod sender;
 mod types;
 
+pub use email::EmailSender;
 pub use locale::{
     DEFAULT_LOCALE, NotificationStrings, VALID_LOCALES, format_message, get_strings,
     validate_locale,
 };
 pub use sender::{
-    CommsSender, DiscordSender, EmailSender, SendError, SignalSender, TelegramSender,
-    is_valid_phone_number, is_valid_signal_username, mime_encode_header, sanitize_header_value,
+    CommsSender, DiscordSender, SendError, SignalSender, TelegramSender, is_valid_phone_number,
+    is_valid_signal_username,
 };
 pub use types::{CommsChannel, CommsStatus, CommsType, NewComms, QueuedComms};

crates/tranquil-comms/src/sender.rs

@@ -1,11 +1,7 @@
 use async_trait::async_trait;
-use base64::{Engine, engine::general_purpose::STANDARD as BASE64};
 use reqwest::Client;
 use serde_json::json;
-use std::process::Stdio;
 use std::time::Duration;
-use tokio::io::AsyncWriteExt;
-use tokio::process::Command;
 
 use super::types::{CommsChannel, QueuedComms};
 
@@ -21,25 +17,51 @@ pub trait CommsSender: Send + Sync {
 
 #[derive(Debug, thiserror::Error)]
 pub enum SendError {
-    #[error("Failed to spawn {command}: {source}")]
-    ProcessSpawn {
-        command: String,
-        source: std::io::Error,
-    },
-    #[error("{command} exited with non-zero status: {detail}")]
-    ProcessFailed { command: String, detail: String },
     #[error("Channel not configured: {0:?}")]
     NotConfigured(CommsChannel),
-    #[error("External service error: {0}")]
-    ExternalService(String),
+    #[error("Email configuration invalid: {0}")]
+    ConfigInvalid(String),
     #[error("Invalid recipient format: {0}")]
     InvalidRecipient(String),
+    #[error("Message construction failed: {0}")]
+    MessageBuild(String),
+    #[error("transient DNS lookup failure: {0}")]
+    DnsTransient(String),
+    #[error("permanent DNS lookup failure: {0}")]
+    DnsPermanent(String),
+    #[error("SMTP transient error: {0}")]
+    SmtpTransient(String),
+    #[error("SMTP permanent error: {0}")]
+    SmtpPermanent(String),
+    #[error("DKIM signing failed: {0}")]
+    DkimSign(String),
+    #[error("External service error: {0}")]
+    ExternalService(String),
     #[error("Request timeout")]
     Timeout,
     #[error("Max retries exceeded: {0}")]
     MaxRetriesExceeded(String),
 }
 
+impl SendError {
+    pub fn is_permanent(&self) -> bool {
+        match self {
+            Self::SmtpPermanent(_)
+            | Self::DnsPermanent(_)
+            | Self::InvalidRecipient(_)
+            | Self::MessageBuild(_)
+            | Self::DkimSign(_)
+            | Self::ConfigInvalid(_) => true,
+            Self::SmtpTransient(_)
+            | Self::DnsTransient(_)
+            | Self::Timeout
+            | Self::ExternalService(_)
+            | Self::MaxRetriesExceeded(_)
+            | Self::NotConfigured(_) => false,
+        }
+    }
+}
+
 fn create_http_client() -> Client {
     Client::builder()
         .timeout(Duration::from_secs(HTTP_TIMEOUT_SECS))
@@ -100,19 +122,6 @@ where
     ))
 }
 
-pub fn sanitize_header_value(value: &str) -> String {
-    value.replace(['\r', '\n'], " ").trim().to_string()
-}
-
-pub fn mime_encode_header(value: &str) -> String {
-    if value.is_ascii() {
-        sanitize_header_value(value)
-    } else {
-        let sanitized = sanitize_header_value(value);
-        format!("=?UTF-8?B?{}?=", BASE64.encode(sanitized.as_bytes()))
-    }
-}
-
 pub fn escape_html(text: &str) -> String {
     text.replace('&', "&")
         .replace('<', "&lt;")
@@ -135,93 +144,6 @@ pub fn is_valid_signal_username(username: &str) -> bool {
     tranquil_signal::SignalUsername::parse(username).is_ok()
 }
 
-pub struct EmailSender {
-    from_address: String,
-    from_name: String,
-    sendmail_path: String,
-}
-
-impl EmailSender {
-    pub fn new(from_address: String, from_name: String, sendmail_path: String) -> Self {
-        Self {
-            from_address,
-            from_name,
-            sendmail_path,
-        }
-    }
-
-    pub fn from_config(cfg: &tranquil_config::TranquilConfig) -> Option<Self> {
-        let from_address = cfg.email.from_address.clone()?;
-        let from_name = cfg.email.from_name.clone();
-        let sendmail_path = cfg.email.sendmail_path.clone();
-        Some(Self::new(from_address, from_name, sendmail_path))
-    }
-
-    pub fn format_email(&self, notification: &QueuedComms) -> String {
-        let subject = mime_encode_header(notification.subject.as_deref().unwrap_or("Notification"));
-        let recipient = sanitize_header_value(&notification.recipient);
-        let from_header = if self.from_name.is_empty() {
-            self.from_address.clone()
-        } else {
-            format!(
-                "{} <{}>",
-                sanitize_header_value(&self.from_name),
-                self.from_address
-            )
-        };
-        format!(
-            "From: {}\r\nTo: {}\r\nSubject: {}\r\nContent-Type: text/plain; charset=utf-8\r\nMIME-Version: 1.0\r\n\r\n{}",
-            from_header, recipient, subject, notification.body
-        )
-    }
-}
-
-#[async_trait]
-impl CommsSender for EmailSender {
-    fn channel(&self) -> CommsChannel {
-        CommsChannel::Email
-    }
-
-    async fn send(&self, notification: &QueuedComms) -> Result<(), SendError> {
-        let email_content = self.format_email(notification);
-        let mut child = Command::new(&self.sendmail_path)
-            .arg("-t")
-            .arg("-oi")
-            .stdin(Stdio::piped())
-            .stdout(Stdio::piped())
-            .stderr(Stdio::piped())
-            .spawn()
-            .map_err(|e| SendError::ProcessSpawn {
-                command: self.sendmail_path.clone(),
-                source: e,
-            })?;
-        if let Some(mut stdin) = child.stdin.take() {
-            stdin
-                .write_all(email_content.as_bytes())
-                .await
-                .map_err(|e| SendError::ProcessSpawn {
-                    command: self.sendmail_path.clone(),
-                    source: e,
-                })?;
-        }
-        let output = child
-            .wait_with_output()
-            .await
-            .map_err(|e| SendError::ProcessSpawn {
-                command: self.sendmail_path.clone(),
-                source: e,
-            })?;
-        if !output.status.success() {
-            let stderr = String::from_utf8_lossy(&output.stderr);
-            return Err(SendError::ProcessFailed {
-                command: self.sendmail_path.clone(),
-                detail: stderr.to_string(),
-            });
-        }
-        Ok(())
-    }
-}
-
 const DISCORD_API_BASE: &str = "https://discord.com/api/v10";
 
 #[derive(Clone)]
@@ -610,3 +532,28 @@ impl CommsSender for SignalSender {
         ))
     }
 }
+
+#[cfg(test)]
+mod is_permanent_matrix {
+    use super::{CommsChannel, SendError};
+
+    #[test]
+    fn permanent_variants_are_permanent() {
+        assert!(SendError::SmtpPermanent("x".into()).is_permanent());
+        assert!(SendError::DnsPermanent("x".into()).is_permanent());
+        assert!(SendError::InvalidRecipient("x".into()).is_permanent());
+        assert!(SendError::MessageBuild("x".into()).is_permanent());
+        assert!(SendError::DkimSign("x".into()).is_permanent());
+        assert!(SendError::ConfigInvalid("x".into()).is_permanent());
+    }
+
+    #[test]
+    fn transient_variants_are_not_permanent() {
+        assert!(!SendError::SmtpTransient("x".into()).is_permanent());
+        assert!(!SendError::DnsTransient("x".into()).is_permanent());
+        assert!(!SendError::Timeout.is_permanent());
+        assert!(!SendError::ExternalService("x".into()).is_permanent());
+        assert!(!SendError::MaxRetriesExceeded("x".into()).is_permanent());
+        assert!(!SendError::NotConfigured(CommsChannel::Email).is_permanent());
+    }
+}

crates/tranquil-comms/tests/email_smtp.rs

@@ -0,0 +1,158 @@
+use std::time::Duration;
+
+use chrono::Utc;
+use lettre::message::Mailbox;
+use lettre::transport::smtp::AsyncSmtpTransport;
+use lettre::transport::smtp::extension::ClientId;
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+use tokio::net::{TcpListener, TcpStream};
+use tranquil_comms::email::transport::SendMode;
+use tranquil_comms::email::{EmailSender, types::HeloName};
+use tranquil_comms::{CommsChannel, CommsSender, CommsStatus, CommsType, QueuedComms, SendError};
+use uuid::Uuid;
+
+fn fixture(recipient: &str, subject: &str, body: &str) -> QueuedComms {
+    QueuedComms {
+        id: Uuid::new_v4(),
+        user_id: None,
+        channel: CommsChannel::Email,
+        comms_type: CommsType::Welcome,
+        status: CommsStatus::Pending,
+        recipient: recipient.to_string(),
+        subject: Some(subject.to_string()),
+        body: body.to_string(),
+        metadata: None,
+        attempts: 0,
+        max_attempts: 3,
+        last_error: None,
+        created_at: Utc::now(),
+        updated_at: Utc::now(),
+        scheduled_for: Utc::now(),
+        processed_at: None,
+    }
+}
+
+fn build_smarthost_sender(host: &str, port: u16) -> EmailSender {
+    build_smarthost_sender_with_total_timeout(host, port, Duration::from_secs(10))
+}
+
+fn build_smarthost_sender_with_total_timeout(
+    host: &str,
+    port: u16,
+    total_timeout: Duration,
+) -> EmailSender {
+    let from: Mailbox = "Tranquil Test <noreply@nel.pet>".parse().unwrap();
+    let helo = HeloName::parse("mta.nel.pet").unwrap();
+    let transport = AsyncSmtpTransport::<lettre::Tokio1Executor>::builder_dangerous(host)
+        .port(port)
+        .hello_name(ClientId::Domain(helo.into_inner()))
+        .timeout(Some(Duration::from_secs(5)))
+        .build();
+    EmailSender::new(
+        from,
+        SendMode::Smarthost {
+            transport: Box::new(transport),
+            total_timeout,
+        },
+        None,
+    )
+}
+
+async fn drive_stub(stream: TcpStream, rcpt_response: &'static [u8]) -> std::io::Result<()> {
+    let (read, mut write) = stream.into_split();
+    let mut reader = BufReader::new(read);
+    write.write_all(b"220 stub ESMTP\r\n").await?;
+    let mut line = String::new();
+    loop {
+        line.clear();
+        let n = reader.read_line(&mut line).await?;
+        if n == 0 {
+            return Ok(());
+        }
+        let upper = line.to_ascii_uppercase();
+        let response: &[u8] = match upper.split_whitespace().next() {
+            Some("EHLO") | Some("HELO") => b"250-stub\r\n250 SIZE 10240000\r\n",
+            Some("MAIL") => b"250 OK\r\n",
+            Some("RCPT") => rcpt_response,
+            Some("DATA") => b"354 end with .\r\n",
+            Some("RSET") => b"250 OK\r\n",
+            Some("QUIT") => b"221 bye\r\n",
+            _ => b"500 unknown\r\n",
+        };
+        write.write_all(response).await?;
+        if upper.starts_with("QUIT") {
+            return Ok(());
+        }
+    }
+}
+
+async fn spawn_stub(rcpt_response: &'static [u8]) -> u16 {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+    tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.unwrap();
+        let _ = drive_stub(stream, rcpt_response).await;
+    });
+    port
+}
+
+#[tokio::test]
+async fn rcpt_550_classifies_as_smtp_permanent() {
+    let port = spawn_stub(b"550 5.1.1 user unknown\r\n").await;
+    let sender = build_smarthost_sender("127.0.0.1", port);
+    let result = sender.send(&fixture("nel@nel.pet", "x", "x")).await;
+    match result {
+        Err(SendError::SmtpPermanent(_)) => {}
+        other => panic!("expected SmtpPermanent, got {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn rcpt_421_classifies_as_smtp_transient() {
+    let port = spawn_stub(b"421 4.7.0 try again later\r\n").await;
+    let sender = build_smarthost_sender("127.0.0.1", port);
+    let result = sender.send(&fixture("nel@nel.pet", "x", "x")).await;
+    match result {
+        Err(SendError::SmtpTransient(_)) => {}
+        other => panic!("expected SmtpTransient, got {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn invalid_recipient_classifies_as_invalid_recipient() {
+    let port = spawn_stub(b"250 OK\r\n").await;
+    let sender = build_smarthost_sender("127.0.0.1", port);
+    let result = sender.send(&fixture("not-an-address", "x", "x")).await;
+    match result {
+        Err(SendError::InvalidRecipient(_)) => {}
+        other => panic!("expected InvalidRecipient, got {other:?}"),
+    }
+}
+
+async fn spawn_silent_stub() -> u16 {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let port = listener.local_addr().unwrap().port();
+    tokio::spawn(async move {
+        let (_stream, _) = listener.accept().await.unwrap();
+        std::future::pending::<()>().await;
+    });
+    port
+}
+
+#[tokio::test]
+async fn smarthost_silent_relay_hits_total_timeout() {
+    let port = spawn_silent_stub().await;
+    let sender =
+        build_smarthost_sender_with_total_timeout("127.0.0.1", port, Duration::from_millis(500));
+    let start = std::time::Instant::now();
+    let result = sender.send(&fixture("nel@nel.pet", "x", "x")).await;
+    let elapsed = start.elapsed();
+    match result {
+        Err(SendError::Timeout) => {}
+        other => panic!("expected Timeout, got {other:?}"),
+    }
+    assert!(
+        elapsed < Duration::from_secs(2),
+        "send returned in {elapsed:?}, expected close to 500ms total_timeout"
+    );
+}

crates/tranquil-config/src/lib.rs

@@ -5,6 +5,14 @@ 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 {
@@ -162,6 +170,14 @@ 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 {
@@ -210,6 +226,10 @@ impl TranquilConfig {
             }
         }
 
+        // -- email -----------------------------------------------------------
+        self.email
+            .validate(self.server.hostname_without_port(), &mut errors);
+
         // -- telegram ---------------------------------------------------------
         if self.telegram.bot_token.is_some() && self.telegram.webhook_secret.is_none() {
             errors.push(
@@ -754,9 +774,278 @@ pub struct EmailConfig {
     #[config(env = "MAIL_FROM_NAME", default = "Tranquil PDS")]
     pub from_name: String,
 
-    /// Path to the `sendmail` binary.
-    #[config(env = "SENDMAIL_PATH", default = "/usr/sbin/sendmail")]
-    pub sendmail_path: 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,
+}
+
+impl EmailConfig {
+    pub fn validate(&self, server_hostname: &str, errors: &mut Vec<String>) {
+        match self.smarthost.tls.to_ascii_lowercase().as_str() {
+            "implicit" | "starttls" => {}
+            "none" => {
+                if self.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
+            .smarthost
+            .host
+            .as_deref()
+            .is_some_and(|h| !h.is_empty());
+        let username_set = self.smarthost.username.is_some();
+        let password_set = self.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.smarthost.command_timeout_secs == 0 {
+            errors.push("email.smarthost.command_timeout_secs must be at least 1".to_string());
+        }
+        if self.smarthost.total_timeout_secs == 0 {
+            errors.push("email.smarthost.total_timeout_secs must be at least 1".to_string());
+        }
+        if self.smarthost.pool_size == 0 {
+            errors.push("email.smarthost.pool_size must be at least 1".to_string());
+        }
+
+        if self.direct_mx.max_concurrent_sends == 0 {
+            errors.push("email.direct_mx.max_concurrent_sends must be at least 1".to_string());
+        }
+        if self.direct_mx.command_timeout_secs == 0 {
+            errors.push("email.direct_mx.command_timeout_secs must be at least 1".to_string());
+        }
+        if self.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.dkim.selector.is_some()
+            || self.dkim.domain.is_some()
+            || self.dkim.private_key_path.is_some();
+        if dkim_set {
+            if self.dkim.selector.is_none() {
+                errors
+                    .push("email.dkim.selector is required when any DKIM field is set".to_string());
+            }
+            if self.dkim.domain.is_none() {
+                errors.push("email.dkim.domain is required when any DKIM field is set".to_string());
+            }
+            if self.dkim.private_key_path.is_none() {
+                errors.push(
+                    "email.dkim.private_key_path is required when any DKIM field is set"
+                        .to_string(),
+                );
+            }
+        }
+
+        let Some(from_address) = self.from_address.as_deref().filter(|s| !s.is_empty()) else {
+            return;
+        };
+
+        if !looks_like_email_address(from_address) {
+            errors.push(format!(
+                "email.from_address {from_address:?} is not a valid email address"
+            ));
+        }
+        if self.from_name.chars().any(|c| c.is_control()) {
+            errors.push("email.from_name must not contain control characters".to_string());
+        }
+
+        let helo_raw = self
+            .helo_name
+            .as_deref()
+            .map(str::to_string)
+            .unwrap_or_else(|| server_hostname.to_string());
+        if !is_non_whitespace_token(&helo_raw) {
+            errors.push(format!(
+                "email HELO name {helo_raw:?} must be non-empty and contain no whitespace"
+            ));
+        }
+
+        if smarthost_host_set {
+            let host = self.smarthost.host.as_deref().unwrap_or("");
+            if !is_non_whitespace_token(host) {
+                errors.push(format!(
+                    "email.smarthost.host {host:?} must contain no whitespace"
+                ));
+            }
+            if self.smarthost.port == 0 {
+                errors.push("email.smarthost.port must be non-zero".to_string());
+            }
+            if let Some(u) = self.smarthost.username.as_deref()
+                && u.is_empty()
+            {
+                errors.push("email.smarthost.username must be non-empty".to_string());
+            }
+            if let Some(p) = self.smarthost.password.as_deref()
+                && p.is_empty()
+            {
+                errors.push("email.smarthost.password must be non-empty".to_string());
+            }
+        }
+
+        if let Some(selector) = self.dkim.selector.as_deref()
+            && !is_valid_dkim_selector(selector)
+        {
+            errors.push(format!(
+                "email.dkim.selector {selector:?} must be valid subdomain syntax"
+            ));
+        }
+        if let Some(domain) = self.dkim.domain.as_deref()
+            && !is_non_whitespace_token(domain)
+        {
+            errors.push(format!(
+                "email.dkim.domain {domain:?} must be non-empty and contain no whitespace"
+            ));
+        }
+        if let Some(key_path) = self.dkim.private_key_path.as_deref()
+            && key_path.trim().is_empty()
+        {
+            errors.push("email.dkim.private_key_path must be non-empty".to_string());
+        }
+    }
+}
+
+fn looks_like_email_address(s: &str) -> bool {
+    let trimmed = s.trim();
+    if trimmed.is_empty() || trimmed.chars().any(char::is_whitespace) {
+        return false;
+    }
+    let mut parts = trimmed.split('@');
+    let local = parts.next().unwrap_or("");
+    let domain = parts.next().unwrap_or("");
+    parts.next().is_none() && !local.is_empty() && !domain.is_empty() && domain.contains('.')
+}
+
+fn is_non_whitespace_token(s: &str) -> bool {
+    let trimmed = s.trim();
+    !trimmed.is_empty() && !trimmed.chars().any(char::is_whitespace)
+}
+
+fn is_valid_dkim_selector(s: &str) -> bool {
+    let trimmed = s.trim();
+    !trimmed.is_empty()
+        && trimmed.split('.').all(|seg| {
+            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, 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>,
 }
 
 #[derive(Debug, Config)]
@@ -1196,3 +1485,189 @@ 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
+        );
+    }
+
+    #[test]
+    fn email_address_predicate_accepts_typical_addresses() {
+        assert!(looks_like_email_address("alice@nel.pet"));
+        assert!(looks_like_email_address("a.b+tag@example.co.uk"));
+    }
+
+    #[test]
+    fn email_address_predicate_rejects_malformed() {
+        assert!(!looks_like_email_address(""));
+        assert!(!looks_like_email_address("no-at-sign"));
+        assert!(!looks_like_email_address("@nel.pet"));
+        assert!(!looks_like_email_address("alice@"));
+        assert!(!looks_like_email_address("alice@nel"));
+        assert!(!looks_like_email_address("a@b@c.com"));
+        assert!(!looks_like_email_address("alice @nel.pet"));
+    }
+
+    #[test]
+    fn dkim_selector_predicate_matches_subdomain_syntax() {
+        assert!(is_valid_dkim_selector("default"));
+        assert!(is_valid_dkim_selector("s2024-q1"));
+        assert!(is_valid_dkim_selector("mailo-2024.nel.pet"));
+        assert!(!is_valid_dkim_selector(""));
+        assert!(!is_valid_dkim_selector("a..b"));
+        assert!(!is_valid_dkim_selector("-leading"));
+        assert!(!is_valid_dkim_selector("trailing-"));
+        assert!(!is_valid_dkim_selector("s_under"));
+    }
+
+    #[test]
+    fn email_validate_disabled_when_from_address_unset() {
+        let cfg = email_config_for_test(EmailOverrides::default());
+        let mut errors = Vec::new();
+        cfg.validate("test.local", &mut errors);
+        assert!(errors.is_empty(), "expected no errors, got {errors:?}");
+    }
+
+    #[test]
+    fn email_validate_rejects_bad_from_address() {
+        let cfg = email_config_for_test(EmailOverrides {
+            from_address: Some("not-an-email"),
+            ..Default::default()
+        });
+        let mut errors = Vec::new();
+        cfg.validate("test.local", &mut errors);
+        assert!(
+            errors.iter().any(|e| e.contains("from_address")),
+            "expected from_address error, got {errors:?}"
+        );
+    }
+
+    #[test]
+    fn email_validate_rejects_smarthost_with_bad_credentials() {
+        let cfg = email_config_for_test(EmailOverrides {
+            from_address: Some("alice@nel.pet"),
+            smarthost_host: Some("smtp.nel.pet"),
+            smarthost_username: Some(""),
+            smarthost_password: Some("hunter2"),
+            ..Default::default()
+        });
+        let mut errors = Vec::new();
+        cfg.validate("test.local", &mut errors);
+        assert!(
+            errors.iter().any(|e| e.contains("smarthost.username")),
+            "expected smarthost.username error, got {errors:?}"
+        );
+    }
+
+    #[test]
+    fn email_validate_rejects_bad_dkim_selector() {
+        let cfg = email_config_for_test(EmailOverrides {
+            from_address: Some("alice@nel.pet"),
+            dkim_selector: Some("-bad"),
+            dkim_domain: Some("nel.pet"),
+            dkim_key_path: Some("/etc/dkim.key"),
+            ..Default::default()
+        });
+        let mut errors = Vec::new();
+        cfg.validate("test.local", &mut errors);
+        assert!(
+            errors.iter().any(|e| e.contains("dkim.selector")),
+            "expected dkim.selector error, got {errors:?}"
+        );
+    }
+
+    #[derive(Default)]
+    struct EmailOverrides {
+        from_address: Option<&'static str>,
+        smarthost_host: Option<&'static str>,
+        smarthost_username: Option<&'static str>,
+        smarthost_password: Option<&'static str>,
+        dkim_selector: Option<&'static str>,
+        dkim_domain: Option<&'static str>,
+        dkim_key_path: Option<&'static str>,
+    }
+
+    fn email_config_for_test(o: EmailOverrides) -> EmailConfig {
+        EmailConfig {
+            from_address: o.from_address.map(str::to_string),
+            from_name: "Tranquil PDS".to_string(),
+            helo_name: None,
+            smarthost: SmarthostConfig {
+                host: o.smarthost_host.map(str::to_string),
+                port: 587,
+                username: o.smarthost_username.map(str::to_string),
+                password: o.smarthost_password.map(str::to_string),
+                tls: "starttls".to_string(),
+                pool_size: 4,
+                command_timeout_secs: 30,
+                total_timeout_secs: 60,
+            },
+            direct_mx: DirectMxConfig {
+                command_timeout_secs: 30,
+                total_timeout_secs: 60,
+                max_concurrent_sends: 8,
+                require_tls: false,
+            },
+            dkim: DkimConfig {
+                selector: o.dkim_selector.map(str::to_string),
+                domain: o.dkim_domain.map(str::to_string),
+                private_key_path: o.dkim_key_path.map(str::to_string),
+            },
+        }
+    }
+}

crates/tranquil-db-traits/src/infra.rs

@@ -244,6 +244,8 @@ pub trait InfraRepository: Send + Sync {
 
     async fn mark_comms_failed(&self, id: Uuid, error: &str) -> Result<(), DbError>;
 
+    async fn mark_comms_failed_permanent(&self, id: Uuid, error: &str) -> Result<(), DbError>;
+
     async fn create_invite_code(
         &self,
         code: &str,

crates/tranquil-db/src/postgres/infra.rs

@@ -65,9 +65,13 @@ impl InfraRepository for PostgresInfraRepository {
                SET status = 'processing', updated_at = NOW()
                WHERE id IN (
                    SELECT id FROM comms_queue
-                   WHERE status = 'pending'
+                   WHERE attempts < max_attempts
                      AND scheduled_for <= $1
-                     AND attempts < max_attempts
+                     AND (
+                         status = 'pending'
+                         OR (status = 'processing'
+                             AND updated_at < $1 - INTERVAL '10 minutes')
+                     )
                    ORDER BY scheduled_for ASC
                    LIMIT $2
                    FOR UPDATE SKIP LOCKED
@@ -127,6 +131,24 @@ impl InfraRepository for PostgresInfraRepository {
         Ok(())
     }
 
+    async fn mark_comms_failed_permanent(&self, id: Uuid, error: &str) -> Result<(), DbError> {
+        sqlx::query!(
+            r#"UPDATE comms_queue
+               SET status = 'failed'::comms_status,
+                   attempts = max_attempts,
+                   last_error = $2,
+                   updated_at = NOW()
+               WHERE id = $1"#,
+            id,
+            error
+        )
+        .execute(&self.pool)
+        .await
+        .map_err(map_sqlx_error)?;
+
+        Ok(())
+    }
+
     async fn create_invite_code(
         &self,
         code: &str,

crates/tranquil-db/src/postgres/oauth.rs

@@ -374,7 +374,7 @@ impl OAuthRepository for PostgresOAuthRepository {
             WHERE id IN (
                 SELECT id FROM oauth_token
                 WHERE did = $1
-                ORDER BY updated_at ASC
+                ORDER BY created_at DESC
                 OFFSET $2
             )
             "#,

crates/tranquil-pds/Cargo.toml

@@ -55,6 +55,7 @@ metrics-exporter-prometheus = { workspace = true }
 multibase = { workspace = true }
 multihash = { workspace = true }
 p256 = { workspace = true }
+parking_lot = { workspace = true }
 rand = { workspace = true }
 redis = { workspace = true, optional = true }
 regex = { workspace = true }

crates/tranquil-pds/src/api/error.rs

@@ -21,6 +21,8 @@ pub enum ApiError {
     InvalidToken(Option<String>),
     ExpiredToken(Option<String>),
     OAuthExpiredToken(Option<String>),
+    UseDpopNonce(String),
+    InvalidDpopProof(String),
     TokenRequired,
     AccountDeactivated,
     AccountTakedown,
@@ -137,6 +139,8 @@ impl ApiError {
             | Self::InvalidToken(_)
             | Self::PasskeyCounterAnomaly
             | Self::OAuthExpiredToken(_)
+            | Self::UseDpopNonce(_)
+            | Self::InvalidDpopProof(_)
             | Self::ReauthRequired { .. } => StatusCode::UNAUTHORIZED,
             Self::InvalidCode(_) => StatusCode::BAD_REQUEST,
             Self::ExpiredToken(_) => StatusCode::BAD_REQUEST,
@@ -236,6 +240,8 @@ impl ApiError {
             Self::AuthenticationFailed(_) => Cow::Borrowed("AuthenticationFailed"),
             Self::InvalidToken(_) => Cow::Borrowed("InvalidToken"),
             Self::ExpiredToken(_) | Self::OAuthExpiredToken(_) => Cow::Borrowed("ExpiredToken"),
+            Self::UseDpopNonce(_) => Cow::Borrowed("use_dpop_nonce"),
+            Self::InvalidDpopProof(_) => Cow::Borrowed("invalid_dpop_proof"),
             Self::TokenRequired => Cow::Borrowed("TokenRequired"),
             Self::AccountDeactivated => Cow::Borrowed("AccountDeactivated"),
             Self::AccountTakedown => Cow::Borrowed("AccountTakedown"),
@@ -335,6 +341,8 @@ impl ApiError {
             Self::ExpiredToken(msg) | Self::OAuthExpiredToken(msg) => {
                 msg.clone().unwrap_or_else(|| "Token has expired".into())
             }
+            Self::UseDpopNonce(_) => "DPoP nonce required".into(),
+            Self::InvalidDpopProof(msg) => msg.clone(),
             Self::RepoNotFound(msg) => msg
                 .clone()
                 .unwrap_or_else(|| "Repository not found".into()),
@@ -560,6 +568,36 @@ impl IntoResponse for ApiError {
                     ),
                 );
             }
+            Self::UseDpopNonce(nonce) => {
+                match HeaderValue::from_str(nonce) {
+                    Ok(val) => {
+                        response
+                            .headers_mut()
+                            .insert(crate::util::HEADER_DPOP_NONCE, val);
+                    }
+                    Err(err) => {
+                        tracing::error!(
+                            ?err,
+                            nonce_len = nonce.len(),
+                            "generated DPoP nonce is not a valid header value"
+                        );
+                    }
+                }
+                response.headers_mut().insert(
+                    http::header::WWW_AUTHENTICATE,
+                    HeaderValue::from_static(
+                        "DPoP error=\"use_dpop_nonce\", error_description=\"Resource server requires nonce in DPoP proof\"",
+                    ),
+                );
+            }
+            Self::InvalidDpopProof(_) => {
+                response.headers_mut().insert(
+                    http::header::WWW_AUTHENTICATE,
+                    HeaderValue::from_static(
+                        "DPoP error=\"invalid_dpop_proof\", error_description=\"Invalid DPoP proof\"",
+                    ),
+                );
+            }
             _ => {}
         }
         response
@@ -596,6 +634,8 @@ impl From<crate::auth::TokenValidationError> for ApiError {
             crate::auth::TokenValidationError::InvalidToken => {
                 Self::AuthenticationFailed(Some("Invalid token format".to_string()))
             }
+            crate::auth::TokenValidationError::UseDpopNonce(nonce) => Self::UseDpopNonce(nonce),
+            crate::auth::TokenValidationError::InvalidDpopProof(msg) => Self::InvalidDpopProof(msg),
         }
     }
 }
@@ -625,10 +665,8 @@ impl From<crate::auth::extractor::AuthError> for ApiError {
             crate::auth::extractor::AuthError::OAuthExpiredToken(msg) => {
                 Self::OAuthExpiredToken(Some(msg))
             }
-            crate::auth::extractor::AuthError::UseDpopNonce(_)
-            | crate::auth::extractor::AuthError::InvalidDpopProof(_) => {
-                Self::AuthenticationFailed(None)
-            }
+            crate::auth::extractor::AuthError::UseDpopNonce(nonce) => Self::UseDpopNonce(nonce),
+            crate::auth::extractor::AuthError::InvalidDpopProof(msg) => Self::InvalidDpopProof(msg),
         }
     }
 }

crates/tranquil-pds/src/auth/extractor.rs

@@ -2,7 +2,7 @@ use std::marker::PhantomData;
 
 use axum::{
     extract::{FromRequestParts, OptionalFromRequestParts, OriginalUri},
-    http::{StatusCode, header::AUTHORIZATION, request::Parts},
+    http::{header::AUTHORIZATION, request::Parts},
     response::{IntoResponse, Response},
 };
 use tracing::{debug, error, info};
@@ -35,32 +35,7 @@ pub enum AuthError {
 
 impl IntoResponse for AuthError {
     fn into_response(self) -> Response {
-        match self {
-            Self::UseDpopNonce(nonce) => (
-                StatusCode::UNAUTHORIZED,
-                [
-                    ("DPoP-Nonce", nonce.as_str()),
-                    ("WWW-Authenticate", "DPoP error=\"use_dpop_nonce\""),
-                ],
-                axum::Json(serde_json::json!({
-                    "error": "use_dpop_nonce",
-                    "message": "DPoP nonce required"
-                })),
-            )
-                .into_response(),
-            Self::OAuthExpiredToken(msg) => ApiError::OAuthExpiredToken(Some(msg)).into_response(),
-            Self::InvalidDpopProof(msg) => (
-                StatusCode::UNAUTHORIZED,
-                [("WWW-Authenticate", "DPoP error=\"invalid_dpop_proof\"")],
-                axum::Json(serde_json::json!({
-                    "error": "invalid_dpop_proof",
-                    "message": msg
-                })),
-            )
-                .into_response(),
-            Self::InsufficientScope(msg) => ApiError::InsufficientScope(Some(msg)).into_response(),
-            other => ApiError::from(other).into_response(),
-        }
+        ApiError::from(self).into_response()
     }
 }
 

crates/tranquil-pds/src/auth/mod.rs

@@ -106,7 +106,7 @@ struct CachedUserStatus {
     is_admin: bool,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub enum TokenValidationError {
     AccountDeactivated,
     AccountTakedown,
@@ -115,6 +115,8 @@ pub enum TokenValidationError {
     TokenExpired,
     OAuthTokenExpired,
     InvalidToken,
+    UseDpopNonce(String),
+    InvalidDpopProof(String),
 }
 
 impl fmt::Display for TokenValidationError {
@@ -126,6 +128,8 @@ impl fmt::Display for TokenValidationError {
             Self::AuthenticationFailed => write!(f, "AuthenticationFailed"),
             Self::TokenExpired | Self::OAuthTokenExpired => write!(f, "ExpiredToken"),
             Self::InvalidToken => write!(f, "InvalidToken"),
+            Self::UseDpopNonce(_) => write!(f, "use_dpop_nonce"),
+            Self::InvalidDpopProof(_) => write!(f, "invalid_dpop_proof"),
         }
     }
 }
@@ -613,6 +617,12 @@ pub async fn validate_token_with_dpop(
         Err(crate::oauth::OAuthError::ExpiredToken(_)) => {
             Err(TokenValidationError::OAuthTokenExpired)
         }
+        Err(crate::oauth::OAuthError::UseDpopNonce(nonce)) => {
+            Err(TokenValidationError::UseDpopNonce(nonce))
+        }
+        Err(crate::oauth::OAuthError::InvalidDpopProof(msg)) => {
+            Err(TokenValidationError::InvalidDpopProof(msg))
+        }
         Err(_) => Err(TokenValidationError::AuthenticationFailed),
     }
 }

crates/tranquil-pds/src/comms/mod.rs

@@ -4,7 +4,7 @@ pub use tranquil_comms::{
     CommsChannel, CommsSender, CommsStatus, CommsType, DEFAULT_LOCALE, DiscordSender, EmailSender,
     NewComms, NotificationStrings, QueuedComms, SendError, SignalSender, TelegramSender,
     VALID_LOCALES, format_message, get_strings, is_valid_phone_number, is_valid_signal_username,
-    mime_encode_header, sanitize_header_value, validate_locale,
+    validate_locale,
 };
 
 pub use service::{CommsService, repo as comms_repo, resolve_delivery_channel};

crates/tranquil-pds/src/comms/service.rs

@@ -149,13 +149,19 @@ impl CommsService {
                 }
             }
             Err(e) => {
+                let permanent = e.is_permanent();
                 let error_msg = e.to_string();
                 warn!(
                     comms_id = %comms_id,
                     error = %error_msg,
+                    permanent,
                     "Failed to send comms"
                 );
-                if let Err(db_err) = self.mark_failed(comms_id, &error_msg).await {
+                let db_result = match permanent {
+                    true => self.mark_failed_permanent(comms_id, &error_msg).await,
+                    false => self.mark_failed(comms_id, &error_msg).await,
+                };
+                if let Err(db_err) = db_result {
                     error!(
                         comms_id = %comms_id,
                         error = %db_err,
@@ -173,6 +179,14 @@ impl CommsService {
     async fn mark_failed(&self, id: Uuid, error: &str) -> Result<(), tranquil_db_traits::DbError> {
         self.infra_repo.mark_comms_failed(id, error).await
     }
+
+    async fn mark_failed_permanent(
+        &self,
+        id: Uuid,
+        error: &str,
+    ) -> Result<(), tranquil_db_traits::DbError> {
+        self.infra_repo.mark_comms_failed_permanent(id, error).await
+    }
 }
 
 struct ResolvedRecipient {

crates/tranquil-pds/src/repo_ops.rs

@@ -526,10 +526,8 @@ pub async fn commit_and_log(
 
     let obsolete_bytes: Vec<Vec<u8>> = obsolete_cids.iter().map(|c| c.to_bytes()).collect();
 
-    let final_ops: HashMap<(&Nsid, &Rkey), &RecordOp> = ops
-        .iter()
-        .map(|op| (op.collection_rkey(), op))
-        .collect();
+    let final_ops: HashMap<(&Nsid, &Rkey), &RecordOp> =
+        ops.iter().map(|op| (op.collection_rkey(), op)).collect();
 
     let final_record_uris: HashSet<AtUri> = final_ops
         .iter()

crates/tranquil-pds/src/scheduled.rs

@@ -399,6 +399,10 @@ pub async fn start_scheduled_tasks(
     let mut compaction_ticker = interval(compaction_interval);
     compaction_ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
 
+    let compaction_blocklist = Arc::new(parking_lot::Mutex::new(CompactionBlocklist::new(
+        Duration::from_secs(300),
+    )));
+
     let mut reachability_ticker = interval(reachability_interval);
     reachability_ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
 
@@ -464,8 +468,9 @@ pub async fn start_scheduled_tasks(
                     let store = store.clone();
                     let threshold = cfg.scheduled.compaction_liveness_threshold;
                     let grace_ms = cfg.scheduled.compaction_grace_period_ms;
+                    let blocklist = Arc::clone(&compaction_blocklist);
                     if let Err(e) = tokio::task::spawn_blocking(move || {
-                        run_compaction_pass(&store, threshold, grace_ms)
+                        run_compaction_pass(&store, threshold, grace_ms, &blocklist)
                     }).await.unwrap_or_else(|e| Err(anyhow::anyhow!("compaction task panicked: {e}"))) {
                         error!("Compaction error: {e}");
                     }
@@ -485,6 +490,8 @@ pub async fn start_scheduled_tasks(
                                 live_refcounted = result.live_refcounted,
                                 leaked_blocks = result.leaked_blocks,
                                 repaired_blocks = result.repaired_blocks,
+                                phantom_files_purged = result.phantom_files_purged,
+                                phantom_blocks_purged = result.phantom_blocks_purged,
                                 bloom_heap_mb = result.bloom_heap_bytes / (1024 * 1024),
                                 "reachability walk complete"
                             );
@@ -536,11 +543,44 @@ pub async fn start_scheduled_tasks(
     }
 }
 
+pub struct CompactionBlocklist {
+    entries: std::collections::HashMap<tranquil_store::blockstore::DataFileId, std::time::Instant>,
+    cool_off: Duration,
+}
+
+impl CompactionBlocklist {
+    pub fn new(cool_off: Duration) -> Self {
+        Self {
+            entries: std::collections::HashMap::new(),
+            cool_off,
+        }
+    }
+
+    pub fn record_failure(&mut self, file_id: tranquil_store::blockstore::DataFileId) {
+        self.entries.insert(file_id, std::time::Instant::now());
+    }
+
+    pub fn is_blocked(&self, file_id: tranquil_store::blockstore::DataFileId) -> bool {
+        self.entries
+            .get(&file_id)
+            .is_some_and(|recorded| recorded.elapsed() < self.cool_off)
+    }
+
+    pub fn prune_expired(&mut self) {
+        let cool_off = self.cool_off;
+        self.entries
+            .retain(|_, recorded| recorded.elapsed() < cool_off);
+    }
+}
+
 fn run_compaction_pass(
     store: &tranquil_store::blockstore::TranquilBlockStore,
     liveness_threshold: f64,
     grace_period_ms: u64,
+    blocklist: &parking_lot::Mutex<CompactionBlocklist>,
 ) -> anyhow::Result<()> {
+    blocklist.lock().prune_expired();
+
     match store.cleanup_gc_meta() {
         Ok(0) => {}
         Ok(n) => info!(count = n, "cleaned up stale gc_meta entries"),
@@ -553,7 +593,11 @@ fn run_compaction_pass(
 
     let candidate = liveness_map
         .iter()
-        .filter(|(_, info)| info.total_blocks > 0 && info.ratio() < liveness_threshold)
+        .filter(|(fid, info)| {
+            info.total_blocks > 0
+                && info.ratio() < liveness_threshold
+                && !blocklist.lock().is_blocked(**fid)
+        })
         .min_by(|(_, a), (_, b)| {
             a.ratio()
                 .partial_cmp(&b.ratio())
@@ -574,21 +618,35 @@ fn run_compaction_pass(
                 "compacting data file"
             );
             match store.compact_file(file_id, grace_period_ms) {
-                Ok(result) => {
+                Ok(tranquil_store::blockstore::CompactionResult::Compacted(stats)) => {
                     info!(
-                        file_id = %result.file_id,
-                        reclaimed_bytes = result.reclaimed_bytes,
-                        live_blocks = result.live_blocks,
-                        dead_blocks = result.dead_blocks,
+                        file_id = %stats.file_id,
+                        reclaimed_bytes = stats.reclaimed_bytes,
+                        live_blocks = stats.live_blocks,
+                        dead_blocks = stats.dead_blocks,
                         "compaction complete"
                     );
                     Ok(())
                 }
+                Ok(tranquil_store::blockstore::CompactionResult::Purged {
+                    file_id,
+                    phantom_blocks,
+                }) => {
+                    warn!(
+                        file_id = %file_id,
+                        phantom_blocks,
+                        "compaction target missing on disk, purged phantom index entries"
+                    );
+                    Ok(())
+                }
                 Err(tranquil_store::blockstore::CompactionError::ActiveFileCannotBeCompacted) => {
                     debug!(file_id = %file_id, "skipped active file");
                     Ok(())
                 }
-                Err(e) => Err(anyhow::anyhow!("compaction failed: {e}")),
+                Err(e) => {
+                    blocklist.lock().record_failure(file_id);
+                    Err(anyhow::anyhow!("compaction failed: {e}"))
+                }
             }
         }
     }
@@ -693,6 +751,24 @@ pub async fn generate_repo_car(
             .await
             .context("Failed to fetch blocks")?;
 
+        let missing: Vec<Cid> = chunk
+            .iter()
+            .zip(blocks.iter())
+            .filter_map(|(cid, block_opt)| block_opt.is_none().then_some(*cid))
+            .collect();
+        if !missing.is_empty() {
+            anyhow::bail!(
+                "repo CAR is incomplete: {} block(s) referenced by the MST are missing from storage. First 5: {}",
+                missing.len(),
+                missing
+                    .iter()
+                    .take(5)
+                    .map(|c| c.to_string())
+                    .collect::<Vec<_>>()
+                    .join(", ")
+            );
+        }
+
         chunk
             .iter()
             .zip(blocks.iter())
@@ -746,6 +822,8 @@ pub struct ReachabilityResult {
     pub leaked_blocks: u64,
     pub repaired_blocks: u64,
     pub bloom_heap_bytes: usize,
+    pub phantom_files_purged: u64,
+    pub phantom_blocks_purged: u64,
 }
 
 const REPO_PAGE_SIZE: i64 = 500;
@@ -761,6 +839,7 @@ fn walk_repo_dag_sync(
     store: &tranquil_store::blockstore::TranquilBlockStore,
     head_cid: &Cid,
     reachable: &mut std::collections::HashSet<CidBytes>,
+    phantom_files: &mut std::collections::HashSet<tranquil_store::blockstore::DataFileId>,
 ) -> anyhow::Result<()> {
     let mut to_visit = vec![cid_to_bytes(head_cid)?];
 
@@ -769,15 +848,56 @@ fn walk_repo_dag_sync(
             continue;
         }
 
-        let block = match store.get_block_sync(&cid_bytes)? {
-            Some(b) => b,
-            None => {
+        let block = match store.get_block_sync(&cid_bytes) {
+            Ok(Some(b)) => b,
+            Ok(None) => {
                 tracing::warn!(
                     ?cid_bytes,
                     "referenced block missing during reachability walk"
                 );
                 continue;
             }
+            Err(e) => {
+                let Some(entry) = store.block_index().get(&cid_bytes) else {
+                    tracing::warn!(
+                        ?cid_bytes,
+                        error = %e,
+                        "reachability walk: index entry vanished between read attempt and re-check"
+                    );
+                    continue;
+                };
+                let file_path = store.data_file_path(entry.location.file_id);
+                match file_path.try_exists() {
+                    Ok(false) => {
+                        tracing::warn!(
+                            ?cid_bytes,
+                            file_id = %entry.location.file_id,
+                            error = %e,
+                            "indexed block points at missing data file, scheduling phantom purge"
+                        );
+                        phantom_files.insert(entry.location.file_id);
+                        continue;
+                    }
+                    Ok(true) => {
+                        return Err(anyhow::anyhow!(
+                            "reachability walk read error on present data file {}: {e}",
+                            entry.location.file_id
+                        ));
+                    }
+                    Err(probe_err) => {
+                        tracing::warn!(
+                            ?cid_bytes,
+                            file_id = %entry.location.file_id,
+                            existence_probe_error = %probe_err,
+                            "could not probe data file existence after read error"
+                        );
+                        return Err(anyhow::anyhow!(
+                            "reachability walk read error on file {}: {e}",
+                            entry.location.file_id
+                        ));
+                    }
+                }
+            }
         };
 
         if let Ok(commit) = Commit::from_cbor(&block) {
@@ -858,13 +978,15 @@ pub fn run_reachability_walk(
 
     let mut repos_walked: u64 = 0;
     let mut seen_heads: std::collections::HashMap<Did, CidLink> = std::collections::HashMap::new();
+    let mut phantom_files: std::collections::HashSet<tranquil_store::blockstore::DataFileId> =
+        std::collections::HashSet::new();
 
     paginate_repos(&rt, repo_repo, |page| {
         page.iter().try_for_each(|repo| -> anyhow::Result<()> {
             let cid =
                 Cid::from_str(repo.repo_root_cid.as_str()).context("invalid repo_root_cid")?;
             seen_heads.insert(repo.did.clone(), repo.repo_root_cid.clone());
-            walk_repo_dag_sync(store, &cid, &mut visited)?;
+            walk_repo_dag_sync(store, &cid, &mut visited, &mut phantom_files)?;
             repos_walked = repos_walked.saturating_add(1);
             if repos_walked.is_multiple_of(1000) {
                 info!(
@@ -894,7 +1016,7 @@ pub fn run_reachability_walk(
             let cid =
                 Cid::from_str(repo.repo_root_cid.as_str()).context("invalid repo_root_cid")?;
             let mut extra = std::collections::HashSet::new();
-            walk_repo_dag_sync(store, &cid, &mut extra)?;
+            walk_repo_dag_sync(store, &cid, &mut extra, &mut phantom_files)?;
             extra.iter().for_each(|c| reachable.insert(c));
             seen_heads.insert(repo.did.clone(), repo.repo_root_cid.clone());
             stale_repos = stale_repos.saturating_add(1);
@@ -922,7 +1044,7 @@ pub fn run_reachability_walk(
                 let cid =
                     Cid::from_str(repo.repo_root_cid.as_str()).context("invalid repo_root_cid")?;
                 let mut extra = std::collections::HashSet::new();
-                walk_repo_dag_sync(store, &cid, &mut extra)?;
+                walk_repo_dag_sync(store, &cid, &mut extra, &mut phantom_files)?;
                 extra.iter().for_each(|c| reachable.insert(c));
                 quiesced_stale = quiesced_stale.saturating_add(1);
                 Ok(())
@@ -958,6 +1080,19 @@ pub fn run_reachability_walk(
         }
     };
 
+    let phantom_files_purged = u64::try_from(phantom_files.len()).unwrap_or(u64::MAX);
+    let phantom_blocks_purged = phantom_files
+        .iter()
+        .map(|fid| store.block_index().purge_by_file_id(*fid))
+        .sum::<u64>();
+
+    if phantom_files_purged > 0 {
+        warn!(
+            phantom_files_purged,
+            phantom_blocks_purged, "purged phantom index entries from unreadable data files"
+        );
+    }
+
     Ok(ReachabilityResult {
         repos_walked,
         blocks_visited,
@@ -965,5 +1100,7 @@ pub fn run_reachability_walk(
         leaked_blocks,
         repaired_blocks,
         bloom_heap_bytes,
+        phantom_files_purged,
+        phantom_blocks_purged,
     })
 }

crates/tranquil-pds/src/state.rs

@@ -523,13 +523,16 @@ fn wire_tranquil_store(
     let metastore =
         Metastore::open(&metastore_dir, metastore_config).expect("failed to open metastore");
 
-    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,
-    })
+    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(),
+    )
     .expect("failed to open blockstore");
 
     let event_log = EventLog::open(
@@ -577,6 +580,18 @@ fn wire_tranquil_store(
                     "repaired orphan data files"
                 );
             }
+            if repair.orphan_hints_removed > 0 {
+                tracing::info!(
+                    removed = repair.orphan_hints_removed,
+                    "repaired orphan hint files"
+                );
+            }
+            if repair.phantom_index_entries_purged > 0 {
+                tracing::info!(
+                    purged = repair.phantom_index_entries_purged,
+                    "purged phantom index entries pointing at missing data files"
+                );
+            }
             if repair.had_errors() {
                 tracing::warn!(errors = repair.repair_errors, "some repairs failed");
             }

crates/tranquil-pds/tests/oauth_token_eviction.rs

@@ -0,0 +1,142 @@
+mod common;
+mod helpers;
+
+use chrono::{DateTime, Duration, Utc};
+use common::{base_url, client, get_test_db_pool, get_test_repos};
+use helpers::verify_new_account;
+use reqwest::StatusCode;
+use serde_json::{Value, json};
+use tranquil_types::Did;
+
+async fn create_account_and_get_did(handle: &str, email: &str, password: &str) -> Did {
+    let client = client();
+    let res = client
+        .post(format!(
+            "{}/xrpc/com.atproto.server.createAccount",
+            base_url().await
+        ))
+        .json(&json!({
+            "handle": handle,
+            "email": email,
+            "password": password,
+        }))
+        .send()
+        .await
+        .expect("createAccount request failed");
+    assert_eq!(res.status(), StatusCode::OK, "createAccount failed");
+    let body: Value = res.json().await.expect("invalid createAccount JSON");
+    let did_str = body["did"]
+        .as_str()
+        .expect("no did in response")
+        .to_string();
+    let _ = verify_new_account(&client, &did_str).await;
+    Did::new(did_str).expect("invalid DID format")
+}
+
+async fn insert_token_with_created_at(
+    pool: &sqlx::PgPool,
+    did: &Did,
+    token_id: &str,
+    created_at: DateTime<Utc>,
+) {
+    sqlx::query(
+        r#"
+        INSERT INTO oauth_token (
+            did, token_id, created_at, updated_at, expires_at,
+            client_id, client_auth, parameters
+        ) VALUES ($1, $2, $3, $3, $4, $5, $6::jsonb, $7::jsonb)
+        "#,
+    )
+    .bind(did.as_str())
+    .bind(token_id)
+    .bind(created_at)
+    .bind(created_at + Duration::hours(1))
+    .bind("https://test.example/client")
+    .bind(r#"{"method":"none"}"#)
+    .bind(
+        r#"{"response_type":"code","client_id":"https://test.example/client","redirect_uri":"https://test.example/cb","code_challenge":"x","code_challenge_method":"S256"}"#,
+    )
+    .execute(pool)
+    .await
+    .expect("token insert failed");
+}
+
+#[tokio::test]
+async fn delete_oldest_tokens_evicts_lowest_created_at() {
+    let ts = Utc::now().timestamp_millis();
+    let handle = format!("tok-evict-{}.test", ts);
+    let email = format!("tok-evict-{}@test.com", ts);
+    let did = create_account_and_get_did(&handle, &email, "EvictTest123!").await;
+
+    let pool = get_test_db_pool().await;
+    let repos = get_test_repos().await;
+
+    let base = Utc::now();
+    let token_ids: Vec<String> = (0..5).map(|i| format!("tok-{}-{}", ts, i)).collect();
+
+    for (i, tid) in token_ids.iter().enumerate() {
+        let created = base + Duration::seconds(i as i64);
+        insert_token_with_created_at(pool, &did, tid, created).await;
+    }
+
+    let count_before = repos
+        .oauth
+        .count_tokens_for_user(&did)
+        .await
+        .expect("count failed");
+    assert_eq!(count_before, 5, "all 5 tokens should be present");
+
+    let deleted = repos
+        .oauth
+        .delete_oldest_tokens_for_user(&did, 3)
+        .await
+        .expect("delete failed");
+    assert_eq!(deleted, 2, "two oldest tokens should be deleted");
+
+    let remaining = repos
+        .oauth
+        .list_tokens_for_user(&did)
+        .await
+        .expect("list failed");
+    assert_eq!(remaining.len(), 3, "three newest tokens should remain");
+
+    let remaining_ids: std::collections::HashSet<String> =
+        remaining.iter().map(|t| t.token_id.0.clone()).collect();
+    let expected_ids: std::collections::HashSet<String> = token_ids[2..].iter().cloned().collect();
+    assert_eq!(
+        remaining_ids, expected_ids,
+        "surviving tokens must be the three newest by created_at"
+    );
+}
+
+#[tokio::test]
+async fn delete_oldest_tokens_no_op_when_under_keep_count() {
+    let ts = Utc::now().timestamp_millis();
+    let handle = format!("tok-evict-noop-{}.test", ts);
+    let email = format!("tok-evict-noop-{}@test.com", ts);
+    let did = create_account_and_get_did(&handle, &email, "EvictTest123!").await;
+
+    let pool = get_test_db_pool().await;
+    let repos = get_test_repos().await;
+
+    let base = Utc::now();
+    for i in 0..2 {
+        let tid = format!("noop-tok-{}-{}", ts, i);
+        let created = base + Duration::seconds(i);
+        insert_token_with_created_at(pool, &did, &tid, created).await;
+    }
+
+    let deleted = repos
+        .oauth
+        .delete_oldest_tokens_for_user(&did, 5)
+        .await
+        .expect("delete failed");
+    assert_eq!(deleted, 0, "nothing to delete when count <= keep");
+
+    let remaining = repos
+        .oauth
+        .list_tokens_for_user(&did)
+        .await
+        .expect("list failed");
+    assert_eq!(remaining.len(), 2);
+}

crates/tranquil-pds/tests/repo_batch.rs

@@ -402,10 +402,7 @@ async fn test_apply_writes_delete_then_create_same_rkey() {
             "{}/xrpc/com.atproto.repo.listRecords",
             base_url().await
         ))
-        .query(&[
-            ("repo", did.as_str()),
-            ("collection", "app.bsky.feed.post"),
-        ])
+        .query(&[("repo", did.as_str()), ("collection", "app.bsky.feed.post")])
         .send()
         .await
         .expect("Failed to list records");
@@ -484,10 +481,7 @@ async fn test_apply_writes_create_then_delete_same_rkey() {
             "{}/xrpc/com.atproto.repo.listRecords",
             base_url().await
         ))
-        .query(&[
-            ("repo", did.as_str()),
-            ("collection", "app.bsky.feed.post"),
-        ])
+        .query(&[("repo", did.as_str()), ("collection", "app.bsky.feed.post")])
         .send()
         .await
         .expect("Failed to list records");

crates/tranquil-pds/tests/security_fixes.rs

@@ -1,43 +1,7 @@
 mod common;
-use tranquil_pds::comms::{
-    SendError, is_valid_phone_number, is_valid_signal_username, sanitize_header_value,
-};
+use tranquil_pds::comms::{SendError, is_valid_phone_number, is_valid_signal_username};
 use tranquil_pds::image::{ImageError, ImageProcessor};
 
-#[test]
-fn test_header_injection_sanitization() {
-    let malicious = "Injected\r\nBcc: attacker@evil.com";
-    let sanitized = sanitize_header_value(malicious);
-    assert!(!sanitized.contains('\r') && !sanitized.contains('\n'));
-    assert!(sanitized.contains("Injected") && sanitized.contains("Bcc:"));
-
-    let normal = "Normal Subject Line";
-    assert_eq!(sanitize_header_value(normal), "Normal Subject Line");
-
-    let padded = "  Subject  ";
-    assert_eq!(sanitize_header_value(padded), "Subject");
-
-    let multi_newline = "Line1\r\nLine2\nLine3\rLine4";
-    let sanitized = sanitize_header_value(multi_newline);
-    assert!(!sanitized.contains('\r') && !sanitized.contains('\n'));
-    assert!(sanitized.contains("Line1") && sanitized.contains("Line4"));
-
-    let header_injection = "Normal Subject\r\nBcc: attacker@evil.com\r\nX-Injected: value";
-    let sanitized = sanitize_header_value(header_injection);
-    assert_eq!(sanitized.split("\r\n").count(), 1);
-    assert!(
-        sanitized.contains("Normal Subject")
-            && sanitized.contains("Bcc:")
-            && sanitized.contains("X-Injected:")
-    );
-
-    let with_null = "client\0id";
-    assert!(sanitize_header_value(with_null).contains("client"));
-
-    let long_input = "x".repeat(10000);
-    assert!(!sanitize_header_value(&long_input).is_empty());
-}
-
 #[test]
 fn test_phone_number_validation() {
     assert!(is_valid_phone_number("+1234567890"));

crates/tranquil-server/src/main.rs

@@ -53,16 +53,12 @@ async fn main() -> ExitCode {
                         return ExitCode::FAILURE;
                     }
                 };
-                match config.validate(*ignore_secrets) {
-                    Ok(()) => {
-                        println!("Configuration is valid.");
-                        ExitCode::SUCCESS
-                    }
-                    Err(e) => {
-                        eprint!("{e}");
-                        ExitCode::FAILURE
-                    }
+                if let Err(e) = config.validate(*ignore_secrets) {
+                    eprint!("{e}");
+                    return ExitCode::FAILURE;
                 }
+                println!("Configuration is valid.");
+                ExitCode::SUCCESS
             }
         };
     }
@@ -141,11 +137,18 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
 
     let cfg = tranquil_config::get();
 
-    if let Some(email_sender) = EmailSender::from_config(cfg) {
-        info!("Email comms enabled");
-        comms_service = comms_service.register_sender(email_sender);
-    } else {
-        warn!("Email comms disabled (MAIL_FROM_ADDRESS not set)");
+    match EmailSender::from_config(cfg) {
+        Ok(Some(email_sender)) => {
+            info!("Email comms enabled");
+            comms_service = comms_service.register_sender(email_sender);
+        }
+        Ok(None) => {
+            warn!("Email comms disabled (MAIL_FROM_ADDRESS unset)");
+        }
+        Err(e) => {
+            error!(error = %e, "Email configuration invalid");
+            return Err(e.into());
+        }
     }
 
     if let Some(discord_sender) = DiscordSender::from_config(cfg) {

crates/tranquil-store/fuzz/fuzz_targets/gauntlet_micro.rs

@@ -76,6 +76,7 @@ fn tiny_config() -> GauntletConfig {
         },
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     }
 }
 

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

@@ -64,6 +64,16 @@ 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,
@@ -95,6 +105,13 @@ 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,
@@ -159,6 +176,8 @@ struct ConfigFile {
     #[serde(default)]
     dump_regressions: Option<PathBuf>,
     #[serde(default)]
+    scratch_roots: Vec<PathBuf>,
+    #[serde(default)]
     overrides: ConfigOverrides,
 }
 
@@ -178,6 +197,8 @@ struct SweepConfigFile {
     #[serde(default)]
     dump_regressions: Option<PathBuf>,
     #[serde(default)]
+    scratch_roots: Vec<PathBuf>,
+    #[serde(default)]
     base_overrides: ConfigOverrides,
     #[serde(default)]
     axes: SweepAxes,
@@ -405,6 +426,7 @@ struct FarmPlan {
     seeds: u64,
     hours: Option<f64>,
     dump_regressions: Option<PathBuf>,
+    scratch_roots: Vec<PathBuf>,
     overrides: ConfigOverrides,
     shrink: bool,
     shrink_budget: usize,
@@ -418,6 +440,7 @@ 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> {
@@ -443,6 +466,11 @@ 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,
@@ -450,12 +478,50 @@ 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"));
@@ -564,6 +630,7 @@ fn main() -> ExitCode {
             hours,
             dump_regressions,
             config,
+            scratch_root,
             no_shrink,
             shrink_budget,
         } => {
@@ -574,6 +641,7 @@ fn main() -> ExitCode {
                 hours,
                 dump_regressions,
                 config,
+                scratch_root,
                 !no_shrink,
                 shrink_budget,
             ) {
@@ -625,6 +693,7 @@ fn main() -> ExitCode {
             seed_start,
             seeds,
             dump_regressions,
+            scratch_root,
             no_shrink,
             shrink_budget,
             max_runs,
@@ -634,6 +703,7 @@ fn main() -> ExitCode {
                 seed_start,
                 seeds,
                 dump_regressions,
+                scratch_root,
                 !no_shrink,
                 shrink_budget,
                 max_runs,
@@ -659,17 +729,20 @@ 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,
@@ -687,6 +760,7 @@ 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());
@@ -705,6 +779,7 @@ fn resolve_sweep(
         seed_start,
         seeds,
         dump_regressions,
+        scratch_roots,
         shrink,
         shrink_budget,
         base_overrides: file.base_overrides,
@@ -749,6 +824,7 @@ fn run_sweep(plan: SweepPlan, rt: &Runtime, interrupt: Arc<AtomicBool>) -> ExitC
         seed_start,
         seeds,
         dump_regressions,
+        scratch_roots,
         shrink,
         shrink_budget,
         base_overrides,
@@ -782,12 +858,13 @@ 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(
+        let reports = farm::run_many_timed_with_scratch_roots(
             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();
@@ -840,6 +917,7 @@ fn run_farm(plan: FarmPlan, rt: &Runtime, interrupt: Arc<AtomicBool>) -> ExitCod
         seeds,
         hours,
         dump_regressions,
+        scratch_roots,
         overrides,
         shrink,
         shrink_budget,
@@ -871,12 +949,13 @@ 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(
+        let reports = farm::run_many_timed_with_scratch_roots(
             |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

@@ -7,7 +7,9 @@ use super::group_commit::{ActiveFileSet, FileIdAllocator};
 use super::hash_index::{BlockIndex, BlockIndexError};
 use super::hint::{HintFileWriter, hint_file_path};
 use super::manager::DataFileManager;
-use super::types::{BlockLocation, CidBytes, CommitEpoch, CompactionResult, DataFileId};
+use super::types::{
+    BlockLocation, CidBytes, CommitEpoch, CompactionResult, CompactionStats, DataFileId,
+};
 
 #[derive(Debug)]
 pub enum CompactionError {
@@ -68,7 +70,13 @@ pub(super) fn compact_on_writer_thread<S: StorageIO>(
         return Err(CompactionError::ActiveFileCannotBeCompacted);
     }
 
-    let source_handle = manager.open_for_read(source_file_id)?;
+    let source_handle = match manager.open_for_read(source_file_id) {
+        Ok(handle) => handle,
+        Err(e) if e.kind() == io::ErrorKind::NotFound => {
+            return purge_phantom_file(manager, index, hint_positions, epoch, source_file_id);
+        }
+        Err(e) => return Err(CompactionError::Io(e)),
+    };
     let source_size = manager.io().file_size(source_handle.fd())?;
 
     let new_file_id = file_ids.allocate();
@@ -92,10 +100,16 @@ pub(super) fn compact_on_writer_thread<S: StorageIO>(
                 .ok();
             Err(e)
         }
-        Ok((new_size, live_count, dead_count)) => {
-            if let Err(e) = index.write_checkpoint(epoch.current(), hint_positions) {
-                tracing::warn!(error = %e, "pre-delete checkpoint failed during compaction");
+        Ok((new_size, live_count, dead_count, new_hint_offset)) => {
+            match live_count {
+                0 => hint_positions.forget_extra(new_file_id),
+                _ => hint_positions.record_extra(new_file_id, new_hint_offset),
             }
+            hint_positions.forget_extra(source_file_id);
+
+            index
+                .write_checkpoint(epoch.current(), hint_positions)
+                .map_err(CompactionError::Io)?;
 
             manager.delete_data_file(source_file_id)?;
             manager
@@ -124,18 +138,51 @@ pub(super) fn compact_on_writer_thread<S: StorageIO>(
                 "compaction complete"
             );
 
-            Ok(CompactionResult {
+            Ok(CompactionResult::Compacted(CompactionStats {
                 file_id: source_file_id,
                 old_size: source_size,
                 new_size,
                 live_blocks: live_count,
                 dead_blocks: dead_count,
                 reclaimed_bytes,
-            })
+            }))
         }
     }
 }
 
+fn purge_phantom_file<S: StorageIO>(
+    manager: &DataFileManager<S>,
+    index: &BlockIndex,
+    hint_positions: &super::group_commit::ShardHintPositions,
+    epoch: &super::types::EpochCounter,
+    source_file_id: DataFileId,
+) -> Result<CompactionResult, CompactionError> {
+    let phantom_blocks = index.purge_by_file_id(source_file_id);
+
+    tracing::warn!(
+        file_id = %source_file_id,
+        phantom_blocks,
+        "source data file missing on disk, purged phantom index entries"
+    );
+
+    hint_positions.forget_extra(source_file_id);
+
+    manager
+        .io()
+        .delete(&hint_file_path(manager.data_dir(), source_file_id))
+        .ok();
+    manager.io().sync_dir(manager.data_dir()).ok();
+
+    index
+        .write_checkpoint(epoch.current(), hint_positions)
+        .map_err(CompactionError::Io)?;
+
+    Ok(CompactionResult::Purged {
+        file_id: source_file_id,
+        phantom_blocks,
+    })
+}
+
 fn stream_compact<S: StorageIO>(
     manager: &DataFileManager<S>,
     index: &BlockIndex,
@@ -144,7 +191,7 @@ fn stream_compact<S: StorageIO>(
     new_file_id: DataFileId,
     current_epoch: CommitEpoch,
     grace_period_ms: u64,
-) -> Result<(u64, u64, u64), CompactionError> {
+) -> Result<(u64, u64, u64, super::types::HintOffset), CompactionError> {
     let mut reader = DataFileReader::open(manager.io(), source_fd)?;
     let now = crate::wall_clock_ms();
 
@@ -222,8 +269,10 @@ 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 final_hint_offset = hint_writer.position();
     let _ = manager.io().close(hint_fd);
 
     finalize_result?;
@@ -232,5 +281,5 @@ fn stream_compact<S: StorageIO>(
 
     index.apply_compaction(&relocations, &dead_cids);
 
-    Ok((new_size, live_count, dead_count))
+    Ok((new_size, live_count, dead_count, final_hint_offset))
 }

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

@@ -74,30 +74,49 @@ impl ActiveFileSet {
 }
 
 pub struct ShardHintPositions {
-    positions: RwLock<Vec<(DataFileId, HintOffset)>>,
+    shard_positions: RwLock<Vec<(DataFileId, HintOffset)>>,
+    extra_positions: RwLock<HashMap<DataFileId, HintOffset>>,
 }
 
 impl ShardHintPositions {
     pub fn new(shard_count: u8) -> Self {
         Self {
-            positions: RwLock::new(
+            shard_positions: RwLock::new(
                 (0..shard_count as usize)
                     .map(|_| (DataFileId::new(0), HintOffset::new(0)))
                     .collect(),
             ),
+            extra_positions: RwLock::new(HashMap::new()),
         }
     }
 
     pub fn update(&self, shard_id: ShardId, file_id: DataFileId, offset: HintOffset) {
-        let mut positions = self.positions.write();
+        let mut positions = self.shard_positions.write();
         let idx = shard_id.as_usize();
         if idx < positions.len() {
             positions[idx] = (file_id, offset);
         }
     }
 
+    pub fn record_extra(&self, file_id: DataFileId, offset: HintOffset) {
+        self.extra_positions.write().insert(file_id, offset);
+    }
+
+    pub fn forget_extra(&self, file_id: DataFileId) {
+        self.extra_positions.write().remove(&file_id);
+    }
+
     pub fn snapshot(&self) -> CheckpointPositions {
-        CheckpointPositions(self.positions.read().clone())
+        let shard = self.shard_positions.read().clone();
+        let extra = self.extra_positions.read().clone();
+        debug_assert!(
+            shard
+                .iter()
+                .filter(|(fid, _)| fid.raw() != 0)
+                .all(|(fid, _)| !extra.contains_key(fid)),
+            "shard_positions and extra_positions must not overlap on the same DataFileId"
+        );
+        CheckpointPositions(shard.into_iter().chain(extra).collect())
     }
 }
 
@@ -1343,6 +1362,10 @@ 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/hash_index.rs

@@ -606,6 +606,34 @@ impl HashTable {
         });
     }
 
+    pub fn cids_in_file(&self, file_id: DataFileId) -> Vec<CidBytes> {
+        self.iter()
+            .filter(|s| s.file_id == file_id)
+            .map(|s| s.cid)
+            .collect()
+    }
+
+    pub fn purge_by_file_id(&mut self, file_id: DataFileId) -> u64 {
+        let victims: Vec<(CidBytes, RefCount)> = self
+            .iter()
+            .filter(|s| s.file_id == file_id)
+            .map(|s| (s.cid, s.refcount))
+            .collect();
+
+        let live_discarded = victims.iter().filter(|(_, rc)| !rc.is_zero()).count();
+        if live_discarded > 0 {
+            tracing::warn!(
+                file_id = %file_id,
+                live_discarded,
+                total_purged = victims.len(),
+                "discarding live index entries for missing data file"
+            );
+        }
+
+        let removed = victims.iter().filter(|(cid, _)| self.remove(cid)).count();
+        u64::try_from(removed).unwrap_or(u64::MAX)
+    }
+
     pub fn cleanup_stale_gc(&mut self) -> u64 {
         self.slots
             .iter_mut()
@@ -1503,6 +1531,14 @@ impl BlockIndex {
             })
     }
 
+    pub fn cids_in_file(&self, file_id: DataFileId) -> Vec<CidBytes> {
+        self.table.read().cids_in_file(file_id)
+    }
+
+    pub fn purge_by_file_id(&self, file_id: DataFileId) -> u64 {
+        self.table.write().purge_by_file_id(file_id)
+    }
+
     pub fn read_write_cursor(&self) -> Option<WriteCursor> {
         self.table.read().write_cursor()
     }

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

@@ -27,11 +27,11 @@ 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, TranquilBlockStore};
+pub use store::{BlockStoreConfig, DEFAULT_SHARD_COUNT, OpenRetryPolicy, TranquilBlockStore};
 pub use types::{
     BlockLength, BlockLocation, BlockOffset, BlockstoreSnapshot, CidBytes, CollectionResult,
-    CommitEpoch, CompactionResult, DataFileId, EpochCounter, HintOffset, IndexEntry, LivenessInfo,
-    MAX_BLOCK_SIZE, RefCount, ShardId, WallClockMs, WriteCursor,
+    CommitEpoch, CompactionResult, CompactionStats, DataFileId, EpochCounter, HintOffset,
+    IndexEntry, LivenessInfo, MAX_BLOCK_SIZE, RefCount, ShardId, WallClockMs, WriteCursor,
 };
 
 use std::io;

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

@@ -1,7 +1,9 @@
 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;
@@ -150,6 +152,24 @@ 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)
@@ -161,6 +181,50 @@ 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> {
@@ -331,15 +395,7 @@ 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) => {
-                    tracing::warn!(
-                        file_id = %file_id,
-                        offset = scan_pos.raw(),
-                        error = %e,
-                        "IO error during recovery scan, stopping"
-                    );
-                    None
-                }
+                Err(e) => Some(Err(e)),
                 Ok(None) => None,
                 Ok(Some(ReadBlockRecord::Valid {
                     offset,
@@ -354,7 +410,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((
+                    Some(Ok((
                         cid_bytes,
                         BlockLocation {
                             file_id,
@@ -362,20 +418,30 @@ impl<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
                             length,
                         },
                         new_end,
-                    ))
+                    )))
                 }
                 Ok(Some(ReadBlockRecord::Corrupted { .. } | ReadBlockRecord::Truncated { .. })) => {
                     None
                 }
             }
         })
-        .fold(
+        .try_fold(
             (Vec::new(), start_offset),
-            |(mut entries, _), (cid, loc, new_end)| {
+            |(mut entries, _), item: io::Result<_>| {
+                let (cid, loc, new_end) = item?;
                 entries.push((cid, loc));
-                (entries, new_end)
+                Ok::<_, io::Error>((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!(
@@ -533,6 +599,16 @@ impl<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
             .map_err(RepoError::storage)
     }
 
+    pub fn list_hint_files(&self) -> Result<Vec<DataFileId>, RepoError> {
+        let io = self.reader.manager().io();
+        super::list_files_by_extension(io, &self.data_dir, super::hint::HINT_FILE_EXTENSION)
+            .map_err(RepoError::storage)
+    }
+
+    pub fn hint_file_path(&self, file_id: DataFileId) -> std::path::PathBuf {
+        super::hint::hint_file_path(&self.data_dir, file_id)
+    }
+
     pub fn put_blocks_blocking(
         &self,
         blocks: Vec<([u8; CID_SIZE], Vec<u8>)>,
@@ -713,3 +789,210 @@ 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/blockstore/types.rs

@@ -68,7 +68,8 @@ pub struct CollectionResult {
     pub total_bytes: u64,
 }
 
-pub struct CompactionResult {
+#[derive(Debug)]
+pub struct CompactionStats {
     pub file_id: DataFileId,
     pub old_size: u64,
     pub new_size: u64,
@@ -77,6 +78,24 @@ pub struct CompactionResult {
     pub reclaimed_bytes: u64,
 }
 
+#[derive(Debug)]
+pub enum CompactionResult {
+    Compacted(CompactionStats),
+    Purged {
+        file_id: DataFileId,
+        phantom_blocks: u64,
+    },
+}
+
+impl CompactionResult {
+    pub fn file_id(&self) -> DataFileId {
+        match self {
+            Self::Compacted(stats) => stats.file_id,
+            Self::Purged { file_id, .. } => *file_id,
+        }
+    }
+}
+
 pub struct LivenessInfo {
     pub live_bytes: u64,
     pub total_bytes: u64,

crates/tranquil-store/src/consistency.rs

@@ -28,6 +28,8 @@ pub struct ConsistencyReport {
     pub orphaned_user_repos: Vec<OrphanedUserRepo>,
     pub inconsistent_handles: Vec<InconsistentHandle>,
     pub orphan_data_files: Vec<DataFileId>,
+    pub orphan_hint_files: Vec<DataFileId>,
+    pub missing_indexed_files: Vec<DataFileId>,
     pub deserialization_failures: u64,
     pub eventlog_contiguity: Option<SequenceContiguityResult>,
     pub cursor_ahead_of_eventlog: bool,
@@ -74,6 +76,8 @@ impl ConsistencyReport {
             && self.orphaned_user_repos.is_empty()
             && self.inconsistent_handles.is_empty()
             && self.orphan_data_files.is_empty()
+            && self.orphan_hint_files.is_empty()
+            && self.missing_indexed_files.is_empty()
             && self.deserialization_failures == 0
             && self
                 .eventlog_contiguity
@@ -84,6 +88,8 @@ impl ConsistencyReport {
 
     pub fn has_repairable_issues(&self) -> bool {
         !self.orphan_data_files.is_empty()
+            || !self.orphan_hint_files.is_empty()
+            || !self.missing_indexed_files.is_empty()
     }
 
     pub fn has_unrecoverable_issues(&self) -> bool {
@@ -136,6 +142,20 @@ impl ConsistencyReport {
                 "orphan data files with no index references"
             );
         }
+        if !self.orphan_hint_files.is_empty() {
+            tracing::warn!(
+                count = self.orphan_hint_files.len(),
+                files = ?self.orphan_hint_files,
+                "orphan hint files with no matching data file"
+            );
+        }
+        if !self.missing_indexed_files.is_empty() {
+            tracing::warn!(
+                count = self.missing_indexed_files.len(),
+                files = ?self.missing_indexed_files,
+                "index references data files that are missing on disk"
+            );
+        }
         if self.deserialization_failures > 0 {
             tracing::error!(
                 count = self.deserialization_failures,
@@ -181,13 +201,15 @@ impl fmt::Display for ConsistencyReport {
         write!(
             f,
             "INCONSISTENT: dangling_roots={}, dangling_records={}, orphaned_repos={}, \
-             inconsistent_handles={}, orphan_files={}, deserialize_failures={}, \
-             eventlog_gaps={}, cursor_ahead={}",
+             inconsistent_handles={}, orphan_files={}, orphan_hints={}, missing_indexed_files={}, \
+             deserialize_failures={}, eventlog_gaps={}, cursor_ahead={}",
             self.dangling_root_cids.len(),
             self.dangling_record_cids.len(),
             self.orphaned_user_repos.len(),
             self.inconsistent_handles.len(),
             self.orphan_data_files.len(),
+            self.orphan_hint_files.len(),
+            self.missing_indexed_files.len(),
             self.deserialization_failures,
             self.eventlog_contiguity
                 .as_ref()
@@ -204,6 +226,8 @@ pub struct ConsistencyCheckOptions {
     pub check_user_blocks: bool,
     pub check_eventlog: bool,
     pub check_orphan_files: bool,
+    pub check_missing_indexed_files: bool,
+    pub check_orphan_hint_files: bool,
 }
 
 impl Default for ConsistencyCheckOptions {
@@ -214,6 +238,8 @@ impl Default for ConsistencyCheckOptions {
             check_user_blocks: true,
             check_eventlog: true,
             check_orphan_files: true,
+            check_missing_indexed_files: true,
+            check_orphan_hint_files: true,
         }
     }
 }
@@ -269,6 +295,14 @@ pub fn verify_store_consistency_with_options<S: StorageIO + 'static>(
         check_orphan_data_files(blockstore, block_index, &mut report);
     }
 
+    if options.check_missing_indexed_files {
+        check_missing_indexed_files(blockstore, block_index, &mut report);
+    }
+
+    if options.check_orphan_hint_files {
+        check_orphan_hint_files(blockstore, &mut report);
+    }
+
     report
 }
 
@@ -565,6 +599,52 @@ fn check_orphan_data_files(
     });
 }
 
+fn check_missing_indexed_files(
+    blockstore: &TranquilBlockStore,
+    block_index: &BlockIndex,
+    report: &mut ConsistencyReport,
+) {
+    let disk_files: HashSet<DataFileId> = match blockstore.list_data_files() {
+        Ok(files) => files.into_iter().collect(),
+        Err(e) => {
+            tracing::warn!(error = %e, "failed to list data files for missing-file check");
+            return;
+        }
+    };
+
+    let epoch = blockstore.epoch().current();
+    let now = crate::wall_clock_ms();
+    let indexed_files = block_index.liveness_by_file(epoch, now, 0);
+
+    indexed_files
+        .iter()
+        .filter(|(fid, _)| !disk_files.contains(fid))
+        .for_each(|(fid, _)| report.missing_indexed_files.push(*fid));
+}
+
+fn check_orphan_hint_files(blockstore: &TranquilBlockStore, report: &mut ConsistencyReport) {
+    let data_files: HashSet<DataFileId> = match blockstore.list_data_files() {
+        Ok(files) => files.into_iter().collect(),
+        Err(e) => {
+            tracing::warn!(error = %e, "failed to list data files for orphan-hint check");
+            return;
+        }
+    };
+
+    let hint_files = match blockstore.list_hint_files() {
+        Ok(files) => files,
+        Err(e) => {
+            tracing::warn!(error = %e, "failed to list hint files for orphan-hint check");
+            return;
+        }
+    };
+
+    hint_files
+        .iter()
+        .filter(|fid| !data_files.contains(fid))
+        .for_each(|fid| report.orphan_hint_files.push(*fid));
+}
+
 fn try_cid_bytes_to_fixed(bytes: &[u8]) -> Option<[u8; CID_SIZE]> {
     bytes.try_into().ok()
 }
@@ -621,12 +701,39 @@ pub fn repair_known_issues(
         }
     });
 
+    report.orphan_hint_files.iter().for_each(|&file_id| {
+        let path = blockstore.hint_file_path(file_id);
+        match std::fs::remove_file(&path) {
+            Ok(()) => {
+                tracing::info!(%file_id, "removed orphan hint file");
+                result.orphan_hints_removed = result.orphan_hints_removed.saturating_add(1);
+            }
+            Err(e) => {
+                tracing::warn!(%file_id, error = %e, "failed to remove orphan hint file");
+                result.repair_errors = result.repair_errors.saturating_add(1);
+            }
+        }
+    });
+
+    report.missing_indexed_files.iter().for_each(|&file_id| {
+        let purged = blockstore.block_index().purge_by_file_id(file_id);
+        tracing::info!(
+            %file_id,
+            purged,
+            "purged phantom index entries for missing data file"
+        );
+        result.phantom_index_entries_purged =
+            result.phantom_index_entries_purged.saturating_add(purged);
+    });
+
     result
 }
 
 #[derive(Debug, Default)]
 pub struct RepairResult {
     pub orphan_files_removed: u64,
+    pub orphan_hints_removed: u64,
+    pub phantom_index_entries_purged: u64,
     pub repair_errors: u64,
 }
 

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

@@ -3,16 +3,27 @@ use std::sync::Arc;
 
 use tracing::warn;
 
-use crate::io::StorageIO;
+use crate::io::{FileId, StorageIO};
 
 use super::manager::SegmentManager;
-use super::segment_file::{SEGMENT_HEADER_SIZE, SegmentWriter, ValidEvent};
+use super::segment_file::{
+    SEGMENT_HEADER_SIZE, SEGMENT_MAGIC, SegmentWriter, ValidEvent, ValidateEventRecord,
+    validate_event_record,
+};
 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,
@@ -31,7 +42,8 @@ pub struct EventLogWriter<S: StorageIO> {
     max_payload: u32,
     event_count_in_segment: usize,
     last_event_offset: Option<SegmentOffset>,
-    pending_events: Vec<ValidEvent>,
+    pending: Vec<PendingAppend>,
+    poisoned: bool,
 }
 
 impl<S: StorageIO> EventLogWriter<S> {
@@ -83,10 +95,25 @@ impl<S: StorageIO> EventLogWriter<S> {
             max_payload,
             event_count_in_segment: 0,
             last_event_offset: None,
-            pending_events: Vec::new(),
+            pending: Vec::new(),
+            poisoned: false,
         })
     }
 
+    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],
@@ -97,6 +124,19 @@ 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,
@@ -107,15 +147,11 @@ impl<S: StorageIO> EventLogWriter<S> {
             Err(rebuild_err) => {
                 let file_size = manager.io().file_size(fd)?;
                 if file_size <= SEGMENT_HEADER_SIZE as u64 {
-                    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(
+                    return Self::truncate_and_init_fresh(
                         Arc::clone(&manager),
+                        fd,
                         active_id,
-                        next_seq,
+                        prev_segments,
                         index_interval,
                         max_payload,
                     );
@@ -131,8 +167,6 @@ 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) =
@@ -196,7 +230,8 @@ impl<S: StorageIO> EventLogWriter<S> {
             max_payload,
             event_count_in_segment,
             last_event_offset,
-            pending_events: Vec::new(),
+            pending: Vec::new(),
+            poisoned: false,
         })
     }
 
@@ -227,28 +262,20 @@ impl<S: StorageIO> EventLogWriter<S> {
             payload,
         };
 
-        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)
+        self.append_inner(event).map(|_| 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
@@ -265,21 +292,52 @@ 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_events.push(event);
+        self.pending.push(PendingAppend { event, offset });
 
         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.pending_events.is_empty() {
-            self.active_writer.sync(self.manager.io())?;
+        if self.poisoned {
+            return Err(io::Error::other(
+                "writer poisoned by partial-valid sync; reopen required",
+            ));
         }
 
-        let flushed = std::mem::take(&mut self.pending_events);
+        if !self.pending.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();
+
         self.synced_seq = flushed.last().map(|e| e.seq).unwrap_or(self.synced_seq);
 
         Ok(SyncResult {
@@ -290,12 +348,22 @@ 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_events.is_empty() {
+        if !self.pending.is_empty() {
             return Ok(None);
         }
 
@@ -386,6 +454,40 @@ 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],
@@ -1094,4 +1196,62 @@ 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/chaos_walker.rs

@@ -0,0 +1,113 @@
+use std::collections::HashSet;
+
+use cid::Cid;
+use jacquard_repo::mst::NodeData;
+
+use super::oracle::{hex_short, try_cid_to_fixed};
+use crate::StorageIO;
+use crate::blockstore::{CidBytes, TranquilBlockStore};
+
+pub enum LookupResult {
+    Found(Cid),
+    NotFound,
+    LostPath,
+}
+
+pub fn walk_mst_node_cids_tolerant<S: StorageIO + Send + Sync + 'static>(
+    store: &TranquilBlockStore<S>,
+    root: Cid,
+    lost: &HashSet<CidBytes>,
+) -> Result<Vec<CidBytes>, String> {
+    let mut visited: HashSet<CidBytes> = HashSet::new();
+    let mut to_visit: Vec<Cid> = vec![root];
+    let mut result: Vec<CidBytes> = Vec::new();
+
+    while let Some(cid) = to_visit.pop() {
+        let cid_bytes = try_cid_to_fixed(&cid).map_err(|e| format!("cid format: {e}"))?;
+        if !visited.insert(cid_bytes) {
+            continue;
+        }
+        if lost.contains(&cid_bytes) {
+            continue;
+        }
+        let node = read_node(store, &cid_bytes)?;
+        result.push(cid_bytes);
+        if let Some(left) = node.left {
+            to_visit.push(left);
+        }
+        node.entries
+            .into_iter()
+            .filter_map(|e| e.tree)
+            .for_each(|t| to_visit.push(t));
+    }
+
+    Ok(result)
+}
+
+pub fn mst_get_tolerant<S: StorageIO + Send + Sync + 'static>(
+    store: &TranquilBlockStore<S>,
+    root: Cid,
+    target: &str,
+    lost: &HashSet<CidBytes>,
+) -> Result<LookupResult, String> {
+    let mut cursor = root;
+    loop {
+        let cursor_bytes = try_cid_to_fixed(&cursor).map_err(|e| format!("cid format: {e}"))?;
+        if lost.contains(&cursor_bytes) {
+            return Ok(LookupResult::LostPath);
+        }
+        let node = read_node(store, &cursor_bytes)?;
+        let keys = full_keys(&node)?;
+        let index = keys
+            .iter()
+            .position(|k| k.as_str() >= target)
+            .unwrap_or(keys.len());
+        if index < keys.len() && keys[index] == target {
+            return Ok(LookupResult::Found(node.entries[index].value));
+        }
+        let subtree = match index {
+            0 => node.left,
+            n => node.entries[n - 1].tree,
+        };
+        match subtree {
+            Some(child) => cursor = child,
+            None => return Ok(LookupResult::NotFound),
+        }
+    }
+}
+
+fn read_node<S: StorageIO + Send + Sync + 'static>(
+    store: &TranquilBlockStore<S>,
+    cid_bytes: &CidBytes,
+) -> Result<NodeData, String> {
+    let bytes = match store.get_block_sync(cid_bytes) {
+        Ok(Some(b)) => b,
+        Ok(None) => return Err(format!("missing block: {}", hex_short(cid_bytes))),
+        Err(e) => return Err(format!("read {}: {e}", hex_short(cid_bytes))),
+    };
+    serde_ipld_dagcbor::from_slice(&bytes)
+        .map_err(|e| format!("deserialize node {}: {e}", hex_short(cid_bytes)))
+}
+
+fn full_keys(node: &NodeData) -> Result<Vec<String>, String> {
+    node.entries
+        .iter()
+        .scan(String::new(), |last_key, entry| {
+            let suffix = match std::str::from_utf8(&entry.key_suffix) {
+                Ok(s) => s,
+                Err(e) => return Some(Err(format!("invalid utf-8 in key suffix: {e}"))),
+            };
+            let prefix_len = entry.prefix_len as usize;
+            if prefix_len > last_key.len() {
+                return Some(Err(format!(
+                    "prefix length {} exceeds last key length {}",
+                    prefix_len,
+                    last_key.len()
+                )));
+            }
+            let full = format!("{}{}", &last_key[..prefix_len], suffix);
+            *last_key = full.clone();
+            Some(Ok(full))
+        })
+        .collect()
+}

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

@@ -1,5 +1,6 @@
 use std::cell::RefCell;
 use std::panic::{AssertUnwindSafe, catch_unwind};
+use std::path::PathBuf;
 use std::time::{Duration, Instant};
 
 use rayon::prelude::*;
@@ -44,6 +45,17 @@ 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,
 {
@@ -51,10 +63,14 @@ 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 gauntlet = Gauntlet::new(cfg).expect("build gauntlet");
+                let mut gauntlet = Gauntlet::new(cfg).expect("build gauntlet");
+                if let Some(root) = scratch {
+                    gauntlet = gauntlet.with_scratch_root(root);
+                }
                 with_runtime(|rt| rt.block_on(gauntlet.run()))
             }));
             let report = outcome.unwrap_or_else(|payload| {
@@ -66,6 +82,14 @@ 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>()
@@ -84,3 +108,59 @@ 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,6 +405,8 @@ 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/invariants.rs

@@ -31,6 +31,9 @@ impl InvariantSet {
     pub const MONOTONIC_SEQ: Self = Self(1 << 10);
     pub const FSYNC_ORDERING: Self = Self(1 << 11);
     pub const TOMBSTONE_BOUND: Self = Self(1 << 12);
+    pub const INDEX_BACKED_BY_DISK: Self = Self(1 << 13);
+    pub const HINT_BACKED_BY_DATA: Self = Self(1 << 14);
+    pub const INDEX_BLOCKS_READABLE: Self = Self(1 << 15);
 
     const ALL_KNOWN: u32 = Self::REFCOUNT_CONSERVATION.0
         | Self::REACHABILITY.0
@@ -44,7 +47,10 @@ impl InvariantSet {
         | Self::CHECKSUM_COVERAGE.0
         | Self::MONOTONIC_SEQ.0
         | Self::FSYNC_ORDERING.0
-        | Self::TOMBSTONE_BOUND.0;
+        | Self::TOMBSTONE_BOUND.0
+        | Self::INDEX_BACKED_BY_DISK.0
+        | Self::HINT_BACKED_BY_DATA.0
+        | Self::INDEX_BLOCKS_READABLE.0;
 
     pub const fn contains(self, other: Self) -> bool {
         (self.0 & other.0) == other.0
@@ -377,6 +383,164 @@ fn compact_by_liveness<S: StorageIO + Send + Sync + 'static>(
         })
 }
 
+pub struct HintBackedByData;
+
+#[async_trait]
+impl<S: StorageIO + Send + Sync + 'static> Invariant<S> for HintBackedByData {
+    fn name(&self) -> &'static str {
+        "HintBackedByData"
+    }
+
+    async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> {
+        let store_c = ctx.store.clone();
+        let result = tokio::task::spawn_blocking(move || {
+            let data: std::collections::HashSet<_> = store_c
+                .list_data_files()
+                .map_err(|e| e.to_string())?
+                .into_iter()
+                .collect();
+            let hints = store_c.list_hint_files().map_err(|e| e.to_string())?;
+            let orphans: Vec<String> = hints
+                .iter()
+                .filter(|fid| !data.contains(fid))
+                .map(|fid| fid.to_string())
+                .collect();
+            Ok::<_, String>(orphans)
+        })
+        .await
+        .map_err(|e| InvariantViolation {
+            invariant: "HintBackedByData",
+            detail: format!("join: {e}"),
+        })?;
+
+        let orphans = result.map_err(|e| InvariantViolation {
+            invariant: "HintBackedByData",
+            detail: e,
+        })?;
+
+        if orphans.is_empty() {
+            Ok(())
+        } else {
+            Err(InvariantViolation {
+                invariant: "HintBackedByData",
+                detail: format!(
+                    "hint files without matching data file (orphan hints): {}",
+                    orphans.join(", ")
+                ),
+            })
+        }
+    }
+}
+
+pub struct IndexBlocksReadable;
+
+#[async_trait]
+impl<S: StorageIO + Send + Sync + 'static> Invariant<S> for IndexBlocksReadable {
+    fn name(&self) -> &'static str {
+        "IndexBlocksReadable"
+    }
+
+    async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> {
+        let store_c = ctx.store.clone();
+        let result = tokio::task::spawn_blocking(move || {
+            let entries = store_c.block_index().live_entries_snapshot();
+            let unreadable: Vec<String> = entries
+                .iter()
+                .take(INDEX_READABLE_SAMPLE_CAP)
+                .filter_map(|(cid, _)| match store_c.get_block_sync(cid) {
+                    Ok(Some(_)) => None,
+                    Ok(None) => Some(format!(
+                        "{}: index says present but reader missed",
+                        hex_short(cid)
+                    )),
+                    Err(e) => Some(format!("{}: read error {e}", hex_short(cid))),
+                })
+                .take(INDEX_READABLE_REPORT_CAP)
+                .collect();
+            Ok::<_, String>(unreadable)
+        })
+        .await
+        .map_err(|e| InvariantViolation {
+            invariant: "IndexBlocksReadable",
+            detail: format!("join: {e}"),
+        })?;
+
+        let unreadable = result.map_err(|e| InvariantViolation {
+            invariant: "IndexBlocksReadable",
+            detail: e,
+        })?;
+
+        if unreadable.is_empty() {
+            Ok(())
+        } else {
+            Err(InvariantViolation {
+                invariant: "IndexBlocksReadable",
+                detail: format!(
+                    "live index entries cannot be read back (first {INDEX_READABLE_REPORT_CAP}): {}",
+                    unreadable.join("; ")
+                ),
+            })
+        }
+    }
+}
+
+const INDEX_READABLE_SAMPLE_CAP: usize = 512;
+const INDEX_READABLE_REPORT_CAP: usize = 20;
+
+pub struct IndexBackedByDisk;
+
+#[async_trait]
+impl<S: StorageIO + Send + Sync + 'static> Invariant<S> for IndexBackedByDisk {
+    fn name(&self) -> &'static str {
+        "IndexBackedByDisk"
+    }
+
+    async fn check(&self, ctx: &InvariantCtx<'_, S>) -> Result<(), InvariantViolation> {
+        let store_c = ctx.store.clone();
+        let result = tokio::task::spawn_blocking(move || {
+            let disk: std::collections::HashSet<_> = store_c
+                .list_data_files()
+                .map_err(|e| e.to_string())?
+                .into_iter()
+                .collect();
+            let liveness = store_c.compaction_liveness(0).map_err(|e| e.to_string())?;
+            let missing: Vec<String> = liveness
+                .iter()
+                .filter(|(fid, _)| !disk.contains(fid))
+                .map(|(fid, info)| {
+                    format!(
+                        "{fid} (live_blocks={}, total_blocks={})",
+                        info.live_blocks, info.total_blocks
+                    )
+                })
+                .collect();
+            Ok::<_, String>(missing)
+        })
+        .await
+        .map_err(|e| InvariantViolation {
+            invariant: "IndexBackedByDisk",
+            detail: format!("join: {e}"),
+        })?;
+
+        let missing = result.map_err(|e| InvariantViolation {
+            invariant: "IndexBackedByDisk",
+            detail: e,
+        })?;
+
+        if missing.is_empty() {
+            Ok(())
+        } else {
+            Err(InvariantViolation {
+                invariant: "IndexBackedByDisk",
+                detail: format!(
+                    "index references data files missing on disk (iris-shaped corruption): {}",
+                    missing.join(", ")
+                ),
+            })
+        }
+    }
+}
+
 pub struct NoOrphanFiles;
 
 #[async_trait]
@@ -766,6 +930,18 @@ pub fn invariants_for<S: StorageIO + Send + Sync + 'static>(
             Box::new(CompactionIdempotent),
         ),
         (InvariantSet::NO_ORPHAN_FILES, Box::new(NoOrphanFiles)),
+        (
+            InvariantSet::INDEX_BACKED_BY_DISK,
+            Box::new(IndexBackedByDisk),
+        ),
+        (
+            InvariantSet::HINT_BACKED_BY_DATA,
+            Box::new(HintBackedByData),
+        ),
+        (
+            InvariantSet::INDEX_BLOCKS_READABLE,
+            Box::new(IndexBlocksReadable),
+        ),
         (InvariantSet::BYTE_BUDGET, Box::new(ByteBudget::default())),
         (
             InvariantSet::MANIFEST_EQUALS_REALITY,

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

@@ -1,3 +1,4 @@
+pub mod chaos_walker;
 pub mod farm;
 pub mod flaky;
 pub mod invariants;
@@ -17,13 +18,14 @@ pub use flaky::{
     BackingMegabytes, DownIntervalSecs, FlakyConfig, FlakyError, FlakyMount, UpIntervalSecs,
 };
 pub use invariants::{
-    EventLogSnapshot, Invariant, InvariantSet, InvariantViolation, SnapshotEvent, invariants_for,
+    EventLogSnapshot, HintBackedByData, IndexBackedByDisk, IndexBlocksReadable, Invariant,
+    InvariantCtx, InvariantSet, InvariantViolation, SnapshotEvent, invariants_for,
 };
 pub use leak::{LeakGateBuildError, LeakGateConfig, LeakViolation, evaluate as evaluate_leak_gate};
 pub use metrics::{MetricName, MetricsSample, sample_harness};
 pub use op::{
-    CollectionName, DidSeed, EventKind, Op, OpStream, PayloadSeed, RecordKey, RetentionSecs, Seed,
-    ValueSeed,
+    CollectionName, DidSeed, EventKind, FileChoice, Op, OpStream, PayloadSeed, RecordKey,
+    RetentionSecs, Seed, ValueSeed,
 };
 pub use oracle::{EventExpectation, Oracle};
 pub use overrides::{ConfigOverrides, GroupCommitOverrides, StoreOverrides};

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

@@ -29,6 +29,9 @@ pub enum EventKind {
     Sync,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
+pub struct FileChoice(pub u32);
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum Op {
     AddRecord {
@@ -58,6 +61,9 @@ pub enum Op {
     ReadBlock {
         value_seed: ValueSeed,
     },
+    ExternalDeleteDataFile {
+        choice: FileChoice,
+    },
 }
 
 impl Op {

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

@@ -1,4 +1,4 @@
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 
 use cid::Cid;
 
@@ -29,6 +29,7 @@ pub struct Oracle {
     unsynced_events: Vec<EventExpectation>,
     last_synced_seq: Option<EventSequence>,
     last_retention_cutoff_us: Option<u64>,
+    lost_blocks: HashSet<CidBytes>,
 }
 
 impl Oracle {
@@ -93,6 +94,27 @@ impl Oracle {
         self.unsynced_events.push(event);
     }
 
+    pub fn mark_blocks_lost(&mut self, cids: impl IntoIterator<Item = CidBytes>) -> usize {
+        let added: HashSet<CidBytes> = cids.into_iter().collect();
+        let added_count = added.len();
+        self.live
+            .retain(|_, record_cid| !added.contains(record_cid));
+        self.lost_blocks.extend(added);
+        added_count
+    }
+
+    pub fn lost_blocks(&self) -> &HashSet<CidBytes> {
+        &self.lost_blocks
+    }
+
+    pub fn is_block_lost(&self, cid: &CidBytes) -> bool {
+        self.lost_blocks.contains(cid)
+    }
+
+    pub fn has_lost_blocks(&self) -> bool {
+        !self.lost_blocks.is_empty()
+    }
+
     pub fn record_event_sync(&mut self, synced_through: EventSequence) {
         let (promoted, remaining): (Vec<_>, Vec<_>) = self
             .unsynced_events

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, SimulatedIO};
+use crate::sim::{FaultConfig, PristineGuard, SimulatedIO};
 
 #[derive(Debug, Clone, Copy)]
 pub enum IoBackend {
@@ -93,6 +93,7 @@ pub struct GauntletConfig {
     pub store: StoreConfig,
     pub eventlog: Option<EventLogConfig>,
     pub writer_concurrency: WriterConcurrency,
+    pub tolerate_op_errors: bool,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
@@ -178,6 +179,7 @@ pub struct SharedState<S: StorageIO + Send + Sync + 'static> {
 
 pub struct Gauntlet {
     config: GauntletConfig,
+    scratch_root: Option<PathBuf>,
 }
 
 #[derive(Debug, thiserror::Error)]
@@ -185,7 +187,15 @@ pub enum GauntletBuildError {}
 
 impl Gauntlet {
     pub fn new(config: GauntletConfig) -> Result<Self, GauntletBuildError> {
-        Ok(Self { config })
+        Ok(Self {
+            config,
+            scratch_root: None,
+        })
+    }
+
+    pub fn with_scratch_root(mut self, root: PathBuf) -> Self {
+        self.scratch_root = Some(root);
+        self
     }
 
     pub fn generate_ops(&self) -> OpStream {
@@ -211,6 +221,7 @@ 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(
@@ -219,6 +230,7 @@ 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,
@@ -269,9 +281,14 @@ 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 = tempfile::TempDir::new().expect("tempdir");
+    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 root = dir.path().to_path_buf();
+    let tolerate = config.tolerate_op_errors;
     let report = run_inner_real_on_root(
         config,
         root,
@@ -279,7 +296,7 @@ async fn run_inner_real(
         ops_counter,
         op_errors_counter,
         restarts_counter,
-        false,
+        tolerate,
         Duration::ZERO,
     )
     .await;
@@ -364,7 +381,7 @@ async fn run_inner_real_on_root(
     let segments_dir = segments_subdir(&root);
     let open = {
         let segments_dir = segments_dir.clone();
-        move || -> Result<Harness<RealIO>, String> {
+        move |_attempt: usize| -> Result<Harness<RealIO>, String> {
             let store = TranquilBlockStore::open(cfg.clone())
                 .map(Arc::new)
                 .map_err(|e| e.to_string())?;
@@ -417,14 +434,15 @@ async fn run_inner_simulated(
 ) -> GauntletReport {
     let dir = tempfile::TempDir::new().expect("tempdir");
     let cfg = blockstore_config(dir.path(), &config.store);
-    let tolerate_errors = fault.injects_errors();
+    let tolerate_errors = fault.injects_errors() || config.tolerate_op_errors;
     let eventlog_cfg = config.eventlog;
     let segments_dir = segments_subdir(dir.path());
     let sim: Arc<SimulatedIO> = Arc::new(SimulatedIO::new(config.seed.0, fault));
     let sim_for_open = Arc::clone(&sim);
     let open = {
         let segments_dir = segments_dir.clone();
-        move || -> Result<Harness<Arc<SimulatedIO>>, String> {
+        move |attempt: usize| -> Result<Harness<Arc<SimulatedIO>>, String> {
+            let _pristine = PristineGuard::new(Arc::clone(&sim_for_open), attempt > 0);
             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)
@@ -512,28 +530,30 @@ async fn run_inner_generic<S, Open, Crash>(
 ) -> GauntletReport
 where
     S: StorageIO + Send + Sync + 'static,
-    Open: FnMut() -> Result<Harness<S>, String>,
+    Open: FnMut(usize) -> 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 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 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 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));
@@ -750,7 +770,7 @@ async fn reopen_with_recovery<S, Open, Crash>(
 ) -> Result<Harness<S>, String>
 where
     S: StorageIO + Send + Sync + 'static,
-    Open: FnMut() -> Result<Harness<S>, String>,
+    Open: FnMut(usize) -> Result<Harness<S>, String>,
     Crash: FnMut(),
 {
     let mut errors: Vec<String> = Vec::new();
@@ -758,7 +778,7 @@ where
         if attempt > 0 && !backoff.is_zero() {
             tokio::time::sleep(backoff).await;
         }
-        match open() {
+        match open(attempt) {
             Ok(h) => return Ok(h),
             Err(e) => {
                 errors.push(format!("attempt {attempt}: {e}"));
@@ -810,7 +830,6 @@ async fn run_quick_check<S: StorageIO + Send + Sync + 'static>(
         };
     };
 
-    let mst = Mst::load(store.clone(), r, None);
     let live: Vec<(super::op::CollectionName, super::op::RecordKey, CidBytes)> = oracle
         .live_records()
         .map(|(c, k, v)| (c.clone(), k.clone(), *v))
@@ -822,24 +841,39 @@ async fn run_quick_check<S: StorageIO + Send + Sync + 'static>(
         sample_distinct(rng, total, sample_size)
     };
 
-    let mut violations: Vec<String> = Vec::new();
-    for idx in picks {
-        let (coll, rkey, expected) = &live[idx];
-        let key = format!("{}/{}", coll.0, rkey.0);
-        match mst.get(&key).await {
-            Ok(Some(cid)) => match try_cid_to_fixed(&cid) {
-                Ok(actual) if actual == *expected => {}
-                Ok(actual) => violations.push(format!(
-                    "{key}: MST cid {} != oracle cid {}",
-                    hex_short(&actual),
-                    hex_short(expected)
-                )),
-                Err(e) => violations.push(format!("{key}: cid format: {e}")),
-            },
-            Ok(None) => violations.push(format!("{key}: missing after reopen")),
-            Err(e) => violations.push(format!("{key}: mst.get error: {e}")),
-        }
-    }
+    let store_c = store.clone();
+    let lost_clone = oracle.lost_blocks().clone();
+    let live_clone = live.clone();
+    let picks_c = picks.clone();
+    let violations: Vec<String> = tokio::task::spawn_blocking(move || {
+        picks_c
+            .iter()
+            .filter_map(|&idx| {
+                let (coll, rkey, expected) = &live_clone[idx];
+                let key = format!("{}/{}", coll.0, rkey.0);
+                match super::chaos_walker::mst_get_tolerant(&store_c, r, &key, &lost_clone) {
+                    Ok(super::chaos_walker::LookupResult::Found(cid)) => {
+                        match try_cid_to_fixed(&cid) {
+                            Ok(actual) if actual == *expected => None,
+                            Ok(actual) => Some(format!(
+                                "{key}: MST cid {} != oracle cid {}",
+                                hex_short(&actual),
+                                hex_short(expected)
+                            )),
+                            Err(e) => Some(format!("{key}: cid format: {e}")),
+                        }
+                    }
+                    Ok(super::chaos_walker::LookupResult::NotFound) => {
+                        Some(format!("{key}: missing after reopen"))
+                    }
+                    Ok(super::chaos_walker::LookupResult::LostPath) => None,
+                    Err(e) => Some(format!("{key}: mst.get error: {e}")),
+                }
+            })
+            .collect()
+    })
+    .await
+    .unwrap_or_else(|e| vec![format!("quick_check join: {e}")]);
 
     if violations.is_empty() {
         Vec::new()
@@ -949,16 +983,13 @@ pub(super) async fn refresh_oracle_graph<S: StorageIO + Send + Sync + 'static>(
             Ok(())
         }
         Some(r) => {
-            let settled = Mst::load(store.clone(), r, None);
-            let cids = settled
-                .collect_node_cids()
-                .await
-                .map_err(|e| format!("collect_node_cids: {e}"))?;
-            let fixed: Vec<CidBytes> = cids
-                .iter()
-                .map(try_cid_to_fixed)
-                .collect::<Result<_, _>>()
-                .map_err(|e| format!("mst node cid: {e}"))?;
+            let store_c = store.clone();
+            let lost_clone = oracle.lost_blocks().clone();
+            let fixed = tokio::task::spawn_blocking(move || {
+                super::chaos_walker::walk_mst_node_cids_tolerant(&store_c, r, &lost_clone)
+            })
+            .await
+            .map_err(|e| format!("refresh join: {e}"))??;
             oracle.set_root(r);
             oracle.set_mst_node_cids(fixed);
             Ok(())
@@ -1182,6 +1213,39 @@ pub(super) async fn apply_op<S: StorageIO + Send + Sync + 'static>(
             let _ = harness.store.get_block_sync(&record_cid);
             Ok(())
         }
+        Op::ExternalDeleteDataFile { choice } => {
+            let s = harness.store.clone();
+            let pick = choice.0;
+            let lost_cids = tokio::task::spawn_blocking(move || externally_delete_data_file(&s, pick))
+                .await
+                .map_err(|e| OpError::Join(e.to_string()))??;
+            if !lost_cids.is_empty() {
+                oracle.mark_blocks_lost(lost_cids);
+            }
+            Ok(())
+        }
+    }
+}
+
+fn externally_delete_data_file(
+    store: &std::sync::Arc<TranquilBlockStore<impl StorageIO + 'static>>,
+    pick: u32,
+) -> Result<Vec<CidBytes>, OpError> {
+    let active = store.block_index().read_write_cursor().map(|c| c.file_id);
+    let mut candidates = match store.list_data_files() {
+        Ok(files) => files,
+        Err(_) => return Ok(Vec::new()),
+    };
+    candidates.retain(|fid| active.is_none_or(|a| *fid < a));
+    if candidates.is_empty() {
+        return Ok(Vec::new());
+    }
+    let idx = (pick as usize) % candidates.len();
+    let victim = candidates[idx];
+    let cids = store.block_index().cids_in_file(victim);
+    match std::fs::remove_file(store.data_file_path(victim)) {
+        Ok(()) => Ok(cids),
+        Err(_) => Ok(Vec::new()),
     }
 }
 
@@ -1191,6 +1255,10 @@ 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;
@@ -1489,6 +1557,18 @@ async fn apply_op_concurrent<S: StorageIO + Send + Sync + 'static>(
             let _ = shared.store.get_block_sync(&record_cid);
             Ok(())
         }
+        Op::ExternalDeleteDataFile { choice } => {
+            let mut guard = shared.write.lock().await;
+            let s = shared.store.clone();
+            let pick = choice.0;
+            let lost_cids = tokio::task::spawn_blocking(move || externally_delete_data_file(&s, pick))
+                .await
+                .map_err(|e| OpError::Join(e.to_string()))??;
+            if !lost_cids.is_empty() {
+                guard.oracle.mark_blocks_lost(lost_cids);
+            }
+            Ok(())
+        }
     }
 }
 
@@ -1564,7 +1644,7 @@ async fn run_inner_generic_concurrent<S, Open, Crash>(
 ) -> GauntletReport
 where
     S: StorageIO + Send + Sync + 'static,
-    Open: FnMut() -> Result<Harness<S>, String>,
+    Open: FnMut(usize) -> Result<Harness<S>, String>,
     Crash: FnMut(),
 {
     let ops: Vec<Op> = op_stream.into_vec();
@@ -1577,22 +1657,24 @@ 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 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 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 root: Option<Cid> = None;
     let mut oracle = Oracle::new();
     let mut halt_ops = false;
@@ -1801,3 +1883,126 @@ 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),
+            tolerate_op_errors: false,
+        }
+    }
+
+    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;
+use crate::blockstore::{GroupCommitConfig, MAX_BLOCK_SIZE};
 use crate::sim::FaultConfig;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -31,6 +31,7 @@ pub enum Scenario {
     ContendedReaders,
     ContendedWriters,
     FlakyDevice,
+    ExternalCorruption,
 }
 
 impl Scenario {
@@ -53,6 +54,7 @@ impl Scenario {
             Self::ContendedReaders => "ContendedReaders",
             Self::ContendedWriters => "ContendedWriters",
             Self::FlakyDevice => "FlakyDevice",
+            Self::ExternalCorruption => "ExternalCorruption",
         }
     }
 
@@ -75,6 +77,7 @@ impl Scenario {
             Self::ContendedReaders => "contended-readers",
             Self::ContendedWriters => "contended-writers",
             Self::FlakyDevice => "flaky-device",
+            Self::ExternalCorruption => "external-corruption",
         }
     }
 
@@ -109,6 +112,9 @@ impl Scenario {
             Self::FlakyDevice => {
                 "Real IO on ext4 atop dm-flakey. Requires root with dm-flakey available, skips otherwise."
             }
+            Self::ExternalCorruption => {
+                "Rare external data-file deletion mid-workload. Validates phantom-purge self-heal under chaos."
+            }
         }
     }
 
@@ -138,6 +144,7 @@ impl Scenario {
         Self::ContendedReaders,
         Self::ContendedWriters,
         Self::FlakyDevice,
+        Self::ExternalCorruption,
     ];
 }
 
@@ -211,6 +218,7 @@ pub fn config_for(scenario: Scenario, seed: Seed) -> GauntletConfig {
         Scenario::ContendedReaders => contended_readers(seed),
         Scenario::ContendedWriters => contended_writers(seed),
         Scenario::FlakyDevice => flaky_device(seed),
+        Scenario::ExternalCorruption => external_corruption(seed),
     }
 }
 
@@ -280,6 +288,7 @@ fn smoke_pr(seed: Seed) -> GauntletConfig {
         store: tiny_store(),
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     }
 }
 
@@ -305,6 +314,7 @@ fn mst_churn(seed: Seed) -> GauntletConfig {
         store: tiny_store(),
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     }
 }
 
@@ -330,6 +340,7 @@ fn mst_restart_churn(seed: Seed) -> GauntletConfig {
         store: tiny_store(),
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     }
 }
 
@@ -359,6 +370,7 @@ fn full_stack_restart(seed: Seed) -> GauntletConfig {
         },
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     }
 }
 
@@ -372,6 +384,9 @@ fn phase2_invariants() -> InvariantSet {
         | InvariantSet::BYTE_BUDGET
         | InvariantSet::MANIFEST_EQUALS_REALITY
         | InvariantSet::CHECKSUM_COVERAGE
+        | InvariantSet::INDEX_BACKED_BY_DISK
+        | InvariantSet::HINT_BACKED_BY_DATA
+        | InvariantSet::INDEX_BLOCKS_READABLE
 }
 
 fn catastrophic_churn(seed: Seed) -> GauntletConfig {
@@ -392,6 +407,7 @@ fn catastrophic_churn(seed: Seed) -> GauntletConfig {
         store: tiny_store(),
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     }
 }
 
@@ -402,7 +418,7 @@ fn huge_values(seed: Seed) -> GauntletConfig {
         workload: block_workload(
             block_weights(85, 5, 8, 2),
             SizeDistribution::HeavyTail(
-                ByteRange::new(ValueBytes(256), ValueBytes(16 * 1024 * 1024))
+                ByteRange::new(ValueBytes(256), ValueBytes(MAX_BLOCK_SIZE))
                     .expect("huge_values ByteRange"),
             ),
             KeySpaceSize(64),
@@ -424,6 +440,7 @@ fn huge_values(seed: Seed) -> GauntletConfig {
         },
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     }
 }
 
@@ -454,6 +471,7 @@ fn tiny_batches(seed: Seed) -> GauntletConfig {
         },
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     }
 }
 
@@ -484,6 +502,7 @@ fn giant_batches(seed: Seed) -> GauntletConfig {
         },
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     }
 }
 
@@ -509,6 +528,7 @@ fn many_files(seed: Seed) -> GauntletConfig {
         },
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     }
 }
 
@@ -521,6 +541,9 @@ fn sim_invariants() -> InvariantSet {
         | InvariantSet::NO_ORPHAN_FILES
         | InvariantSet::BYTE_BUDGET
         | InvariantSet::CHECKSUM_COVERAGE
+        | InvariantSet::INDEX_BACKED_BY_DISK
+        | InvariantSet::HINT_BACKED_BY_DATA
+        | InvariantSet::INDEX_BLOCKS_READABLE
 }
 
 fn sim_microbench_workload() -> WorkloadModel {
@@ -558,6 +581,7 @@ fn moderate_faults(seed: Seed) -> GauntletConfig {
         store: sim_store(),
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     }
 }
 
@@ -577,6 +601,7 @@ fn aggressive_faults(seed: Seed) -> GauntletConfig {
         store: sim_store(),
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     }
 }
 
@@ -596,6 +621,7 @@ fn torn_pages(seed: Seed) -> GauntletConfig {
         store: sim_store(),
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     }
 }
 
@@ -615,6 +641,7 @@ fn fsyncgate(seed: Seed) -> GauntletConfig {
         store: sim_store(),
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     }
 }
 
@@ -654,6 +681,7 @@ fn firehose_fanout(seed: Seed) -> GauntletConfig {
             max_segment_size: MaxSegmentSize(64 * 1024),
         }),
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     }
 }
 
@@ -688,6 +716,7 @@ fn contended_readers(seed: Seed) -> GauntletConfig {
         store: sim_store(),
         eventlog: None,
         writer_concurrency: WriterConcurrency(64),
+        tolerate_op_errors: false,
     }
 }
 
@@ -719,6 +748,7 @@ fn flaky_device(seed: Seed) -> GauntletConfig {
         store: tiny_store(),
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     }
 }
 
@@ -753,5 +783,35 @@ fn contended_writers(seed: Seed) -> GauntletConfig {
         store: sim_store(),
         eventlog: None,
         writer_concurrency: WriterConcurrency(32),
+        tolerate_op_errors: false,
+    }
+}
+
+fn external_corruption(seed: Seed) -> GauntletConfig {
+    GauntletConfig {
+        seed,
+        io: IoBackend::Real,
+        workload: block_workload(
+            OpWeights {
+                add: 50,
+                delete: 30,
+                compact: 18,
+                checkpoint: 1,
+                external_delete_data_file: 1,
+                ..OpWeights::default()
+            },
+            SizeDistribution::Fixed(ValueBytes(128)),
+            KeySpaceSize(200),
+        ),
+        op_count: OpCount(2_000),
+        invariants: InvariantSet::NO_ORPHAN_FILES | InvariantSet::BYTE_BUDGET,
+        limits: RunLimits {
+            max_wall_ms: Some(WallMs(60_000)),
+        },
+        restart_policy: RestartPolicy::EveryNOps(OpInterval(1_000)),
+        store: tiny_store(),
+        eventlog: None,
+        writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: true,
     }
 }

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

@@ -142,6 +142,7 @@ mod tests {
             },
             eventlog: None,
             writer_concurrency: WriterConcurrency(1),
+            tolerate_op_errors: false,
         }
     }
 

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

@@ -1,6 +1,6 @@
 use super::op::{
-    CollectionName, DidSeed, EventKind, Op, OpStream, PayloadSeed, RecordKey, RetentionSecs, Seed,
-    ValueSeed,
+    CollectionName, DidSeed, EventKind, FileChoice, Op, OpStream, PayloadSeed, RecordKey,
+    RetentionSecs, Seed, ValueSeed,
 };
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
@@ -23,6 +23,7 @@ pub struct OpWeights {
     pub run_retention: u32,
     pub read_record: u32,
     pub read_block: u32,
+    pub external_delete_data_file: u32,
 }
 
 impl OpWeights {
@@ -36,6 +37,7 @@ impl OpWeights {
             + self.run_retention
             + self.read_record
             + self.read_block
+            + self.external_delete_data_file
     }
 
     pub const fn touches_eventlog(&self) -> bool {
@@ -103,6 +105,7 @@ impl Default for WorkloadModel {
                 run_retention: 0,
                 read_record: 0,
                 read_block: 0,
+                external_delete_data_file: 0,
             },
             size_distribution: SizeDistribution::Fixed(ValueBytes(64)),
             collections: vec![CollectionName("app.bsky.feed.post".to_string())],
@@ -138,6 +141,7 @@ impl WorkloadModel {
                 let t6 = t5 + w.sync_event_log;
                 let t7 = t6 + w.run_retention;
                 let t8 = t7 + w.read_record;
+                let t9 = t8 + w.read_block;
 
                 match bucket {
                     b if b < t1 => Op::AddRecord {
@@ -166,9 +170,12 @@ impl WorkloadModel {
                         collection: coll,
                         rkey,
                     },
-                    _ => Op::ReadBlock {
+                    b if b < t9 => Op::ReadBlock {
                         value_seed: ValueSeed(rng.next_u32()),
                     },
+                    _ => Op::ExternalDeleteDataFile {
+                        choice: FileChoice(rng.next_u32()),
+                    },
                 }
             })
             .collect();

crates/tranquil-store/src/io.rs

@@ -104,6 +104,10 @@ 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(()))
@@ -190,6 +194,9 @@ 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, SimulatedIO, SyncReorderWindow,
+    FaultConfig, LatencyNs, OpRecord, PristineGuard, Probability, SimulatedIO, SyncReorderWindow,
     sim_proptest_cases, sim_seed_count, sim_seed_range, sim_single_seed,
 };
 

crates/tranquil-store/src/metastore/client.rs

@@ -1860,6 +1860,18 @@ impl<S: StorageIO + 'static> tranquil_db_traits::InfraRepository for MetastoreCl
         recv(rx).await
     }
 
+    async fn mark_comms_failed_permanent(&self, id: Uuid, error: &str) -> Result<(), DbError> {
+        let (tx, rx) = oneshot::channel();
+        self.pool.send(MetastoreRequest::Infra(
+            InfraRequest::MarkCommsFailedPermanent {
+                id,
+                error: error.to_owned(),
+                tx,
+            },
+        ))?;
+        recv(rx).await
+    }
+
     async fn create_invite_code(
         &self,
         code: &str,

crates/tranquil-store/src/metastore/handler.rs

@@ -1789,6 +1789,11 @@ pub enum InfraRequest {
         error: String,
         tx: Tx<()>,
     },
+    MarkCommsFailedPermanent {
+        id: Uuid,
+        error: String,
+        tx: Tx<()>,
+    },
     CreateInviteCode {
         code: String,
         use_count: i32,
@@ -3888,6 +3893,14 @@ fn dispatch_infra<S: StorageIO>(state: &HandlerState<S>, req: InfraRequest) {
                 .map_err(metastore_to_db);
             let _ = tx.send(result);
         }
+        InfraRequest::MarkCommsFailedPermanent { id, error, tx } => {
+            let result = state
+                .metastore
+                .infra_ops()
+                .mark_comms_failed_permanent(id, &error)
+                .map_err(metastore_to_db);
+            let _ = tx.send(result);
+        }
         InfraRequest::CreateInviteCode {
             code,
             use_count,

crates/tranquil-store/src/metastore/infra_ops.rs

@@ -247,7 +247,6 @@ impl InfraOps {
 
         val.status = status_to_u8(CommsStatus::Sent);
         val.sent_at_ms = Some(Utc::now().timestamp_millis());
-        val.attempts = val.attempts.saturating_add(1);
 
         let mut batch = self.db.batch();
         batch.insert(&self.infra, key.as_slice(), val.serialize());
@@ -272,9 +271,46 @@ impl InfraOps {
         )?
         .ok_or(MetastoreError::InvalidInput("comms entry not found"))?;
 
+        let next_attempts = val.attempts.saturating_add(1);
+        let exhausted = next_attempts >= val.max_attempts;
+        let next_status = match exhausted {
+            true => CommsStatus::Failed,
+            false => CommsStatus::Pending,
+        };
+        let now_ms = Utc::now().timestamp_millis();
+        let backoff_ms = i64::from(next_attempts).saturating_mul(60_000);
+
+        val.status = status_to_u8(next_status);
+        val.error_message = Some(error.to_owned());
+        val.attempts = next_attempts;
+        val.scheduled_for_ms = now_ms.saturating_add(backoff_ms);
+
+        let mut batch = self.db.batch();
+        batch.insert(&self.infra, key.as_slice(), val.serialize());
+
+        if let Some((hk, mut hv)) =
+            self.find_history_entry(val.user_id.unwrap_or(Uuid::nil()), val.id)?
+        {
+            hv.status = status_to_u8(next_status);
+            batch.insert(&self.infra, hk.as_slice(), hv.serialize());
+        }
+
+        batch.commit().map_err(MetastoreError::Fjall)
+    }
+
+    pub fn mark_comms_failed_permanent(&self, id: Uuid, error: &str) -> Result<(), MetastoreError> {
+        let key = comms_queue_key(id);
+        let mut val: QueuedCommsValue = point_lookup(
+            &self.infra,
+            key.as_slice(),
+            QueuedCommsValue::deserialize,
+            "corrupt comms queue entry",
+        )?
+        .ok_or(MetastoreError::InvalidInput("comms entry not found"))?;
+
         val.status = status_to_u8(CommsStatus::Failed);
         val.error_message = Some(error.to_owned());
-        val.attempts = val.attempts.saturating_add(1);
+        val.attempts = val.max_attempts;
 
         let mut batch = self.db.batch();
         batch.insert(&self.infra, key.as_slice(), val.serialize());

crates/tranquil-store/src/sim.rs

@@ -1,8 +1,9 @@
 use std::collections::{HashMap, HashSet, VecDeque};
 use std::io;
 use std::path::{Path, PathBuf};
+use std::sync::Arc;
 use std::sync::Mutex;
-use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::time::Duration;
 
 use crate::io::{FileId, OpenOptions, StorageIO};
@@ -226,6 +227,7 @@ pub enum OpRecord {
     SyncDir {
         path: PathBuf,
     },
+    Barrier,
 }
 
 struct PendingSync {
@@ -326,6 +328,7 @@ impl SimState {
 pub struct SimulatedIO {
     state: Mutex<SimState>,
     fault_config: FaultConfig,
+    pristine_mode: AtomicBool,
     rng_seed: u64,
     latency_counter: AtomicU64,
 }
@@ -346,13 +349,26 @@ 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.fault_config.latency_distribution_ns.0;
+        let max_ns = self.effective_fault_config().latency_distribution_ns.0;
         if max_ns == 0 {
             return;
         }
@@ -429,12 +445,30 @@ 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, self.fault_config.io_error_probability) {
+        if state.should_fault(seed, fault.io_error_probability) {
             return Err(io::Error::other("simulated EIO on open"));
         }
 
@@ -514,6 +548,7 @@ 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;
@@ -522,22 +557,21 @@ impl StorageIO for SimulatedIO {
             return Err(io::Error::other("simulated EIO after delayed sync fault"));
         }
 
-        if state.should_fault(seed, self.fault_config.io_error_probability) {
+        if state.should_fault(seed, fault.io_error_probability) {
             return Err(io::Error::other("simulated EIO on read"));
         }
 
-        let read_offset =
-            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 {
-                    offset.saturating_sub(drift)
-                } else {
-                    offset.saturating_add(drift)
-                }
+        let read_offset = if state.should_fault(seed, fault.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 {
+                offset.saturating_sub(drift)
             } else {
-                offset
-            };
+                offset.saturating_add(drift)
+            }
+        } else {
+            offset
+        };
 
         let storage = state.storage.get(&sid).unwrap();
 
@@ -556,7 +590,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, self.fault_config.bit_flip_on_read_probability) && to_read > 0 {
+        if state.should_fault(seed, fault.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;
@@ -572,6 +606,7 @@ 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;
@@ -580,36 +615,33 @@ impl StorageIO for SimulatedIO {
             return Err(io::Error::other("simulated EIO after delayed sync fault"));
         }
 
-        if state.should_fault(seed, self.fault_config.io_error_probability) {
+        if state.should_fault(seed, fault.io_error_probability) {
             return Err(io::Error::other("simulated EIO on write"));
         }
 
-        let torn_len =
-            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());
-                let max_sectors = cap / SECTOR_BYTES;
-                (max_sectors >= 2).then(|| {
-                    let n = state.next_random_usize(seed, max_sectors - 1) + 1;
-                    n * SECTOR_BYTES
-                })
-            } else {
-                None
-            };
+        let torn_len = if buf.len() > 1 && state.should_fault(seed, fault.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());
+            let max_sectors = cap / SECTOR_BYTES;
+            (max_sectors >= 2).then(|| {
+                let n = state.next_random_usize(seed, max_sectors - 1) + 1;
+                n * SECTOR_BYTES
+            })
+        } else {
+            None
+        };
 
         let actual_len = match torn_len {
             Some(n) => n,
-            None if buf.len() > 1
-                && state.should_fault(seed, self.fault_config.partial_write_probability) =>
-            {
+            None if buf.len() > 1 && state.should_fault(seed, fault.partial_write_probability) => {
                 let partial = state.next_random_usize(seed, buf.len());
                 partial.max(1)
             }
             None => buf.len(),
         };
 
-        let misdirected = state.should_fault(seed, self.fault_config.misdirected_write_probability);
+        let misdirected = state.should_fault(seed, fault.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;
@@ -643,6 +675,7 @@ 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;
@@ -651,16 +684,22 @@ impl StorageIO for SimulatedIO {
             return Err(io::Error::other("simulated EIO after delayed sync fault"));
         }
 
-        if state.should_fault(seed, self.fault_config.io_error_probability) {
+        if state.should_fault(seed, fault.io_error_probability) {
             return Err(io::Error::other("simulated EIO on sync"));
         }
 
-        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;
+        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 evicted = if sync_succeeded && reorder_window > 0 {
+        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 snapshot = state.storage.get(&sid).unwrap().buffered.clone();
             state.pending_syncs.push_back(PendingSync {
                 storage_id: sid,
@@ -686,7 +725,7 @@ impl StorageIO for SimulatedIO {
 
         let storage = state.storage.get_mut(&sid).unwrap();
 
-        if sync_succeeded && reorder_window == 0 {
+        if reorder_window == 0 {
             storage.durable = storage.buffered.clone();
         }
         if poison_after {
@@ -695,7 +734,7 @@ impl StorageIO for SimulatedIO {
 
         state.op_log.push(OpRecord::Sync {
             fd: id,
-            succeeded: sync_succeeded,
+            succeeded: true,
         });
         Ok(())
     }
@@ -774,17 +813,30 @@ 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, self.fault_config.io_error_probability) {
+        if state.should_fault(seed, fault.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, self.fault_config.dir_sync_failure_probability);
+        let actually_persisted = !state.should_fault(seed, fault.dir_sync_failure_probability);
 
         if actually_persisted {
             state.dirs_durable.insert(dir_path.clone());

crates/tranquil-store/tests/external_corruption_recovery.rs

@@ -0,0 +1,235 @@
+mod common;
+
+use std::fs;
+
+use common::{block_data, test_cid, tiny_blockstore_config, with_runtime};
+use tranquil_store::blockstore::{
+    CompactionResult, DataFileId, TranquilBlockStore, hint_file_path,
+};
+
+fn data_file_path(dir: &std::path::Path, file_id: DataFileId) -> std::path::PathBuf {
+    dir.join(format!("{file_id}.tqb"))
+}
+
+fn populate_with_compaction_history(store: &TranquilBlockStore, live_cids: &[u32]) {
+    live_cids.iter().for_each(|&seed| {
+        store
+            .put_blocks_blocking(vec![(test_cid(seed), block_data(seed))])
+            .unwrap();
+    });
+
+    (0..200u32).for_each(|round| {
+        let churn = test_cid(50_000 + round);
+        store
+            .put_blocks_blocking(vec![(churn, block_data(50_000 + round))])
+            .unwrap();
+        store.apply_commit_blocking(vec![], vec![churn]).unwrap();
+
+        if round % 4 == 0 {
+            common::compact_lowest_liveness(store);
+        }
+    });
+}
+
+#[test]
+fn deleting_indexed_data_file_externally_self_heals_on_compaction() {
+    with_runtime(|| {
+        let dir = tempfile::TempDir::new().unwrap();
+        let live_cids: Vec<u32> = (0..12u32).collect();
+
+        {
+            let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap();
+            populate_with_compaction_history(&store, &live_cids);
+            drop(store);
+        }
+
+        let data_dir = dir.path().join("data");
+        let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap();
+
+        let liveness = store.compaction_liveness(0).unwrap();
+        let victim_fid = liveness
+            .iter()
+            .filter(|(_, info)| info.live_blocks > 0)
+            .map(|(&fid, _)| fid)
+            .next()
+            .expect("expected at least one file with live blocks");
+
+        drop(store);
+
+        let victim_path = data_file_path(&data_dir, victim_fid);
+        assert!(
+            victim_path.exists(),
+            "victim data file should exist before deletion"
+        );
+        fs::remove_file(&victim_path).unwrap();
+
+        let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap();
+
+        let liveness_before = store.compaction_liveness(0).unwrap();
+        assert!(
+            liveness_before.contains_key(&victim_fid),
+            "index should still claim the deleted file before compaction self-heal"
+        );
+
+        let result = store.compact_file(victim_fid, 0).unwrap();
+        match result {
+            CompactionResult::Purged {
+                file_id,
+                phantom_blocks,
+            } => {
+                assert_eq!(file_id, victim_fid);
+                assert!(
+                    phantom_blocks > 0,
+                    "expected to purge non-zero phantom entries"
+                );
+            }
+            CompactionResult::Compacted(stats) => {
+                panic!(
+                    "expected purge for missing source file, got compaction with {stats:?} live={} dead={}",
+                    stats.live_blocks, stats.dead_blocks
+                );
+            }
+        }
+
+        let liveness_after = store.compaction_liveness(0).unwrap();
+        assert!(
+            !liveness_after.contains_key(&victim_fid),
+            "compaction-purge must remove all index entries pointing at the deleted file"
+        );
+    });
+}
+
+#[test]
+fn external_hint_orphan_cleaned_by_consistency_repair() {
+    with_runtime(|| {
+        let dir = tempfile::TempDir::new().unwrap();
+
+        {
+            let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap();
+            populate_with_compaction_history(&store, &(0..6u32).collect::<Vec<_>>());
+            drop(store);
+        }
+
+        let data_dir = dir.path().join("data");
+
+        let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap();
+        let any_existing_fid = store
+            .list_data_files()
+            .unwrap()
+            .into_iter()
+            .next()
+            .expect("expected at least one data file after populate");
+        drop(store);
+
+        let orphan_hint_fid = DataFileId::new(any_existing_fid.raw().saturating_add(10_000));
+        let orphan_path = hint_file_path(&data_dir, orphan_hint_fid);
+        fs::write(&orphan_path, b"\x00").unwrap();
+
+        let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap();
+
+        let metastore_dir = dir.path().join("metastore");
+        std::fs::create_dir_all(&metastore_dir).unwrap();
+        let metastore = tranquil_store::metastore::Metastore::open(
+            &metastore_dir,
+            tranquil_store::metastore::MetastoreConfig::default(),
+        )
+        .unwrap();
+
+        let segments_dir = dir.path().join("eventlog").join("segments");
+        std::fs::create_dir_all(&segments_dir).unwrap();
+        let eventlog = tranquil_store::eventlog::EventLog::open(
+            tranquil_store::eventlog::EventLogConfig {
+                segments_dir,
+                ..tranquil_store::eventlog::EventLogConfig::default()
+            },
+            tranquil_store::RealIO::new(),
+        )
+        .unwrap();
+
+        let report =
+            tranquil_store::consistency::verify_store_consistency(&store, &metastore, &eventlog);
+
+        assert!(
+            report.orphan_hint_files.contains(&orphan_hint_fid),
+            "consistency check should flag the synthetic orphan hint file"
+        );
+
+        let repair = tranquil_store::consistency::repair_known_issues(&store, &report);
+        assert!(
+            repair.orphan_hints_removed >= 1,
+            "repair should remove the orphan hint file"
+        );
+        assert!(
+            !orphan_path.exists(),
+            "orphan hint file should be unlinked after repair"
+        );
+    });
+}
+
+#[test]
+fn consistency_check_flags_and_repairs_missing_indexed_file() {
+    with_runtime(|| {
+        let dir = tempfile::TempDir::new().unwrap();
+        let live_cids: Vec<u32> = (0..10u32).collect();
+
+        {
+            let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap();
+            populate_with_compaction_history(&store, &live_cids);
+            drop(store);
+        }
+
+        let data_dir = dir.path().join("data");
+        let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap();
+        let victim_fid = store
+            .compaction_liveness(0)
+            .unwrap()
+            .iter()
+            .filter(|(_, info)| info.live_blocks > 0)
+            .map(|(&fid, _)| fid)
+            .next()
+            .expect("expected at least one indexed file");
+        drop(store);
+
+        fs::remove_file(data_file_path(&data_dir, victim_fid)).unwrap();
+        let _ = fs::remove_file(hint_file_path(&data_dir, victim_fid));
+
+        let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap();
+        let metastore_dir = dir.path().join("metastore");
+        std::fs::create_dir_all(&metastore_dir).unwrap();
+        let metastore = tranquil_store::metastore::Metastore::open(
+            &metastore_dir,
+            tranquil_store::metastore::MetastoreConfig::default(),
+        )
+        .unwrap();
+
+        let segments_dir = dir.path().join("eventlog").join("segments");
+        std::fs::create_dir_all(&segments_dir).unwrap();
+        let eventlog = tranquil_store::eventlog::EventLog::open(
+            tranquil_store::eventlog::EventLogConfig {
+                segments_dir,
+                ..tranquil_store::eventlog::EventLogConfig::default()
+            },
+            tranquil_store::RealIO::new(),
+        )
+        .unwrap();
+
+        let report =
+            tranquil_store::consistency::verify_store_consistency(&store, &metastore, &eventlog);
+        assert!(
+            report.missing_indexed_files.contains(&victim_fid),
+            "consistency check should flag the missing indexed file"
+        );
+
+        let repair = tranquil_store::consistency::repair_known_issues(&store, &report);
+        assert!(
+            repair.phantom_index_entries_purged > 0,
+            "repair should purge phantom entries"
+        );
+
+        let post_repair_liveness = store.compaction_liveness(0).unwrap();
+        assert!(
+            !post_repair_liveness.contains_key(&victim_fid),
+            "no index entries should remain for the missing file after repair"
+        );
+    });
+}

crates/tranquil-store/tests/gauntlet_index_backed.rs

@@ -0,0 +1,130 @@
+mod common;
+
+use std::sync::Arc;
+
+use common::with_runtime;
+use tranquil_store::blockstore::{BlockStoreConfig, GroupCommitConfig, TranquilBlockStore};
+use tranquil_store::gauntlet::{
+    Gauntlet, IndexBackedByDisk, Invariant, InvariantCtx, InvariantSet, Oracle, Scenario, Seed,
+    config_for,
+};
+
+#[test]
+fn index_backed_by_disk_invariant_catches_phantom_after_external_delete() {
+    with_runtime(|| {
+        let dir = tempfile::TempDir::new().unwrap();
+        let cfg = BlockStoreConfig {
+            data_dir: dir.path().join("data"),
+            index_dir: dir.path().join("index"),
+            max_file_size: 4096,
+            group_commit: GroupCommitConfig::default(),
+            shard_count: 1,
+        };
+        let store = Arc::new(TranquilBlockStore::open(cfg).expect("open store"));
+
+        let cids: Vec<[u8; 36]> = (0..20u32)
+            .map(|seed| {
+                let le = seed.to_le_bytes();
+                std::array::from_fn(|i| match i {
+                    0 => 0x01,
+                    1 => 0x71,
+                    2 => 0x12,
+                    3 => 0x20,
+                    4..8 => le[i - 4],
+                    _ => (seed as u8).wrapping_add(i as u8),
+                })
+            })
+            .collect();
+
+        cids.iter().for_each(|cid| {
+            store
+                .put_blocks_blocking(vec![(*cid, vec![0xAA; 96])])
+                .expect("put block");
+        });
+
+        let victim_fid = store
+            .compaction_liveness(0)
+            .unwrap()
+            .iter()
+            .filter(|(_, info)| info.live_blocks > 0)
+            .map(|(&fid, _)| fid)
+            .next()
+            .expect("expected at least one indexed file");
+
+        let victim_path = dir.path().join("data").join(format!("{victim_fid}.tqb"));
+        std::fs::remove_file(&victim_path).unwrap();
+
+        let oracle = Oracle::new();
+        let ctx = InvariantCtx {
+            store: &store,
+            oracle: &oracle,
+            root: None,
+            eventlog: None,
+        };
+        let runtime = tokio::runtime::Handle::current();
+        let result = runtime.block_on(IndexBackedByDisk.check(&ctx));
+
+        let violation = result.expect_err("phantom index entry must trigger violation");
+        assert_eq!(violation.invariant, "IndexBackedByDisk");
+        assert!(
+            violation.detail.contains(&victim_fid.to_string()),
+            "violation detail must reference the deleted file_id: {}",
+            violation.detail
+        );
+    });
+}
+
+#[tokio::test]
+async fn external_corruption_scenario_survives_many_seeds() {
+    let failures: Vec<String> = futures::future::join_all((0..5).map(Seed).map(|seed| async move {
+        let cfg = config_for(Scenario::ExternalCorruption, seed);
+        let report = Gauntlet::new(cfg).expect("build gauntlet").run().await;
+        (seed, report)
+    }))
+    .await
+    .into_iter()
+    .filter(|(_, r)| !r.is_clean())
+    .map(|(seed, r)| {
+        format!(
+            "seed {}: {} violations\n  {}",
+            seed.0,
+            r.violations.len(),
+            r.violations
+                .iter()
+                .map(|v| format!("{}: {}", v.invariant, v.detail))
+                .collect::<Vec<_>>()
+                .join("\n  ")
+        )
+    })
+    .collect();
+    assert!(failures.is_empty(), "{}", failures.join("\n---\n"));
+}
+
+#[tokio::test]
+#[ignore = "long running, validates iris-class regression over many seeds"]
+async fn iris_class_regression_30_seeds() {
+    let failures: Vec<String> =
+        futures::future::join_all((0..30).map(Seed).map(|seed| async move {
+            let mut cfg = config_for(Scenario::SmokePR, seed);
+            cfg.invariants = cfg.invariants | InvariantSet::INDEX_BACKED_BY_DISK;
+            let report = Gauntlet::new(cfg).expect("build gauntlet").run().await;
+            (seed, report)
+        }))
+        .await
+        .into_iter()
+        .filter(|(_, r)| !r.is_clean())
+        .map(|(seed, r)| {
+            format!(
+                "seed {}: {} violations\n  {}",
+                seed.0,
+                r.violations.len(),
+                r.violations
+                    .iter()
+                    .map(|v| format!("{}: {}", v.invariant, v.detail))
+                    .collect::<Vec<_>>()
+                    .join("\n  ")
+            )
+        })
+        .collect();
+    assert!(failures.is_empty(), "{}", failures.join("\n---\n"));
+}

crates/tranquil-store/tests/gauntlet_smoke.rs

@@ -82,6 +82,7 @@ fn fast_sanity_config(seed: Seed) -> GauntletConfig {
         },
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     }
 }
 
@@ -142,6 +143,7 @@ async fn compaction_idempotent_sanity() {
         },
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     };
     let report = Gauntlet::new(cfg).expect("build gauntlet").run().await;
     assert_clean(&report);
@@ -181,6 +183,7 @@ async fn no_orphan_files_sanity() {
         },
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     };
     let report = Gauntlet::new(cfg).expect("build gauntlet").run().await;
     assert_clean(&report);
@@ -225,6 +228,7 @@ async fn simulated_pristine_roundtrip() {
         },
         eventlog: None,
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     };
     let report = Gauntlet::new(cfg).expect("build gauntlet").run().await;
     assert_clean(&report);
@@ -279,6 +283,7 @@ async fn firehose_fanout_pristine_smoke() {
             max_segment_size: MaxSegmentSize(32 * 1024),
         }),
         writer_concurrency: WriterConcurrency(1),
+        tolerate_op_errors: false,
     };
     let report = Gauntlet::new(cfg).expect("build gauntlet").run().await;
     assert_clean(&report);
@@ -325,6 +330,7 @@ async fn contended_readers_pristine_smoke() {
         },
         eventlog: None,
         writer_concurrency: WriterConcurrency(16),
+        tolerate_op_errors: false,
     };
     let report = Gauntlet::new(cfg).expect("build gauntlet").run().await;
     assert_clean(&report);
@@ -372,6 +378,7 @@ async fn contended_writers_pristine_smoke() {
         },
         eventlog: None,
         writer_concurrency: WriterConcurrency(8),
+        tolerate_op_errors: false,
     };
     let report = Gauntlet::new(cfg).expect("build gauntlet").run().await;
     assert_clean(&report);
@@ -460,3 +467,113 @@ 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),
+        tolerate_op_errors: false,
+    };
+    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(fast_sanity_config, &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/gc.rs

@@ -193,10 +193,15 @@ fn compact_data_file_preserves_live_removes_dead() {
             .unwrap();
         std::thread::sleep(std::time::Duration::from_millis(5));
 
-        let result = store.compact_file(first_file, 0).unwrap();
-        assert!(result.dead_blocks > 0, "should have removed dead blocks");
-        assert!(result.live_blocks > 0, "should have preserved live blocks");
-        assert!(result.reclaimed_bytes > 0, "should have reclaimed space");
+        let stats = match store.compact_file(first_file, 0).unwrap() {
+            tranquil_store::blockstore::CompactionResult::Compacted(s) => s,
+            tranquil_store::blockstore::CompactionResult::Purged { .. } => {
+                panic!("expected compaction, got phantom purge")
+            }
+        };
+        assert!(stats.dead_blocks > 0, "should have removed dead blocks");
+        assert!(stats.live_blocks > 0, "should have preserved live blocks");
+        assert!(stats.reclaimed_bytes > 0, "should have reclaimed space");
 
         [1u8, 3].iter().for_each(|&seed| {
             let data = store.get_block_sync(&test_cid(seed)).unwrap();

crates/tranquil-store/tests/gc_stress.rs

@@ -579,9 +579,14 @@ fn all_dead_file_compaction() {
         advance_epoch(&store);
         std::thread::sleep(std::time::Duration::from_millis(5));
 
-        let result = store.compact_file(first_file, 0).unwrap();
-        assert_eq!(result.live_blocks, 0);
-        assert!(result.dead_blocks > 0);
+        let stats = match store.compact_file(first_file, 0).unwrap() {
+            tranquil_store::blockstore::CompactionResult::Compacted(s) => s,
+            tranquil_store::blockstore::CompactionResult::Purged { .. } => {
+                panic!("expected compaction, got phantom purge")
+            }
+        };
+        assert_eq!(stats.live_blocks, 0);
+        assert!(stats.dead_blocks > 0);
 
         (0u32..5).for_each(|seed| {
             let data = store.get_block_sync(&test_cid_u32(seed)).unwrap();
@@ -838,10 +843,15 @@ fn grace_period_prevents_collection_during_active_write() {
 
         let files = store.list_data_files().unwrap();
         let first_file = files[0];
-        let result = store.compact_file(first_file, 600_000).unwrap();
+        let stats = match store.compact_file(first_file, 600_000).unwrap() {
+            tranquil_store::blockstore::CompactionResult::Compacted(s) => s,
+            tranquil_store::blockstore::CompactionResult::Purged { .. } => {
+                panic!("expected compaction, got phantom purge")
+            }
+        };
 
         assert_eq!(
-            result.dead_blocks, 0,
+            stats.dead_blocks, 0,
             "grace period should prevent any collection"
         );
 

crates/tranquil-store/tests/sim_blockstore.rs

@@ -12,7 +12,9 @@ use tranquil_store::blockstore::{
     GroupCommitConfig, HINT_RECORD_SIZE, HintFileWriter, HintOffset, TranquilBlockStore,
     WallClockMs, WriteCursor, hint_file_path,
 };
-use tranquil_store::{FaultConfig, OpenOptions, SimulatedIO, StorageIO, sim_seed_range};
+use tranquil_store::{
+    FaultConfig, OpenOptions, SimulatedIO, StorageIO, SyncReorderWindow, sim_seed_range,
+};
 
 use common::{Rng, advance_epoch, block_data, test_cid, with_runtime};
 
@@ -691,3 +693,53 @@ 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,10 +5,12 @@ use std::sync::Arc;
 
 use rayon::prelude::*;
 use tranquil_store::eventlog::{
-    DidHash, EVENT_RECORD_OVERHEAD, EventLogWriter, EventSequence, EventTypeTag, MAX_EVENT_PAYLOAD,
-    SEGMENT_HEADER_SIZE, SegmentId, SegmentManager, SegmentReader, ValidEvent,
+    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,
 };
-use tranquil_store::{FaultConfig, Probability, SimulatedIO, StorageIO, sim_seed_range};
 
 use common::Rng;
 
@@ -204,13 +206,13 @@ fn crash_mid_rotation_with_faults() {
             EventLogWriter::open(Arc::clone(&mgr), 256, MAX_EVENT_PAYLOAD)
         }));
 
-        if let Ok(Ok(writer)) = recovery
-            && let Ok(synced_before) = write_result
-        {
+        if let Ok(Ok(writer)) = recovery {
+            let recovered = writer.synced_seq().raw();
             assert!(
-                writer.synced_seq().raw() <= synced_before,
-                "seed {seed}: recovered more events than were synced"
+                recovered <= events_per_seg as u64,
+                "seed {seed}: recovered {recovered} > written {events_per_seg}"
             );
+            let _ = write_result;
         }
     });
 }
@@ -1020,3 +1022,172 @@ 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-store/tests/sim_soak.rs

@@ -124,6 +124,7 @@ fn verify_integrity(stores: &TestStores, oracle: &SoakOracle) {
 }
 
 #[test]
+#[ignore = "wall-clock soak, 30min runtime. Reenable once we mock time for deterministic soak"]
 fn sim_soak_continuous_operations_with_crash_recovery() {
     let rt = tokio::runtime::Runtime::new().unwrap();
     let _guard = rt.enter();

crates/tranquil-sync/Cargo.toml

@@ -22,9 +22,3 @@ 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

@@ -85,14 +85,13 @@ async fn recover_lagged_events(
         };
         for event in events {
             *last_seen = event.seq;
-            let bytes =
-                match format_event_with_prefetched_blocks(state, event, &prefetched).await {
-                    Ok(b) => b,
-                    Err(e) => {
-                        warn!("Lag recovery format failed: {}", e);
-                        return Err(());
-                    }
-                };
+            let bytes = match format_event_with_prefetched_blocks(state, event, &prefetched).await {
+                Ok(b) => b,
+                Err(e) => {
+                    warn!("Lag recovery format failed: {}", e);
+                    return Err(());
+                }
+            };
             if let Err(e) = socket.send(Message::Binary(bytes.into())).await {
                 warn!("Lag recovery send failed: {}", e);
                 return Err(());
@@ -302,8 +301,6 @@ async fn handle_socket_inner(
                         break;
                     };
 
-                    info!("{msg:?}");
-
                     if let Message::Close(_) = msg {
                         info!("Client closed connection");
                         break;
@@ -314,44 +311,3 @@ 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);
-    }
-}

docs/install-nix.md

@@ -38,18 +38,189 @@ In `configuration.nix`:
 {
   services.tranquil-pds = {
     enable = true;
-    database.createLocally = true;
+    database.createLocally = true; # set to false if you prefer to manually manage postgres. You must then set settings.database.url.
     settings = {
       server.hostname = "pds.example.com";
+      # database.url = "postgresql://user:postgres@example.com" -- Only if database.createLocally is set to false.
       # see example.toml for all options
     };
+    # see Secrets section for more information.
+    environmentFiles = [ "/etc/secrets/tranquil.env.production" ];
   };
 }
 ```
 
-This will set up the local postgres database for you automatically. If you prefer to manage postgres yourself, leave `database.createLocally` at its default (`false`) and set `settings.database.url` manually.
+You will also likely want to configure Caddy or nginx to actually serve traffic to the service. An example Caddy config is provided below.
 
-See [example.toml](../example.toml) for the full set of configuration options.
+See [example.toml](https://tangled.org/tranquil.farm/tranquil-pds/blob/main/example.toml) at the repository root for the full set of configuration options.
+
+### Example Caddy config
+
+```nix
+{
+  services.caddy = {
+    enable = true;
+
+    virtualHosts = {
+      "pds.example.com" = {
+        # by default, tranquil runs on port 3000.
+        # You can change this with the tranquil-pds.settings.server.port option in the service config.
+        extraConfig = ''
+          reverse_proxy localhost:3000
+        '';
+      };
+    };
+  };
+
+  networking.firewall.allowedTCPPorts = [
+    80
+    443
+  ];
+}
+```
+
+### Secrets
+
+Additionally, you will need to provide the following environment variables.
+
+```env
+# use `openssl rand -base64 32` to generate these.
+JWT_SECRET=<secret_here>
+DPOP_SECRET=<secret_here>
+MASTER_KEY=<secret_here>
+```
+
+The simplest (least secure and least reproducible) option is to provide these secrets in a `.env` file using the `environmentFiles` option as shown above.
+
+It is recommended that you use something like [`agenix`](https://github.com/ryantm/agenix) or [`sops-nix`](https://github.com/Mic92/sops-nix) for proper secrets management on a NixOS machine instead.
+
+Example `sops-nix` config:
+
+```nix
+let
+  inherit (config.sops) secrets;
+in
+{
+  services.tranquil-pds = {
+    enable = true;
+    database.createLocally = true;
+    settings = {
+      server.hostname = "pds.example.com";
+    };
+    environmentFiles = [ secrets.tranquils-secrets.path ];
+  };
+}
+```
+
+## Communications
+
+To actually be able to receive communications from the PDS for things like verification codes, or PLC operations, you must set at least one of the following options.
+
+### Email
+
+```nix
+let
+  inherit (config.sops) secrets;
+in
+{
+  services.tranquil-pds = {
+    enable = true;
+    database.createLocally = true;
+    settings = {
+      server.hostname = "pds.example.com";
+      email.from_address = "tranquil_admin@pds.example.com";
+      # email.from_name = "Tranquil PDS";
+    };
+    environmentFiles = [ secrets.tranquils-secrets.path ];
+  };
+}
+```
+
+For DKIM options, please consult the [example.toml](https://tangled.org/tranquil.farm/tranquil-pds/blob/main/example.toml) at the repository root.
+
+### Discord
+
+```nix
+let
+  inherit (config.sops) secrets;
+in
+{
+  services.tranquil-pds = {
+    enable = true;
+    database.createLocally = true;
+    settings = {
+      server.hostname = "pds.example.com";
+      # if you're using proper secrets management, you should provide DISCORD_BOT_TOKEN in the environment file instead.
+      discord.bot_token = "whatever";
+    };
+    environmentFiles = [ secrets.tranquils-secrets.path ];
+  };
+}
+```
+
+### Telegram
+
+```nix
+let
+  inherit (config.sops) secrets;
+in
+{
+  services.tranquil-pds = {
+    enable = true;
+    database.createLocally = true;
+    settings = {
+      server.hostname = "pds.example.com";
+      telegram = {
+        # if you're using proper secrets management, you should provide TELEGRAM_BOT_TOKEN in the environment file instead.
+        bot_token = "whatever";
+        # if you're using proper secrets management, you should provide TELEGRAM_WEBHOOK_SECRET in the environment file instead.
+        webhook_secret = "whatever2";
+      };
+    };
+    environmentFiles = [ secrets.tranquils-secrets.path ];
+  };
+}
+```
+
+### Signal
+
+```nix
+let
+  inherit (config.sops) secrets;
+in
+{
+  services.tranquil-pds = {
+    enable = true;
+    database.createLocally = true;
+    settings = {
+      server.hostname = "pds.example.com";
+      # you must link a device using the admin API before enabling this option.
+      signal.enabled = true;
+    };
+    environmentFiles = [ secrets.tranquils-secrets.path ];
+  };
+}
+```
+
+### No comms channel
+
+If you have not set up any of these, you can technically still access any relevant information by querying the database directly.
+
+Please keep in mind this is a last-ditch attempt and it is highly recommended that you do in fact specify some channel for communications.
+
+Running `sudo -u tranquil-pds psql` will give you command line access to the PostgreSQL database.
+
+From there, you can run `SELECT * FROM comms_queue;` which will return all communications sent from the PDS. From there, you can extract any relevant information.
+
+## Bootstrap
+
+When the PDS service is able to properly run for the first time, you will be given a bootstrap invite code to migrate your repository to this PDS.
+
+The simplest way to find this invite code is to check the service logs by doing `journalctl -u tranquil-pds`.
+
+The log entry shold look something like this:
+
+`INFO tranquil_pds::state: No users exist and invite codes are required. Bootstrap invite code: <invite_code_here>`
 
 ## Binary cache
 

example.toml

@@ -373,12 +373,117 @@
 # Default value: "Tranquil PDS"
 #from_name = "Tranquil PDS"
 
-# Path to the `sendmail` binary.
+# HELO/EHLO name announced to remote SMTP servers. Applies to both
+# smarthost and direct-MX modes. Defaults to the server hostname.
 #
-# Can also be specified via environment variable `SENDMAIL_PATH`.
+# 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.
 #
-# Default value: "/usr/sbin/sendmail"
-#sendmail_path = "/usr/sbin/sendmail"
+# 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 =
 
 [discord]
 # Discord bot token. When unset, Discord integration is disabled.

flake.lock

@@ -2,18 +2,15 @@
   "nodes": {
     "nixpkgs": {
       "locked": {
-        "lastModified": 1775888245,
-        "narHash": "sha256-nwASzrRDD1JBEu/o8ekKYEXm/oJW6EMCzCRdrwcLe90=",
-        "owner": "nixos",
-        "repo": "nixpkgs",
-        "rev": "13043924aaa7375ce482ebe2494338e058282925",
-        "type": "github"
+        "lastModified": 1778672786,
+        "narHash": "sha256-Tf7q8/0aIwg4Btd/GUMNHXDTqlrWTr5o/vZ8euFUAQY=",
+        "rev": "eef00dfd8a712b34af845f9350bac681b1228bd1",
+        "type": "tarball",
+        "url": "https://releases.nixos.org/nixpkgs/nixpkgs-26.05pre997230.eef00dfd8a71/nixexprs.tar.xz?lastModified=1778672786&rev=eef00dfd8a712b34af845f9350bac681b1228bd1"
       },
       "original": {
-        "owner": "nixos",
-        "ref": "nixpkgs-unstable",
-        "repo": "nixpkgs",
-        "type": "github"
+        "type": "tarball",
+        "url": "https://channels.nixos.org/nixpkgs-unstable/nixexprs.tar.xz"
       }
     },
     "root": {

flake.nix

@@ -1,6 +1,6 @@
 {
   inputs = {
-    nixpkgs.url = "github:nixos/nixpkgs/nixpkgs-unstable";
+    nixpkgs.url = "https://channels.nixos.org/nixpkgs-unstable/nixexprs.tar.xz";
   };
 
   outputs =

frontend.nix

@@ -19,7 +19,7 @@ stdenvNoCC.mkDerivation (finalAttrs: {
   pnpmDeps = fetchPnpmDeps {
     inherit (finalAttrs) pname version src;
     fetcherVersion = 3;
-    hash = "sha256-E0S8dOaTOpY9m7Ft59tUQ6CLlLriWPE4WE1+S45vomY=";
+    hash = "sha256-qbmIAvE/3u/NB5x9bERCGQqwiDLkzjff3QchgR+ZDFs=";
   };
 
   nativeBuildInputs = [

frontend/Dockerfile

@@ -1,7 +1,7 @@
 FROM node:24-alpine AS builder
 RUN corepack enable && corepack prepare pnpm@latest --activate
 WORKDIR /app
-COPY package.json pnpm-lock.yaml ./
+COPY package.json pnpm-lock.yaml pnpm-workspace.yaml ./
 RUN pnpm install --frozen-lockfile
 COPY . ./
 RUN pnpm build

frontend/pnpm-lock.yaml

@@ -1311,7 +1311,6 @@ packages:
   whatwg-encoding@3.1.1:
     resolution: {integrity: sha512-6qN4hJdMwfYBtE3YBTTHhoeuUrDBPZmbQaxWAqSALV/MeEnR5z1xd8UKud2RAkFoPkmB+hli1TZSnyi84xz1vQ==}
     engines: {node: '>=18'}
-    deprecated: Use @exodus/bytes instead for a more spec-conformant and faster implementation
 
   whatwg-mimetype@4.0.0:
     resolution: {integrity: sha512-QaKxh0eNIi2mE9p2vEdzfagOKHCcj1pJ56EEHGQOVxp8r9/iszLUUV7v89x9O1p/T+NlTM5W7jW6+cz4Fq1YVg==}

frontend/pnpm-workspace.yaml

@@ -0,0 +1,4 @@
+confirmModulesPurge: false
+allowBuilds:
+  es5-ext: true
+  esbuild: true

module.nix

@@ -120,14 +120,6 @@ in
               description = "Directory for storing blobs";
             };
           };
-
-          email = {
-            sendmail_path = mkOption {
-              type = types.path;
-              default = lib.getExe pkgs.system-sendmail;
-              description = "Path to the sendmail executable to use for sending emails.";
-            };
-          };
         };
       };