diff --git a/codex-rs/thread-store/src/local/thread_history_materialization.rs b/codex-rs/thread-store/src/local/thread_history_materialization.rs index 58ea278803..34eeede51b 100644 --- a/codex-rs/thread-store/src/local/thread_history_materialization.rs +++ b/codex-rs/thread-store/src/local/thread_history_materialization.rs @@ -16,6 +16,9 @@ use super::thread_history::RolloutProjectionStep; use crate::ThreadStoreError; use crate::ThreadStoreResult; +const SQLITE_PROJECTION_METRIC: &str = "codex.thread_history.sqlite_projection"; +const SQLITE_PROJECTION_ANOMALY_METRIC: &str = "codex.thread_history.sqlite_projection.anomaly"; + pub(super) async fn materialize_to_sqlite( store: &LocalThreadStore, thread_id: ThreadId, @@ -24,6 +27,16 @@ pub(super) async fn materialize_to_sqlite( if store.state_db.is_none() { return Ok(()); } + let result = materialize_to_sqlite_with_state_db(store, thread_id, rollout_path).await; + record_projection_outcome(&result); + result +} + +async fn materialize_to_sqlite_with_state_db( + store: &LocalThreadStore, + thread_id: ThreadId, + rollout_path: &Path, +) -> ThreadStoreResult<()> { let projection_state = super::thread_history::projection_state(store, thread_id).await?; let start_offset = projection_state .as_ref() @@ -116,10 +129,7 @@ async fn read_projection_steps( let mut projections = Vec::new(); let mut next_ordinal = expected_ordinal; let mut next_offset = start_offset; - let mut pending_rejected_line_count = 0; let mut line_start_offset = start_offset; - // Keep rejected lines pending until a later valid ordinal proves whether they consumed history. - // This lets a same-ordinal retry replace a failed write without advancing only one checkpoint. for line_bytes in bytes[..complete_byte_count].split_inclusive(|byte| *byte == b'\n') { let line_end_offset = line_start_offset .checked_add(u64::try_from(line_bytes.len()).map_err(|_| { @@ -131,9 +141,7 @@ async fn read_projection_steps( message: "durable rollout byte offset overflow".to_string(), })?; if line_bytes.iter().all(u8::is_ascii_whitespace) { - if pending_rejected_line_count == 0 { - next_offset = line_end_offset; - } + next_offset = line_end_offset; line_start_offset = line_end_offset; continue; } @@ -147,16 +155,17 @@ async fn read_projection_steps( line_end_byte_offset = line_end_offset, expected_ordinal = next_ordinal, error = %err, - "deferring rejected rollout line until a later ordinal resolves it" + "skipping malformed rollout line during projection" ); - pending_rejected_line_count += 1; + record_projection_anomaly(ProjectionAnomaly::MalformedJson); + next_offset = line_end_offset; line_start_offset = line_end_offset; continue; } }; - let value_ordinal = value.get("ordinal").and_then(serde_json::Value::as_u64); + let raw_ordinal = value.get("ordinal").and_then(serde_json::Value::as_u64); let line = match codex_rollout::decode_rollout_line(value) { - Ok(line) => Some(line), + Ok(line) => line, Err(err) => { warn!( thread_id = %thread_id, @@ -164,51 +173,50 @@ async fn read_projection_steps( line_start_byte_offset = line_start_offset, line_end_byte_offset = line_end_offset, expected_ordinal = next_ordinal, - line_ordinal = ?value_ordinal, + line_ordinal = ?raw_ordinal, error = %err, - "deferring unknown rollout line until a later ordinal resolves it" + "skipping unknown rollout line during projection" ); - None - } - }; - let ordinal = match line - .as_ref() - .and_then(|line| line.ordinal) - .or(value_ordinal) - { - Some(ordinal) => ordinal, - None if line.is_none() => { - pending_rejected_line_count += 1; + record_projection_anomaly(ProjectionAnomaly::UnknownLine); + // Unknown records only advance the byte checkpoint. A writer that could not + // decode this line may have reused its ordinal, so let the next decoded line + // resolve reuse versus a gap. + next_offset = line_end_offset; line_start_offset = line_end_offset; continue; } + }; + let ordinal = match line.ordinal { + Some(ordinal) => ordinal, None => { - return Err(ThreadStoreError::Internal { - message: format!( - "paginated rollout line for {thread_id} is missing an ordinal" - ), - }); + warn!( + thread_id = %thread_id, + rollout_path = %rollout_path.display(), + line_start_byte_offset = line_start_offset, + line_end_byte_offset = line_end_offset, + expected_ordinal = next_ordinal, + "skipping paginated rollout line without an ordinal" + ); + record_projection_anomaly(ProjectionAnomaly::MissingOrdinal); + next_offset = line_end_offset; + line_start_offset = line_end_offset; + continue; } }; if ordinal < next_ordinal { - return Err(ThreadStoreError::Internal { - message: format!( - "thread history projection for {thread_id} expected ordinal {next_ordinal}, got {ordinal}" - ), - }); - } - let Some(line) = line else { - pending_rejected_line_count += 1; + warn!( + thread_id = %thread_id, + rollout_path = %rollout_path.display(), + line_start_byte_offset = line_start_offset, + line_end_byte_offset = line_end_offset, + expected_ordinal = next_ordinal, + line_ordinal = ordinal, + "skipping duplicate or regressed rollout ordinal during projection" + ); + record_projection_anomaly(ProjectionAnomaly::DuplicateOrRegressedOrdinal); + next_offset = line_end_offset; line_start_offset = line_end_offset; continue; - }; - let skipped_ordinal_count = ordinal - next_ordinal; - if skipped_ordinal_count > pending_rejected_line_count { - return Err(ThreadStoreError::Internal { - message: format!( - "thread history projection for {thread_id} expected ordinal {next_ordinal}, got {ordinal}; {pending_rejected_line_count} rejected rollout lines cannot cover that gap" - ), - }); } let is_inherited_subagent_history = subagent_history_start_ordinal.is_some_and(|start| ordinal < start); @@ -235,9 +243,21 @@ async fn read_projection_steps( expected_ordinal = next_ordinal, line_ordinal = ordinal, error = %err, - "deferring rollout line with invalid timestamp until a later ordinal resolves it" + "skipping rollout line with invalid timestamp during projection" ); - pending_rejected_line_count += 1; + record_projection_anomaly(ProjectionAnomaly::InvalidTimestamp); + let end_ordinal_exclusive = + ordinal + .checked_add(1) + .ok_or_else(|| ThreadStoreError::Internal { + message: "rollout ordinal exceeds SQLite integer range".to_string(), + })?; + projections.push(RolloutProjectionStep::SkippedOrdinalRange { + start_ordinal: next_ordinal, + end_ordinal_exclusive, + }); + next_ordinal = end_ordinal_exclusive; + next_offset = line_end_offset; line_start_offset = line_end_offset; continue; } @@ -245,7 +265,7 @@ async fn read_projection_steps( } else { None }; - if skipped_ordinal_count > 0 { + if ordinal > next_ordinal { warn!( thread_id = %thread_id, rollout_path = %rollout_path.display(), @@ -255,14 +275,14 @@ async fn read_projection_steps( line_ordinal = ordinal, skipped_ordinal_start = next_ordinal, skipped_ordinal_end_exclusive = ordinal, - "skipping rollout ordinal range after rejected lines" + "skipping missing rollout ordinal range during projection" ); + record_projection_anomaly(ProjectionAnomaly::ForwardOrdinalGap); projections.push(RolloutProjectionStep::SkippedOrdinalRange { start_ordinal: next_ordinal, end_ordinal_exclusive: ordinal, }); } - pending_rejected_line_count = 0; let next_line_ordinal = ordinal .checked_add(1) @@ -289,6 +309,52 @@ async fn read_projection_steps( Ok((projections, next_offset)) } +#[derive(Clone, Copy)] +enum ProjectionAnomaly { + MalformedJson, + UnknownLine, + MissingOrdinal, + DuplicateOrRegressedOrdinal, + ForwardOrdinalGap, + InvalidTimestamp, +} + +impl ProjectionAnomaly { + fn tag(self) -> &'static str { + match self { + Self::MalformedJson => "malformed_json", + Self::UnknownLine => "unknown_line", + Self::MissingOrdinal => "missing_ordinal", + Self::DuplicateOrRegressedOrdinal => "duplicate_or_regressed_ordinal", + Self::ForwardOrdinalGap => "forward_ordinal_gap", + Self::InvalidTimestamp => "invalid_timestamp", + } + } +} + +fn record_projection_outcome(result: &ThreadStoreResult<()>) { + let Some(metrics) = codex_otel::global() else { + return; + }; + let outcome = if result.is_ok() { "success" } else { "error" }; + let _ = metrics.counter( + SQLITE_PROJECTION_METRIC, + /*inc*/ 1, + &[("outcome", outcome)], + ); +} + +fn record_projection_anomaly(anomaly: ProjectionAnomaly) { + let Some(metrics) = codex_otel::global() else { + return; + }; + let _ = metrics.counter( + SQLITE_PROJECTION_ANOMALY_METRIC, + /*inc*/ 1, + &[("kind", anomaly.tag())], + ); +} + fn thread_store_io_error(err: std::io::Error) -> ThreadStoreError { ThreadStoreError::Internal { message: err.to_string(), diff --git a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs index 373419c16e..1a5a43d9b2 100644 --- a/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs +++ b/codex-rs/thread-store/src/local/thread_history_materialization_tests.rs @@ -2020,7 +2020,7 @@ async fn catch_up_preserves_trailing_partial_line_boundaries() { } #[tokio::test] -async fn catch_up_rejects_invalid_complete_suffixes_without_advancing_state() { +async fn catch_up_skips_invalid_complete_suffixes_and_projects_later_history() { let cases = [ ( "missing ordinal", @@ -2028,28 +2028,28 @@ async fn catch_up_rejects_invalid_complete_suffixes_without_advancing_state() { "{}\n", rollout_line(/*ordinal*/ None, turn_started("turn-1")) ), + 1, + Vec::new(), ), ( "duplicate ordinal", format!( - "{}\n{}\n", + "{}\n{}\n{}\n", rollout_line(Some(1), turn_started("turn-1")), - rollout_line(Some(1), turn_started("turn-2")), + rollout_line(Some(1), turn_started("skipped-turn")), + rollout_line(Some(2), turn_started("turn-2")), ), + 3, + vec!["turn-1", "turn-2"], ), ( - "out of order ordinal", + "forward ordinal gap", format!("{}\n", rollout_line(Some(2), turn_started("turn-1"))), - ), - ( - "gap larger than rejected prefix", - format!( - "{{not json}}\n{}\n", - rollout_line(Some(3), turn_started("turn-1")), - ), + 3, + vec!["turn-1"], ), ]; - for (name, suffix) in cases { + for (name, suffix, expected_next_ordinal, expected_turn_ids) in cases { let home = TempDir::new().expect("temp dir"); let store = projection_store(home.path()).await; let thread_id = ThreadId::default(); @@ -2073,26 +2073,22 @@ async fn catch_up_rejects_invalid_complete_suffixes_without_advancing_state() { super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path()) .await - .expect_err(name); + .expect(name); + let expected_offset = before.0 + i64::try_from(suffix.len()).expect("suffix byte count"); assert_eq!( projection_state(&pool, thread_id).await, - before, - "{name} should not advance projection state" + (expected_offset, expected_next_ordinal), + "{name} should advance projection state" ); - let counts = sqlx::query_as::<_, (i64, i64)>( - r#" -SELECT - (SELECT COUNT(*) FROM thread_turns WHERE thread_id = ?), - (SELECT COUNT(*) FROM thread_items WHERE thread_id = ?) - "#, + let turn_ids = sqlx::query_scalar::<_, String>( + "SELECT turn_id FROM thread_turns WHERE thread_id = ? ORDER BY rollout_ordinal", ) .bind(thread_id.to_string()) - .bind(thread_id.to_string()) - .fetch_one(&pool) + .fetch_all(&pool) .await - .expect("read projected row counts"); - assert_eq!(counts, (0, 0), "{name} should not project rows"); + .expect("read projected turns"); + assert_eq!(turn_ids, expected_turn_ids, "{name}"); } } @@ -2178,7 +2174,7 @@ async fn sqlite_failure_does_not_fail_durable_jsonl_write() { } #[tokio::test] -async fn blank_and_rejected_rollout_lines_do_not_poison_projection() { +async fn blank_and_rejected_rollout_lines_advance_projection() { let home = TempDir::new().expect("temp dir"); let store = projection_store(home.path()).await; let thread_id = ThreadId::default(); @@ -2202,15 +2198,21 @@ async fn blank_and_rejected_rollout_lines_do_not_poison_projection() { .append(true) .open(rollout_path.as_path()) .expect("open rollout for rejected line"); - file.write_all(b"\n \t\r\n{not json}\n\xff\n") + file.write_all(b"{not json}\n\n") .expect("append blank and rejected lines"); file.flush().expect("flush rejected line"); + let rejected_tail_len = i64::try_from( + fs::metadata(rollout_path.as_path()) + .expect("rollout metadata") + .len(), + ) + .expect("rollout length"); super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path()) .await - .expect("leave rejected tail pending"); + .expect("skip rejected tail"); assert_eq!( projection_state(&pool, thread_id).await, - (before.0 + 5, before.1) + (rejected_tail_len, before.1) ); let recorder = store @@ -2248,7 +2250,7 @@ async fn blank_and_rejected_rollout_lines_do_not_poison_projection() { } #[tokio::test] -async fn unprojectable_rollout_lines_wait_for_later_ordinals() { +async fn unknown_rollout_lines_leave_ordinals_pending_for_valid_retries() { let unknown_line = |ordinal| { format!( concat!( @@ -2258,76 +2260,67 @@ async fn unprojectable_rollout_lines_wait_for_later_ordinals() { ordinal = ordinal ) }; - let cases = [ - ("unknown payload", unknown_line(1), unknown_line(2)), - ( - "structurally invalid JSON", - "{}".to_string(), - "null".to_string(), - ), - ]; - for (name, pending_line, skipped_line) in cases { - let home = TempDir::new().expect("temp dir"); - let store = projection_store(home.path()).await; - let thread_id = ThreadId::default(); - create_paginated_thread(&store, thread_id).await; - store - .persist_thread(thread_id, PersistContext::Standard) - .await - .expect("persist session metadata"); - - let pool = codex_state::open_thread_history_db( - &codex_state::SqliteConfig::new_for_testing(home.path().abs()), - ) + let home = TempDir::new().expect("temp dir"); + let store = projection_store(home.path()).await; + let thread_id = ThreadId::default(); + create_paginated_thread(&store, thread_id).await; + store + .persist_thread(thread_id, PersistContext::Standard) .await - .expect("open thread history db"); - let before = projection_state(&pool, thread_id).await; - let rollout_path = store - .live_rollout_path(thread_id) - .await - .expect("rollout path"); - append_suffix(rollout_path.as_path(), format!("{pending_line}\n").as_str()); + .expect("persist session metadata"); - super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path()) - .await - .expect("leave unprojectable tail pending"); - assert_eq!(projection_state(&pool, thread_id).await, before, "{name}"); - - append_suffix( - rollout_path.as_path(), - format!( - "{}\n{skipped_line}\n{}\n", - rollout_line(Some(1), turn_started("retry-turn")), - rollout_line(Some(3), turn_started("turn-1")), - ) - .as_str(), - ); - - super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path()) - .await - .expect(name); - - let rollout_len = - i64::try_from(fs::metadata(rollout_path).expect("rollout metadata").len()) - .expect("rollout length"); - assert_eq!( - projection_state(&pool, thread_id).await, - (rollout_len, 4), - "{name}" - ); - let turn_ordinals = sqlx::query_as::<_, (String, i64)>( - "SELECT turn_id, rollout_ordinal FROM thread_turns WHERE thread_id = ? ORDER BY rollout_ordinal", - ) - .bind(thread_id.to_string()) - .fetch_all(&pool) + let pool = codex_state::open_thread_history_db(&codex_state::SqliteConfig::new_for_testing( + home.path().abs(), + )) + .await + .expect("open thread history db"); + let before = projection_state(&pool, thread_id).await; + let rollout_path = store + .live_rollout_path(thread_id) .await - .expect("read projected turns"); - assert_eq!( - turn_ordinals, - vec![("retry-turn".to_string(), 1), ("turn-1".to_string(), 3)], - "{name}" - ); - } + .expect("rollout path"); + let pending_line = unknown_line(1); + append_suffix(rollout_path.as_path(), format!("{pending_line}\n").as_str()); + + super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path()) + .await + .expect("skip unknown tail"); + let expected_pending_offset = + before.0 + i64::try_from(pending_line.len() + 1).expect("pending line byte count"); + assert_eq!( + projection_state(&pool, thread_id).await, + (expected_pending_offset, 1) + ); + + append_suffix( + rollout_path.as_path(), + format!( + "{}\n{}\n{}\n", + rollout_line(Some(1), turn_started("retry-turn")), + unknown_line(2), + rollout_line(Some(3), turn_started("turn-1")), + ) + .as_str(), + ); + + super::materialize_to_sqlite(&store, thread_id, rollout_path.as_path()) + .await + .expect("project valid retry and later history"); + + let rollout_len = i64::try_from(fs::metadata(rollout_path).expect("rollout metadata").len()) + .expect("rollout length"); + assert_eq!(projection_state(&pool, thread_id).await, (rollout_len, 4)); + let turn_ordinals = sqlx::query_as::<_, (String, i64)>( + "SELECT turn_id, rollout_ordinal FROM thread_turns WHERE thread_id = ? ORDER BY rollout_ordinal", + ) + .bind(thread_id.to_string()) + .fetch_all(&pool) + .await + .expect("read projected turns"); + assert_eq!( + turn_ordinals, + vec![("retry-turn".to_string(), 1), ("turn-1".to_string(), 3)] + ); } #[tokio::test]