diff --git a/crates/tracedecay-global-db/src/registered_source_rewrite.rs b/crates/tracedecay-global-db/src/registered_source_rewrite.rs index b59ddf4ecb..2393bc63e3 100644 --- a/crates/tracedecay-global-db/src/registered_source_rewrite.rs +++ b/crates/tracedecay-global-db/src/registered_source_rewrite.rs @@ -15,6 +15,12 @@ use tracedecay_store::{ProjectionStoreError, ProjectionStoreResult}; use super::RegisteredGlobalDb; use super::observation_projection::{ensure_projection_output_state_cache, retire_source_record}; +/// Unoffered-record reads are keyset-paginated at this page size. The exact- +/// SQL engine refuses a query that materializes more rows than its hard cap, +/// and one long host session can hold more unoffered records than that cap, +/// which made rewrite completion fail deterministically for those sources. +const UNOFFERED_RECORD_PAGE_SIZE: i64 = 1_000; + /// One record a source layout offered, under the identity admission derived. pub struct ObservationSourcePresenceV1 { pub observation_id: CanonicalObservationIdV1, @@ -207,31 +213,57 @@ impl RegisteredGlobalDb { .map_err(|error| storage("release source rewrite completion", error))?; return Ok(0); } - let mut rows = transaction - .query( - "SELECT observation_id FROM observation_source_presence - WHERE source_key = ?1 AND generation <> ?2", - params![key.as_str(), generation.as_str()], - ) - .await - .map_err(|error| storage("read unoffered source records", error))?; - let mut unoffered = Vec::new(); - while let Some(row) = rows - .next() - .await - .map_err(|error| storage("read unoffered source records", error))? - { - unoffered.push( - row.get::(0) - .map_err(|error| storage("read unoffered source records", error))?, - ); - } - drop(rows); ensure_projection_output_state_cache(&transaction).await?; let mut retired = 0_u64; - for observation_id in &unoffered { - if retire_source_record(&transaction, observation_id).await? { - retired += 1; + // Keyset-paginate the unoffered ids instead of materializing the whole + // set in one statement: a single rewritten source can hold more + // unoffered records than the engine's per-query materialization cap, + // and one oversized read wedged every completion attempt forever. + // `retire_source_record` never touches `observation_source_presence`, + // so the ordered id cursor stays stable across pages; the settle + // deletes below remain the authoritative cleanup. + let mut after_observation_id = String::new(); + loop { + let mut rows = transaction + .query( + "SELECT observation_id FROM observation_source_presence + WHERE source_key = ?1 AND generation <> ?2 + AND observation_id > ?3 + ORDER BY observation_id + LIMIT ?4", + params![ + key.as_str(), + generation.as_str(), + after_observation_id, + UNOFFERED_RECORD_PAGE_SIZE + ], + ) + .await + .map_err(|error| storage("read unoffered source records", error))?; + let mut page = Vec::new(); + while let Some(row) = rows + .next() + .await + .map_err(|error| storage("read unoffered source records", error))? + { + page.push( + row.get::(0) + .map_err(|error| storage("read unoffered source records", error))?, + ); + } + drop(rows); + if page.is_empty() { + break; + } + let page_len = page.len(); + after_observation_id = page[page_len - 1].clone(); + for observation_id in &page { + if retire_source_record(&transaction, observation_id).await? { + retired += 1; + } + } + if (page_len as i64) < UNOFFERED_RECORD_PAGE_SIZE { + break; } } for sql in [ diff --git a/crates/tracedecay-global-db/src/tests.rs b/crates/tracedecay-global-db/src/tests.rs index dc9488ff86..f9602f7467 100644 --- a/crates/tracedecay-global-db/src/tests.rs +++ b/crates/tracedecay-global-db/src/tests.rs @@ -17,6 +17,8 @@ mod payload_gc_cost; mod schema_install_cancellation; #[cfg(test)] mod session_sync; +#[cfg(test)] +mod source_rewrite; #[cfg(test)] use super::registered_analytics::ANALYTICS_INSERT_ROWS_PER_STATEMENT; diff --git a/crates/tracedecay-global-db/src/tests/source_rewrite.rs b/crates/tracedecay-global-db/src/tests/source_rewrite.rs new file mode 100644 index 0000000000..b99769d831 --- /dev/null +++ b/crates/tracedecay-global-db/src/tests/source_rewrite.rs @@ -0,0 +1,94 @@ +//! Regression tests for registered source-rewrite completion. +//! +//! Rewrite completion retires every record the rewritten source no longer +//! offers. One long host session can hold more unoffered records than the +//! exact-SQL engine's per-query materialization cap, so completion must read +//! them in keyset pages rather than one unbounded statement; the unbounded +//! read failed deterministically for exactly those sources and wedged the +//! host catch-up in a retry loop that never converged. + +use tracedecay_domain::{ + ObservationScopeV1, ObservationSourceGenerationV1, ObservationSourceIdentityV1, SessionId, +}; +use tracedecay_runtime_core::db::engine::params; + +use crate::tests::harness::RegisteredGlobalDbHarness; + +/// More unoffered records than one keyset page *and* more than the engine's +/// per-query materialization cap, so the test fails against the unbounded +/// read and exercises repeated pagination against the fix. +const UNOFFERED_RECORDS: i64 = 10_003; + +#[tokio::test] +async fn rewrite_completion_retires_unoffered_records_beyond_the_query_cap() { + let harness = RegisteredGlobalDbHarness::open("source-rewrite-unoffered-cap").await; + let source = + ObservationSourceIdentityV1::new(SessionId::new("session.rewrite-cap").unwrap()).unwrap(); + let scope = ObservationScopeV1::Profile; + let previous = ObservationSourceGenerationV1::new(1).unwrap(); + let generation = ObservationSourceGenerationV1::new(2).unwrap(); + + let key = serde_json::to_string(&(source.clone(), scope.clone())).unwrap(); + let writer = harness.registered.writer_connection().unwrap(); + for index in 0..UNOFFERED_RECORDS { + writer + .execute( + "INSERT INTO observation_source_presence ( + source_key, observation_id, generation, start_offset + ) VALUES (?1, ?2, ?3, 0)", + params![ + key.as_str(), + format!("sha256:unoffered-{index:06}"), + previous.generation_id().to_string() + ], + ) + .await + .unwrap(); + } + + harness + .registered + .begin_observation_source_rewrite( + &source, + &scope, + previous, + generation.clone(), + 0, + ) + .await + .unwrap(); + + // None of the seeded records are re-offered under the new generation, so + // completion must retire-path all of them. The `observations` table is + // empty, so each retire is a no-op returning false; what matters is that + // the reads stay under the materialization cap and the settle deletes + // still clear the source. + let retired = harness + .registered + .complete_observation_source_rewrite(&source, &scope, generation) + .await + .unwrap_or_else(|error| panic!("rewrite completion must not exceed query caps: {error}")); + assert_eq!(retired, 0); + + let mut remaining = writer + .query( + "SELECT COUNT(*) FROM observation_source_presence WHERE source_key = ?1", + params![key.as_str()], + ) + .await + .unwrap(); + let count = remaining.next().await.unwrap().unwrap(); + let presence: i64 = count.get(0).unwrap(); + assert_eq!(presence, 0, "unoffered presence rows must be settled"); + + let mut pending = writer + .query( + "SELECT COUNT(*) FROM observation_source_rewrites WHERE source_key = ?1", + params![key.as_str()], + ) + .await + .unwrap(); + let row = pending.next().await.unwrap().unwrap(); + let rewrites: i64 = row.get(0).unwrap(); + assert_eq!(rewrites, 0, "completed rewrite row must be removed"); +} diff --git a/crates/tracedecay-session-temporal-store/src/rebuild.rs b/crates/tracedecay-session-temporal-store/src/rebuild.rs index 3f1652516b..aac5d91612 100644 --- a/crates/tracedecay-session-temporal-store/src/rebuild.rs +++ b/crates/tracedecay-session-temporal-store/src/rebuild.rs @@ -233,10 +233,49 @@ pub(super) async fn validate_candidate_frontier( expected.insert(occurrence_id.clone()); } if expected.is_empty() && source_frontier != base_frontier { - return Err(storage_message( - ACTIVATE_OPERATION, - "candidate generation has no canonical message outputs past its base frontier", - )); + // A frontier advance whose every observation was deterministically + // skipped (effects recorded with zero outputs, e.g. host + // non-conversational records) legitimately introduces no canonical + // outputs. Refusing those activations wedged the refresh worker in a + // permanent retry loop. Only an advance that processed no + // observation at all is a validation failure. + let mut processed_any = conn + .query( + "SELECT EXISTS ( + SELECT 1 FROM session_temporal_observation_effects + WHERE session_id = ?1 + AND observation_sequence > ?2 + AND observation_sequence <= ?3 + )", + params![ + session_id, + i64::try_from(base_frontier) + .map_err(|error| storage(ACTIVATE_OPERATION, error))?, + i64::try_from(source_frontier) + .map_err(|error| storage(ACTIVATE_OPERATION, error))? + ], + ) + .await + .map_err(|error| storage(ACTIVATE_OPERATION, error))?; + let processed = match processed_any + .next() + .await + .map_err(|error| storage(ACTIVATE_OPERATION, error))? + { + Some(row) => { + row.get::(0) + .map_err(|error| storage(ACTIVATE_OPERATION, error))? + != 0 + } + None => false, + }; + drop(processed_any); + if !processed { + return Err(storage_message( + ACTIVATE_OPERATION, + "candidate generation advanced its source frontier without processing any observations", + )); + } } require_unsettled_message_ids(conn, session_id, generation, &parent_resolver).await?;