Add bencher_replica: in-process SQLite replication to replace Litestream

Litestream 0.5.13 blocked all API writes for ~5.5 minutes in production
(2026-07-10): when it decides a full re-snapshot is needed, it copies the
entire database into a local LTX file while holding the SQLite write lock,
and no configuration reaches that code path. Its LTX compaction also churns
whole-database S3 transfers several times a day.

bencher_replica is an in-process replacement built around six invariants
(documented in src/lib.rs), the prime one being that the SQLite write lock
is only ever held for O(WAL-tail) work, never O(database):

- WAL parser with full salt and cumulative checksum-chain verification
- Local filesystem XOR S3-compatible storage behind one contract
- Step-driven sync engine; checkpoints are PASSIVE while the replicator
  itself holds BEGIN IMMEDIATE, closing the ship-vs-checkpoint race without
  ever needing RESTART or TRUNCATE checkpoints
- Generation-based snapshots via a single-step SQLite online backup into a
  scratch file (transactionally consistent under concurrent checkpoints),
  throttled zstd multipart upload, snapshot.json as the atomic commit marker
- Latest-only restore in the same startup handshake slot Litestream used,
  with chain pre-validation and checkpoint-consumption verification
- Restore-and-compare verification (default daily) and shadow mode: with
  both plus.litestream and plus.replica configured, Litestream keeps
  checkpoint ownership and restore precedence during the burn-in

Also included:
- Fix: standalone sweep connections (stats, credit grants) now disable
  wal_autocheckpoint when replication is configured; previously the credit
  sweep could checkpoint and restart the WAL behind Litestream's back
- plus.replica config (JsonReplication), otel Replica* counters, main.rs
  lifecycle wiring (restore precedence, fatal race arm, final ship inside
  the Fly kill budget), TestServer::new_with_replica, Dockerfile stubs
- JsonLitestream.metrics_port and [[metrics]] in the Fly configs so
  litestream_* Prometheus metrics are scraped during the shadow period

Testing: 275 crate tests (WAL fixtures cross-validated against SQLite
itself, a three-backend storage contract suite, 8 fault-injection
scenarios, 6 crash kill points, 8 seeded 200-op equivalence workloads,
ignored soak and live-S3 tiers) plus 4 server-level integration tests.
An adversarial multi-agent review confirmed 18 findings, all fixed with
regression tests, including a silent data-loss gap in resume (salt-match
resume now proves content against the replica tip, and the meta-verified
path requires salt1 continuity).

Author: epompeii

Cargo.lock

@@ -67,6 +67,7 @@ bencher_otel_provider = { path = "plus/bencher_otel_provider" }
 bencher_oci_storage = { path = "plus/bencher_oci_storage" }
 bencher_rate_limiter = { path = "plus/bencher_rate_limiter" }
 bencher_recaptcha = { path = "plus/bencher_recaptcha" }
+bencher_replica = { path = "plus/bencher_replica" }
 # plus - runner
 bencher_runner = { path = "plus/bencher_runner" }
 bencher_oci = { path = "plus/bencher_oci" }

Cargo.toml

@@ -61,6 +61,7 @@ RUN cargo init --lib bencher_otel
 RUN cargo init --lib bencher_otel_provider
 RUN cargo init --lib bencher_rate_limiter
 RUN cargo init --lib bencher_recaptcha
+RUN cargo init --lib bencher_replica
 RUN cargo init --lib bencher_rootfs
 RUN cargo init --lib bencher_runner
 RUN cargo init --lib bencher_init

docker/bench.Dockerfile

@@ -42,8 +42,12 @@ slog.workspace = true
 
 [dev-dependencies]
 bencher_api_tests.workspace = true
-bencher_json = { workspace = true, features = ["server", "schema"] }
+bencher_json = { workspace = true, features = ["server", "schema", "test-clock"] }
+bencher_replica = { workspace = true, features = ["plus", "testing"] }
+camino.workspace = true
 http.workspace = true
+rusqlite.workspace = true
+tempfile.workspace = true
 tokio = { workspace = true, features = ["rt-multi-thread", "macros"] }
 
 [lints]

lib/api_server/Cargo.toml

@@ -2,8 +2,16 @@
 #[cfg(test)]
 use bencher_api_tests as _;
 #[cfg(test)]
+use bencher_replica as _;
+#[cfg(test)]
+use camino as _;
+#[cfg(test)]
 use http as _;
 #[cfg(test)]
+use rusqlite as _;
+#[cfg(test)]
+use tempfile as _;
+#[cfg(test)]
 use tokio as _;
 
 mod backup;

lib/api_server/src/lib.rs

