Add process-scoped SQLite telemetry

Co-authored-by: Codex <noreply@openai.com>

codex-rs/rollout/src/lib.rs

@@ -10,6 +10,7 @@ pub(crate) mod metadata;
 pub(crate) mod policy;
 pub(crate) mod recorder;
 pub(crate) mod session_index;
+mod sqlite_metrics;
 pub mod state_db;
 
 pub(crate) mod default_client {
@@ -63,6 +64,7 @@ pub use session_index::find_thread_meta_by_name_str;
 pub use session_index::find_thread_name_by_id;
 pub use session_index::find_thread_names_by_ids;
 pub use state_db::StateDbHandle;
+pub use state_db::sqlite_telemetry_recorder;
 
 #[cfg(test)]
 mod tests;

codex-rs/rollout/src/list.rs

@@ -1255,47 +1255,57 @@ async fn find_thread_path_by_id_str_in_subdir(
     };
     let thread_id = ThreadId::from_string(id_str).ok();
     let mut unverified_db_path = None;
+    let mut fallback_reason = state_db_ctx.is_none().then_some("db_unavailable");
     if let Some(state_db_ctx) = state_db_ctx
         && let Some(thread_id) = thread_id
-        && let Some(db_path) = state_db::find_rollout_path_by_id(
-            Some(state_db_ctx),
-            thread_id,
-            archived_only,
-            "find_path_query",
-        )
-        .await
     {
-        if tokio::fs::try_exists(&db_path).await.unwrap_or(false) {
-            match read_session_meta_line(&db_path).await {
-                Ok(meta_line) if meta_line.meta.id == thread_id => {
-                    return Ok(Some(db_path));
-                }
-                Ok(meta_line) => {
+        match state_db_ctx
+            .find_rollout_path_by_id(thread_id, archived_only)
+            .await
+        {
+            Ok(Some(db_path)) => {
+                if tokio::fs::try_exists(&db_path).await.unwrap_or(false) {
+                    match read_session_meta_line(&db_path).await {
+                        Ok(meta_line) if meta_line.meta.id == thread_id => {
+                            return Ok(Some(db_path));
+                        }
+                        Ok(meta_line) => {
+                            tracing::error!(
+                                "state db returned rollout path for thread {id_str} but file belongs to thread {}: {}",
+                                meta_line.meta.id,
+                                db_path.display()
+                            );
+                            tracing::warn!(
+                                "state db discrepancy during find_thread_path_by_id_str_in_subdir: mismatched_db_path"
+                            );
+                            codex_state::record_fallback(None, "find_thread_path", "mismatch");
+                        }
+                        Err(err) => {
+                            tracing::debug!(
+                                "state db returned rollout path for thread {id_str} that could not be verified: {}: {err}",
+                                db_path.display()
+                            );
+                            unverified_db_path = Some(db_path);
+                        }
+                    }
+                } else {
                     tracing::error!(
-                        "state db returned rollout path for thread {id_str} but file belongs to thread {}: {}",
-                        meta_line.meta.id,
+                        "state db returned stale rollout path for thread {id_str}: {}",
                         db_path.display()
                     );
                     tracing::warn!(
-                        "state db discrepancy during find_thread_path_by_id_str_in_subdir: mismatched_db_path"
+                        "state db discrepancy during find_thread_path_by_id_str_in_subdir: stale_db_path"
                     );
-                }
-                Err(err) => {
-                    tracing::debug!(
-                        "state db returned rollout path for thread {id_str} that could not be verified: {}: {err}",
-                        db_path.display()
-                    );
-                    unverified_db_path = Some(db_path);
+                    codex_state::record_fallback(None, "find_thread_path", "stale_path");
                 }
             }
-        } else {
-            tracing::error!(
-                "state db returned stale rollout path for thread {id_str}: {}",
-                db_path.display()
-            );
-            tracing::warn!(
-                "state db discrepancy during find_thread_path_by_id_str_in_subdir: stale_db_path"
-            );
+            Ok(None) => fallback_reason = Some("missing_row"),
+            Err(err) => {
+                tracing::warn!(
+                    "state db find_rollout_path_by_id failed during find_path_query: {err}"
+                );
+                fallback_reason = Some("db_error");
+            }
         }
     }
 
@@ -1323,6 +1333,9 @@ async fn find_thread_path_by_id_str_in_subdir(
         tracing::warn!(
             "state db discrepancy during find_thread_path_by_id_str_in_subdir: falling_back"
         );
+        if let Some(reason) = fallback_reason {
+            codex_state::record_fallback(None, "find_thread_path", reason);
+        }
         state_db::read_repair_rollout_path(
             state_db_ctx,
             thread_id,

codex-rs/rollout/src/recorder.rs

@@ -450,6 +450,7 @@ impl RolloutRecorder {
         if state_db_ctx.is_none() {
             // Keep legacy behavior when SQLite is unavailable: return filesystem results
             // at the requested page size.
+            codex_state::record_fallback(None, "list_threads", "db_unavailable");
             return Ok(page_from_filesystem_scan(
                 fs_page,
                 sort_direction,
@@ -558,6 +559,7 @@ impl RolloutRecorder {
                     )
                     .await;
                 }
+                codex_state::record_fallback(None, "list_threads", "metadata_filter");
                 let page = page_from_filesystem_scan(fs_page, sort_direction, page_size, sort_key);
                 return Ok(fill_missing_thread_item_metadata_from_state_db(
                     state_db_ctx.as_deref(),
@@ -569,6 +571,7 @@ impl RolloutRecorder {
         }
         if listing_has_metadata_filters {
             let page = page_from_filesystem_scan(fs_page, sort_direction, page_size, sort_key);
+            codex_state::record_fallback(None, "list_threads", "db_error");
             return Ok(fill_missing_thread_item_metadata_from_state_db(
                 state_db_ctx.as_deref(),
                 page,
@@ -578,6 +581,7 @@ impl RolloutRecorder {
         // If SQLite listing still fails, return the filesystem page rather than failing the list.
         tracing::error!("Falling back on rollout system");
         tracing::warn!("state db discrepancy during list_threads_with_db_fallback: falling_back");
+        codex_state::record_fallback(None, "list_threads", "db_error");
         Ok(page_from_filesystem_scan(
             fs_page,
             sort_direction,
@@ -601,6 +605,7 @@ impl RolloutRecorder {
     ) -> std::io::Result<Option<PathBuf>> {
         let codex_home = config.codex_home();
         let cwd_filter = filter_cwd.map(Path::to_path_buf);
+        let mut fallback_reason = state_db_ctx.is_none().then_some("db_unavailable");
         if state_db_ctx.is_some() {
             let mut db_cursor = cursor.cloned();
             loop {
@@ -619,6 +624,7 @@ impl RolloutRecorder {
                 )
                 .await
                 else {
+                    fallback_reason = Some("db_error");
                     break;
                 };
                 if let Some(path) =
@@ -628,10 +634,14 @@ impl RolloutRecorder {
                 }
                 db_cursor = db_page.next_anchor.map(Into::into);
                 if db_cursor.is_none() {
+                    fallback_reason = Some("missing_row");
                     break;
                 }
             }
         }
+        if let Some(reason) = fallback_reason {
+            codex_state::record_fallback(None, "find_latest_thread_path", reason);
+        }
 
         let mut cursor = cursor.cloned();
         loop {

codex-rs/rollout/src/sqlite_metrics.rs

@@ -0,0 +1,40 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use codex_otel::ORIGINATOR_TAG;
+use codex_otel::bounded_originator_tag_value;
+use codex_state::DbTelemetry;
+use codex_state::DbTelemetryHandle;
+
+struct OtelSqliteTelemetry {
+    metrics: codex_otel::MetricsClient,
+    originator: &'static str,
+}
+
+impl DbTelemetry for OtelSqliteTelemetry {
+    fn counter(&self, name: &str, inc: i64, tags: &[(&str, &str)]) {
+        let tags = with_originator(tags, self.originator);
+        let _ = self.metrics.counter(name, inc, &tags);
+    }
+
+    fn record_duration(&self, name: &str, duration: Duration, tags: &[(&str, &str)]) {
+        let tags = with_originator(tags, self.originator);
+        let _ = self.metrics.record_duration(name, duration, &tags);
+    }
+}
+
+pub(crate) fn recorder(metrics: codex_otel::MetricsClient, originator: &str) -> DbTelemetryHandle {
+    Arc::new(OtelSqliteTelemetry {
+        metrics,
+        originator: bounded_originator_tag_value(originator),
+    })
+}
+
+fn with_originator<'a>(
+    tags: &[(&'a str, &'a str)],
+    originator: &'static str,
+) -> Vec<(&'a str, &'a str)> {
+    let mut tags = tags.to_vec();
+    tags.push((ORIGINATOR_TAG, originator));
+    tags
+}

codex-rs/rollout/src/state_db.rs

@@ -4,6 +4,7 @@ use crate::list::Cursor;
 use crate::list::SortDirection;
 use crate::list::ThreadSortKey;
 use crate::metadata;
+use crate::sqlite_metrics;
 use chrono::DateTime;
 use chrono::Utc;
 use codex_protocol::ThreadId;
@@ -115,6 +116,25 @@ async fn try_init_with_roots_inner(
                     sqlite_home.display()
                 )
             })?;
+    let backfill_gate_started = Instant::now();
+    let backfill_gate_result = wait_for_backfill_gate(
+        runtime.as_ref(),
+        codex_home.as_path(),
+        default_model_provider_id.as_str(),
+        backfill_lease_seconds,
+    )
+    .await;
+    codex_state::record_backfill_gate(None, backfill_gate_started.elapsed(), &backfill_gate_result);
+    backfill_gate_result?;
+    Ok(runtime)
+}
+
+async fn wait_for_backfill_gate(
+    runtime: &codex_state::StateRuntime,
+    codex_home: &Path,
+    default_model_provider_id: &str,
+    backfill_lease_seconds: Option<i64>,
+) -> anyhow::Result<()> {
     let wait_started = Instant::now();
     let mut reported_wait = false;
     loop {
@@ -125,24 +145,19 @@ async fn try_init_with_roots_inner(
             )
         })?;
         if backfill_state.status == codex_state::BackfillStatus::Complete {
-            return Ok(runtime);
+            return Ok(());
         }
 
         if let Some(backfill_lease_seconds) = backfill_lease_seconds {
             metadata::backfill_sessions_with_lease(
-                runtime.as_ref(),
-                codex_home.as_path(),
-                default_model_provider_id.as_str(),
+                runtime,
+                codex_home,
+                default_model_provider_id,
                 backfill_lease_seconds,
             )
             .await;
         } else {
-            metadata::backfill_sessions(
-                runtime.as_ref(),
-                codex_home.as_path(),
-                default_model_provider_id.as_str(),
-            )
-            .await;
+            metadata::backfill_sessions(runtime, codex_home, default_model_provider_id).await;
         }
         let backfill_state = runtime.get_backfill_state().await.map_err(|err| {
             anyhow::anyhow!(
@@ -151,7 +166,7 @@ async fn try_init_with_roots_inner(
             )
         })?;
         if backfill_state.status == codex_state::BackfillStatus::Complete {
-            return Ok(runtime);
+            return Ok(());
         }
         if wait_started.elapsed() >= STARTUP_BACKFILL_WAIT_TIMEOUT {
             return Err(anyhow::anyhow!(
@@ -195,17 +210,32 @@ fn emit_startup_warning(message: &str) {
 pub async fn get_state_db(config: &impl RolloutConfigView) -> Option<StateDbHandle> {
     let state_path = codex_state::state_db_path(config.sqlite_home());
     if !tokio::fs::try_exists(&state_path).await.unwrap_or(false) {
+        codex_state::record_fallback(None, "get_state_db", "db_unavailable");
         return None;
     }
-    let runtime = codex_state::StateRuntime::init(
+    let runtime = match codex_state::StateRuntime::init(
         config.sqlite_home().to_path_buf(),
         config.model_provider_id().to_string(),
     )
     .await
-    .ok()?;
+    {
+        Ok(runtime) => runtime,
+        Err(_) => {
+            codex_state::record_fallback(None, "get_state_db", "db_error");
+            return None;
+        }
+    };
     require_backfill_complete(runtime, config.sqlite_home()).await
 }
 
+/// Build a SQLite telemetry recorder backed by an OTEL metrics client.
+pub fn sqlite_telemetry_recorder(
+    metrics: codex_otel::MetricsClient,
+    originator: &str,
+) -> codex_state::DbTelemetryHandle {
+    sqlite_metrics::recorder(metrics, originator)
+}
+
 async fn require_backfill_complete(
     runtime: StateDbHandle,
     codex_home: &Path,
@@ -218,6 +248,7 @@ async fn require_backfill_complete(
                 codex_home.display(),
                 state.status.as_str()
             );
+            codex_state::record_fallback(None, "get_state_db", "backfill_incomplete");
             None
         }
         Err(err) => {
@@ -225,6 +256,7 @@ async fn require_backfill_complete(
                 "failed to read backfill state at {}: {err}",
                 codex_home.display()
             );
+            codex_state::record_fallback(None, "get_state_db", "db_error");
             None
         }
     }