fix(lexicons): cache ttl

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

.config/nextest.toml

@@ -72,6 +72,10 @@ test-group = "io-heavy-sim"
 filter = "test(/test_scale_/) | test(/full_backup_and_restore/)"
 slow-timeout = { period = "120s", terminate-after = 4 }
 
+[[profile.default.overrides]]
+filter = "binary(gauntlet_smoke)"
+slow-timeout = { period = "300s", terminate-after = 8 }
+
 [[profile.default.overrides]]
 filter = "binary(compaction_restart) | binary(mst_refcount_integrity) | binary(gc_compaction_restart)"
 slow-timeout = { period = "120s", terminate-after = 4 }

Cargo.lock

@@ -7405,7 +7405,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-api"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "anyhow",
  "axum",
@@ -7456,7 +7456,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-auth"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "anyhow",
  "base32",
@@ -7479,7 +7479,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-cache"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "async-trait",
  "base64 0.22.1",
@@ -7493,7 +7493,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-comms"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "async-trait",
  "base64 0.22.1",
@@ -7511,7 +7511,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-config"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "confique",
  "serde",
@@ -7519,7 +7519,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-crypto"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "aes-gcm",
  "base64 0.22.1",
@@ -7535,7 +7535,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-db"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "async-trait",
  "chrono",
@@ -7552,7 +7552,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-db-traits"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "async-trait",
  "base64 0.22.1",
@@ -7568,7 +7568,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-infra"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "async-trait",
  "bytes",
