fix(session-ingest): unblock catch-up convergence wedged by oversized reads and all-skipped refreshes

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::<String>(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::<String>(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 [

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;

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");
+}

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::<i64>(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?;