From 5eb4eb6d140e6262e536664146854e971230b4b6 Mon Sep 17 00:00:00 2001 From: spa5k <79936503+spa5k@users.noreply.github.com> Date: Thu, 8 Oct 2026 16:02:57 +0530 Subject: [PATCH 1/2] fix(global-db): paginate unoffered source-record reads during rewrite completion Rewrite completion materialized every unoffered observation id for a source in one statement. The exact-SQL engine refuses queries that materialize more rows than its hard cap, and one long host session can hold more unoffered records than that cap, so completing that source's rewrite failed deterministically. Host catch-up mapped the failure to a retryable storage outcome and re-attempted forever: every pass re-read the full host transcript corpus, re-processed collision skips, and never converged, which also kept the retained session authority unmounted. Read the unoffered ids in ordered keyset pages inside the completion transaction instead. Retiring never touches observation_source_presence, so the cursor stays stable across pages; the settle deletes remain the authoritative cleanup. Adds a regression test that seeds more unoffered records than the engine cap and fails against the unbounded read with the production error. --- .../src/registered_source_rewrite.rs | 78 ++++++++++----- crates/tracedecay-global-db/src/tests.rs | 2 + .../src/tests/source_rewrite.rs | 94 +++++++++++++++++++ 3 files changed, 151 insertions(+), 23 deletions(-) create mode 100644 crates/tracedecay-global-db/src/tests/source_rewrite.rs 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"); +} From e841b14fb1c350f0a71e3602dd4cd8d8dc765e0c Mon Sep 17 00:00:00 2001 From: spa5k <79936503+spa5k@users.noreply.github.com> Date: Thu, 8 Oct 2026 16:03:09 +0530 Subject: [PATCH 2/2] fix(session-temporal): accept refresh frontiers whose observations were all skipped Candidate-generation validation refused any frontier advance that produced no canonical message outputs. A refresh whose every new observation was deterministically skipped (recorded effects with zero outputs, e.g. host non-conversational records) legitimately produces no outputs, so activating that generation failed forever and the temporal refresh worker retried the same session without bound, stalling the convergence queue behind it. Keep the refusal only for an advance that processed no observation at all, detected via effects recorded past the base frontier. A frontier that moved over all-skipped observations now activates normally. --- .../src/rebuild.rs | 47 +++++++++++++++++-- 1 file changed, 43 insertions(+), 4 deletions(-) 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?;