@@ -4,7 +4,7 @@ use bencher_json::{
 };
 use bencher_schema::{
     auth_conn,
-    context::{ApiContext, DbConnection},
+    context::{ApiContext, DbConnection, configure_standalone_connection},
     error::{forbidden_error, issue_error, not_found_error},
     model::{
         server::QueryServer,
@@ -49,8 +49,23 @@ pub async fn server_stats_get(
 
 async fn get_one_inner(log: &Logger, context: &ApiContext) -> Result<JsonServerStats, HttpError> {
     let query_server = QueryServer::get_server(auth_conn!(context))?;
-    let conn = DbConnection::establish(context.database.path.to_string_lossy().as_ref())
+    let mut conn = DbConnection::establish(context.database.path.to_string_lossy().as_ref())
         .map_err(not_found_error)?;
+    // Route this standalone connection through the shared configuration so it
+    // does not bypass the busy_timeout / autocheckpoint settings the pools and
+    // the writer use: under replication a stray write must never checkpoint.
+    configure_standalone_connection(
+        &mut conn,
+        context.database.busy_timeout,
+        context.database.replicated,
+    )
+    .map_err(|e| {
+        issue_error(
+            "Failed to configure server stats connection",
+            "Failed to configure the server stats database connection PRAGMAs",
+            e,
+        )
+    })?;
     query_server
         .get_stats(log.clone(), conn, context.is_bencher_cloud)
         .await

lib/api_server/src/stats.rs

@@ -0,0 +1,215 @@
+#![cfg(feature = "plus")]
+#![expect(unused_crate_dependencies, reason = "integration test file")]
+//! Integration tests for in-process replication (`bencher_replica`) against
+//! a full API server: real Dropshot server, real writes through the API,
+//! step-driven replication engine, restore-and-compare verification.
+
+#[cfg(test)]
+mod cases {
+    use bencher_api_tests::TestServer;
+    use bencher_json::system::config::{JsonReplication, ReplicationTarget};
+    use bencher_replica::testing::assert_replica_equivalent;
+    use bencher_replica::{
+        CheckpointOutcome, EngineState, ReplicaConfig, RestoreOutcome, SyncEngine,
+        restore_if_missing,
+    };
+    use camino::{Utf8Path, Utf8PathBuf};
+
+    fn dir_path(tmp: &tempfile::TempDir) -> Utf8PathBuf {
+        Utf8Path::from_path(tmp.path())
+            .expect("tempdir path is UTF-8")
+            .to_path_buf()
+    }
+
+    fn replica_json(root: &Utf8Path) -> JsonReplication {
+        JsonReplication {
+            target: ReplicationTarget::File {
+                path: root.to_path_buf().into_std_path_buf(),
+            },
+            sync_interval_secs: None,
+            checkpoint_interval_secs: None,
+            min_checkpoint_pages: None,
+            snapshot_interval_secs: None,
+            snapshot_throttle_mib: None,
+            retention_generations: None,
+            verification_interval_secs: None,
+            shutdown_sync_timeout_secs: None,
+        }
+    }
+
+    fn logger() -> slog::Logger {
+        slog::Logger::root(slog::Discard, slog::o!())
+    }
+
+    // A fixed, injected clock. The repo rule is that time-based tests inject a
+    // clock instead of racing the wall clock, and a frozen clock is also
+    // strictly safer here: the SyncEngine's only clock-driven work is the
+    // due-ness of periodic checkpoints, verification, and snapshots. With time
+    // frozen, none of those become due, so these tests exercise exactly the
+    // steps they drive explicitly (the bootstrap snapshot, ship, and
+    // checkpoint). The initial-generation snapshot is driven by
+    // `pending_new_generation` at construction, not the clock, so freezing time
+    // does not gate `until_streaming` into a hang.
+    fn test_clock() -> bencher_json::Clock {
+        bencher_json::Clock::Custom(std::sync::Arc::new(|| bencher_json::DateTime::TEST))
+    }
+
+    /// Drive the engine through the fresh-replica bootstrap snapshot.
+    async fn until_streaming<C: Send + 'static>(engine: &mut SyncEngine<C>) {
+        for _ in 0u8..64 {
+            if engine.state() == EngineState::Streaming {
+                return;
+            }
+            engine.sync_once().await.expect("bootstrap sync");
+        }
+        panic!(
+            "engine never reached Streaming; state: {:?}",
+            engine.state()
+        );
+    }
+
+    // Real API writes replicate, and the replica restores to an equivalent
+    // database.
+    #[tokio::test]
+    async fn server_writes_replicate_and_restore() {
+        let replica_tmp = tempfile::tempdir().expect("replica tempdir");
+        let replica_root = dir_path(&replica_tmp);
+        let (server, mut engine) =
+            TestServer::new_with_replica(replica_json(&replica_root), Some(test_clock()), false)
+                .await;
+        until_streaming(&mut engine).await;
+
+        // Writes through the real API.
+        let _admin = server.signup("Replica Admin", "replica@example.com").await;
+        let _user = server
+            .signup("Replica User", "replicauser@example.com")
+            .await;
+        engine.sync_once().await.expect("ship API writes");
+
+        // The replica restores to a logically equivalent database.
+        let restore_tmp = tempfile::tempdir().expect("restore tempdir");
+        let target_db = dir_path(&restore_tmp).join("restored.db");
+        let config = ReplicaConfig::try_from(replica_json(&replica_root)).expect("config");
+        let outcome = restore_if_missing(&logger(), &config, &target_db)
+            .await
+            .expect("restore");
+        assert!(
+            matches!(outcome, RestoreOutcome::Restored { .. }),
+            "expected Restored, got {outcome:?}"
+        );
+        assert_replica_equivalent(
+            Utf8Path::from_path(server.db_path()).expect("db path is UTF-8"),
+            &target_db,
+        );
+    }
+
+    // The startup handshake: a missing database file is rebuilt from the
+    // replica, and the signed-up user is present in the restored database.
+    #[tokio::test]
+    async fn server_reboot_missing_db_auto_restores() {
+        let replica_tmp = tempfile::tempdir().expect("replica tempdir");
+        let replica_root = dir_path(&replica_tmp);
+        let (server, mut engine) =
+            TestServer::new_with_replica(replica_json(&replica_root), Some(test_clock()), false)
+                .await;
+        until_streaming(&mut engine).await;
+        let _admin = server.signup("Reboot Admin", "reboot@example.com").await;
+        engine.sync_once().await.expect("ship API writes");
+        drop(engine);
+        drop(server);
+
+        // "Reboot": the volume is gone; the same restore-if-missing handshake
+        // that main.rs runs rebuilds the database before any connection opens.
+        let boot_tmp = tempfile::tempdir().expect("boot tempdir");
+        let new_db = dir_path(&boot_tmp).join("bencher.db");
+        let config = ReplicaConfig::try_from(replica_json(&replica_root)).expect("config");
+        let outcome = restore_if_missing(&logger(), &config, &new_db)
+            .await
+            .expect("restore");
+        assert!(
+            matches!(outcome, RestoreOutcome::Restored { .. }),
+            "expected Restored, got {outcome:?}"
+        );
+        let conn = rusqlite::Connection::open(&new_db).expect("open restored db");
+        let users: i64 = conn
+            .query_row(
+                "SELECT count(*) FROM user WHERE email = 'reboot@example.com'",
+                [],
+                |row| row.get(0),
+            )
+            .expect("query restored user");
+        assert_eq!(users, 1, "signed-up user must survive the reboot restore");
+    }
+
+    // A fresh server over an empty replica is a clean fresh start, and the
+    // existing database is never touched.
+    #[tokio::test]
+    async fn server_boot_empty_replica_fresh_start() {
+        let replica_tmp = tempfile::tempdir().expect("replica tempdir");
+        let replica_root = dir_path(&replica_tmp);
+        let config = ReplicaConfig::try_from(replica_json(&replica_root)).expect("config");
+
+        // Empty replica, missing database: NoReplica (fresh server).
+        let boot_tmp = tempfile::tempdir().expect("boot tempdir");
+        let new_db = dir_path(&boot_tmp).join("bencher.db");
+        let outcome = restore_if_missing(&logger(), &config, &new_db)
+            .await
+            .expect("restore");
+        assert!(
+            matches!(outcome, RestoreOutcome::NoReplica),
+            "expected NoReplica, got {outcome:?}"
+        );
+        assert!(!new_db.exists(), "fresh start must not create a database");
+
+        // Existing database: Skipped, file untouched.
+        let (server, mut engine) =
+            TestServer::new_with_replica(replica_json(&replica_root), Some(test_clock()), false)
+                .await;
+        until_streaming(&mut engine).await;
+        let outcome = restore_if_missing(
+            &logger(),
+            &config,
+            Utf8Path::from_path(server.db_path()).expect("db path is UTF-8"),
+        )
+        .await
+        .expect("restore");
+        assert!(
+            matches!(outcome, RestoreOutcome::Skipped),
+            "expected Skipped, got {outcome:?}"
+        );
+    }
+
+    // Shadow mode: the replica ships and restores alongside a (nominal)
+    // Litestream, but never checkpoints.
+    #[tokio::test]
+    async fn server_shadow_mode_ships_without_checkpoints() {
+        let replica_tmp = tempfile::tempdir().expect("replica tempdir");
+        let replica_root = dir_path(&replica_tmp);
+        let (server, mut engine) =
+            TestServer::new_with_replica(replica_json(&replica_root), Some(test_clock()), true)
+                .await;
+        until_streaming(&mut engine).await;
+        let _admin = server.signup("Shadow Admin", "shadow@example.com").await;
+        engine.sync_once().await.expect("ship API writes");
+
+        // Shadow never checkpoints: Litestream keeps checkpoint ownership.
+        let outcome = engine.checkpoint_once().await.expect("checkpoint");
+        assert_eq!(outcome, CheckpointOutcome::SkippedShadow);
+
+        // The shadow replica still restores to an equivalent database.
+        let restore_tmp = tempfile::tempdir().expect("restore tempdir");
+        let target_db = dir_path(&restore_tmp).join("restored.db");
+        let config = ReplicaConfig::try_from(replica_json(&replica_root)).expect("config");
+        let outcome = restore_if_missing(&logger(), &config, &target_db)
+            .await
+            .expect("restore");
+        assert!(
+            matches!(outcome, RestoreOutcome::Restored { .. }),
+            "expected Restored, got {outcome:?}"
+        );
+        assert_replica_equivalent(
+            Utf8Path::from_path(server.db_path()).expect("db path is UTF-8"),
+            &target_db,
+        );
+    }
+}