diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index fa5a5e2803..be969b08cd 100644 --- a/codex-rs/state/src/lib.rs +++ b/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; diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 376ad682ed..14fcd2d167 100644 --- a/codex-rs/state/src/runtime.rs +++ b/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, thread_recency_at_millis: Arc, + reclamation: Arc, } 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() { diff --git a/codex-rs/state/src/runtime/reclamation.rs b/codex-rs/state/src/runtime/reclamation.rs new file mode 100644 index 0000000000..29c46e4fb1 --- /dev/null +++ b/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 { + 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::>(); + 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> { + 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) -> 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, + /// Amount of pages to reclaim per transaction + pages: u32, +} + +struct ReclamationOptions { + budget: Budget, + batch_pages: u32, +} + +impl From for ReclamationOptions { + fn from(budget: Budget) -> Self { + Self { + budget, + batch_pages: BATCH_PAGES, + } + } +} + +async fn reclaim( + path: &Path, + options: impl Into, + shutdown: &watch::Receiver<()>, +) -> anyhow::Result { + 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, + shutdown: &watch::Receiver<()>, +) -> anyhow::Result { + 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 { + error + .as_database_error() + .and_then(sqlx::error::DatabaseError::code) + .and_then( + |code| match code.parse::().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 { + // 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; diff --git a/codex-rs/state/src/runtime/reclamation_tests.rs b/codex-rs/state/src/runtime/reclamation_tests.rs new file mode 100644 index 0000000000..4ae711acf9 --- /dev/null +++ b/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(()) +} diff --git a/codex-rs/state/src/sqlite.rs b/codex-rs/state/src/sqlite.rs index b0a9294c95..6487395c33 100644 --- a/codex-rs/state/src/sqlite.rs +++ b/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() } diff --git a/codex-rs/state/src/telemetry.rs b/codex-rs/state/src/telemetry.rs index 0d9495ba79..d9692b1943 100644 --- a/codex-rs/state/src/telemetry.rs +++ b/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, +) { + 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);