Add telemetry for SQLite log persistence (#40726)

## Why

SQLite log persistence needs visibility into batch size, write latency,
failures, and entries dropped before they reach the writer. Exporter
diagnostics must not feed back into the SQLite log sink and keep metric
exports active indefinitely.

## What changed

- Record write count, duration, batch bytes, entry count, largest-entry size,
  and queue drops with bounded outcome tags.
- Add explicit byte histogram boundaries for log batch metrics.
- Filter OTLP transport noise and the benign unsolicited HTTP/2 PING warning
  from SQLite log persistence to prevent export cycles while retaining other
  HTTP/2 warnings.

## Testing

- Cover metric values, tags, queue-drop reasons, and histogram boundaries.
- Verify successful and failed OTLP HTTP and gRPC exports become idle instead
  of generating recurring SQLite log-write metrics.

GitOrigin-RevId: a63224ae15945fe84adc977a8b7d55bb47a4de53

codex-rs/app-server/tests/suite/logging.rs

@@ -30,7 +30,14 @@ use core_test_support::skip_if_no_network;
 use pretty_assertions::assert_eq;
 use serde_json::Value;
 use serde_json::json;
+use std::collections::BTreeMap;
 use tempfile::TempDir;
+use test_case::test_case;
+use tokio::io::AsyncReadExt;
+use tokio::io::AsyncWriteExt;
+use tokio::net::TcpListener;
+use tokio::net::TcpStream;
+use tokio::sync::watch;
 use tokio::time::Duration;
 use tokio::time::timeout;
 use wiremock::Mock;
@@ -263,6 +270,308 @@ fn standalone_app_server_emits_json_info_events() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn sqlite_log_metrics_exports_do_not_create_log_cycles() -> Result<()> {
+    let quiet_period = codex_state::log_db::LogSinkQueueConfig::default().flush_interval
+        + Duration::from_millis(600);
+    let export_timeout = quiet_period * 3;
+
+    for status in [200, 500] {
+        let collector = MockServer::start().await;
+        Mock::given(method("POST"))
+            .and(path("/v1/metrics"))
+            .respond_with(ResponseTemplate::new(status))
+            .mount(&collector)
+            .await;
+
+        let codex_home = TempDir::new()?;
+        let endpoint = format!("{}/v1/metrics", collector.uri());
+        std::fs::write(
+            codex_home.path().join("config.toml"),
+            format!(
+                "[analytics]\nenabled = true\n\n[otel.metrics_exporter.otlp-http]\nendpoint = {endpoint:?}\nprotocol = \"json\"\n"
+            ),
+        )?;
+
+        let _app_server = TestAppServer::builder()
+            .with_codex_home(codex_home.path())
+            .with_env_overrides(&[
+                ("OTEL_METRIC_EXPORT_INTERVAL", Some("200")),
+                (
+                    codex_app_server_transport::REMOTE_CONTROL_DISABLED_ENV_VAR,
+                    Some("1"),
+                ),
+            ])
+            .build_initialized_with_timeout(export_timeout)
+            .await?;
+
+        let exports = timeout(export_timeout, async {
+            loop {
+                let requests = collector.received_requests().await.unwrap_or_default();
+                if requests.iter().any(|request| {
+                    String::from_utf8_lossy(&request.body).contains(codex_state::LOG_WRITE_METRIC)
+                }) {
+                    break requests.len();
+                }
+                tokio::time::sleep(Duration::from_millis(25)).await;
+            }
+        })
+        .await?;
+
+        let requests = timeout(export_timeout, async {
+            let mut observed_exports = exports;
+            loop {
+                tokio::time::sleep(quiet_period).await;
+                let requests = collector.received_requests().await.unwrap_or_default();
+                if requests.len() == observed_exports {
+                    break requests;
+                }
+                observed_exports = requests.len();
+            }
+        })
+        .await
+        .with_context(|| {
+            format!("OTLP HTTP {status} exports did not stop after the application became idle")
+        })?;
+
+        let expected_metrics = [
+            (codex_state::LOG_WRITE_METRIC, "sum"),
+            (codex_state::LOG_WRITE_DURATION_METRIC, "histogram"),
+            (codex_state::LOG_WRITE_BYTES_METRIC, "histogram"),
+            (codex_state::LOG_WRITE_ENTRIES_METRIC, "histogram"),
+            (codex_state::LOG_WRITE_MAX_ENTRY_BYTES_METRIC, "histogram"),
+        ];
+        let payloads = requests
+            .iter()
+            .map(|request| serde_json::from_slice::<Value>(&request.body))
+            .collect::<serde_json::Result<Vec<_>>>()?;
+        let mut observed_metrics = BTreeMap::<&str, (u64, f64)>::new();
+        for (name, kind) in expected_metrics {
+            let points = payloads
+                .iter()
+                .flat_map(|payload| payload["resourceMetrics"].as_array().into_iter().flatten())
+                .flat_map(|resource| resource["scopeMetrics"].as_array().into_iter().flatten())
+                .flat_map(|scope| scope["metrics"].as_array().into_iter().flatten())
+                .filter(|metric| metric["name"].as_str() == Some(name))
+                .flat_map(|metric| metric[kind]["dataPoints"].as_array().into_iter().flatten())
+                .collect::<Vec<_>>();
+            assert!(
+                !points.is_empty(),
+                "metric {name} must reach the collector as a {kind}"
+            );
+
+            for point in points {
+                let attributes: BTreeMap<&str, &str> = point["attributes"]
+                    .as_array()
+                    .context("metric data points must include attributes")?
+                    .iter()
+                    .filter_map(|attribute| {
+                        Some((
+                            attribute["key"].as_str()?,
+                            attribute["value"]["stringValue"].as_str()?,
+                        ))
+                    })
+                    .collect();
+                assert_eq!(
+                    attributes,
+                    BTreeMap::from([
+                        ("error", "none"),
+                        ("originator", "codex-app-server"),
+                        ("status", "success"),
+                    ]),
+                    "metric {name} must preserve its production dimensions"
+                );
+
+                let count_field = if kind == "sum" { "asInt" } else { "count" };
+                let count = point[count_field]
+                    .as_u64()
+                    .or_else(|| point[count_field].as_str()?.parse().ok())
+                    .with_context(|| format!("metric {name} must include a valid {count_field}"))?;
+                let value = if kind == "sum" {
+                    count as f64
+                } else {
+                    point["sum"]
+                        .as_f64()
+                        .with_context(|| format!("metric {name} must include its sample sum"))?
+                };
+                let observed = observed_metrics.entry(name).or_default();
+                observed.0 += count;
+                observed.1 += value;
+
+                if matches!(
+                    name,
+                    codex_state::LOG_WRITE_BYTES_METRIC
+                        | codex_state::LOG_WRITE_MAX_ENTRY_BYTES_METRIC
+                ) {
+                    let bounds = point["explicitBounds"]
+                        .as_array()
+                        .context("byte histograms must export explicit bucket bounds")?;
+                    assert_eq!(bounds.first().and_then(Value::as_f64), Some(128.0));
+                    assert_eq!(bounds.last().and_then(Value::as_f64), Some(16_777_216.0));
+                }
+            }
+        }
+
+        let write_count = observed_metrics[codex_state::LOG_WRITE_METRIC].0;
+        assert!(write_count > 0, "at least one SQLite batch must be written");
+        for (name, _) in expected_metrics {
+            assert_eq!(
+                observed_metrics[name].0, write_count,
+                "metric {name} must record exactly one sample per SQLite write"
+            );
+        }
+
+        let state = codex_state::StateRuntime::init(
+            codex_state::SqliteConfig::new_for_testing(codex_home.path().abs()),
+            "test-provider".to_string(),
+        )
+        .await?;
+        let persisted_logs = state.query_logs(&codex_state::LogQuery::default()).await?;
+        assert_eq!(
+            observed_metrics[codex_state::LOG_WRITE_ENTRIES_METRIC].1,
+            persisted_logs.len() as f64,
+            "exported batch sizes must equal the number of persisted SQLite log rows"
+        );
+        assert!(
+            observed_metrics[codex_state::LOG_WRITE_BYTES_METRIC].1
+                >= observed_metrics[codex_state::LOG_WRITE_MAX_ENTRY_BYTES_METRIC].1,
+            "each batch's largest entry cannot exceed the total batch size"
+        );
+    }
+
+    Ok(())
+}
+
+/// Exporting SQLite metrics must eventually stop producing new SQLite metrics,
+/// including when the collector uses HTTP/2 or rejects the gRPC request.
+#[test_case("0"; "success")]
+#[test_case("14"; "unavailable")]
+#[tokio::test]
+async fn sqlite_log_metrics_grpc_exports_do_not_create_log_cycles(grpc_status: &str) -> Result<()> {
+    let quiet_period = codex_state::log_db::LogSinkQueueConfig::default().flush_interval
+        + Duration::from_millis(600);
+    let export_timeout = quiet_period * 3;
+    let (export_count, mut exports) = watch::channel(/*init*/ 0_usize);
+    let (ping_ack_count, mut ping_acknowledgements) = watch::channel(/*init*/ 0_usize);
+    let collector = MockServer::builder()
+        .disable_request_recording()
+        .start()
+        .await;
+    // An uncompressed gRPC frame containing an empty ExportMetricsServiceResponse.
+    // Tonic also accepts grpc-status in the initial response headers.
+    let response = ResponseTemplate::new(/*s*/ 200)
+        .set_body_raw(vec![0; 5], "application/grpc")
+        .insert_header("grpc-status", grpc_status);
+    Mock::given(method("POST"))
+        .and(path(
+            "/opentelemetry.proto.collector.metrics.v1.MetricsService/Export",
+        ))
+        .respond_with(move |request: &wiremock::Request| {
+            let metric_name = codex_state::LOG_WRITE_METRIC.as_bytes();
+            if request
+                .body
+                .windows(metric_name.len())
+                .any(|bytes| bytes == metric_name)
+            {
+                export_count.send_modify(|count| *count += 1);
+            }
+            response.clone()
+        })
+        .mount(&collector)
+        .await;
+
+    let proxy = TcpListener::bind("127.0.0.1:0").await?;
+    let endpoint = format!("http://{}", proxy.local_addr()?);
+    let collector_address = *collector.address();
+    let _proxy = tokio::spawn(async move {
+        let (downstream, _) = proxy.accept().await?;
+        let upstream = TcpStream::connect(collector_address).await?;
+        let (mut downstream_reader, mut downstream_writer) = downstream.into_split();
+        let (mut upstream_reader, mut upstream_writer) = upstream.into_split();
+
+        let requests = tokio::io::copy(&mut downstream_reader, &mut upstream_writer);
+        let responses = async move {
+            loop {
+                let mut header = [0; 9];
+                if upstream_reader.read_exact(&mut header).await.is_err() {
+                    break;
+                }
+                let length = (usize::from(header[0]) << 16)
+                    | (usize::from(header[1]) << 8)
+                    | usize::from(header[2]);
+                let mut payload = vec![0; length];
+                upstream_reader.read_exact(&mut payload).await?;
+                downstream_writer.write_all(&header).await?;
+                downstream_writer.write_all(&payload).await?;
+
+                if header[3] == 0 {
+                    // A valid but unsolicited HTTP/2 PING ACK makes h2 emit a
+                    // warning from its separate connection-driving task.
+                    downstream_writer
+                        .write_all(&[0, 0, 8, 6, 1, 0, 0, 0, 0, 1, 2, 3, 4, 5, 6, 7, 8])
+                        .await?;
+                    ping_ack_count.send_modify(|count| *count += 1);
+                }
+            }
+            Ok::<(), std::io::Error>(())
+        };
+
+        tokio::try_join!(requests, responses)?;
+        Ok::<(), std::io::Error>(())
+    });
+
+    let codex_home = TempDir::new()?;
+    std::fs::write(
+        codex_home.path().join("config.toml"),
+        format!(
+            "[analytics]\nenabled = true\n\n[otel.metrics_exporter.otlp-grpc]\nendpoint = {endpoint:?}\n"
+        ),
+    )?;
+    let _app_server = TestAppServer::builder()
+        .with_codex_home(codex_home.path())
+        .with_env_overrides(&[
+            ("OTEL_METRIC_EXPORT_INTERVAL", Some("200")),
+            (
+                codex_app_server_transport::REMOTE_CONTROL_DISABLED_ENV_VAR,
+                Some("1"),
+            ),
+        ])
+        .build_initialized_with_timeout(export_timeout)
+        .await?;
+
+    timeout(export_timeout, exports.wait_for(|count| *count > 0))
+        .await
+        .context("collector never received a SQLite log-write metric")??;
+    timeout(
+        export_timeout,
+        ping_acknowledgements.wait_for(|count| *count > 0),
+    )
+    .await
+    .context("collector response never produced an unsolicited HTTP/2 PING acknowledgment")??;
+    let first_count = *exports.borrow_and_update();
+
+    // Allow delayed startup batches, but require silence longer than one SQLite
+    // flush interval plus one metric-export interval. Do not retain request bodies
+    // or keep resetting the overall deadline when a broken exporter feeds itself.
+    let quiescence = timeout(export_timeout, async {
+        loop {
+            match timeout(quiet_period, exports.changed()).await {
+                Err(_) => return Ok::<(), watch::error::RecvError>(()),
+                Ok(result) => result?,
+            }
+        }
+    })
+    .await;
+    let final_count = *exports.borrow();
+    quiescence.with_context(|| {
+        format!(
+            "OTLP gRPC status {grpc_status} did not become idle: SQLite log-write exports increased from {first_count} to {final_count}"
+        )
+    })??;
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn app_server_emits_structured_tool_call_timing_event() -> Result<()> {
     skip_if_no_network!(Ok(()));