@@ -7579,9 +7579,10 @@ dependencies = [
 
 [[package]]
 name = "tranquil-lexicon"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "chrono",
+ "futures",
  "hickory-resolver",
  "parking_lot",
  "reqwest",
@@ -7597,7 +7598,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-oauth"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "anyhow",
  "axum",
@@ -7620,7 +7621,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-oauth-server"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "axum",
  "base64 0.22.1",
@@ -7653,7 +7654,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-pds"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "aes-gcm",
  "anyhow",
@@ -7745,7 +7746,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-repo"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "bytes",
  "cid",
@@ -7757,7 +7758,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-ripple"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "async-trait",
  "backon",
@@ -7782,7 +7783,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-scopes"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "axum",
  "futures",
@@ -7798,7 +7799,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-server"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "axum",
  "clap",
@@ -7819,7 +7820,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-signal"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "async-trait",
  "chrono",
@@ -7842,7 +7843,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-storage"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "async-trait",
  "aws-config",
@@ -7859,7 +7860,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-store"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "async-trait",
  "bytes",
@@ -7906,7 +7907,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-sync"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "anyhow",
  "axum",
@@ -7928,7 +7929,7 @@ dependencies = [
 
 [[package]]
 name = "tranquil-types"
-version = "0.5.4"
+version = "0.5.5"
 dependencies = [
  "chrono",
  "cid",

Cargo.toml

@@ -26,7 +26,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.5.4"
+version = "0.5.5"
 edition = "2024"
 license = "AGPL-3.0-or-later"
 

crates/tranquil-api/src/actor/preferences.rs

@@ -12,7 +12,6 @@ use tranquil_pds::auth::{Auth, NotTakendown, Permissive};
 use tranquil_pds::state::AppState;
 
 const APP_BSKY_NAMESPACE: &str = "app.bsky";
-const MAX_PREFERENCES_COUNT: usize = 100;
 const MAX_PREFERENCE_SIZE: usize = 10_000;
 const PERSONAL_DETAILS_PREF: &str = "app.bsky.actor.defs#personalDetailsPref";
 const DECLARED_AGE_PREF: &str = "app.bsky.actor.defs#declaredAgePref";
@@ -92,6 +91,8 @@ pub async fn put_preferences(
     auth: Auth<NotTakendown>,
     Json(input): Json<PutPreferencesInput>,
 ) -> Response {
+    let max_preferences_count: usize = tranquil_config::get().server.max_preferences_count;
+
     let has_full_access = auth.permissions().has_full_access();
     let user_id: uuid::Uuid = match state.repos.user.get_id_by_did(&auth.did).await {
         Ok(Some(id)) => id,
@@ -99,11 +100,11 @@ pub async fn put_preferences(
             return ApiError::InternalError(Some("User not found".into())).into_response();
         }
     };
-    if input.preferences.len() > MAX_PREFERENCES_COUNT {
+    if input.preferences.len() > max_preferences_count {
         return ApiError::InvalidRequest(format!(
             "Too many preferences: {} exceeds limit of {}",
             input.preferences.len(),
-            MAX_PREFERENCES_COUNT
+            max_preferences_count
         ))
         .into_response();
     }

crates/tranquil-config/src/lib.rs

@@ -108,6 +108,9 @@ pub struct TranquilConfig {
     #[config(nested)]
     pub storage: StorageConfig,
 
+    #[config(nested)]
+    pub tranquil_store: TranquilStoreConfig,
+
     #[config(nested)]
     pub cache: CacheConfig,
 
@@ -143,9 +146,6 @@ pub struct TranquilConfig {
 
     #[config(nested)]
     pub scheduled: ScheduledConfig,
-
-    #[config(nested)]
-    pub tranquil_store: TranquilStoreConfig,
 }
 
 impl TranquilConfig {
@@ -451,6 +451,10 @@ pub struct ServerConfig {
     /// Maximum allowed blob size in bytes (default 10 GiB).
     #[config(env = "MAX_BLOB_SIZE", default = 10_737_418_240u64)]
     pub max_blob_size: u64,
+
+    /// Maximum allowed number of preferences
+    #[config(env = "MAX_PREFERENCES_COUNT", default = 1000)]
+    pub max_preferences_count: usize,
 }
 
 impl ServerConfig {
@@ -656,10 +660,12 @@ pub struct StorageConfig {
     #[config(env = "S3_BUCKET")]
     pub s3_bucket: Option<String>,
 
-    /// Custom S3 endpoint URL (for MinIO, R2, etc.).
+    /// Custom S3 endpoint URL.
     #[config(env = "S3_ENDPOINT")]
     pub s3_endpoint: Option<String>,
 
+    /// Repository backend: `postgres` by default, or `tranquil-store`, our embedded db.
+    /// tranquil-store is EXPERIMENTAL!!!! RISK OF TOTAL DATA LOSS.
     #[config(env = "REPO_BACKEND", default = "postgres")]
     pub repo_backend: String,
 }
@@ -674,7 +680,7 @@ impl StorageConfig {
 
 #[derive(Debug, Config)]
 pub struct CacheConfig {
-    /// Cache backend: `ripple` (default, built-in gossip) or `valkey`.
+    /// Cache backend: `ripple` by default, or `valkey`.
     #[config(env = "CACHE_BACKEND", default = "ripple")]
     pub backend: String,
 
@@ -774,6 +780,8 @@ pub struct TelegramConfig {
 
 #[derive(Debug, Config)]
 pub struct SignalConfig {
+    /// Protocol state is stored in postgres' signal_* tables.
+    /// Link a device via the admin API before enabling.
     #[config(env = "SIGNAL_ENABLED", default = false)]
     pub enabled: bool,
 }
@@ -1124,27 +1132,26 @@ pub struct ScheduledConfig {
 
     /// Maximum age of events retained in the eventlog before pruning.
     /// Per the atproto firehose spec, the relay backfill window only needs
-    /// to cover "hours or days". Default: 7 days.
+    /// to cover "hours or days".
     #[config(env = "EVENT_RETENTION_MAX_AGE_SECS", default = 604800)]
     pub event_retention_max_age_secs: u64,
 
     /// Interval in seconds between event retention prune passes.
-    /// Set to 0 to disable. Default: hourly.
+    /// Set to 0 to disable.
     #[config(env = "EVENT_RETENTION_INTERVAL_SECS", default = 3600)]
     pub event_retention_interval_secs: u64,
 }
 
 #[derive(Debug, Config)]
 pub struct TranquilStoreConfig {
-    /// Directory for tranquil-store data (metastore, eventlog).
+    /// Directory for tranquil-store data: the metastore, eventlog, and blockstore.
     #[config(
         env = "TRANQUIL_STORE_DATA_DIR",
         default = "/var/lib/tranquil-pds/store"
     )]
     pub data_dir: String,
 
-    /// Fjall block cache size in megabytes. Defaults to 20% of system RAM
-    /// when unset.
+    /// Fjall block cache size in megabytes. Defaults to 20% of system RAM when unset.
     #[config(env = "TRANQUIL_STORE_MEMORY_BUDGET_MB")]
     pub memory_budget_mb: Option<u64>,
 
@@ -1152,9 +1159,9 @@ pub struct TranquilStoreConfig {
     #[config(env = "TRANQUIL_STORE_HANDLER_THREADS")]
     pub handler_threads: Option<usize>,
 
-    /// Maximum total bytes of pending (unsynced) eventlog payloads. Appenders
-    /// block once this budget is exhausted until in-flight events drain via
-    /// fsync. Set to 0 to disable backpressure (unbounded). Default: 1 GiB.
+    /// Maximum total bytes of pending (unsynced) eventlog payloads. Appenders block
+    /// once this budget is exhausted until in-flight events drain via fsync. Set to
+    /// 0 to disable backpressure. Default: 1 GiB.
     #[config(
         env = "TRANQUIL_STORE_EVENTLOG_PENDING_BYTES_BUDGET",
         default = 1_073_741_824

crates/tranquil-lexicon/Cargo.toml

@@ -24,3 +24,4 @@ urlencoding = { workspace = true, optional = true }
 [dev-dependencies]
 wiremock = { workspace = true }
 tokio = { workspace = true }
+futures = { workspace = true }

crates/tranquil-lexicon/src/dynamic.rs

@@ -5,25 +5,60 @@ use std::collections::{HashMap, VecDeque};
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::{Duration, Instant};
+use tokio::sync::Notify;
 
 const NEGATIVE_CACHE_TTL: Duration = Duration::from_secs(24 * 60 * 60);
+const POSITIVE_CACHE_TTL: Duration = Duration::from_secs(24 * 60 * 60);
+const REFRESH_FAILURE_BACKOFF: Duration = Duration::from_secs(60);
 const MAX_DYNAMIC_SCHEMAS: usize = 1024;
 
 struct NegativeEntry {
     expires_at: Instant,
 }
 
+struct PositiveEntry {
+    doc: Arc<LexiconDoc>,
+    expires_at: Instant,
+}
+
+pub(crate) enum CacheEntry {
+    Fresh(Arc<LexiconDoc>),
+    Stale(Arc<LexiconDoc>),
+}
+
+impl CacheEntry {
+    #[cfg(test)]
+    fn is_fresh(&self) -> bool {
+        matches!(self, Self::Fresh(_))
+    }
+}
+
 struct SchemaStore {
-    schemas: HashMap<String, Arc<LexiconDoc>>,
+    schemas: HashMap<String, PositiveEntry>,
     insertion_order: VecDeque<String>,
 }
 
 pub struct DynamicRegistry {
     store: RwLock<SchemaStore>,
     negative_cache: RwLock<HashMap<String, NegativeEntry>>,
+    in_flight: RwLock<HashMap<String, Arc<Notify>>>,
     network_disabled: AtomicBool,
 }
 
+struct InFlightGuard<'a> {
+    registry: &'a DynamicRegistry,
+    nsid: String,
+}
+
+impl Drop for InFlightGuard<'_> {
+    fn drop(&mut self) {
+        let notify = self.registry.in_flight.write().remove(&self.nsid);
+        if let Some(n) = notify {
+            n.notify_waiters();
+        }
+    }
+}
+
 impl DynamicRegistry {
     pub fn new() -> Self {
         let network_disabled =
@@ -34,6 +69,7 @@ impl DynamicRegistry {
                 insertion_order: VecDeque::new(),
             }),
             negative_cache: RwLock::new(HashMap::new()),
+            in_flight: RwLock::new(HashMap::new()),
             network_disabled: AtomicBool::new(network_disabled),
         }
     }
@@ -43,8 +79,23 @@ impl DynamicRegistry {
         self.network_disabled.store(disabled, Ordering::Relaxed);
     }
 
-    pub fn get(&self, nsid: &str) -> Option<Arc<LexiconDoc>> {
-        self.store.read().schemas.get(nsid).cloned()
+    pub fn get_cached(&self, nsid: &str) -> Option<Arc<LexiconDoc>> {
+        self.store
+            .read()
+            .schemas
+            .get(nsid)
+            .map(|e| Arc::clone(&e.doc))
+    }
+
+    pub(crate) fn get_entry(&self, nsid: &str) -> Option<CacheEntry> {
+        let now = Instant::now();
+        self.store.read().schemas.get(nsid).map(|e| {
+            if e.expires_at > now {
+                CacheEntry::Fresh(Arc::clone(&e.doc))
+            } else {
+                CacheEntry::Stale(Arc::clone(&e.doc))
+            }
+        })
     }
 
     pub fn is_negative_cached(&self, nsid: &str) -> bool {
@@ -56,7 +107,7 @@ impl DynamicRegistry {
 
     fn insert_negative(&self, nsid: &str) {
         let mut cache = self.negative_cache.write();
-        if cache.len() > MAX_DYNAMIC_SCHEMAS {
+        if cache.len() >= MAX_DYNAMIC_SCHEMAS {
             let now = Instant::now();
             cache.retain(|_, entry| entry.expires_at > now);
         }
@@ -87,29 +138,95 @@ impl DynamicRegistry {
             });
         }
 
-        if store
-            .schemas
-            .insert(nsid.clone(), Arc::clone(&arc))
-            .is_some()
-        {
+        let entry = PositiveEntry {
+            doc: Arc::clone(&arc),
+            expires_at: Instant::now() + POSITIVE_CACHE_TTL,
+        };
+        if store.schemas.insert(nsid.clone(), entry).is_some() {
             store.insertion_order.retain(|k| k != &nsid);
         }
         store.insertion_order.push_back(nsid.clone());
+        drop(store);
 
         self.negative_cache.write().remove(&arc.id);
 
         arc
     }
 
+    fn bump_expiry(&self, nsid: &str, duration: Duration) {
+        let mut store = self.store.write();
+        if let Some(entry) = store.schemas.get_mut(nsid) {
+            entry.expires_at = Instant::now() + duration;
+        }
+    }
+
     pub async fn resolve_and_cache(&self, nsid: &str) -> Result<Arc<LexiconDoc>, ResolveError> {
-        if let Some(doc) = self.get(nsid) {
-            return Ok(doc);
+        self.resolve_and_cache_with(nsid, |n| async move { resolve_lexicon(&n).await })
+            .await
+    }
+
+    async fn resolve_and_cache_with<F, Fut>(
+        &self,
+        nsid: &str,
+        resolver: F,
+    ) -> Result<Arc<LexiconDoc>, ResolveError>
+    where
+        F: FnOnce(String) -> Fut,
+        Fut: std::future::Future<Output = Result<LexiconDoc, ResolveError>>,
+    {
+        match self.get_entry(nsid) {
+            Some(CacheEntry::Fresh(doc)) => Ok(doc),
+            Some(CacheEntry::Stale(stale)) => self.refresh_stale(nsid, stale, resolver).await,
+            None => self.resolve_fresh(nsid, resolver).await,
+        }
+    }
+
+    async fn refresh_stale<F, Fut>(
+        &self,
+        nsid: &str,
+        stale: Arc<LexiconDoc>,
+        resolver: F,
+    ) -> Result<Arc<LexiconDoc>, ResolveError>
+    where
+        F: FnOnce(String) -> Fut,
+        Fut: std::future::Future<Output = Result<LexiconDoc, ResolveError>>,
+    {
+        if self.network_disabled.load(Ordering::Relaxed) {
+            return Ok(stale);
         }
 
+        match self.acquire_leadership(nsid) {
+            Some(_guard) => match resolver(nsid.to_string()).await {
+                Ok(doc) => Ok(self.insert_schema(doc)),
+                Err(e) => {
+                    self.bump_expiry(nsid, REFRESH_FAILURE_BACKOFF);
+                    tracing::warn!(
+                        nsid = nsid,
+                        error = %e,
+                        "lexicon refresh failed, serving stale cached entry"
+                    );
+                    Ok(stale)
+                }
+            },
+            None => {
+                self.wait_for_leader(nsid).await;
+                Ok(self.get_cached(nsid).unwrap_or(stale))
+            }
+        }
+    }
+
+    async fn resolve_fresh<F, Fut>(
+        &self,
+        nsid: &str,
+        resolver: F,
+    ) -> Result<Arc<LexiconDoc>, ResolveError>
+    where
+        F: FnOnce(String) -> Fut,
+        Fut: std::future::Future<Output = Result<LexiconDoc, ResolveError>>,
+    {
         if self.network_disabled.load(Ordering::Relaxed) {
             return Err(ResolveError::NetworkDisabled);
         }
-
         if self.is_negative_cached(nsid) {
             return Err(ResolveError::NegativelyCached {
                 nsid: nsid.to_string(),
@@ -117,19 +234,75 @@ impl DynamicRegistry {
             });
         }
 
-        match resolve_lexicon(nsid).await {
-            Ok(doc) => Ok(self.insert_schema(doc)),
-            Err(e) => {
-                tracing::debug!(nsid = nsid, error = %e, "caching negative resolution result");
-                self.insert_negative(nsid);
-                Err(e)
+        match self.acquire_leadership(nsid) {
+            Some(_guard) => match resolver(nsid.to_string()).await {
+                Ok(doc) => Ok(self.insert_schema(doc)),
+                Err(e) => {
+                    self.insert_negative(nsid);
+                    tracing::debug!(nsid = nsid, error = %e, "caching negative resolution result");
+                    Err(e)
+                }
+            },
+            None => {
+                self.wait_for_leader(nsid).await;
+                match self.get_cached(nsid) {
+                    Some(doc) => Ok(doc),
+                    None if self.is_negative_cached(nsid) => {
+                        Err(ResolveError::NegativelyCached {
+                            nsid: nsid.to_string(),
+                            ttl_secs: NEGATIVE_CACHE_TTL.as_secs(),
+                        })
+                    }
+                    None => Err(ResolveError::LeaderAborted {
+                        nsid: nsid.to_string(),
+                    }),
+                }
             }
         }
     }
 
+    fn acquire_leadership(&self, nsid: &str) -> Option<InFlightGuard<'_>> {
+        let mut map = self.in_flight.write();
+        if map.contains_key(nsid) {
+            None
+        } else {
+            map.insert(nsid.to_string(), Arc::new(Notify::new()));
+            Some(InFlightGuard {
+                registry: self,
+                nsid: nsid.to_string(),
+            })
+        }
+    }
+
+    async fn wait_for_leader(&self, nsid: &str) {
+        let notify = {
+            let map = self.in_flight.read();
+            match map.get(nsid) {
+                Some(n) => Arc::clone(n),
+                None => return,
+            }
+        };
+        let notified = notify.notified();
+        tokio::pin!(notified);
+        notified.as_mut().enable();
+        let still_active = self.in_flight.read().contains_key(nsid);
+        if !still_active {
+            return;
+        }
+        notified.as_mut().await;
+    }
+
     pub fn schema_count(&self) -> usize {
         self.store.read().schemas.len()
     }
+
+    #[cfg(test)]
+    fn expire_now(&self, nsid: &str) {
+        let mut store = self.store.write();
+        if let Some(entry) = store.schemas.get_mut(nsid) {
+            entry.expires_at = Instant::now();
+        }
+    }
 }
 
 impl Default for DynamicRegistry {
@@ -171,7 +344,7 @@ mod tests {
     #[test]
     fn test_empty_lookup() {
         let registry = DynamicRegistry::new();
-        assert!(registry.get("com.example.nonexistent").is_none());
+        assert!(registry.get_cached("com.example.nonexistent").is_none());
         assert_eq!(registry.schema_count(), 0);
     }
 
@@ -188,9 +361,12 @@ mod tests {
         assert_eq!(arc.id, "com.example.test");
         assert_eq!(registry.schema_count(), 1);
 
-        let retrieved = registry.get("com.example.test");
+        let retrieved = registry.get_cached("com.example.test");
         assert!(retrieved.is_some());
         assert_eq!(retrieved.unwrap().id, "com.example.test");
+
+        let entry = registry.get_entry("com.example.test").unwrap();
+        assert!(entry.is_fresh(), "freshly inserted entry must be fresh");
     }
 
     #[test]
@@ -210,6 +386,231 @@ mod tests {
         assert!(!registry.is_negative_cached("com.example.test"));
     }
 
+    #[test]
+    fn test_positive_entry_reports_stale_after_ttl() {
+        let registry = DynamicRegistry::new();
+        let doc = LexiconDoc {
+            lexicon: 1,
+            id: "pet.nel.stale".to_string(),
+            defs: HashMap::new(),
+        };
+        registry.insert_schema(doc);
+
+        assert!(registry.get_entry("pet.nel.stale").unwrap().is_fresh());
+
+        registry.expire_now("pet.nel.stale");
+
+        assert!(
+            !registry.get_entry("pet.nel.stale").unwrap().is_fresh(),
+            "entry past expiry must be reported stale"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_stale_served_on_resolve_failure() {
+        let registry = DynamicRegistry::new();
+        let doc = LexiconDoc {
+            lexicon: 1,
+            id: "pet.nel.flaky".to_string(),
+            defs: HashMap::new(),
+        };
+        registry.insert_schema(doc);
+        registry.expire_now("pet.nel.flaky");
+
+        let result = registry
+            .resolve_and_cache_with("pet.nel.flaky", |n| async move {
+                Err::<LexiconDoc, _>(ResolveError::DnsLookup {
+                    domain: n,
+                    reason: "simulated failure".to_string(),
+                })
+            })
+            .await;
+
+        let served = result.expect("stale entry must be served when refresh fails");
+        assert_eq!(served.id, "pet.nel.flaky");
+        assert!(
+            registry
+                .get_entry("pet.nel.flaky")
+                .unwrap()
+                .is_fresh(),
+            "failed refresh must bump expiry so subsequent lookups skip the resolver"
+        );
+        assert!(
+            !registry.is_negative_cached("pet.nel.flaky"),
+            "stale refresh failure must not poison negative cache"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_fresh_hit_skips_resolver() {
+        let registry = DynamicRegistry::new();
+        let doc = LexiconDoc {
+            lexicon: 1,
+            id: "pet.nel.fresh".to_string(),
+            defs: HashMap::new(),
+        };
+        registry.insert_schema(doc);
+
+        let result = registry
+            .resolve_and_cache_with("pet.nel.fresh", |_| async move {
+                panic!("resolver must not run on fresh hit")
+            })
+            .await;
+
+        assert!(result.is_ok());
+    }
+
+    #[tokio::test]
+    async fn test_stale_served_when_network_disabled() {
+        let registry = DynamicRegistry::new();
+        let doc = LexiconDoc {
+            lexicon: 1,
+            id: "pet.nel.offline".to_string(),
+            defs: HashMap::new(),
+        };
+        registry.insert_schema(doc);
+        registry.expire_now("pet.nel.offline");
+        registry.set_network_disabled(true);
+
+        let result = registry
+            .resolve_and_cache_with("pet.nel.offline", |_| async move {
+                panic!("resolver must not run when network disabled")
+            })
+            .await;
+
+        assert!(result.is_ok());
+    }
+
+    #[tokio::test]
+    async fn test_successful_refresh_updates_cached_at() {
+        let registry = DynamicRegistry::new();
+        let doc = LexiconDoc {
+            lexicon: 1,
+            id: "pet.nel.refresh".to_string(),
+            defs: HashMap::new(),
+        };
+        registry.insert_schema(doc);
+        registry.expire_now("pet.nel.refresh");
+
+        assert!(
+            !registry
+                .get_entry("pet.nel.refresh")
+                .unwrap()
+                .is_fresh()
+        );
+
+        let refreshed = registry
+            .resolve_and_cache_with("pet.nel.refresh", |n| async move {
+                Ok(LexiconDoc {
+                    lexicon: 1,
+                    id: n,
+                    defs: HashMap::new(),
+                })
+            })
+            .await
+            .unwrap();
+
+        assert_eq!(refreshed.id, "pet.nel.refresh");
+        assert!(
+            registry
+                .get_entry("pet.nel.refresh")
+                .unwrap()
+                .is_fresh(),
+            "refresh must restore freshness"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_single_flight_dedups_concurrent_resolves() {
+        use std::sync::atomic::AtomicUsize;
+        let registry = Arc::new(DynamicRegistry::new());
+        let calls = Arc::new(AtomicUsize::new(0));
+
+        let tasks: Vec<_> = (0..16)
+            .map(|_| {
+                let registry = Arc::clone(&registry);
+                let calls = Arc::clone(&calls);
+                tokio::spawn(async move {
+                    registry
+                        .resolve_and_cache_with("pet.nel.herd", |n| {
+                            let calls = Arc::clone(&calls);
+                            async move {
+                                calls.fetch_add(1, Ordering::SeqCst);
+                                tokio::time::sleep(Duration::from_millis(50)).await;
+                                Ok(LexiconDoc {
+                                    lexicon: 1,
+                                    id: n,
+                                    defs: HashMap::new(),
+                                })
+                            }
+                        })
+                        .await
+                })
+            })
+            .collect();
+
+        let results = futures_collect(tasks).await;
+        results
+            .iter()
+            .for_each(|r| assert!(r.is_ok(), "all single-flight callers must succeed"));
+        assert_eq!(
+            calls.load(Ordering::SeqCst),
+            1,
+            "single-flight must coalesce concurrent resolves"
+        );
+        assert_eq!(registry.schema_count(), 1);
+    }
+
+    #[tokio::test]
+    async fn test_single_flight_followers_observe_leader_failure() {
+        use std::sync::atomic::AtomicUsize;
+        let registry = Arc::new(DynamicRegistry::new());
+        let calls = Arc::new(AtomicUsize::new(0));
+
+        let tasks: Vec<_> = (0..8)
+            .map(|_| {
+                let registry = Arc::clone(&registry);
+                let calls = Arc::clone(&calls);
+                tokio::spawn(async move {
+                    registry
+                        .resolve_and_cache_with("pet.nel.failHerd", |n| {
+                            let calls = Arc::clone(&calls);
+                            async move {
+                                calls.fetch_add(1, Ordering::SeqCst);
+                                tokio::time::sleep(Duration::from_millis(50)).await;
+                                Err::<LexiconDoc, _>(ResolveError::DnsLookup {
+                                    domain: n,
+                                    reason: "simulated".to_string(),
+                                })
+                            }
+                        })
+                        .await
+                })
+            })
+            .collect();
+
+        let results = futures_collect(tasks).await;
+        results
+            .iter()
+            .for_each(|r| assert!(r.is_err(), "all followers must observe leader failure"));
+        assert_eq!(
+            calls.load(Ordering::SeqCst),
+            1,
+            "single-flight must coalesce failing resolves too"
+        );
+        assert!(registry.is_negative_cached("pet.nel.failHerd"));
+    }
+
+    async fn futures_collect<T>(
+        handles: Vec<tokio::task::JoinHandle<T>>,
+    ) -> Vec<T> {
+        futures::future::join_all(handles)
+            .await
+            .into_iter()
+            .map(|r| r.expect("task panicked"))
+            .collect()
+    }
+
     #[test]
     fn test_eviction_is_fifo() {
         let registry = DynamicRegistry::new();
@@ -217,7 +618,7 @@ mod tests {
         (0..MAX_DYNAMIC_SCHEMAS).for_each(|i| {
             let doc = LexiconDoc {
                 lexicon: 1,
-                id: format!("com.example.schema{}", i),
+                id: format!("pet.nel.schema{}", i),
                 defs: HashMap::new(),
             };
             registry.insert_schema(doc);
@@ -226,23 +627,23 @@ mod tests {
 
         let trigger = LexiconDoc {
             lexicon: 1,
-            id: "com.example.trigger".to_string(),
+            id: "pet.nel.trigger".to_string(),
             defs: HashMap::new(),
         };
         registry.insert_schema(trigger);
 
         assert!(
-            registry.get("com.example.schema0").is_none(),
+            registry.get_cached("pet.nel.schema0").is_none(),
             "oldest entry should be evicted"
         );
         assert!(
-            registry.get("com.example.trigger").is_some(),
+            registry.get_cached("pet.nel.trigger").is_some(),
             "newly inserted entry should exist"
         );
         let evict_count = MAX_DYNAMIC_SCHEMAS / 4;
         assert!(
             registry
-                .get(&format!("com.example.schema{}", evict_count))
+                .get_cached(&format!("pet.nel.schema{}", evict_count))
                 .is_some(),
             "entry after eviction window should survive"
         );
@@ -253,7 +654,7 @@ mod tests {
         let registry = DynamicRegistry::new();
         let doc = LexiconDoc {
             lexicon: 1,
-            id: "com.example.tracked".to_string(),
+            id: "pet.nel.tracked".to_string(),
             defs: HashMap::new(),
         };
         let arc = registry.insert_schema(doc);
@@ -265,7 +666,7 @@ mod tests {
         (0..MAX_DYNAMIC_SCHEMAS).for_each(|i| {
             registry.insert_schema(LexiconDoc {
                 lexicon: 1,
-                id: format!("com.example.filler{}", i),
+                id: format!("pet.nel.filler{}", i),
                 defs: HashMap::new(),
             });
         });

crates/tranquil-lexicon/src/registry.rs

@@ -43,7 +43,7 @@ impl LexiconRegistry {
         self.schemas.get(nsid).cloned().or_else(|| {
             #[cfg(feature = "resolve")]
             {
-                self.dynamic.get(nsid)
+                self.dynamic.get_cached(nsid)
             }
             #[cfg(not(feature = "resolve"))]
             {

crates/tranquil-lexicon/src/resolve.rs

@@ -70,6 +70,8 @@ pub enum ResolveError {
     NegativelyCached { nsid: String, ttl_secs: u64 },
     #[error("network resolution disabled")]
     NetworkDisabled,
+    #[error("leader task for {nsid} aborted before completion")]
+    LeaderAborted { nsid: String },
 }
 
 pub fn nsid_to_authority(nsid: &str) -> Result<String, ResolveError> {

crates/tranquil-oauth-server/src/endpoints/authorize/passkey.rs

@@ -175,10 +175,7 @@ pub async fn passkey_start(
     }
 }
 
-async fn passkey_start_discoverable(
-    state: AppState,
-    request_id: RequestId,
-) -> Response {
+async fn passkey_start_discoverable(state: AppState, request_id: RequestId) -> Response {
     let (rcr, auth_state) = match state.webauthn_config.start_discoverable_authentication() {
         Ok(result) => result,
         Err(e) => {
@@ -570,16 +567,13 @@ pub async fn passkey_finish(
             Err(response) => return response,
         },
         None => {
-            let result = match passkey_finish_discoverable(
-                &state,
-                &credential,
-                &passkey_finish_request_id,
-            )
-            .await
-            {
-                Ok(result) => result,
-                Err(response) => return response,
-            };
+            let result =
+                match passkey_finish_discoverable(&state, &credential, &passkey_finish_request_id)
+                    .await
+                {
+                    Ok(result) => result,
+                    Err(response) => return response,
+                };
             if state
                 .repos
                 .oauth

crates/tranquil-pds/src/repo_ops.rs

@@ -246,9 +246,8 @@ pub async fn finalize_repo_write(
 
     let obsolete_cids = match original_settled.diff(&new_settled).await {
         Ok(diff) => {
-            let mut obsolete: Vec<Cid> = Vec::with_capacity(
-                1 + diff.removed_mst_blocks.len() + diff.removed_cids.len(),
-            );
+            let mut obsolete: Vec<Cid> =
+                Vec::with_capacity(1 + diff.removed_mst_blocks.len() + diff.removed_cids.len());
             obsolete.push(ctx.current_root_cid);
             obsolete.extend(diff.removed_mst_blocks);
             obsolete.extend(diff.removed_cids);

crates/tranquil-pds/tests/gc_compaction_restart.rs

@@ -13,12 +13,10 @@ fn run_compaction(store: &tranquil_store::blockstore::TranquilBlockStore) {
         .map(|(&fid, _)| fid)
         .collect::<Vec<_>>()
         .into_iter()
-        .for_each(|fid| {
-            match store.compact_file(fid, 0) {
-                Ok(_) => {}
-                Err(tranquil_store::blockstore::CompactionError::ActiveFileCannotBeCompacted) => {}
-                Err(e) => eprintln!("compaction: {e}"),
-            }
+        .for_each(|fid| match store.compact_file(fid, 0) {
+            Ok(_) => {}
+            Err(tranquil_store::blockstore::CompactionError::ActiveFileCannotBeCompacted) => {}
+            Err(e) => eprintln!("compaction: {e}"),
         });
 }
 
@@ -84,10 +82,7 @@ async fn mst_blocks_survive_full_store_reopen() {
     }
 
     let data_dir = store.data_dir().to_path_buf();
-    let index_dir = data_dir
-        .parent()
-        .unwrap()
-        .join("index");
+    let index_dir = data_dir.parent().unwrap().join("index");
 
     let store_clone = store.clone();
     tokio::task::spawn_blocking(move || {
@@ -107,10 +102,9 @@ async fn mst_blocks_survive_full_store_reopen() {
 
     let head_cid = cid::Cid::try_from(repo_root_str.as_str()).expect("invalid cid");
 
-    let car_blocks =
-        tranquil_pds::scheduled::collect_current_repo_blocks(block_store, &head_cid)
-            .await
-            .expect("collect blocks");
+    let car_blocks = tranquil_pds::scheduled::collect_current_repo_blocks(block_store, &head_cid)
+        .await
+        .expect("collect blocks");
 
     let block_count_before = car_blocks.len();
 
@@ -131,8 +125,8 @@ async fn mst_blocks_survive_full_store_reopen() {
             group_commit: tranquil_store::blockstore::GroupCommitConfig::default(),
             shard_count: 1,
         };
-        let fresh = tranquil_store::blockstore::TranquilBlockStore::open(config)
-            .expect("reopen failed");
+        let fresh =
+            tranquil_store::blockstore::TranquilBlockStore::open(config).expect("reopen failed");
 
         let missing: Vec<String> = car_blocks
             .iter()

crates/tranquil-pds/tests/lifecycle_session.rs

@@ -626,7 +626,12 @@ async fn create_app_password_session(
         .send()
         .await
         .expect("Failed to login with app password");
-    assert_eq!(login_res.status(), StatusCode::OK, "App password login for '{}' failed", name);
+    assert_eq!(
+        login_res.status(),
+        StatusCode::OK,
+        "App password login for '{}' failed",
+        name
+    );
     let session: Value = login_res.json().await.unwrap();
     let jwt = session["accessJwt"].as_str().unwrap().to_string();
     (jwt, scopes_response)
@@ -635,10 +640,7 @@ async fn create_app_password_session(
 async fn try_chat_service_auth(client: &reqwest::Client, jwt: &str) -> StatusCode {
     let base = base_url().await;
     let res = client
-        .get(format!(
-            "{}/xrpc/com.atproto.server.getServiceAuth",
-            base
-        ))
+        .get(format!("{}/xrpc/com.atproto.server.getServiceAuth", base))
         .bearer_auth(jwt)
         .query(&[
             ("aud", "did:web:api.bsky.app"),

crates/tranquil-pds/tests/mst_diff_equivalence.rs

@@ -15,7 +15,9 @@ fn test_cid(n: u32) -> Cid {
     Cid::new_v1(0x71, mh)
 }
 
-async fn compute_obsolete_full_walk<S: jacquard_repo::storage::BlockStore + Sync + Send + 'static>(
+async fn compute_obsolete_full_walk<
+    S: jacquard_repo::storage::BlockStore + Sync + Send + 'static,
+>(
     old: &Mst<S>,
     new: &Mst<S>,
 ) -> BTreeSet<Cid> {
@@ -34,9 +36,7 @@ async fn compute_obsolete_full_walk<S: jacquard_repo::storage::BlockStore + Sync
         .collect()
 }
 
-fn compute_obsolete_from_diff(
-    diff: &jacquard_repo::mst::diff::MstDiff,
-) -> BTreeSet<Cid> {
+fn compute_obsolete_from_diff(diff: &jacquard_repo::mst::diff::MstDiff) -> BTreeSet<Cid> {
     diff.removed_mst_blocks
         .iter()
         .copied()
@@ -74,12 +74,17 @@ async fn assert_equivalence(
     let diff_obsolete = compute_obsolete_from_diff(&diff);
 
     assert_eq!(
-        full_walk_obsolete, diff_obsolete,
+        full_walk_obsolete,
+        diff_obsolete,
         "MISMATCH in scenario: {scenario}\n  full_walk count: {}\n  diff count: {}\n  in full_walk but not diff: {:?}\n  in diff but not full_walk: {:?}",
         full_walk_obsolete.len(),
         diff_obsolete.len(),
-        full_walk_obsolete.difference(&diff_obsolete).collect::<Vec<_>>(),
-        diff_obsolete.difference(&full_walk_obsolete).collect::<Vec<_>>(),
+        full_walk_obsolete
+            .difference(&diff_obsolete)
+            .collect::<Vec<_>>(),
+        diff_obsolete
+            .difference(&full_walk_obsolete)
+            .collect::<Vec<_>>(),
     );
 }
 
@@ -256,7 +261,12 @@ async fn massive_to_empty() {
 async fn massive_complete_replacement() {
     let old = generate_records("app.bsky.feed.post", 0..1000);
     let new_rec = generate_records("app.bsky.feed.post", 1000..2000);
-    assert_equivalence(&old, &new_rec, "1000 records fully replaced with 1000 different").await;
+    assert_equivalence(
+        &old,
+        &new_rec,
+        "1000 records fully replaced with 1000 different",
+    )
+    .await;
 }
 
 #[tokio::test]
@@ -276,7 +286,12 @@ async fn multi_collection_5_collections_500_each() {
     ];
     let old = generate_multi_collection_records(&collections, 500);
     let new_rec = apply_scattered_updates(&old, 4, 30000);
-    assert_equivalence(&old, &new_rec, "5 collections x 500 records - update every 4th").await;
+    assert_equivalence(
+        &old,
+        &new_rec,
+        "5 collections x 500 records - update every 4th",
+    )
+    .await;
 }
 
 #[tokio::test]
@@ -294,7 +309,12 @@ async fn multi_collection_wipe_one_collection() {
         .filter(|(key, _)| !key.starts_with("app.bsky.feed.repost"))
         .cloned()
         .collect();
-    assert_equivalence(&old, &new_rec, "4 collections x 400 - wipe repost collection").await;
+    assert_equivalence(
+        &old,
+        &new_rec,
+        "4 collections x 400 - wipe repost collection",
+    )
+    .await;
 }
 
 #[tokio::test]
@@ -313,10 +333,7 @@ async fn multi_collection_keep_only_one() {
 
 #[tokio::test]
 async fn multi_collection_add_new_collection() {
-    let old_collections = [
-        "app.bsky.feed.like",
-        "app.bsky.feed.post",
-    ];
+    let old_collections = ["app.bsky.feed.like", "app.bsky.feed.post"];
     let old = generate_multi_collection_records(&old_collections, 500);
     let new_rec = append_records(&old, "app.bsky.graph.follow", 0..500, 40000);
     assert_equivalence(&old, &new_rec, "2 collections x 500 + add 500 follows").await;
@@ -378,7 +395,12 @@ async fn interleaved_keys_disjoint_ranges() {
     let new_rec: Vec<_> = (0..1000u32)
         .map(|i| (make_key("app.bsky.feed.post", i * 2 + 1), i + 10000))
         .collect();
-    assert_equivalence(&old, &new_rec, "1000 even-keyed records replaced by 1000 odd-keyed").await;
+    assert_equivalence(
+        &old,
+        &new_rec,
+        "1000 even-keyed records replaced by 1000 odd-keyed",
+    )
+    .await;
 }
 
 #[tokio::test]
@@ -426,7 +448,12 @@ async fn many_collections_few_records_each() {
         })
         .collect();
 
-    assert_equivalence(&old, &new_rec, "50 collections x 20 records - delete every 15th, update every 7th").await;
+    assert_equivalence(
+        &old,
+        &new_rec,
+        "50 collections x 20 records - delete every 15th, update every 7th",
+    )
+    .await;
 }
 
 #[tokio::test]
@@ -457,7 +484,12 @@ async fn one_to_massive() {
 async fn delete_head_and_tail() {
     let old = generate_records("app.bsky.feed.post", 0..2000);
     let new_rec: Vec<_> = old[200..1800].to_vec();
-    assert_equivalence(&old, &new_rec, "2000 records - delete first 200 and last 200").await;
+    assert_equivalence(
+        &old,
+        &new_rec,
+        "2000 records - delete first 200 and last 200",
+    )
+    .await;
 }
 
 #[tokio::test]
@@ -465,7 +497,12 @@ async fn keep_head_and_tail_only() {
     let old = generate_records("app.bsky.feed.post", 0..2000);
     let mut new_rec: Vec<_> = old[..100].to_vec();
     new_rec.extend_from_slice(&old[1900..]);
-    assert_equivalence(&old, &new_rec, "2000 records - keep only first 100 and last 100").await;
+    assert_equivalence(
+        &old,
+        &new_rec,
+        "2000 records - keep only first 100 and last 100",
+    )
+    .await;
 }
 
 #[tokio::test]
@@ -515,7 +552,12 @@ async fn swiss_cheese_deletions() {
         })
         .map(|(_, r)| r.clone())
         .collect();
-    assert_equivalence(&old, &new_rec, "1500 records - delete every 3rd chunk of 50").await;
+    assert_equivalence(
+        &old,
+        &new_rec,
+        "1500 records - delete every 3rd chunk of 50",
+    )
+    .await;
 }
 
 #[tokio::test]
@@ -529,9 +571,7 @@ async fn mixed_ops_with_key_density_change() {
         .filter(|(_, val)| val % 4 != 0)
         .cloned()
         .collect();
-    new_rec.extend((0..500u32).map(|i| {
-        (make_key("app.bsky.feed.post", i * 3 + 1), i + 100000)
-    }));
+    new_rec.extend((0..500u32).map(|i| (make_key("app.bsky.feed.post", i * 3 + 1), i + 100000)));
     new_rec.sort_by(|(a, _), (b, _)| a.cmp(b));
 
     assert_equivalence(

crates/tranquil-scopes/src/permissions.rs

@@ -164,7 +164,10 @@ impl ScopePermissions {
             if self.has_transition_generic && !self.has_transition_chat {
                 return Err(ScopeError::InsufficientScope {
                     required: "transition:chat.bsky".to_string(),
-                    message: format!("Chat access requires transition:chat.bsky scope to call {}", lxm),
+                    message: format!(
+                        "Chat access requires transition:chat.bsky scope to call {}",
+                        lxm
+                    ),
                 });
             }
         }

crates/tranquil-store/Cargo.toml

@@ -14,7 +14,7 @@ parking_lot = { workspace = true }
 fjall = "3"
 lsm-tree = "3"
 flume = "0.11"
-tokio = { workspace = true, features = ["sync", "rt"] }
+tokio = { workspace = true, features = ["sync", "rt", "time"] }
 bytes = "1"
 memmap2 = "0.9"
 tracing = { workspace = true }
@@ -34,9 +34,10 @@ dashmap = "6"
 rayon = "1"
 smallvec = "1"
 uuid = { workspace = true }
+tempfile = { version = "3", optional = true }
 
 [features]
-test-harness = []
+test-harness = ["dep:tempfile"]
 
 [dev-dependencies]
 tranquil-store = { path = ".", features = ["test-harness"] }

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

@@ -1202,6 +1202,15 @@ impl BlockIndex {
         self.table.read().contains_live(cid)
     }
 
+    pub fn live_entries_snapshot(&self) -> Vec<([u8; CID_SIZE], RefCount)> {
+        self.table
+            .read()
+            .iter()
+            .filter(|s| !s.refcount.is_zero())
+            .map(|s| (s.cid, s.refcount))
+            .collect()
+    }
+
     pub fn batch_put(
         &self,
         entries: &[([u8; CID_SIZE], BlockLocation)],
@@ -1222,7 +1231,14 @@ impl BlockIndex {
         now: WallClockMs,
         position_update: PositionUpdate<'_>,
     ) -> Result<(), BlockIndexError> {
-        self.batch_put_inner(entries, decrements, cursor, epoch, now, Some(position_update))
+        self.batch_put_inner(
+            entries,
+            decrements,
+            cursor,
+            epoch,
+            now,
+            Some(position_update),
+        )
     }
 
     fn batch_put_inner(

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

@@ -57,10 +57,8 @@ fn write_hint_record<S: StorageIO>(
 
 fn encode_location_fields(record: &mut [u8; HINT_RECORD_SIZE], loc: &BlockLocation) {
     record[FIELD_A_OFFSET..FIELD_A_OFFSET + 4].copy_from_slice(&loc.file_id.raw().to_le_bytes());
-    record[FIELD_A_OFFSET + 4..FIELD_A_OFFSET + 8]
-        .copy_from_slice(&loc.length.raw().to_le_bytes());
-    record[FIELD_B_OFFSET..FIELD_B_OFFSET + 8]
-        .copy_from_slice(&loc.offset.raw().to_le_bytes());
+    record[FIELD_A_OFFSET + 4..FIELD_A_OFFSET + 8].copy_from_slice(&loc.length.raw().to_le_bytes());
+    record[FIELD_B_OFFSET..FIELD_B_OFFSET + 8].copy_from_slice(&loc.offset.raw().to_le_bytes());
 }
 
 pub(crate) fn encode_hint_record<S: StorageIO>(
@@ -841,7 +839,11 @@ mod tests {
         let offset = BlockOffset::new(1024);
         let length = BlockLength::new(256);
 
-        let loc = BlockLocation { file_id, offset, length };
+        let loc = BlockLocation {
+            file_id,
+            offset,
+            length,
+        };
         encode_hint_record(&sim, fd, HintOffset::new(0), &cid, &loc).unwrap();
 
         let file_size = sim.file_size(fd).unwrap();

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

@@ -552,7 +552,7 @@ fn decode_mmap_event(
             segment = %segment_id,
             offset = raw,
             file_size,
-            "decode offset past file size (corrupt index?)"
+            "decode offset past file size, index likely corrupt"
         );
         return Ok(MmapDecodeResult::Corrupted);
     }

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

@@ -305,7 +305,7 @@ impl<S: StorageIO> EventLogWriter<S> {
 
         match self.build_sidecar_for_segment(old_id) {
             Ok(()) => {}
-            Err(e) => warn!(segment = %old_id, error = %e, "sidecar build failed (non-fatal)"),
+            Err(e) => warn!(segment = %old_id, error = %e, "non-fatal sidecar build failure"),
         }
 
         let (new_id, new_fd) = self.manager.prepare_rotation(old_id)?;

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

@@ -0,0 +1,41 @@
+use std::cell::RefCell;
+
+use rayon::prelude::*;
+use tokio::runtime::Runtime;
+
+use super::op::Seed;
+use super::runner::{Gauntlet, GauntletConfig, GauntletReport};
+
+thread_local! {
+    static RUNTIME: RefCell<Option<Runtime>> = const { RefCell::new(None) };
+}
+
+fn with_runtime<R>(f: impl FnOnce(&Runtime) -> R) -> R {
+    RUNTIME.with(|cell| {
+        let mut slot = cell.borrow_mut();
+        if slot.is_none() {
+            *slot = Some(
+                tokio::runtime::Builder::new_current_thread()
+                    .enable_all()
+                    .build()
+                    .expect("build rt"),
+            );
+        }
+        f(slot.as_ref().expect("runtime present"))
+    })
+}
+
+pub fn run_many<F>(make_config: F, seeds: impl IntoIterator<Item = Seed>) -> Vec<GauntletReport>
+where
+    F: Fn(Seed) -> GauntletConfig + Sync + Send,
+{
+    let seeds: Vec<Seed> = seeds.into_iter().collect();
+    seeds
+        .into_par_iter()
+        .map(|s| {
+            let cfg = make_config(s);
+            let gauntlet = Gauntlet::new(cfg).expect("build gauntlet");
+            with_runtime(|rt| rt.block_on(gauntlet.run()))
+        })
+        .collect()
+}

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

@@ -0,0 +1,271 @@
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+
+use async_trait::async_trait;
+use cid::Cid;
+use jacquard_repo::mst::Mst;
+
+use super::oracle::{Oracle, hex_short, try_cid_to_fixed};
+use crate::blockstore::{CidBytes, TranquilBlockStore};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct InvariantSet(u32);
+
+impl InvariantSet {
+    pub const EMPTY: Self = Self(0);
+    pub const REFCOUNT_CONSERVATION: Self = Self(1 << 0);
+    pub const REACHABILITY: Self = Self(1 << 1);
+    pub const ACKED_WRITE_PERSISTENCE: Self = Self(1 << 2);
+    pub const READ_AFTER_WRITE: Self = Self(1 << 3);
+    pub const RESTART_IDEMPOTENT: Self = Self(1 << 4);
+
+    const ALL_KNOWN: u32 = Self::REFCOUNT_CONSERVATION.0
+        | Self::REACHABILITY.0
+        | Self::ACKED_WRITE_PERSISTENCE.0
+        | Self::READ_AFTER_WRITE.0
+        | Self::RESTART_IDEMPOTENT.0;
+
+    pub const fn contains(self, other: Self) -> bool {
+        (self.0 & other.0) == other.0
+    }
+
+    pub const fn union(self, other: Self) -> Self {
+        Self(self.0 | other.0)
+    }
+
+    pub const fn without(self, other: Self) -> Self {
+        Self(self.0 & !other.0)
+    }
+
+    pub const fn unknown_bits(self) -> u32 {
+        self.0 & !Self::ALL_KNOWN
+    }
+}
+
+impl std::ops::BitOr for InvariantSet {
+    type Output = Self;
+    fn bitor(self, rhs: Self) -> Self {
+        self.union(rhs)
+    }
+}
+
+#[derive(Debug)]
+pub struct InvariantViolation {
+    pub invariant: &'static str,
+    pub detail: String,
+}
+
+pub struct InvariantCtx<'a> {
+    pub store: &'a Arc<TranquilBlockStore>,
+    pub oracle: &'a Oracle,
+    pub root: Option<Cid>,
+}
+
+#[async_trait]
+pub trait Invariant: Send + Sync {
+    fn name(&self) -> &'static str;
+    async fn check(&self, ctx: &InvariantCtx<'_>) -> Result<(), InvariantViolation>;
+}
+
+pub struct RefcountConservation;
+
+#[async_trait]
+impl Invariant for RefcountConservation {
+    fn name(&self) -> &'static str {
+        "RefcountConservation"
+    }
+
+    async fn check(&self, ctx: &InvariantCtx<'_>) -> Result<(), InvariantViolation> {
+        let live: Vec<(String, CidBytes)> = ctx.oracle.live_cids_labeled();
+        let live_set: HashSet<CidBytes> = live.iter().map(|(_, c)| *c).collect();
+        let index: HashMap<CidBytes, u32> = ctx
+            .store
+            .block_index()
+            .live_entries_snapshot()
+            .into_iter()
+            .map(|(c, r)| (c, r.raw()))
+            .collect();
+
+        let forward: Vec<String> = live
+            .iter()
+            .filter_map(|(label, cid)| match index.get(cid) {
+                Some(&r) if r >= 1 => None,
+                Some(&r) => Some(format!("{label}: refcount {r}")),
+                None => Some(format!("{label}: missing from index")),
+            })
+            .collect();
+
+        let inverse: Vec<String> = index
+            .iter()
+            .filter(|(cid, _)| !live_set.contains(*cid))
+            .map(|(cid, r)| format!("orphan cid {} refcount {}", hex_short(cid), r))
+            .collect();
+
+        let violations: Vec<String> = forward.into_iter().chain(inverse).collect();
+        if violations.is_empty() {
+            Ok(())
+        } else {
+            Err(InvariantViolation {
+                invariant: "RefcountConservation",
+                detail: violations.join("; "),
+            })
+        }
+    }
+}
+
+pub struct Reachability;
+
+#[async_trait]
+impl Invariant for Reachability {
+    fn name(&self) -> &'static str {
+        "Reachability"
+    }
+
+    async fn check(&self, ctx: &InvariantCtx<'_>) -> Result<(), InvariantViolation> {
+        let violations: Vec<String> = ctx
+            .oracle
+            .live_cids_labeled()
+            .into_iter()
+            .filter_map(|(label, fixed)| match ctx.store.get_block_sync(&fixed) {
+                Ok(Some(_)) => None,
+                Ok(None) => Some(format!("{label}: missing")),
+                Err(e) => Some(format!("{label}: read error {e}")),
+            })
+            .collect();
+
+        if violations.is_empty() {
+            Ok(())
+        } else {
+            Err(InvariantViolation {
+                invariant: "Reachability",
+                detail: violations.join("; "),
+            })
+        }
+    }
+}
+
+pub struct AckedWritePersistence;
+
+#[async_trait]
+impl Invariant for AckedWritePersistence {
+    fn name(&self) -> &'static str {
+        "AckedWritePersistence"
+    }
+
+    async fn check(&self, ctx: &InvariantCtx<'_>) -> Result<(), InvariantViolation> {
+        let Some(root) = ctx.root else {
+            if ctx.oracle.live_count() == 0 {
+                return Ok(());
+            }
+            return Err(InvariantViolation {
+                invariant: "AckedWritePersistence",
+                detail: format!(
+                    "oracle has {} live records but reopened store has no root",
+                    ctx.oracle.live_count()
+                ),
+            });
+        };
+        let mst = Mst::load(ctx.store.clone(), root, None);
+        let keys: Vec<String> = ctx
+            .oracle
+            .live_records()
+            .map(|(c, r, _)| format!("{}/{}", c.0, r.0))
+            .collect();
+
+        let mut missing: Vec<String> = Vec::new();
+        for key in &keys {
+            match mst.get(key).await {
+                Ok(Some(_)) => {}
+                Ok(None) => missing.push(format!("{key}: missing after reopen")),
+                Err(e) => missing.push(format!("{key}: mst.get error after reopen: {e}")),
+            }
+        }
+
+        if missing.is_empty() {
+            Ok(())
+        } else {
+            Err(InvariantViolation {
+                invariant: "AckedWritePersistence",
+                detail: missing.join("; "),
+            })
+        }
+    }
+}
+
+pub struct ReadAfterWrite;
+
+#[async_trait]
+impl Invariant for ReadAfterWrite {
+    fn name(&self) -> &'static str {
+        "ReadAfterWrite"
+    }
+
+    async fn check(&self, ctx: &InvariantCtx<'_>) -> Result<(), InvariantViolation> {
+        let Some(root) = ctx.root else {
+            return Ok(());
+        };
+        let mst = Mst::load(ctx.store.clone(), root, None);
+
+        let entries: Vec<(String, CidBytes)> = ctx
+            .oracle
+            .live_records()
+            .map(|(c, r, v)| (format!("{}/{}", c.0, r.0), *v))
+            .collect();
+
+        let mut violations: Vec<String> = Vec::new();
+        for (key, expected) in &entries {
+            match mst.get(key).await {
+                Ok(Some(cid)) => match try_cid_to_fixed(&cid) {
+                    Ok(actual) if actual == *expected => match ctx.store.get_block_sync(&actual) {
+                        Ok(Some(_)) => {}
+                        Ok(None) => violations.push(format!("{key}: block missing for cid")),
+                        Err(e) => violations.push(format!("{key}: block read error {e}")),
+                    },
+                    Ok(actual) => violations.push(format!(
+                        "{key}: MST cid {} != oracle cid {}",
+                        hex_short(&actual),
+                        hex_short(expected),
+                    )),
+                    Err(e) => {
+                        violations.push(format!("{key}: unexpected CID format from MST: {e}"))
+                    }
+                },
+                Ok(None) => violations.push(format!("{key}: MST returned None")),
+                Err(e) => violations.push(format!("{key}: mst.get error {e}")),
+            }
+        }
+
+        if violations.is_empty() {
+            Ok(())
+        } else {
+            Err(InvariantViolation {
+                invariant: "ReadAfterWrite",
+                detail: violations.join("; "),
+            })
+        }
+    }
+}
+
+pub fn invariants_for(set: InvariantSet) -> Vec<Box<dyn Invariant>> {
+    let unknown = set.unknown_bits();
+    assert!(
+        unknown == 0,
+        "invariants_for: unknown InvariantSet bits 0x{unknown:x}; all bits must map to an impl"
+    );
+    let candidates: Vec<(InvariantSet, Box<dyn Invariant>)> = vec![
+        (
+            InvariantSet::REFCOUNT_CONSERVATION,
+            Box::new(RefcountConservation),
+        ),
+        (InvariantSet::REACHABILITY, Box::new(Reachability)),
+        (
+            InvariantSet::ACKED_WRITE_PERSISTENCE,
+            Box::new(AckedWritePersistence),
+        ),
+        (InvariantSet::READ_AFTER_WRITE, Box::new(ReadAfterWrite)),
+    ];
+    candidates
+        .into_iter()
+        .filter_map(|(flag, inv)| set.contains(flag).then_some(inv))
+        .collect()
+}

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

@@ -0,0 +1,20 @@
+pub mod farm;
+pub mod invariants;
+pub mod op;
+pub mod oracle;
+pub mod runner;
+pub mod scenarios;
+pub mod workload;
+
+pub use invariants::{Invariant, InvariantSet, InvariantViolation, invariants_for};
+pub use op::{CollectionName, Op, OpStream, RecordKey, Seed, ValueSeed};
+pub use oracle::Oracle;
+pub use runner::{
+    Gauntlet, GauntletBuildError, GauntletConfig, GauntletReport, IoBackend, MaxFileSize, OpIndex,
+    OpInterval, OpsExecuted, RestartCount, RestartPolicy, RunLimits, ShardCount, StoreConfig,
+    WallMs,
+};
+pub use scenarios::{Scenario, config_for};
+pub use workload::{
+    ByteRange, KeySpaceSize, OpCount, OpWeights, SizeDistribution, ValueBytes, WorkloadModel,
+};

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

@@ -0,0 +1,60 @@
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct Seed(pub u64);
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct CollectionName(pub String);
+
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct RecordKey(pub String);
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct ValueSeed(pub u32);
+
+#[derive(Debug, Clone)]
+pub enum Op {
+    AddRecord {
+        collection: CollectionName,
+        rkey: RecordKey,
+        value_seed: ValueSeed,
+    },
+    DeleteRecord {
+        collection: CollectionName,
+        rkey: RecordKey,
+    },
+    Compact,
+    Checkpoint,
+}
+
+#[derive(Debug, Clone)]
+pub struct OpStream {
+    ops: Vec<Op>,
+}
+
+impl OpStream {
+    pub fn from_vec(ops: Vec<Op>) -> Self {
+        Self { ops }
+    }
+
+    pub fn into_vec(self) -> Vec<Op> {
+        self.ops
+    }
+
+    pub fn iter(&self) -> impl Iterator<Item = &Op> {
+        self.ops.iter()
+    }
+
+    pub fn len(&self) -> usize {
+        self.ops.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.ops.is_empty()
+    }
+
+    pub fn shrink(&self) -> Option<OpStream> {
+        (self.ops.len() >= 2).then(|| {
+            let half = self.ops.len() / 2;
+            OpStream::from_vec(self.ops[..half].to_vec())
+        })
+    }
+}

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

@@ -0,0 +1,87 @@
+use std::collections::HashMap;
+
+use cid::Cid;
+
+use super::op::{CollectionName, RecordKey};
+use crate::blockstore::CidBytes;
+
+#[derive(Debug, thiserror::Error, PartialEq, Eq)]
+#[error("unexpected CID encoding: got {actual} bytes, expected 36 for sha256 CIDv1")]
+pub struct CidFormatError {
+    pub actual: usize,
+}
+
+#[derive(Debug, Default)]
+pub struct Oracle {
+    live: HashMap<(CollectionName, RecordKey), CidBytes>,
+    current_root: Option<Cid>,
+    mst_node_cids: Vec<CidBytes>,
+}
+
+impl Oracle {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub fn add(
+        &mut self,
+        coll: CollectionName,
+        rkey: RecordKey,
+        record_cid: CidBytes,
+    ) -> Option<CidBytes> {
+        self.live.insert((coll, rkey), record_cid)
+    }
+
+    pub fn delete(&mut self, coll: &CollectionName, rkey: &RecordKey) -> Option<CidBytes> {
+        self.live.remove(&(coll.clone(), rkey.clone()))
+    }
+
+    pub fn set_root(&mut self, root: Cid) {
+        self.current_root = Some(root);
+    }
+
+    pub fn root(&self) -> Option<Cid> {
+        self.current_root
+    }
+
+    pub fn set_mst_node_cids(&mut self, cids: Vec<CidBytes>) {
+        self.mst_node_cids = cids;
+    }
+
+    pub fn clear_mst_state(&mut self) {
+        self.current_root = None;
+        self.mst_node_cids.clear();
+    }
+
+    pub fn live_records(&self) -> impl Iterator<Item = (&CollectionName, &RecordKey, &CidBytes)> {
+        self.live.iter().map(|((c, r), v)| (c, r, v))
+    }
+
+    pub fn live_count(&self) -> usize {
+        self.live.len()
+    }
+
+    pub fn live_cids_labeled(&self) -> Vec<(String, CidBytes)> {
+        let nodes = self
+            .mst_node_cids
+            .iter()
+            .map(|bytes| (format!("mst {}", hex_short(bytes)), *bytes));
+        let records = self
+            .live_records()
+            .map(|(c, r, v)| (format!("record {}/{}", c.0, r.0), *v));
+        nodes.chain(records).collect()
+    }
+}
+
+pub(super) fn try_cid_to_fixed(cid: &Cid) -> Result<CidBytes, CidFormatError> {
+    let bytes = cid.to_bytes();
+    let actual = bytes.len();
+    bytes.try_into().map_err(|_| CidFormatError { actual })
+}
+
+pub(super) fn hex_short(cid: &CidBytes) -> String {
+    cid[cid.len() - 6..]
+        .iter()
+        .map(|b| format!("{b:02x}"))
+        .collect()
+}

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

@@ -0,0 +1,593 @@
+use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::Duration;
+
+use cid::Cid;
+use jacquard_repo::mst::Mst;
+use jacquard_repo::storage::BlockStore;
+
+use super::invariants::{InvariantCtx, InvariantSet, InvariantViolation, invariants_for};
+use super::op::{Op, OpStream, Seed, ValueSeed};
+use super::oracle::{CidFormatError, Oracle, hex_short, try_cid_to_fixed};
+use super::workload::{Lcg, OpCount, SizeDistribution, ValueBytes, WorkloadModel};
+use crate::blockstore::{
+    BlockStoreConfig, CidBytes, CompactionError, GroupCommitConfig, TranquilBlockStore,
+};
+
+#[derive(Debug, Clone, Copy)]
+pub enum IoBackend {
+    Real,
+    Simulated,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct OpInterval(pub usize);
+
+#[derive(Debug, Clone, Copy)]
+pub enum RestartPolicy {
+    Never,
+    EveryNOps(OpInterval),
+    PoissonByOps(OpInterval),
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct WallMs(pub u64);
+
+#[derive(Debug, Clone, Copy)]
+pub struct RunLimits {
+    pub max_wall_ms: Option<WallMs>,
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct MaxFileSize(pub u64);
+
+#[derive(Debug, Clone, Copy)]
+pub struct ShardCount(pub u8);
+
+#[derive(Debug, Clone)]
+pub struct StoreConfig {
+    pub max_file_size: MaxFileSize,
+    pub group_commit: GroupCommitConfig,
+    pub shard_count: ShardCount,
+}
+
+#[derive(Debug, Clone)]
+pub struct GauntletConfig {
+    pub seed: Seed,
+    pub io: IoBackend,
+    pub workload: WorkloadModel,
+    pub op_count: OpCount,
+    pub invariants: InvariantSet,
+    pub limits: RunLimits,
+    pub restart_policy: RestartPolicy,
+    pub store: StoreConfig,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct OpsExecuted(pub usize);
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct RestartCount(pub usize);
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct OpIndex(pub usize);
+
+#[derive(Debug)]
+pub struct GauntletReport {
+    pub seed: Seed,
+    pub ops_executed: OpsExecuted,
+    pub restarts: RestartCount,
+    pub violations: Vec<InvariantViolation>,
+}
+
+impl GauntletReport {
+    pub fn is_clean(&self) -> bool {
+        self.violations.is_empty()
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+enum OpError {
+    #[error("put record: {0}")]
+    PutRecord(String),
+    #[error("mst add: {0}")]
+    MstAdd(String),
+    #[error("mst delete: {0}")]
+    MstDelete(String),
+    #[error("mst persist: {0}")]
+    MstPersist(String),
+    #[error("mst diff: {0}")]
+    MstDiff(String),
+    #[error("apply commit: {0}")]
+    ApplyCommit(String),
+    #[error("compact_file: {0}")]
+    CompactFile(String),
+    #[error("join: {0}")]
+    Join(String),
+    #[error("cid format: {0}")]
+    CidFormat(#[from] CidFormatError),
+}
+
+pub struct Gauntlet {
+    config: GauntletConfig,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum GauntletBuildError {
+    #[error("IoBackend::Simulated not wired yet")]
+    UnsupportedIoBackend,
+}
+
+impl Gauntlet {
+    pub fn new(config: GauntletConfig) -> Result<Self, GauntletBuildError> {
+        match config.io {
+            IoBackend::Real => Ok(Self { config }),
+            IoBackend::Simulated => Err(GauntletBuildError::UnsupportedIoBackend),
+        }
+    }
+
+    pub async fn run(self) -> GauntletReport {
+        let deadline = self
+            .config
+            .limits
+            .max_wall_ms
+            .map(|WallMs(ms)| Duration::from_millis(ms));
+
+        let seed = self.config.seed;
+        let ops_counter = Arc::new(AtomicUsize::new(0));
+        let restarts_counter = Arc::new(AtomicUsize::new(0));
+        let fut = run_real_inner(self.config, ops_counter.clone(), restarts_counter.clone());
+        match deadline {
+            Some(d) => match tokio::time::timeout(d, fut).await {
+                Ok(r) => r,
+                Err(_) => GauntletReport {
+                    seed,
+                    ops_executed: OpsExecuted(ops_counter.load(Ordering::Relaxed)),
+                    restarts: RestartCount(restarts_counter.load(Ordering::Relaxed)),
+                    violations: vec![InvariantViolation {
+                        invariant: "WallClockBudget",
+                        detail: format!("exceeded max_wall_ms of {} ms", d.as_millis()),
+                    }],
+                },
+            },
+            None => fut.await,
+        }
+    }
+}
+
+async fn run_real_inner(
+    config: GauntletConfig,
+    ops_counter: Arc<AtomicUsize>,
+    restarts_counter: Arc<AtomicUsize>,
+) -> GauntletReport {
+    let dir = tempfile::TempDir::new().expect("tempdir");
+    let op_stream: OpStream = config.workload.generate(config.seed, config.op_count);
+
+    let mut oracle = Oracle::new();
+    let mut violations: Vec<InvariantViolation> = Vec::new();
+
+    let mut store = Arc::new(
+        TranquilBlockStore::open(blockstore_config(dir.path(), &config.store)).expect("open store"),
+    );
+    let mut root: Option<Cid> = None;
+    let mut restart_rng = Lcg::new(Seed(config.seed.0 ^ 0xA5A5_A5A5_A5A5_A5A5));
+    let mut halt_ops = false;
+
+    let mid_run_set = config
+        .invariants
+        .without(InvariantSet::RESTART_IDEMPOTENT)
+        .without(InvariantSet::ACKED_WRITE_PERSISTENCE);
+    let post_reopen_set = config.invariants.without(InvariantSet::RESTART_IDEMPOTENT);
+
+    for (idx, op) in op_stream.iter().enumerate() {
+        if halt_ops {
+            break;
+        }
+        match apply_op(&store, &mut root, &mut oracle, op, &config.workload).await {
+            Ok(()) => {}
+            Err(e) => {
+                violations.push(InvariantViolation {
+                    invariant: "OpExecution",
+                    detail: format!("op {idx}: {e}"),
+                });
+                halt_ops = true;
+                continue;
+            }
+        }
+        ops_counter.store(idx + 1, Ordering::Relaxed);
+
+        if should_restart(config.restart_policy, OpIndex(idx), &mut restart_rng) {
+            drop(store);
+            store = Arc::new(
+                TranquilBlockStore::open(blockstore_config(dir.path(), &config.store))
+                    .expect("reopen store"),
+            );
+            let n = restarts_counter.fetch_add(1, Ordering::Relaxed) + 1;
+
+            if let Err(e) = refresh_oracle_graph(&store, &mut oracle, root).await {
+                violations.push(InvariantViolation {
+                    invariant: "OpExecution",
+                    detail: format!("refresh after restart {n}: {e}"),
+                });
+                halt_ops = true;
+                continue;
+            }
+            let before = violations.len();
+            violations.extend(run_invariants(&store, &oracle, root, mid_run_set).await);
+            if violations.len() > before {
+                halt_ops = true;
+            }
+        }
+    }
+
+    if !halt_ops {
+        match refresh_oracle_graph(&store, &mut oracle, root).await {
+            Ok(()) => {
+                let before = violations.len();
+                violations.extend(run_invariants(&store, &oracle, root, mid_run_set).await);
+                if violations.len() > before {
+                    halt_ops = true;
+                }
+            }
+            Err(e) => {
+                violations.push(InvariantViolation {
+                    invariant: "OpExecution",
+                    detail: format!("refresh at end: {e}"),
+                });
+                halt_ops = true;
+            }
+        }
+    }
+
+    if config.invariants.contains(InvariantSet::RESTART_IDEMPOTENT) && !halt_ops {
+        let pre_snapshot = snapshot_block_index(&store);
+        drop(store);
+        let reopened = Arc::new(
+            TranquilBlockStore::open(blockstore_config(dir.path(), &config.store))
+                .expect("reopen for RestartIdempotent"),
+        );
+        let post_snapshot = snapshot_block_index(&reopened);
+        if let Some(detail) = diff_snapshots(&pre_snapshot, &post_snapshot) {
+            violations.push(InvariantViolation {
+                invariant: "RestartIdempotent",
+                detail,
+            });
+        } else {
+            violations.extend(run_invariants(&reopened, &oracle, root, post_reopen_set).await);
+        }
+    }
+
+    GauntletReport {
+        seed: config.seed,
+        ops_executed: OpsExecuted(ops_counter.load(Ordering::Relaxed)),
+        restarts: RestartCount(restarts_counter.load(Ordering::Relaxed)),
+        violations,
+    }
+}
+
+async fn run_invariants(
+    store: &Arc<TranquilBlockStore>,
+    oracle: &Oracle,
+    root: Option<Cid>,
+    set: InvariantSet,
+) -> Vec<InvariantViolation> {
+    let ctx = InvariantCtx {
+        store,
+        oracle,
+        root,
+    };
+    let mut out = Vec::new();
+    for inv in invariants_for(set) {
+        if let Err(v) = inv.check(&ctx).await {
+            out.push(v);
+        }
+    }
+    out
+}
+
+fn snapshot_block_index(store: &TranquilBlockStore) -> Vec<(CidBytes, u32)> {
+    let mut v: Vec<(CidBytes, u32)> = store
+        .block_index()
+        .live_entries_snapshot()
+        .into_iter()
+        .map(|(c, r)| (c, r.raw()))
+        .collect();
+    v.sort_unstable_by(|a, b| a.0.cmp(&b.0));
+    v
+}
+
+const SNAPSHOT_DIFF_ITEMS: usize = 16;
+
+fn diff_snapshots(pre: &[(CidBytes, u32)], post: &[(CidBytes, u32)]) -> Option<String> {
+    if pre == post {
+        return None;
+    }
+    let pre_map: std::collections::HashMap<CidBytes, u32> = pre.iter().copied().collect();
+    let post_map: std::collections::HashMap<CidBytes, u32> = post.iter().copied().collect();
+
+    let only_pre: Vec<String> = pre_map
+        .iter()
+        .filter(|(c, _)| !post_map.contains_key(*c))
+        .map(|(c, r)| format!("lost {} refcount {}", hex_short(c), r))
+        .collect();
+    let only_post: Vec<String> = post_map
+        .iter()
+        .filter(|(c, _)| !pre_map.contains_key(*c))
+        .map(|(c, r)| format!("gained {} refcount {}", hex_short(c), r))
+        .collect();
+    let changed: Vec<String> = pre_map
+        .iter()
+        .filter_map(|(c, pre_r)| match post_map.get(c) {
+            Some(post_r) if post_r != pre_r => {
+                Some(format!("{} refcount {} -> {}", hex_short(c), pre_r, post_r))
+            }
+            _ => None,
+        })
+        .collect();
+
+    let total = only_pre.len() + only_post.len() + changed.len();
+    let mut items: Vec<String> = only_pre
+        .into_iter()
+        .chain(only_post)
+        .chain(changed)
+        .take(SNAPSHOT_DIFF_ITEMS)
+        .collect();
+    if total > items.len() {
+        items.push(format!("+{} more", total - items.len()));
+    }
+    Some(format!(
+        "block index changed across clean reopen: pre={} entries, post={} entries; {}",
+        pre.len(),
+        post.len(),
+        items.join("; "),
+    ))
+}
+
+async fn refresh_oracle_graph(
+    store: &Arc<TranquilBlockStore>,
+    oracle: &mut Oracle,
+    root: Option<Cid>,
+) -> Result<(), String> {
+    match root {
+        None => {
+            oracle.clear_mst_state();
+            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}"))?;
+            oracle.set_root(r);
+            oracle.set_mst_node_cids(fixed);
+            Ok(())
+        }
+    }
+}
+
+fn should_restart(policy: RestartPolicy, idx: OpIndex, rng: &mut Lcg) -> bool {
+    match policy {
+        RestartPolicy::Never => false,
+        RestartPolicy::EveryNOps(OpInterval(n)) => n > 0 && (idx.0 + 1).is_multiple_of(n),
+        RestartPolicy::PoissonByOps(OpInterval(n)) => {
+            if n == 0 {
+                false
+            } else {
+                rng.next_u64().is_multiple_of(n as u64)
+            }
+        }
+    }
+}
+
+fn blockstore_config(dir: &std::path::Path, s: &StoreConfig) -> BlockStoreConfig {
+    BlockStoreConfig {
+        data_dir: dir.join("data"),
+        index_dir: dir.join("index"),
+        max_file_size: s.max_file_size.0,
+        group_commit: s.group_commit.clone(),
+        shard_count: s.shard_count.0,
+    }
+}
+
+fn make_record_bytes(value_seed: ValueSeed, dist: SizeDistribution) -> Vec<u8> {
+    let raw = value_seed.0;
+    let target_len: usize = match dist {
+        SizeDistribution::Fixed(ValueBytes(n)) => n as usize,
+        SizeDistribution::Uniform(range) => {
+            let ValueBytes(lo) = range.min();
+            let ValueBytes(hi) = range.max();
+            let span = u64::from(hi.saturating_sub(lo)).max(1);
+            (lo as usize) + (u64::from(raw) % span) as usize
+        }
+    };
+    let target_len = target_len.max(8);
+    let seed_bytes = raw.to_le_bytes();
+    (0..target_len)
+        .map(|i| seed_bytes[i % 4] ^ (i as u8).wrapping_mul(31))
+        .collect()
+}
+
+async fn apply_op(
+    store: &Arc<TranquilBlockStore>,
+    root: &mut Option<Cid>,
+    oracle: &mut Oracle,
+    op: &Op,
+    workload: &WorkloadModel,
+) -> Result<(), OpError> {
+    match op {
+        Op::AddRecord {
+            collection,
+            rkey,
+            value_seed,
+        } => {
+            let record_bytes = make_record_bytes(*value_seed, workload.size_distribution);
+            let record_cid = store
+                .put(&record_bytes)
+                .await
+                .map_err(|e| OpError::PutRecord(e.to_string()))?;
+            let record_cid_bytes = try_cid_to_fixed(&record_cid)?;
+
+            let outcome =
+                add_record_inner(store, *root, collection, rkey, record_cid, record_cid_bytes)
+                    .await;
+            match outcome {
+                Ok((new_root, applied)) => {
+                    *root = Some(new_root);
+                    if applied {
+                        oracle.add(collection.clone(), rkey.clone(), record_cid_bytes);
+                    }
+                    Ok(())
+                }
+                Err(e) => {
+                    if let Err(cleanup_err) =
+                        decrement_obsolete(store, vec![record_cid_bytes]).await
+                    {
+                        tracing::warn!(
+                            op_error = %e,
+                            cleanup_error = %cleanup_err,
+                            "AddRecord cleanup decrement failed; refcount may leak",
+                        );
+                    }
+                    Err(e)
+                }
+            }
+        }
+        Op::DeleteRecord { collection, rkey } => {
+            let Some(old_root) = *root else { return Ok(()) };
+            if oracle.delete(collection, rkey).is_none() {
+                return Ok(());
+            }
+            let key = format!("{}/{}", collection.0, rkey.0);
+            let loaded = Mst::load(store.clone(), old_root, None);
+            let updated = loaded
+                .delete(&key)
+                .await
+                .map_err(|e| OpError::MstDelete(e.to_string()))?;
+            let new_root = updated
+                .persist()
+                .await
+                .map_err(|e| OpError::MstPersist(e.to_string()))?;
+            apply_mst_diff(store, old_root, new_root).await?;
+            *root = Some(new_root);
+            Ok(())
+        }
+        Op::Compact => {
+            let s = store.clone();
+            tokio::task::spawn_blocking(move || compact_by_liveness(&s))
+                .await
+                .map_err(|e| OpError::Join(e.to_string()))?
+        }
+        Op::Checkpoint => {
+            let s = store.clone();
+            tokio::task::spawn_blocking(move || {
+                s.apply_commit_blocking(vec![], vec![])
+                    .map_err(|e| e.to_string())
+            })
+            .await
+            .map_err(|e| OpError::Join(e.to_string()))?
+            .map_err(OpError::ApplyCommit)
+        }
+    }
+}
+
+async fn add_record_inner(
+    store: &Arc<TranquilBlockStore>,
+    root: Option<Cid>,
+    collection: &super::op::CollectionName,
+    rkey: &super::op::RecordKey,
+    record_cid: Cid,
+    record_cid_bytes: CidBytes,
+) -> Result<(Cid, bool), OpError> {
+    let key = format!("{}/{}", collection.0, rkey.0);
+    let loaded = match root {
+        None => Mst::new(store.clone()),
+        Some(r) => Mst::load(store.clone(), r, None),
+    };
+    let updated = loaded
+        .add(&key, record_cid)
+        .await
+        .map_err(|e| OpError::MstAdd(e.to_string()))?;
+    let new_root = updated
+        .persist()
+        .await
+        .map_err(|e| OpError::MstPersist(e.to_string()))?;
+
+    match root {
+        Some(old_root) if old_root == new_root => {
+            decrement_obsolete(store, vec![record_cid_bytes]).await?;
+            Ok((new_root, false))
+        }
+        Some(old_root) => {
+            apply_mst_diff(store, old_root, new_root).await?;
+            Ok((new_root, true))
+        }
+        None => Ok((new_root, true)),
+    }
+}
+
+async fn decrement_obsolete(
+    store: &Arc<TranquilBlockStore>,
+    obsolete: Vec<CidBytes>,
+) -> Result<(), OpError> {
+    let s = store.clone();
+    tokio::task::spawn_blocking(move || {
+        s.apply_commit_blocking(vec![], obsolete)
+            .map_err(|e| e.to_string())
+    })
+    .await
+    .map_err(|e| OpError::Join(e.to_string()))?
+    .map_err(OpError::ApplyCommit)
+}
+
+async fn apply_mst_diff(
+    store: &Arc<TranquilBlockStore>,
+    old_root: Cid,
+    new_root: Cid,
+) -> Result<(), OpError> {
+    let old_m = Mst::load(store.clone(), old_root, None);
+    let new_m = Mst::load(store.clone(), new_root, None);
+    let diff = old_m
+        .diff(&new_m)
+        .await
+        .map_err(|e| OpError::MstDiff(e.to_string()))?;
+    let obsolete: Vec<CidBytes> = diff
+        .removed_mst_blocks
+        .into_iter()
+        .chain(diff.removed_cids.into_iter())
+        .map(|c| try_cid_to_fixed(&c))
+        .collect::<Result<_, _>>()?;
+    let s = store.clone();
+    tokio::task::spawn_blocking(move || {
+        s.apply_commit_blocking(vec![], obsolete)
+            .map_err(|e| e.to_string())
+    })
+    .await
+    .map_err(|e| OpError::Join(e.to_string()))?
+    .map_err(OpError::ApplyCommit)
+}
+
+const COMPACT_LIVENESS_CEILING: f64 = 0.99;
+
+fn compact_by_liveness(store: &TranquilBlockStore) -> Result<(), OpError> {
+    let liveness = store
+        .compaction_liveness(0)
+        .map_err(|e| OpError::CompactFile(format!("compaction_liveness: {e}")))?;
+    let targets: Vec<_> = liveness
+        .iter()
+        .filter(|(_, info)| info.total_blocks > 0 && info.ratio() < COMPACT_LIVENESS_CEILING)
+        .map(|(&fid, _)| fid)
+        .collect();
+    targets
+        .into_iter()
+        .try_for_each(|fid| match store.compact_file(fid, 0) {
+            Ok(_) => Ok(()),
+            Err(CompactionError::ActiveFileCannotBeCompacted) => Ok(()),
+            Err(e) => Err(OpError::CompactFile(format!("{fid}: {e}"))),
+        })
+}

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

@@ -0,0 +1,166 @@
+use super::invariants::InvariantSet;
+use super::op::{CollectionName, Seed};
+use super::runner::{
+    GauntletConfig, IoBackend, MaxFileSize, OpInterval, RestartPolicy, RunLimits, ShardCount,
+    StoreConfig, WallMs,
+};
+use super::workload::{
+    KeySpaceSize, OpCount, OpWeights, SizeDistribution, ValueBytes, WorkloadModel,
+};
+use crate::blockstore::GroupCommitConfig;
+
+#[derive(Debug, Clone, Copy)]
+pub enum Scenario {
+    SmokePR,
+    MstChurn,
+    MstRestartChurn,
+    FullStackRestart,
+}
+
+pub fn config_for(scenario: Scenario, seed: Seed) -> GauntletConfig {
+    match scenario {
+        Scenario::SmokePR => smoke_pr(seed),
+        Scenario::MstChurn => mst_churn(seed),
+        Scenario::MstRestartChurn => mst_restart_churn(seed),
+        Scenario::FullStackRestart => full_stack_restart(seed),
+    }
+}
+
+fn default_collections() -> Vec<CollectionName> {
+    vec![
+        CollectionName("app.bsky.feed.post".to_string()),
+        CollectionName("app.bsky.feed.like".to_string()),
+    ]
+}
+
+fn tiny_store() -> StoreConfig {
+    StoreConfig {
+        max_file_size: MaxFileSize(4096),
+        group_commit: GroupCommitConfig {
+            checkpoint_interval_ms: 100,
+            checkpoint_write_threshold: 10,
+            ..GroupCommitConfig::default()
+        },
+        shard_count: ShardCount(1),
+    }
+}
+
+fn smoke_pr(seed: Seed) -> GauntletConfig {
+    GauntletConfig {
+        seed,
+        io: IoBackend::Real,
+        workload: WorkloadModel {
+            weights: OpWeights {
+                add: 80,
+                delete: 0,
+                compact: 10,
+                checkpoint: 10,
+            },
+            size_distribution: SizeDistribution::Fixed(ValueBytes(64)),
+            collections: default_collections(),
+            key_space: KeySpaceSize(200),
+        },
+        op_count: OpCount(10_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::EveryNOps(OpInterval(2_000)),
+        store: tiny_store(),
+    }
+}
+
+fn mst_churn(seed: Seed) -> GauntletConfig {
+    GauntletConfig {
+        seed,
+        io: IoBackend::Real,
+        workload: WorkloadModel {
+            weights: OpWeights {
+                add: 85,
+                delete: 0,
+                compact: 10,
+                checkpoint: 5,
+            },
+            size_distribution: SizeDistribution::Fixed(ValueBytes(64)),
+            collections: default_collections(),
+            key_space: KeySpaceSize(2_000),
+        },
+        op_count: OpCount(100_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(600_000)),
+        },
+        restart_policy: RestartPolicy::Never,
+        store: tiny_store(),
+    }
+}
+
+fn mst_restart_churn(seed: Seed) -> GauntletConfig {
+    GauntletConfig {
+        seed,
+        io: IoBackend::Real,
+        workload: WorkloadModel {
+            weights: OpWeights {
+                add: 85,
+                delete: 0,
+                compact: 10,
+                checkpoint: 5,
+            },
+            size_distribution: SizeDistribution::Fixed(ValueBytes(64)),
+            collections: default_collections(),
+            key_space: KeySpaceSize(2_000),
+        },
+        op_count: OpCount(100_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(600_000)),
+        },
+        restart_policy: RestartPolicy::PoissonByOps(OpInterval(5_000)),
+        store: tiny_store(),
+    }
+}
+
+fn full_stack_restart(seed: Seed) -> GauntletConfig {
+    GauntletConfig {
+        seed,
+        io: IoBackend::Real,
+        workload: WorkloadModel {
+            weights: OpWeights {
+                add: 80,
+                delete: 0,
+                compact: 15,
+                checkpoint: 5,
+            },
+            size_distribution: SizeDistribution::Fixed(ValueBytes(80)),
+            collections: default_collections(),
+            key_space: KeySpaceSize(500),
+        },
+        op_count: OpCount(5_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(120_000)),
+        },
+        restart_policy: RestartPolicy::EveryNOps(OpInterval(500)),
+        store: StoreConfig {
+            max_file_size: MaxFileSize(4096),
+            group_commit: GroupCommitConfig::default(),
+            shard_count: ShardCount(1),
+        },
+    }
+}

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

@@ -0,0 +1,133 @@
+use super::op::{CollectionName, Op, OpStream, RecordKey, Seed, ValueSeed};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct ValueBytes(pub u32);
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct KeySpaceSize(pub u32);
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub struct OpCount(pub usize);
+
+#[derive(Debug, Clone, Copy)]
+pub struct OpWeights {
+    pub add: u32,
+    pub delete: u32,
+    pub compact: u32,
+    pub checkpoint: u32,
+}
+
+impl OpWeights {
+    pub const fn total(&self) -> u32 {
+        self.add + self.delete + self.compact + self.checkpoint
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+pub struct ByteRange {
+    min: ValueBytes,
+    max: ValueBytes,
+}
+
+impl ByteRange {
+    pub fn new(min: ValueBytes, max: ValueBytes) -> Result<Self, String> {
+        if max.0 < min.0 {
+            Err(format!("ByteRange: max {} < min {}", max.0, min.0))
+        } else {
+            Ok(Self { min, max })
+        }
+    }
+
+    pub fn min(&self) -> ValueBytes {
+        self.min
+    }
+
+    pub fn max(&self) -> ValueBytes {
+        self.max
+    }
+}
+
+#[derive(Debug, Clone, Copy)]
+pub enum SizeDistribution {
+    Fixed(ValueBytes),
+    Uniform(ByteRange),
+}
+
+#[derive(Debug, Clone)]
+pub struct WorkloadModel {
+    pub weights: OpWeights,
+    pub size_distribution: SizeDistribution,
+    pub collections: Vec<CollectionName>,
+    pub key_space: KeySpaceSize,
+}
+
+impl WorkloadModel {
+    pub fn generate(&self, seed: Seed, op_count: OpCount) -> OpStream {
+        let mut rng = Lcg::new(seed);
+        let total = self.weights.total();
+        assert!(total > 0, "workload weights must sum to > 0");
+        assert!(
+            !self.collections.is_empty(),
+            "workload needs at least 1 collection"
+        );
+
+        let ops: Vec<Op> = (0..op_count.0)
+            .map(|_| {
+                let bucket = rng.next_u32() % total;
+                let coll = self.collections[rng.next_usize() % self.collections.len()].clone();
+                let rkey = RecordKey(format!("{:06}", rng.next_u32() % self.key_space.0.max(1)));
+
+                let (a, d, c) = (
+                    self.weights.add,
+                    self.weights.add + self.weights.delete,
+                    self.weights.add + self.weights.delete + self.weights.compact,
+                );
+                match bucket {
+                    b if b < a => Op::AddRecord {
+                        collection: coll,
+                        rkey,
+                        value_seed: ValueSeed(rng.next_u32()),
+                    },
+                    b if b < d => Op::DeleteRecord {
+                        collection: coll,
+                        rkey,
+                    },
+                    b if b < c => Op::Compact,
+                    _ => Op::Checkpoint,
+                }
+            })
+            .collect();
+        OpStream::from_vec(ops)
+    }
+}
+
+pub struct Lcg {
+    state: u64,
+}
+
+impl Lcg {
+    pub fn new(seed: Seed) -> Self {
+        Self {
+            state: seed
+                .0
+                .wrapping_mul(6364136223846793005)
+                .wrapping_add(1442695040888963407),
+        }
+    }
+
+    pub fn next_u64(&mut self) -> u64 {
+        self.state = self
+            .state
+            .wrapping_mul(6364136223846793005)
+            .wrapping_add(1442695040888963407);
+        self.state
+    }
+
+    pub fn next_u32(&mut self) -> u32 {
+        (self.next_u64() >> 16) as u32
+    }
+
+    pub fn next_usize(&mut self) -> usize {
+        self.next_u32() as usize
+    }
+}

crates/tranquil-store/src/lib.rs

@@ -6,6 +6,8 @@ pub mod consistency;
 pub mod eventlog;
 pub mod fsync_order;
 #[cfg(any(test, feature = "test-harness"))]
+pub mod gauntlet;
+#[cfg(any(test, feature = "test-harness"))]
 mod harness;
 mod io;
 pub mod metastore;

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

@@ -5964,7 +5964,7 @@ fn handler_loop<S: StorageIO + 'static>(
                         None => "unknown panic payload".to_owned(),
                     },
                 };
-                tracing::error!(thread_index, msg, "metastore handler panic (recovered)");
+                tracing::error!(thread_index, msg, "recovered metastore handler panic");
             }
         }
     });

crates/tranquil-store/tests/checkpoint_race.rs

@@ -4,10 +4,10 @@ use std::io;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 
+use tranquil_store::PostBlockstoreHook;
 use tranquil_store::blockstore::{
     BlockStoreConfig, BlocksSynced, CidBytes, GroupCommitConfig, TranquilBlockStore,
 };
-use tranquil_store::PostBlockstoreHook;
 
 struct SlowHook;
 
@@ -72,9 +72,7 @@ fn write_phase(base: &std::path::Path, use_hook: bool) -> Vec<CidBytes> {
                     store
                         .put_blocks_blocking(vec![(cid, vec![shard; 60])])
                         .unwrap();
-                    store
-                        .apply_commit_blocking(vec![], vec![cid])
-                        .unwrap();
+                    store.apply_commit_blocking(vec![], vec![cid]).unwrap();
                     targets.push(cid);
                     seq += 1;
                     total_cycles.fetch_add(1, Ordering::Relaxed);
@@ -208,7 +206,9 @@ fn __crash_write_phase() {
         Ok(d) => d,
         Err(_) => return,
     };
-    let use_hook = std::env::var("CRASH_TEST_HOOK").map(|v| v == "1").unwrap_or(false);
+    let use_hook = std::env::var("CRASH_TEST_HOOK")
+        .map(|v| v == "1")
+        .unwrap_or(false);
     let base = std::path::Path::new(&dir);
 
     let rt = tokio::runtime::Runtime::new().unwrap();

crates/tranquil-store/tests/compaction_liveness.rs

@@ -226,7 +226,7 @@ fn stress_create_delete_restart_cycle_matches_bug_report() {
                     live.insert(seed_a);
                     live.insert(seed_b);
 
-                    if rng.next_u32() % 2 == 0 {
+                    if rng.next_u32().is_multiple_of(2) {
                         let victim: Option<u32> = live.iter().copied().next();
                         if let Some(v) = victim {
                             store

crates/tranquil-store/tests/compaction_restart.rs

@@ -26,9 +26,14 @@ fn open_full_stack(base_dir: &Path) -> FullStack {
     let blockstore_data = base_dir.join("blockstore").join("data");
     let blockstore_index = base_dir.join("blockstore").join("index");
 
-    [&metastore_dir, &segments_dir, &blockstore_data, &blockstore_index]
-        .iter()
-        .for_each(|d| std::fs::create_dir_all(d).unwrap());
+    [
+        &metastore_dir,
+        &segments_dir,
+        &blockstore_data,
+        &blockstore_index,
+    ]
+    .iter()
+    .for_each(|d| std::fs::create_dir_all(d).unwrap());
 
     let metastore = Metastore::open(&metastore_dir, MetastoreConfig::default()).unwrap();
 
@@ -59,9 +64,7 @@ fn open_full_stack(base_dir: &Path) -> FullStack {
 
     let indexes = metastore.partition(Partition::Indexes).clone();
     let event_ops = metastore.event_ops(Arc::clone(&bridge));
-    let recovered = event_ops
-        .recover_metastore_mutations(&indexes)
-        .unwrap();
+    let recovered = event_ops.recover_metastore_mutations(&indexes).unwrap();
     if recovered > 0 {
         eprintln!("replayed {recovered} metastore mutations from eventlog");
     }
@@ -240,9 +243,7 @@ fn commit_style_decrements() {
 
                 if round > 0 {
                     let prev_mst = test_cid(6000 + round - 1);
-                    store
-                        .apply_commit_blocking(vec![], vec![prev_mst])
-                        .unwrap();
+                    store.apply_commit_blocking(vec![], vec![prev_mst]).unwrap();
                 }
 
                 prev_commit = new_commit;
@@ -408,8 +409,7 @@ fn multiple_restart_cycles_blockstore() {
 
         (0..10u32).for_each(|cycle| {
             {
-                let store =
-                    TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap();
+                let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap();
 
                 (0..50u32).for_each(|round| {
                     let churn = test_cid(9000 + cycle * 100 + round);

crates/tranquil-store/tests/eventlog_properties.rs

@@ -579,7 +579,7 @@ fn index_checkpoint_accelerates_recovery() {
 
     assert!(
         reads_with_index < reads_without_index,
-        "read with index ({reads_with_index} reads) should require fewer reads than without ({reads_without_index} reads)"
+        "indexed read took {reads_with_index} reads but unindexed took only {reads_without_index}"
     );
 }
 
@@ -654,7 +654,7 @@ fn fsync_ordering_blocks_before_events() {
     assert_eq!(
         event_writer.synced_seq(),
         EventSequence::BEFORE_ALL,
-        "crash between blockstore sync and eventlog sync must not persist the event (blocks exist, event does not = orphan, not inconsistency)"
+        "crash between blockstore sync and eventlog sync must leave blocks orphaned rather than persist the event"
     );
 
     drop(event_writer);

crates/tranquil-store/tests/eventlog_retention.rs

@@ -72,7 +72,7 @@ fn run_retention_at_deletes_sealed_segments_past_cutoff() {
     );
     assert!(
         segments_after >= 1,
-        "active segment must remain (got {segments_after})"
+        "active segment must remain, got {segments_after}"
     );
 }
 

crates/tranquil-store/tests/gauntlet_smoke.rs

@@ -0,0 +1,132 @@
+use tranquil_store::blockstore::GroupCommitConfig;
+use tranquil_store::gauntlet::{
+    CollectionName, Gauntlet, GauntletConfig, InvariantSet, IoBackend, KeySpaceSize, MaxFileSize,
+    OpCount, OpInterval, OpWeights, RestartPolicy, RunLimits, Scenario, Seed, ShardCount,
+    SizeDistribution, StoreConfig, ValueBytes, WallMs, WorkloadModel, config_for, farm,
+};
+
+#[test]
+#[ignore = "long running, 30 seeds of 10k ops each"]
+fn smoke_pr_30_seeds() {
+    let reports = farm::run_many(
+        |seed| config_for(Scenario::SmokePR, seed),
+        (0..30).map(Seed),
+    );
+    let failures: Vec<String> = reports
+        .iter()
+        .filter(|r| !r.is_clean())
+        .map(|r| {
+            format!(
+                "seed {}: {} violations\n  {}",
+                r.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"));
+}
+
+fn fast_sanity_config(seed: Seed) -> GauntletConfig {
+    GauntletConfig {
+        seed,
+        io: IoBackend::Real,
+        workload: WorkloadModel {
+            weights: OpWeights {
+                add: 80,
+                delete: 0,
+                compact: 10,
+                checkpoint: 10,
+            },
+            size_distribution: SizeDistribution::Fixed(ValueBytes(64)),
+            collections: vec![CollectionName("app.bsky.feed.post".to_string())],
+            key_space: KeySpaceSize(100),
+        },
+        op_count: OpCount(200),
+        invariants: InvariantSet::REFCOUNT_CONSERVATION
+            | InvariantSet::REACHABILITY
+            | InvariantSet::ACKED_WRITE_PERSISTENCE
+            | InvariantSet::READ_AFTER_WRITE
+            | InvariantSet::RESTART_IDEMPOTENT,
+        limits: RunLimits {
+            max_wall_ms: Some(WallMs(30_000)),
+        },
+        restart_policy: RestartPolicy::EveryNOps(OpInterval(80)),
+        store: StoreConfig {
+            max_file_size: MaxFileSize(512),
+            group_commit: GroupCommitConfig {
+                checkpoint_interval_ms: 50,
+                checkpoint_write_threshold: 8,
+                ..GroupCommitConfig::default()
+            },
+            shard_count: ShardCount(1),
+        },
+    }
+}
+
+#[tokio::test]
+async fn gauntlet_fast_sanity() {
+    let report = Gauntlet::new(fast_sanity_config(Seed(7)))
+        .expect("build gauntlet")
+        .run()
+        .await;
+    assert!(
+        report.is_clean(),
+        "violations: {:?}",
+        report
+            .violations
+            .iter()
+            .map(|v| format!("{}: {}", v.invariant, v.detail))
+            .collect::<Vec<_>>()
+    );
+    assert!(
+        report.restarts.0 >= 2,
+        "expected at least 2 restarts, got {}",
+        report.restarts.0
+    );
+    assert_eq!(report.ops_executed.0, 200);
+}
+
+#[tokio::test]
+async fn full_stack_restart_port() {
+    let cfg = config_for(Scenario::FullStackRestart, Seed(1));
+    let report = Gauntlet::new(cfg).expect("build gauntlet").run().await;
+    assert!(
+        report.is_clean(),
+        "violations: {:?}",
+        report
+            .violations
+            .iter()
+            .map(|v| format!("{}: {}", v.invariant, v.detail))
+            .collect::<Vec<_>>()
+    );
+    assert_eq!(
+        report.restarts.0, 10,
+        "FullStackRestart with EveryNOps(500) over 5000 ops must restart exactly 10 times",
+    );
+}
+
+#[tokio::test]
+#[ignore = "long running, 100k ops with around 20 restarts"]
+async fn mst_restart_churn_single_seed() {
+    let cfg = config_for(Scenario::MstRestartChurn, Seed(42));
+    let report = Gauntlet::new(cfg).expect("build gauntlet").run().await;
+    assert!(
+        report.is_clean(),
+        "violations: {:?}",
+        report
+            .violations
+            .iter()
+            .map(|v| format!("{}: {}", v.invariant, v.detail))
+            .collect::<Vec<_>>()
+    );
+    assert!(
+        report.restarts.0 >= 1,
+        "PoissonByOps(5000) over 100k ops should fire at least 1 restart, got {}",
+        report.restarts.0
+    );
+}

crates/tranquil-store/tests/gc.rs

@@ -160,7 +160,7 @@ fn collect_dead_blocks_respects_epoch_gating() {
             .collect();
         assert!(
             all_cids.contains(&cid_a),
-            "cid_a should be collectible (epoch advanced by subsequent commit)"
+            "cid_a should be collectible after subsequent commit advanced the epoch"
         );
     });
 }

crates/tranquil-store/tests/mst_refcount_integrity.rs

@@ -80,8 +80,7 @@ async fn mst_shared_subtrees_survive_incremental_writes_compaction_restart() {
 
                 let obsolete =
                     compute_obsolete_from_diff(&old_settled, &new_settled, prev_commit).await;
-                let obsolete_fixed: Vec<[u8; 36]> =
-                    obsolete.iter().map(|c| cid_to_fixed(c)).collect();
+                let obsolete_fixed: Vec<[u8; 36]> = obsolete.iter().map(cid_to_fixed).collect();
                 let s = store.clone();
                 tokio::task::spawn_blocking(move || {
                     s.apply_commit_blocking(vec![], obsolete_fixed).unwrap();
@@ -115,8 +114,7 @@ async fn mst_shared_subtrees_survive_incremental_writes_compaction_restart() {
 
             let obsolete =
                 compute_obsolete_from_diff(&old_settled, &new_settled, prev_commit).await;
-            let obsolete_fixed: Vec<[u8; 36]> =
-                obsolete.iter().map(|c| cid_to_fixed(c)).collect();
+            let obsolete_fixed: Vec<[u8; 36]> = obsolete.iter().map(cid_to_fixed).collect();
             let s = store.clone();
             tokio::task::spawn_blocking(move || {
                 s.apply_commit_blocking(vec![], obsolete_fixed).unwrap();
@@ -168,9 +166,7 @@ async fn mst_shared_subtrees_survive_incremental_writes_compaction_restart() {
     }
 
     {
-        let store = Arc::new(
-            TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap(),
-        );
+        let store = Arc::new(TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap());
 
         let missing: Vec<String> = final_node_cids
             .iter()

crates/tranquil-store/tests/sim_blockstore.rs

@@ -74,9 +74,7 @@ impl SimHarness {
                 let cid = test_cid(seed as u32);
                 let data = vec![seed as u8; data_size];
                 let loc = writer.append_block(&cid, &data).unwrap();
-                hint_writer
-                    .append_hint(&cid, &loc)
-                    .unwrap();
+                hint_writer.append_hint(&cid, &loc).unwrap();
                 (cid, loc)
             })
             .collect();
@@ -538,9 +536,7 @@ fn sim_aggressive_faults_data_integrity() {
                 let cid = test_cid(i as u32);
                 let data = vec![i as u8; 64];
                 let loc = writer.append_block(&cid, &data).ok()?;
-                hint_writer
-                    .append_hint(&cid, &loc)
-                    .ok()?;
+                hint_writer.append_hint(&cid, &loc).ok()?;
                 Some(())
             })?;
 

example.toml

@@ -26,12 +26,6 @@
 # Can also be specified via environment variable `PDS_USER_HANDLE_DOMAINS`.
 #user_handle_domains =
 
-# List of domains available for user registration.
-# Defaults to the PDS hostname when not set.
-#
-# Can also be specified via environment variable `AVAILABLE_USER_DOMAINS`.
-#available_user_domains =
-
 # Enable PDS-hosted did:web identities.  Hosting did:web requires a
 # long-term commitment to serve DID documents; opt-in only.
 #
@@ -95,6 +89,29 @@
 # Default value: 10737418240
 #max_blob_size = 10737418240
 
+# Maximum allowed number of preferences
+#
+# Can also be specified via environment variable `MAX_PREFERENCES_COUNT`.
+#
+# Default value: 1000
+#max_preferences_count = 1000
+
+[frontend]
+# Whether to enable the built in serving of the frontend.
+#
+# Can also be specified via environment variable `FRONTEND_ENABLED`.
+#
+# Default value: true
+#enabled = true
+
+# Directory to serve as the frontend. The oauth_client_metadata.json will have any references to
+# the frontend hostname replaced by the configured frontend hostname.
+#
+# Can also be specified via environment variable `FRONTEND_DIR`.
+#
+# Default value: "/var/lib/tranquil-pds/frontend"
+#dir = "/var/lib/tranquil-pds/frontend"
+
 [database]
 # PostgreSQL connection URL.
 #
@@ -210,8 +227,7 @@
 # once this budget is exhausted until in-flight events drain via fsync. Set to
 # 0 to disable backpressure. Default: 1 GiB.
 #
-# Can also be specified via environment variable
-# `TRANQUIL_STORE_EVENTLOG_PENDING_BYTES_BUDGET`.
+# Can also be specified via environment variable `TRANQUIL_STORE_EVENTLOG_PENDING_BYTES_BUDGET`.
 #
 # Default value: 1073741824
 #eventlog_pending_bytes_budget = 1073741824
@@ -219,12 +235,29 @@
 # Maximum size of an individual eventlog payload in bytes. Single events
 # larger than this are rejected at append time. Default: 256 MiB.
 #
-# Can also be specified via environment variable
-# `TRANQUIL_STORE_EVENTLOG_MAX_EVENT_PAYLOAD`.
+# Can also be specified via environment variable `TRANQUIL_STORE_EVENTLOG_MAX_EVENT_PAYLOAD`.
 #
 # Default value: 268435456
 #eventlog_max_event_payload = 268435456
 
+# Maximum size of an individual blockstore data file in bytes. When the
+# active data file reaches this size it is rolled over and becomes
+# eligible for compaction. Default: 256 MiB.
+#
+# Can also be specified via environment variable `TRANQUIL_STORE_MAX_BLOCKSTORE_FILE_SIZE`.
+#
+# Default value: 268435456
+#max_blockstore_file_size = 268435456
+
+# Maximum size of an individual eventlog segment file in bytes. When the
+# active segment reaches this size it is sealed and a new one is created.
+# Safe to change on a running instance. Default: 256 MiB.
+#
+# Can also be specified via environment variable `TRANQUIL_STORE_MAX_EVENTLOG_SEGMENT_SIZE`.
+#
+# Default value: 268435456
+#max_eventlog_segment_size = 268435456
+
 [cache]
 # Cache backend: `ripple` by default, or `valkey`.
 #
@@ -322,6 +355,13 @@
 # Default value: 5000
 #max_lag = 5000
 
+# Maximum concurrent full-repo exports, eg. getRepo without `since`.
+#
+# Can also be specified via environment variable `MAX_CONCURRENT_REPO_EXPORTS`.
+#
+# Default value: 4
+#max_concurrent_repo_exports = 4
+
 # List of relay / crawler notification URLs.
 #
 # Can also be specified via environment variable `CRAWLERS`.
@@ -369,6 +409,8 @@
 # Link a device via the admin API before enabling.
 #
 # Can also be specified via environment variable `SIGNAL_ENABLED`.
+#
+# Default value: false
 #enabled = false
 
 [notifications]
@@ -528,13 +570,58 @@
 # Default value: 3600
 #delete_check_interval_secs = 3600
 
+# Interval in seconds between data file compaction scans (tranquil-store only).
+# Set to 0 to disable.
+#
+# Can also be specified via environment variable `COMPACTION_INTERVAL_SECS`.
+#
+# Default value: 3600
+#compaction_interval_secs = 3600
+
+# Liveness ratio threshold below which a data file is compacted (0.0-1.0).
+#
+# Can also be specified via environment variable `COMPACTION_LIVENESS_THRESHOLD`.
+#
+# Default value: 0.7
+#compaction_liveness_threshold = 0.7
+
+# Grace period in milliseconds before a zero-refcount block can be removed by compaction.
+#
+# Can also be specified via environment variable `COMPACTION_GRACE_PERIOD_MS`.
+#
+# Default value: 600000
+#compaction_grace_period_ms = 600000
+
+# Interval in seconds between reachability walk runs (tranquil-store only).
+# Set to 0 to disable. Default: weekly.
+#
+# Can also be specified via environment variable `REACHABILITY_WALK_INTERVAL_SECS`.
+#
+# Default value: 604800
+#reachability_walk_interval_secs = 604800
+
+# Interval in seconds between continuous archival passes (tranquil-store only).
+# Sealed eventlog segments are copied to the archival destination each tick.
+# Set to 0 to disable. Default: 60 seconds.
+#
+# Can also be specified via environment variable `ARCHIVAL_INTERVAL_SECS`.
+#
+# Default value: 60
+#archival_interval_secs = 60
+
+# Archival destination directory for sealed eventlog segments.
+# If unset, archival is disabled.
+#
+# Can also be specified via environment variable `ARCHIVAL_DEST_DIR`.
+#archival_dest_dir =
+
 # Maximum age of events retained in the eventlog before pruning.
 # Per the atproto firehose spec, the relay backfill window only needs
 # to cover "hours or days".
 #
 # Can also be specified via environment variable `EVENT_RETENTION_MAX_AGE_SECS`.
 #
-# Default value: 604800 (7 days)
+# Default value: 604800
 #event_retention_max_age_secs = 604800
 
 # Interval in seconds between event retention prune passes.

flake.nix

@@ -18,6 +18,7 @@
     {
       packages = forAllSystems (pkgs: {
         tranquil-pds = pkgs.callPackage ./default.nix { };
+        tranquil-pds-aarch64 = pkgs.pkgsCross.aarch64-multiplatform.callPackage ./default.nix { };
         tranquil-frontend = pkgs.callPackage ./frontend.nix { };
         default = self.packages.${pkgs.stdenv.hostPlatform.system}.tranquil-pds;
       });