Reclaim unused SQLite log database pages in the background (#49069)

## Why

Deleting logs leaves free SQLite pages available for reuse without shrinking the database file. Reclaim excess free space while preserving a reserve for future writes.

## What changed

- Start a background incremental-vacuum worker with `StateRuntime`, enabling reclamation only for the logs database. Use a per-home file lock to select one worker and wait for it to finish during shutdown.
- Reclaim pages when at least 64 MiB and 25% of the database are free, preserving a reserve of at least 16 MiB or 10% of the database.
- Use short transactions with page and cooperative time budgets, defer work on contention, and reduce batch sizes with exponential backoff when interrupted passes make no progress.
- Record reclamation counts, durations, and reclaimed pages in telemetry.

## Testing

Add a regression test that interrupts a vacuum commit, verifies foreground writes can resume without data loss, then confirms a resumed pass shrinks the database and passes `PRAGMA integrity_check`.

GitOrigin-RevId: 8421f87b0aeef6540eda5690c84216906673c72e

codex-rs/state/src/lib.rs

@@ -82,13 +82,13 @@ pub use runtime::GoalStore;
 pub use runtime::GoalUpdate;
 pub use runtime::MemoryStore;
 pub use runtime::RemoteControlEnrollmentRecord;
-pub use runtime::RuntimeDbBackup;
 pub use runtime::SqliteIntegrityCheck;
 pub use runtime::SqliteQueueStore;
 pub use runtime::ThreadFilterOptions;
 pub use runtime::backup_runtime_db_for_fresh_start;
 pub use runtime::is_sqlite_corruption_error;
 pub use runtime::open_thread_history_db;
+pub use runtime::recovery::RuntimeDbBackup;
 pub use runtime::runtime_db_path_for_corruption_error;
 pub use runtime::sqlite_error_detail_is_corruption;
 pub use runtime::sqlite_error_detail_is_lock;

codex-rs/state/src/runtime.rs

@@ -47,7 +47,8 @@ mod memories;
 mod memory_versions;
 mod projects;
 mod queued_items;
-mod recovery;
+pub(crate) mod reclamation;
+pub(crate) mod recovery;
 mod remote_control;
 mod rollout_migration;
 #[cfg(test)]
@@ -67,8 +68,6 @@ pub use goals::GoalStore;
 pub use goals::GoalUpdate;
 pub use memories::MemoryStore;
 pub use queued_items::SqliteQueueStore;
-pub use recovery::RuntimeDbBackup;
-pub(super) use recovery::RuntimeDbInitError;
 pub use recovery::backup_runtime_db_for_fresh_start;
 pub use recovery::is_sqlite_corruption_error;
 pub use recovery::runtime_db_path_for_corruption_error;
@@ -98,6 +97,7 @@ pub struct StateRuntime {
     thread_queue: SqliteQueueStore,
     thread_updated_at_millis: Arc<AtomicI64>,
     thread_recency_at_millis: Arc<AtomicI64>,
+    reclamation: Arc<reclamation::SqliteReclamationWorker>,
 }
 
 impl StateRuntime {
@@ -253,6 +253,7 @@ impl StateRuntime {
         let thread_updated_at_millis = thread_updated_at_millis.unwrap_or(0);
         let thread_recency_at_millis = thread_recency_at_millis.unwrap_or(0);
         let runtime = Arc::new(Self {
+            reclamation: reclamation::SqliteReclamationWorker::spawn(sqlite.clone()),
             thread_goals: GoalStore::new(Arc::clone(&goals_pool)),
             memories: MemoryStore::new(Arc::clone(&memories_pool), Arc::clone(&pool)),
             memories_v2: Arc::new(tokio::sync::OnceCell::new()),
@@ -303,6 +304,7 @@ impl StateRuntime {
 
     /// Close all SQLite pools and wait for outstanding pool workers to exit.
     pub async fn close(&self) {
+        self.reclamation.close().await;
         self.thread_queue.close().await;
         self.memories.close().await;
         if let Some(memories) = self.memories_v2.get() {

codex-rs/state/src/runtime/reclamation.rs

@@ -0,0 +1,415 @@
+//! Reclaims unused SQLite pages in opted-in databases, reducing batches after stalled passes.
+use crate::RuntimeDbPath;
+use crate::SqliteConfig;
+use log::LevelFilter;
+use sqlx::ConnectOptions;
+use sqlx::Connection;
+use sqlx::SqliteConnection;
+use sqlx::sqlite::SqliteConnectOptions;
+use sqlx::sqlite::SqliteSynchronous;
+use std::fs::File;
+use std::path::Path;
+use std::sync::Arc;
+use std::time::Duration;
+use std::time::Instant;
+use tokio::sync::watch;
+
+const IDLE_INTERVAL: Duration = Duration::from_secs(60);
+const ACTIVE_INTERVAL: Duration = Duration::from_millis(50);
+const CONTENTION_INTERVAL: Duration = Duration::from_millis(500);
+const PASS_DURATION: Duration = Duration::from_millis(100);
+const MIN_FREE_BYTES: i64 = 64 * 1024 * 1024;
+const RESERVE_BYTES: i64 = 16 * 1024 * 1024;
+const PASS_PAGES: u32 = 1024;
+const BATCH_PAGES: u32 = 64;
+
+/// Owns background reclamation for opted-in databases registered with `SqliteConfig`.
+///
+/// A per-home file lock elects one active worker, which visits databases serially
+/// on individual retry schedules. Shutdown waits for the dedicated SQLite
+/// connection to close before the worker releases the lock.
+pub(crate) struct SqliteReclamationWorker {
+    shutdown: watch::Sender<()>,
+    finished: watch::Receiver<()>,
+}
+
+impl SqliteReclamationWorker {
+    pub(crate) fn spawn(sqlite: SqliteConfig) -> Arc<Self> {
+        let (shutdown, mut receiver) = watch::channel(());
+        let (finish, finished) = watch::channel(());
+        tokio::spawn(async move {
+            // Declared first so completion is signaled after all worker resources drop.
+            let _finish = finish;
+            let mut scheduled = sqlite
+                .runtime_db_paths()
+                .into_iter()
+                .filter_map(|db| {
+                    if db.background_reclamation {
+                        Some((db, Instant::now(), ReclamationState::default()))
+                    } else {
+                        None
+                    }
+                })
+                .collect::<Vec<_>>();
+            let mut owner = None;
+            let mut delay = IDLE_INTERVAL;
+            loop {
+                tokio::select! {
+                    _ = receiver.changed() => break,
+                    _ = tokio::time::sleep(delay) => {}
+                }
+                if owner.is_none() {
+                    owner = try_ownership(sqlite.home()).await.ok().flatten();
+                }
+                if owner.is_none() {
+                    continue;
+                }
+                delay = visit(&mut scheduled, &receiver).await;
+            }
+        });
+
+        Arc::new(Self { shutdown, finished })
+    }
+
+    pub(crate) async fn close(&self) {
+        self.shutdown.send_replace(());
+        // Every caller waits for task completion, even if another close is canceled.
+        let _ = self.finished.clone().changed().await;
+    }
+}
+
+async fn visit(
+    scheduled: &mut [(RuntimeDbPath, Instant, ReclamationState)],
+    shutdown: &watch::Receiver<()>,
+) -> Duration {
+    for (db, due, state) in &mut *scheduled {
+        let started = Instant::now();
+        if shutdown.has_changed().unwrap_or(true) {
+            break;
+        }
+        if *due > started {
+            continue;
+        }
+        let result = reclaim(
+            &db.path,
+            ReclamationOptions {
+                budget: Budget {
+                    deadline: Some(started + PASS_DURATION),
+                    pages: PASS_PAGES,
+                },
+                batch_pages: state.batch_pages,
+            },
+            shutdown,
+        )
+        .await;
+        crate::telemetry::record_reclamation(db.label, started.elapsed(), &result);
+        *due = Instant::now() + state.retry_after(&result);
+    }
+    scheduled
+        .iter()
+        .map(|(_, due, _)| due.saturating_duration_since(Instant::now()))
+        .min()
+        .unwrap_or(IDLE_INTERVAL)
+}
+
+async fn try_ownership(home: &Path) -> std::io::Result<Option<File>> {
+    let lock = tokio::fs::OpenOptions::new()
+        .read(true)
+        .write(true)
+        .create(true)
+        .truncate(false)
+        .open(home.join(".sqlite-maintenance.lock"))
+        .await?
+        .into_std()
+        .await;
+    match lock.try_lock() {
+        Ok(()) => Ok(Some(lock)),
+        Err(std::fs::TryLockError::WouldBlock) => Ok(None),
+        Err(std::fs::TryLockError::Error(error)) => Err(error),
+    }
+}
+
+pub(crate) struct ReclamationPass {
+    pub(crate) pages: u32,
+    outcome: PassOutcome,
+}
+
+#[derive(Debug, PartialEq, Eq)]
+enum PassOutcome {
+    Idle,
+    Active,
+    Contended,
+    Interrupted,
+    Shutdown,
+}
+
+#[derive(Debug, PartialEq, Eq)]
+struct ReclamationState {
+    batch_pages: u32,
+    stalled_passes: u32,
+}
+
+impl Default for ReclamationState {
+    fn default() -> Self {
+        Self {
+            batch_pages: BATCH_PAGES,
+            stalled_passes: 0,
+        }
+    }
+}
+
+impl ReclamationState {
+    fn retry_after(&mut self, result: &anyhow::Result<ReclamationPass>) -> Duration {
+        if let Ok(pass) = result
+            && pass.outcome == PassOutcome::Interrupted
+            && pass.pages == 0
+        {
+            self.stalled_passes = self.stalled_passes.saturating_add(1);
+            self.batch_pages = (self.batch_pages / 2).max(1);
+
+            // exponentially backoffs from CONTENTION_INTERVAL to IDLE_INTERVAL (500ms -> 60s)
+            return (CONTENTION_INTERVAL * (1 << (self.stalled_passes - 1).min(7)))
+                .min(IDLE_INTERVAL);
+        }
+
+        // Keep the smaller batch after progress; growing it would repeat the stall.
+        self.stalled_passes = 0;
+        match result {
+            Ok(pass) => match pass.outcome {
+                PassOutcome::Idle | PassOutcome::Shutdown => IDLE_INTERVAL,
+                PassOutcome::Active | PassOutcome::Interrupted => ACTIVE_INTERVAL,
+                PassOutcome::Contended => CONTENTION_INTERVAL,
+            },
+            Err(_) => IDLE_INTERVAL,
+        }
+    }
+}
+
+struct Budget {
+    deadline: Option<Instant>,
+    /// Amount of pages to reclaim per transaction
+    pages: u32,
+}
+
+struct ReclamationOptions {
+    budget: Budget,
+    batch_pages: u32,
+}
+
+impl From<Budget> for ReclamationOptions {
+    fn from(budget: Budget) -> Self {
+        Self {
+            budget,
+            batch_pages: BATCH_PAGES,
+        }
+    }
+}
+
+async fn reclaim(
+    path: &Path,
+    options: impl Into<ReclamationOptions>,
+    shutdown: &watch::Receiver<()>,
+) -> anyhow::Result<ReclamationPass> {
+    if !tokio::fs::try_exists(path).await? {
+        return Ok(ReclamationPass {
+            pages: 0,
+            outcome: PassOutcome::Idle,
+        });
+    }
+    #[expect(
+        clippy::disallowed_methods,
+        reason = "maintenance needs a dedicated connection with no busy wait or automatic checkpoints"
+    )]
+    let mut connection = match SqliteConnection::connect_with(
+        &SqliteConnectOptions::new()
+            .filename(path)
+            .create_if_missing(false)
+            .busy_timeout(Duration::ZERO)
+            .synchronous(SqliteSynchronous::Normal)
+            // Keep commits from doing checkpoint I/O while holding up writers.
+            .pragma("wal_autocheckpoint", "0")
+            // A 16 MiB cache reduced CPU time in large-database measurements.
+            .pragma("cache_size", "-16384")
+            .log_statements(LevelFilter::Off)
+            .log_slow_statements(LevelFilter::Off, Duration::ZERO),
+    )
+    .await
+    {
+        Ok(connection) => connection,
+        Err(error) => match outcome_after_error(&error) {
+            Some(outcome) => {
+                return Ok(ReclamationPass { pages: 0, outcome });
+            }
+            None => return Err(error.into()),
+        },
+    };
+
+    let result = reclaim_pages(&mut connection, options, shutdown).await;
+    connection.close().await?;
+    result
+}
+
+/// Reclaims free pages in short transactions while preserving a reusable reserve.
+///
+/// Each batch checks the reserve under `BEGIN IMMEDIATE` and commits before
+/// checkpointing or sleeping. The pass bounds reclaimed freelist pages and checks
+/// a cooperative deadline; storage I/O can overrun that deadline.
+async fn reclaim_pages(
+    connection: &mut SqliteConnection,
+    options: impl Into<ReclamationOptions>,
+    shutdown: &watch::Receiver<()>,
+) -> anyhow::Result<ReclamationPass> {
+    let ReclamationOptions {
+        budget,
+        batch_pages,
+    } = options.into();
+    if budget
+        .deadline
+        .is_some_and(|deadline| Instant::now() >= deadline)
+        || shutdown.has_changed().unwrap_or(true)
+    {
+        return Ok(ReclamationPass {
+            pages: 0,
+            outcome: if shutdown.has_changed().unwrap_or(true) {
+                PassOutcome::Shutdown
+            } else {
+                PassOutcome::Interrupted
+            },
+        });
+    }
+    let mut reclaimed = 0;
+    let mut outcome = PassOutcome::Contended;
+    let result: Result<(), sqlx::Error> = async {
+        let stop = shutdown.clone();
+        connection
+            .lock_handle()
+            .await?
+            .set_progress_handler(/*num_ops*/ 1, move || {
+                !stop.has_changed().unwrap_or(true)
+                    && budget
+                        .deadline
+                        .is_none_or(|deadline| Instant::now() < deadline)
+            });
+
+        let (mode, page_size, pages, free): (i64, i64, i64, i64) = sqlx::query_as(
+            "SELECT auto_vacuum, page_size, page_count, freelist_count \
+         FROM pragma_auto_vacuum, pragma_page_size, pragma_page_count, pragma_freelist_count",
+        )
+        .fetch_one(&mut *connection)
+        .await?;
+
+        // Require incremental auto-vacuum and at least 64 MiB and 25% free space.
+        if mode != 2 || free * page_size < MIN_FREE_BYTES || free < pages / 4 {
+            outcome = PassOutcome::Idle;
+            return Ok(());
+        }
+        if !checkpoint_complete(connection).await? {
+            return Ok(());
+        }
+
+        outcome = PassOutcome::Active;
+        let reserve = (RESERVE_BYTES / page_size).max(pages / 10);
+        let version: i64 = sqlx::query_scalar("PRAGMA data_version")
+            .fetch_one(&mut *connection)
+            .await?;
+
+        while reclaimed < budget.pages
+            && budget
+                .deadline
+                .is_none_or(|deadline| Instant::now() < deadline)
+            && !shutdown.has_changed().unwrap_or(true)
+        {
+            // Check inside the same write transaction, so a foreground writer cannot
+            // consume the reserve between our free-page check and vacuum.
+            let mut transaction = connection.begin_with("BEGIN IMMEDIATE").await?;
+            let free: i64 = sqlx::query_scalar("PRAGMA freelist_count")
+                .fetch_one(&mut *transaction)
+                .await?;
+            if free <= reserve {
+                transaction.rollback().await?;
+                outcome = PassOutcome::Idle;
+                break;
+            }
+
+            let batch = i64::from(batch_pages.min(budget.pages - reclaimed)).min(free - reserve);
+            sqlx::query(sqlx::AssertSqlSafe(format!(
+                "PRAGMA incremental_vacuum({batch})"
+            )))
+            .execute(&mut *transaction)
+            .await?;
+            transaction.commit().await?;
+            reclaimed += batch as u32;
+            // A reader arriving after admission can pin at most one batch.
+            if !checkpoint_complete(connection).await? {
+                outcome = PassOutcome::Contended;
+                break;
+            }
+
+            tokio::time::sleep(Duration::from_millis(5)).await;
+            // Back off if another connection committed during this pass.
+            let current: i64 = sqlx::query_scalar("PRAGMA data_version")
+                .fetch_one(&mut *connection)
+                .await?;
+            if current != version {
+                outcome = PassOutcome::Contended;
+                break;
+            }
+        }
+        Ok(())
+    }
+    .await;
+    // Contention and deadlines defer work, including when a prefix has committed.
+    // The connection is dedicated and closed by the caller, so an interrupted
+    // transaction cannot leak into foreground work.
+    match result {
+        Ok(()) => {}
+        Err(error) => match outcome_after_error(&error) {
+            Some(reason) => outcome = reason,
+            None => return Err(error.into()),
+        },
+    }
+    if shutdown.has_changed().unwrap_or(true) {
+        outcome = PassOutcome::Shutdown;
+    } else if outcome == PassOutcome::Active
+        && budget
+            .deadline
+            .is_some_and(|deadline| Instant::now() >= deadline)
+    {
+        // The deadline can also expire between statements, without SQLITE_INTERRUPT.
+        outcome = PassOutcome::Interrupted;
+    }
+
+    Ok(ReclamationPass {
+        pages: reclaimed,
+        outcome,
+    })
+}
+
+fn outcome_after_error(error: &sqlx::Error) -> Option<PassOutcome> {
+    error
+        .as_database_error()
+        .and_then(sqlx::error::DatabaseError::code)
+        .and_then(
+            |code| match code.parse::<i32>().ok().map(|code| code & 0xff) {
+                Some(libsqlite3_sys::SQLITE_BUSY | libsqlite3_sys::SQLITE_LOCKED) => {
+                    Some(PassOutcome::Contended)
+                }
+                Some(libsqlite3_sys::SQLITE_INTERRUPT) => Some(PassOutcome::Interrupted),
+                _ => None,
+            },
+        )
+}
+
+async fn checkpoint_complete(connection: &mut SqliteConnection) -> Result<bool, sqlx::Error> {
+    // PASSIVE does not take the writer lock or wait for readers. It can still
+    // exceed the cooperative deadline on slow storage. Do not reject a large
+    // WAL forever: a quiet database may have no other writer to checkpoint it.
+    let (busy, frames, done): (i64, i64, i64) = sqlx::query_as("PRAGMA wal_checkpoint(PASSIVE)")
+        .fetch_one(connection)
+        .await?;
+    Ok(busy == 0 && frames >= 0 && frames == done)
+}
+
+#[cfg(test)]
+#[path = "reclamation_tests.rs"]
+mod tests;

codex-rs/state/src/runtime/reclamation_tests.rs

@@ -0,0 +1,132 @@
+use super::*;
+use crate::migrations::runtime_logs_migrator;
+use crate::runtime::test_support::unique_temp_dir;
+use codex_utils_absolute_path::test_support::PathExt;
+use pretty_assertions::assert_eq;
+
+#[tokio::test]
+async fn interrupted_reclamation_releases_writer_and_resumes_without_data_loss()
+-> anyhow::Result<()> {
+    let sqlite = SqliteConfig::new_for_testing(unique_temp_dir().abs());
+    tokio::fs::create_dir_all(sqlite.home()).await?;
+    let pool = sqlite
+        .open_logs_db(&runtime_logs_migrator(), /*telemetry_override*/ None)
+        .await?;
+    sqlx::query(
+        "WITH RECURSIVE n(i) AS (VALUES(1) UNION ALL SELECT i+1 FROM n WHERE i < 4096) \
+         INSERT INTO logs(ts,ts_nanos,level,target,feedback_log_body,thread_id,process_uuid) \
+         SELECT unixepoch(),0,'INFO','fixture',printf('%032768d',i),'thread-'||i,'process-'||(i%17) FROM n",
+    )
+    .execute(&pool)
+    .await?;
+    sqlx::query("DELETE FROM logs WHERE id%4 != 0")
+        .execute(&pool)
+        .await?;
+    sqlx::query("PRAGMA wal_checkpoint(TRUNCATE)")
+        .execute(&pool)
+        .await?;
+    let rows = "SELECT id, feedback_log_body FROM logs ORDER BY id";
+    let mut expected: Vec<(i64, String)> = sqlx::query_as(rows).fetch_all(&pool).await?;
+    let free_before: i64 = sqlx::query_scalar("PRAGMA freelist_count")
+        .fetch_one(&pool)
+        .await?;
+    let (shutdown, receiver) = watch::channel(());
+    let mut connection = pool.acquire().await?.detach();
+    sqlx::raw_sql(
+        "PRAGMA busy_timeout = 0; PRAGMA wal_autocheckpoint = 0; PRAGMA cache_size = -16384;",
+    )
+    .execute(&mut connection)
+    .await?;
+    // Request shutdown inside the first vacuum commit, while SQLite still owns the
+    // writer lock. The production progress handler must stop this pass safely.
+    connection.lock_handle().await?.set_commit_hook(move || {
+        shutdown.send_replace(());
+        true
+    });
+    let interrupted = reclaim_pages(
+        &mut connection,
+        Budget {
+            // Shutdown, rather than machine speed, determines the interruption point.
+            deadline: Some(Instant::now() + Duration::from_secs(30)),
+            pages: PASS_PAGES,
+        },
+        &receiver,
+    )
+    .await?;
+    assert!(receiver.has_changed()?, "must reach the vacuum commit");
+    assert!(interrupted.pages <= BATCH_PAGES);
+    connection.close().await?;
+    let free_after: i64 = sqlx::query_scalar("PRAGMA freelist_count")
+        .fetch_one(&pool)
+        .await?;
+    // An interruption at COMMIT may leave the first batch committed or rolled back.
+    assert!((0..=i64::from(BATCH_PAGES)).contains(&(free_before - free_after)));
+    assert_eq!(
+        sqlx::query_as::<_, (i64, String)>(rows)
+            .fetch_all(&pool)
+            .await?,
+        expected
+    );
+
+    // A foreground write must acquire the lock immediately, without a busy retry.
+    let mut writer = pool.acquire().await?;
+    sqlx::query("PRAGMA busy_timeout = 0")
+        .execute(&mut *writer)
+        .await?;
+    let mut transaction = writer.begin_with("BEGIN IMMEDIATE").await?;
+    expected[0].1 = "foreground write after interruption".to_string();
+    sqlx::query("UPDATE logs SET feedback_log_body = ? WHERE id = ?")
+        .bind(&expected[0].1)
+        .bind(expected[0].0)
+        .execute(&mut *transaction)
+        .await?;
+    transaction.commit().await?;
+    drop(writer);
+
+    // Flush earlier writes so only the resumed pass can shrink the main file.
+    sqlx::query("PRAGMA wal_checkpoint(TRUNCATE)")
+        .execute(&pool)
+        .await?;
+    let bytes_before_retry = tokio::fs::metadata(sqlite.logs_db_path()).await?.len();
+    let free_before_retry: i64 = sqlx::query_scalar("PRAGMA freelist_count")
+        .fetch_one(&pool)
+        .await?;
+    let (_shutdown, receiver) = watch::channel(());
+    let resumed = reclaim(
+        &sqlite.logs_db_path(),
+        Budget {
+            deadline: Some(Instant::now() + Duration::from_secs(30)),
+            pages: BATCH_PAGES,
+        },
+        &receiver,
+    )
+    .await?;
+    let free_after_retry: i64 = sqlx::query_scalar("PRAGMA freelist_count")
+        .fetch_one(&pool)
+        .await?;
+    assert!(resumed.pages > 0);
+    assert_eq!(
+        free_before_retry - free_after_retry,
+        i64::from(resumed.pages)
+    );
+    let bytes_after_retry = tokio::fs::metadata(sqlite.logs_db_path()).await?.len();
+    assert!(
+        bytes_after_retry < bytes_before_retry,
+        "reclamation must shrink the SQLite file: {bytes_before_retry} -> {bytes_after_retry} bytes"
+    );
+    assert_eq!(
+        sqlx::query_as::<_, (i64, String)>(rows)
+            .fetch_all(&pool)
+            .await?,
+        expected
+    );
+    assert_eq!(
+        sqlx::query_scalar::<_, String>("PRAGMA integrity_check")
+            .fetch_all(&pool)
+            .await?,
+        vec!["ok"]
+    );
+    pool.close().await;
+    tokio::fs::remove_dir_all(sqlite.home()).await?;
+    Ok(())
+}

codex-rs/state/src/sqlite.rs

@@ -7,7 +7,7 @@
 
 use crate::DbTelemetry;
 use crate::migrations::repair_legacy_recency_migration_version;
-use crate::runtime::RuntimeDbInitError;
+use crate::runtime::recovery::RuntimeDbInitError;
 use crate::telemetry;
 use crate::telemetry::DbKind;
 use codex_utils_absolute_path::AbsolutePathBuf;
@@ -38,6 +38,9 @@ struct RuntimeDbSpec {
     kind: DbKind,
     open_phase: &'static str,
     migrate_phase: &'static str,
+    /// Opt in only after auditing writers for deferred read-to-write upgrades:
+    /// an intervening reclamation commit can make those fail with SQLITE_BUSY_SNAPSHOT.
+    background_reclamation: bool,
 }
 
 impl RuntimeDbSpec {
@@ -52,6 +55,7 @@ const STATE_DB: RuntimeDbSpec = RuntimeDbSpec {
     kind: DbKind::State,
     open_phase: "open_state",
     migrate_phase: "migrate_state",
+    background_reclamation: false,
 };
 
 const LOGS_DB: RuntimeDbSpec = RuntimeDbSpec {
@@ -60,6 +64,8 @@ const LOGS_DB: RuntimeDbSpec = RuntimeDbSpec {
     kind: DbKind::Logs,
     open_phase: "open_logs",
     migrate_phase: "migrate_logs",
+    // Log transactions write before reading, so they already hold the writer lock.
+    background_reclamation: true,
 };
 
 const GOALS_DB: RuntimeDbSpec = RuntimeDbSpec {
@@ -68,6 +74,7 @@ const GOALS_DB: RuntimeDbSpec = RuntimeDbSpec {
     kind: DbKind::Goals,
     open_phase: "open_goals",
     migrate_phase: "migrate_goals",
+    background_reclamation: false,
 };
 
 const MEMORIES_DB: RuntimeDbSpec = RuntimeDbSpec {
@@ -76,11 +83,13 @@ const MEMORIES_DB: RuntimeDbSpec = RuntimeDbSpec {
     kind: DbKind::Memories,
     open_phase: "open_memories",
     migrate_phase: "migrate_memories",
+    background_reclamation: false,
 };
 
 const MEMORIES_V2_DB: RuntimeDbSpec = RuntimeDbSpec {
     label: "memories v2 DB",
     filename: "memories_v2_1.sqlite",
+    background_reclamation: false,
     ..MEMORIES_DB
 };
 
@@ -90,6 +99,7 @@ const QUEUE_DB: RuntimeDbSpec = RuntimeDbSpec {
     kind: DbKind::Queue,
     open_phase: "open_queue",
     migrate_phase: "migrate_queue",
+    background_reclamation: false,
 };
 
 const THREAD_HISTORY_DB: RuntimeDbSpec = RuntimeDbSpec {
@@ -98,6 +108,7 @@ const THREAD_HISTORY_DB: RuntimeDbSpec = RuntimeDbSpec {
     kind: DbKind::ThreadHistory,
     open_phase: "open_thread_history",
     migrate_phase: "migrate_thread_history",
+    background_reclamation: false,
 };
 
 const RUNTIME_DBS: [RuntimeDbSpec; 7] = [
@@ -114,6 +125,7 @@ const RUNTIME_DBS: [RuntimeDbSpec; 7] = [
 pub struct RuntimeDbPath {
     pub label: &'static str,
     pub path: PathBuf,
+    pub(crate) background_reclamation: bool,
 }
 
 /// Resolved configuration shared by all Codex SQLite connections.
@@ -185,6 +197,7 @@ impl SqliteConfig {
             .map(|spec| RuntimeDbPath {
                 label: spec.label,
                 path: spec.path(self.home()),
+                background_reclamation: spec.background_reclamation,
             })
             .collect()
     }

codex-rs/state/src/telemetry.rs

@@ -13,9 +13,11 @@ use crate::LOG_WRITE_ENTRIES_METRIC;
 use crate::LOG_WRITE_MAX_ENTRY_BYTES_METRIC;
 use crate::LOG_WRITE_METRIC;
 use crate::LogEntry;
+use crate::runtime::reclamation::ReclamationPass;
+
 use tracing::debug;
 
-/// Low-cardinality sink for SQLite startup, fallback, and log-write telemetry.
+/// Low-cardinality sink for SQLite startup, fallback, log-write, and reclamation telemetry.
 ///
 /// Implementations should absorb delivery failures locally. Database behavior
 /// must not depend on whether telemetry export succeeds.
@@ -136,6 +138,31 @@ pub(crate) fn record_log_queue_drop(reason: &'static str, telemetry: Option<&dyn
     record_counter(telemetry, LOG_QUEUE_DROPPED_METRIC, &[("reason", reason)]);
 }
 
+pub(crate) fn record_reclamation(
+    db: &'static str,
+    duration: Duration,
+    result: &anyhow::Result<ReclamationPass>,
+) {
+    let Some(telemetry) = resolve_telemetry(/*telemetry*/ None) else {
+        return;
+    };
+    let outcome = DbOutcomeTags::from_result(result);
+    let tags = [
+        ("db", db),
+        ("status", outcome.status),
+        ("error", outcome.error),
+    ];
+    telemetry.counter("codex.sqlite.reclamation.count", /*inc*/ 1, &tags);
+    telemetry.record_duration("codex.sqlite.reclamation.duration_ms", duration, &tags);
+    if let Ok(pass) = result {
+        telemetry.histogram(
+            "codex.sqlite.reclamation.pages",
+            i64::from(pass.pages),
+            &tags,
+        );
+    }
+}
+
 fn record_counter(telemetry: Option<&dyn DbTelemetry>, name: &str, tags: &[(&str, &str)]) {
     if let Some(telemetry) = resolve_telemetry(telemetry) {
         telemetry.counter(name, /*inc*/ 1, tags);