diff --git a/crates/tracedecay-global-db/src/observation_projection.rs b/crates/tracedecay-global-db/src/observation_projection.rs index 743ec5da540..cffc0be8705 100644 --- a/crates/tracedecay-global-db/src/observation_projection.rs +++ b/crates/tracedecay-global-db/src/observation_projection.rs @@ -18,7 +18,7 @@ pub(super) use apply::{ derive_projection_with_alias, retire_source_record, verify_workflow_effects, }; pub use rebuild::{project_observation, project_queued_observations, rebuild_projection}; -pub(crate) use schema::OBSERVATION_PROJECTION_PERFORMANCE_INDEX_SQL; +pub(crate) use schema::{ACTIVE_QUEUE_INDEX_SQL, OBSERVATION_PROJECTION_PERFORMANCE_INDEX_SQL}; pub(super) use schema::{ ensure_observation_projection_performance_indexes, ensure_observation_projection_schema, }; diff --git a/crates/tracedecay-global-db/src/observation_projection/apply.rs b/crates/tracedecay-global-db/src/observation_projection/apply.rs index 793d9e7076e..63a346c5467 100644 --- a/crates/tracedecay-global-db/src/observation_projection/apply.rs +++ b/crates/tracedecay-global-db/src/observation_projection/apply.rs @@ -1677,7 +1677,8 @@ pub(in super::super) async fn retire_source_record( }; supersede_projected_message(conn, &message).await?; } - request_session_temporal_reset(conn, observation_id).await?; + // Source-rewrite completion coalesces temporal reset requests for the + // retired page after all output ownership has been reconciled. Ok(true) } diff --git a/crates/tracedecay-global-db/src/observation_projection/rearm_tests.rs b/crates/tracedecay-global-db/src/observation_projection/rearm_tests.rs index 37d8654ebdf..5fe4ca7a890 100644 --- a/crates/tracedecay-global-db/src/observation_projection/rearm_tests.rs +++ b/crates/tracedecay-global-db/src/observation_projection/rearm_tests.rs @@ -1,3 +1,5 @@ +mod restart; + use serde_json::Value; use tracedecay_domain::{ CanonicalObservationEnvelopeV1, ComponentVersion, DurableObservationV1, ObservationId, @@ -8,8 +10,11 @@ use tracedecay_domain::{ }; use tracedecay_runtime_core::db::engine::params; +use super::state::{ + RetryRearmWindows, apply_rearm_projection_retry_candidates, schedule_projection_retry, +}; use super::{RearmedProjectionRetries, rearm_queued_projection_retries}; -use crate::tests::harness::RegisteredGlobalDbHarness; +use crate::tests::harness::{RegisteredGlobalDbHarness, writer_telemetry}; const FAR_FUTURE_MICROS: i64 = 9_000_000_000_000_000; @@ -72,67 +77,70 @@ fn queued_observation(index: i64) -> (DurableObservationV1, ObservationSourceCur /// eligible, interleaved so batches must skip due rows. async fn seed_queue(harness: &RegisteredGlobalDbHarness, deferred: i64, due: i64) { let transaction = harness.registered.begin_write_transaction().await.unwrap(); - for index in 0..deferred + due { - let (observation, cursor) = queued_observation(index); - let receipt = observation.receipt(); - transaction - .execute( - "INSERT INTO sanitization_receipts - (receipt_id, sanitizer_version, payload_digest, receipt_json) - VALUES (?1, ?2, ?3, ?4)", - params![ - receipt.receipt().receipt_id().as_str(), - receipt.receipt().sanitizer_version().as_str(), - observation.payload_reference().digest().as_str(), - serde_json::to_string(receipt).unwrap() - ], - ) - .await - .unwrap(); - transaction - .execute( - "INSERT INTO observations - (observation_id, payload_digest, receipt_id, observation_json, - committed_cursor_json) - VALUES (?1, ?2, ?3, ?4, ?5)", - params![ - observation.observation_id().as_str(), - observation.payload_reference().digest().as_str(), - receipt.receipt().receipt_id().as_str(), - serde_json::to_string(&observation).unwrap(), - serde_json::to_string(&cursor).unwrap() - ], - ) - .await - .unwrap(); - transaction - .execute( - "INSERT INTO source_cursors(source_json, scope_json, cursor_json) - VALUES (?1, ?2, ?3)", - params![ - serde_json::to_string(cursor.source()).unwrap(), - serde_json::to_string(cursor.scope()).unwrap(), - serde_json::to_string(&cursor).unwrap() - ], - ) - .await - .unwrap(); - let next_retry = if index % 2 == 1 && index / 2 < due { - 0 - } else { - FAR_FUTURE_MICROS - }; - transaction - .execute( - "INSERT INTO projection_queue - (observation_id, observation_sequence, attempt_count, next_retry_at_micros, - last_error) - SELECT observation_id, sequence, 3, ?2, 'recorded failure' - FROM observations WHERE observation_id = ?1", - params![observation.observation_id().as_str(), next_retry], - ) - .await - .unwrap(); + // Fixture preparation must not dominate the recovery being measured with + // four asynchronous writer round-trips per observation. Keep every typed + // record and the atomic setup, but submit bounded, explicitly ordered pages. + for start in (0..deferred + due).step_by(128) { + let end = (start + 128).min(deferred + due); + let page = (start..end) + .map(|index| { + let (observation, cursor) = queued_observation(index); + let receipt = observation.receipt(); + let next_retry = if index % 2 == 1 && index / 2 < due { + 0 + } else { + FAR_FUTURE_MICROS + }; + serde_json::json!({ + "receipt_id": receipt.receipt().receipt_id().as_str(), + "sanitizer_version": receipt.receipt().sanitizer_version().as_str(), + "payload_digest": observation.payload_reference().digest().as_str(), + "receipt_json": serde_json::to_string(receipt).unwrap(), + "observation_id": observation.observation_id().as_str(), + "observation_json": serde_json::to_string(&observation).unwrap(), + "cursor_json": serde_json::to_string(&cursor).unwrap(), + "source_json": serde_json::to_string(cursor.source()).unwrap(), + "scope_json": serde_json::to_string(cursor.scope()).unwrap(), + "next_retry": next_retry, + }) + }) + .collect::>(); + let expected_rows = u64::try_from(page.len()).unwrap(); + let encoded = serde_json::to_string(&page).unwrap(); + for sql in [ + "INSERT INTO sanitization_receipts + (receipt_id, sanitizer_version, payload_digest, receipt_json) + SELECT json_extract(value, '$.receipt_id'), + json_extract(value, '$.sanitizer_version'), + json_extract(value, '$.payload_digest'), json_extract(value, '$.receipt_json') + FROM json_each(?1) ORDER BY CAST(key AS INTEGER)", + "INSERT INTO observations + (observation_id, payload_digest, receipt_id, observation_json, committed_cursor_json) + SELECT json_extract(value, '$.observation_id'), json_extract(value, '$.payload_digest'), + json_extract(value, '$.receipt_id'), json_extract(value, '$.observation_json'), + json_extract(value, '$.cursor_json') + FROM json_each(?1) ORDER BY CAST(key AS INTEGER)", + "INSERT INTO source_cursors(source_json, scope_json, cursor_json) + SELECT json_extract(value, '$.source_json'), json_extract(value, '$.scope_json'), + json_extract(value, '$.cursor_json') + FROM json_each(?1) ORDER BY CAST(key AS INTEGER)", + "INSERT INTO projection_queue + (observation_id, observation_sequence, attempt_count, next_retry_at_micros, last_error) + SELECT observation.observation_id, observation.sequence, 3, + json_extract(input.value, '$.next_retry'), 'recorded failure' + FROM json_each(?1) AS input CROSS JOIN observations AS observation + WHERE observation.observation_id = json_extract(input.value, '$.observation_id') + ORDER BY CAST(input.key AS INTEGER)", + ] { + assert_eq!( + transaction + .execute(sql, params![encoded.as_str()]) + .await + .unwrap(), + expected_rows, + "every fixture page must preserve all receipt, observation, cursor and queue rows" + ); + } } transaction.commit().await.unwrap(); } @@ -157,11 +165,35 @@ async fn queue_state(harness: &RegisteredGlobalDbHarness) -> (i64, i64, i64) { ) } +#[asupersync::test] +async fn rearm_writer_work_scales_with_deferred_rows() { + async fn writer_steps(rows: i64) -> u64 { + let harness = RegisteredGlobalDbHarness::open("rearm-dense-queue").await; + seed_queue(&harness, rows, 0).await; + let steps = || writer_telemetry(&harness.registered).sqlite_vm.vm_steps; + let before = steps(); + let rearmed = rearm_queued_projection_retries(harness.registered.runtime_database(), 4) + .await + .unwrap(); + assert_eq!(rearmed.rows, u64::try_from(rows).unwrap()); + steps() - before + } + + let base = writer_steps(32).await; + let grown = writer_steps(256).await; + assert!(base > 0); + assert!( + grown < base * 12, + "an eightfold larger dense queue must not rescan its remainder per window: {base} -> {grown}" + ); +} + #[asupersync::test] async fn rearm_walks_deferred_retries_in_bounded_batches() { let harness = RegisteredGlobalDbHarness::open("rearm-bounded-batches").await; seed_queue(&harness, 9, 3).await; assert_eq!(queue_state(&harness).await, (12, 9, 12)); + let before = writer_telemetry(&harness.registered).transactions; let rearmed = rearm_queued_projection_retries(harness.registered.runtime_database(), 4) .await @@ -175,6 +207,10 @@ async fn rearm_walks_deferred_retries_in_bounded_batches() { }, "nine deferred rows re-arm in ceil(9 / 4) statements, never one queue-sized one" ); + // The exact-SQL writer counts one dispatched command per leased transaction, + // including a rollback. The persisted queue below proves the windows committed. + let after = writer_telemetry(&harness.registered).transactions; + assert_eq!(after.commands - before.commands, 3); assert_eq!( queue_state(&harness).await, (12, 0, 12), @@ -187,6 +223,697 @@ async fn rearm_walks_deferred_retries_in_bounded_batches() { RearmedProjectionRetries::default(), "an armed queue costs no write batches" ); + assert_eq!( + writer_telemetry(&harness.registered).transactions, + after, + "zero reported batches must also mean zero actual writer transactions" + ); +} + +#[asupersync::test] +async fn expired_retry_deadlines_survive_restarts_without_rearm_writes() { + let harness = RegisteredGlobalDbHarness::open("rearm-expired-deadlines").await; + seed_queue(&harness, 8, 0).await; + + // These positive deadlines have already elapsed. Normal projection may + // consume them directly; a restart must not rewrite the entire overdue + // queue just to change each deadline to zero. + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + assert_eq!( + transaction + .execute("UPDATE projection_queue SET next_retry_at_micros = 1", ()) + .await + .unwrap(), + 8 + ); + transaction.commit().await.unwrap(); + + let (harness, _convergence) = harness.restart_for_daemon().await; + let before = writer_telemetry(&harness.registered).transactions; + for _ in 0..2 { + assert_eq!( + rearm_queued_projection_retries(harness.registered.runtime_database(), 2) + .await + .unwrap(), + RearmedProjectionRetries::default(), + "expired retries already admit projection and need no restart writes" + ); + assert_eq!( + writer_telemetry(&harness.registered).transactions, + before, + "an overdue-only queue must not acquire any SQLite writer" + ); + } + assert_eq!(queue_state(&harness).await, (8, 8, 8)); + + // Interleave actual deferred retries and a terminal collision. Each + // bounded transaction must touch just its one deferred row, preserving + // the other deadlines and the retry history. + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + assert_eq!( + transaction + .execute( + "UPDATE projection_queue SET next_retry_at_micros = ?1 + WHERE observation_sequence IN (2, 6)", + params![FAR_FUTURE_MICROS], + ) + .await + .unwrap(), + 2 + ); + assert_eq!( + transaction + .execute( + "UPDATE projection_queue SET next_retry_at_micros = ?1 + WHERE observation_sequence = 4", + params![i64::MAX], + ) + .await + .unwrap(), + 1 + ); + transaction.commit().await.unwrap(); + + let before = writer_telemetry(&harness.registered).transactions; + assert_eq!( + rearm_queued_projection_retries(harness.registered.runtime_database(), 1) + .await + .unwrap(), + RearmedProjectionRetries { + rows: 2, + batches: 2, + } + ); + let after = writer_telemetry(&harness.registered).transactions; + assert_eq!(after.commands - before.commands, 2); + assert_eq!(queue_state(&harness).await, (8, 6, 8)); + + let snapshot = harness.registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT observation_sequence, next_retry_at_micros + FROM projection_queue ORDER BY observation_sequence", + (), + ) + .await + .unwrap(); + for sequence in 1..=8_i64 { + let row = rows.next().await.unwrap().unwrap(); + assert_eq!(row.get::(0).unwrap(), sequence); + let deadline = row.get::(1).unwrap(); + assert_eq!( + deadline, + if sequence == 4 { + i64::MAX + } else if sequence == 2 || sequence == 6 { + 0 + } else { + 1 + }, + "rearm must not rewrite expired or terminal deadlines" + ); + } + assert!(rows.next().await.unwrap().is_none()); +} + +#[asupersync::test] +async fn rearm_ignores_parked_terminal_collisions_without_empty_write_batches() { + let harness = RegisteredGlobalDbHarness::open("rearm-terminal-collisions").await; + seed_queue(&harness, 9, 3).await; + + // Six permanently parked collisions are interleaved with three deferred + // retries and three already-due rows. Neither parked nor due entries may + // consume a restart re-arm write window. + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + assert_eq!( + transaction + .execute( + "UPDATE projection_queue SET next_retry_at_micros = ?1 + WHERE observation_sequence IN (1, 3, 5, 7, 9, 11)", + params![i64::MAX], + ) + .await + .unwrap(), + 6 + ); + transaction.commit().await.unwrap(); + assert_eq!(queue_state(&harness).await, (12, 9, 12)); + let before = writer_telemetry(&harness.registered).transactions; + + let rearmed = rearm_queued_projection_retries(harness.registered.runtime_database(), 2) + .await + .unwrap(); + assert_eq!( + rearmed, + RearmedProjectionRetries { + rows: 3, + batches: 2, + }, + "parked collisions must not consume re-arm write windows" + ); + let after = writer_telemetry(&harness.registered).transactions; + assert_eq!(after.commands - before.commands, 2); + assert_eq!( + queue_state(&harness).await, + (12, 6, 12), + "only retryable deadlines clear; terminal collisions remain parked" + ); + + // Reopen the durable database through real daemon admission. Do not run + // background convergence before measuring its retry re-arm directly. + let (harness, _convergence) = harness.restart_for_daemon().await; + assert_eq!(queue_state(&harness).await, (12, 6, 12)); + let before = writer_telemetry(&harness.registered).transactions; + for _ in 0..2 { + assert_eq!( + rearm_queued_projection_retries(harness.registered.runtime_database(), 2) + .await + .unwrap(), + RearmedProjectionRetries::default(), + "restarts with only terminal collisions cannot schedule empty write batches" + ); + assert_eq!(writer_telemetry(&harness.registered).transactions, before); + } + + // An idle observation must not cache away a subsequently deferred row. + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + assert_eq!( + transaction + .execute( + "UPDATE projection_queue SET next_retry_at_micros = ?1 + WHERE observation_sequence = 2", + params![FAR_FUTURE_MICROS], + ) + .await + .unwrap(), + 1 + ); + transaction.commit().await.unwrap(); + let before = writer_telemetry(&harness.registered).transactions; + assert_eq!( + rearm_queued_projection_retries(harness.registered.runtime_database(), 2) + .await + .unwrap(), + RearmedProjectionRetries { + rows: 1, + batches: 1, + } + ); + let after = writer_telemetry(&harness.registered).transactions; + assert_eq!(after.commands - before.commands, 1); + assert_eq!(queue_state(&harness).await, (12, 6, 12)); +} + +#[asupersync::test] +async fn rearm_rejects_nonpositive_windows_before_acquiring_the_writer() { + let harness = RegisteredGlobalDbHarness::open("rearm-invalid-window").await; + seed_queue(&harness, 5, 0).await; + let before = writer_telemetry(&harness.registered).transactions; + for batch_rows in [0, -1] { + assert!( + rearm_queued_projection_retries(harness.registered.runtime_database(), batch_rows) + .await + .is_err(), + "a nonpositive limit must not silently skip or unbound recovery" + ); + assert_eq!(queue_state(&harness).await, (5, 5, 5)); + assert_eq!(writer_telemetry(&harness.registered).transactions, before); + } + assert_eq!( + rearm_queued_projection_retries(harness.registered.runtime_database(), 2) + .await + .unwrap(), + RearmedProjectionRetries { + rows: 5, + batches: 3, + } + ); + let after = writer_telemetry(&harness.registered).transactions; + assert_eq!(after.commands - before.commands, 3); + assert_eq!(queue_state(&harness).await, (5, 0, 5)); +} + +#[asupersync::test] +async fn sparse_retry_rearm_writer_work_is_bounded_by_updated_rows() { + async fn writer_steps(prefix: i64, fixture: &'static str) -> u64 { + let harness = RegisteredGlobalDbHarness::open(fixture).await; + seed_queue(&harness, prefix + 4, 0).await; + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + assert_eq!( + transaction + .execute( + "UPDATE projection_queue SET next_retry_at_micros = 0 + WHERE observation_sequence <= ?1", + params![prefix], + ) + .await + .unwrap(), + u64::try_from(prefix).unwrap() + ); + transaction.commit().await.unwrap(); + let before = writer_telemetry(&harness.registered); + let rearmed = rearm_queued_projection_retries(harness.registered.runtime_database(), 4) + .await + .unwrap(); + assert_eq!( + rearmed, + RearmedProjectionRetries { + rows: 4, + batches: 1, + } + ); + let after = writer_telemetry(&harness.registered); + assert_eq!( + after.transactions.commands - before.transactions.commands, + 1 + ); + assert_eq!(queue_state(&harness).await.1, 0); + after.sqlite_vm.vm_steps - before.sqlite_vm.vm_steps + } + + let small = writer_steps(32, "rearm-sparse-small").await; + let large = writer_steps(256, "rearm-sparse-large").await; + assert!(small > 0); + assert!( + large < small * 2, + "eight times as many inert queue rows must not multiply SQLite writer work: {small} -> {large}" + ); +} + +#[asupersync::test] +async fn rearm_reader_snapshot_cannot_clear_a_new_retry_generation() { + let harness = RegisteredGlobalDbHarness::open("rearm-stale-candidates").await; + seed_queue(&harness, 4, 0).await; + let now = 1_000_000_i64; + let mut windows = RetryRearmWindows::new(harness.registered.runtime_database(), 4, now) + .await + .unwrap(); + let candidates = windows.next().await.unwrap().unwrap(); + + // The reader's window is now stale: a foreground owner schedules a new + // generation for one row and another row has become due. + let transaction = harness + .registered + .runtime_database() + .begin_write_transaction("schedule projection retry") + .await + .unwrap(); + assert_eq!( + transaction + .execute( + "UPDATE projection_queue SET attempt_count = attempt_count + 1 + WHERE observation_sequence = 2", + (), + ) + .await + .unwrap(), + 1 + ); + assert_eq!( + transaction + .execute( + "UPDATE projection_queue SET next_retry_at_micros = 1 + WHERE observation_sequence = 3", + (), + ) + .await + .unwrap(), + 1 + ); + // An identical attempt count and deadline can leave every old CAS + // field unchanged. The foreground mount still owns the new retry. + schedule_projection_retry( + &transaction, + queued_observation(3).0.observation_id(), + 3, + FAR_FUTURE_MICROS, + "new mount", + &windows.mount, + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + + assert_eq!( + apply_rearm_projection_retry_candidates( + harness.registered.runtime_database(), + &candidates, + now, + &windows.mount, + ) + .await + .unwrap(), + 1, + "only the unchanged candidate can be reset" + ); + let snapshot = harness.registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT observation_sequence, attempt_count, next_retry_at_micros + FROM projection_queue ORDER BY observation_sequence", + (), + ) + .await + .unwrap(); + for (sequence, attempts, deadline) in [ + (1_i64, 3_i64, 0_i64), + (2, 4, FAR_FUTURE_MICROS), + (3, 3, 1), + (4, 3, FAR_FUTURE_MICROS), + ] { + let row = rows.next().await.unwrap().unwrap(); + assert_eq!(row.get::(0).unwrap(), sequence); + assert_eq!(row.get::(1).unwrap(), attempts); + assert_eq!(row.get::(2).unwrap(), deadline); + } + assert!(rows.next().await.unwrap().is_none()); +} + +#[asupersync::test] +async fn retry_rearm_does_not_erase_a_live_retry_between_windows() { + let harness = RegisteredGlobalDbHarness::open("rearm-live-between-windows").await; + seed_queue(&harness, 2, 1).await; + let database = harness.registered.runtime_database(); + let now = 1_000_000_i64; + let mut windows = RetryRearmWindows::new(database, 1, now).await.unwrap(); + + // Apply the first old retry, then schedule a foreground retry between pages. + assert_eq!( + apply_rearm_projection_retry_candidates( + database, + &windows.next().await.unwrap().expect("first retry window"), + now, + &windows.mount, + ) + .await + .unwrap(), + 1 + ); + let transaction = harness + .registered + .runtime_database() + .begin_write_transaction("schedule projection retry") + .await + .unwrap(); + schedule_projection_retry( + &transaction, + queued_observation(1).0.observation_id(), + 4, + FAR_FUTURE_MICROS, + "fresh live retry", + &windows.mount, + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + + // The next reader sees the live retry and leaves its deadline intact. + assert_eq!( + apply_rearm_projection_retry_candidates( + database, + &windows.next().await.unwrap().expect("next old retry"), + now, + &windows.mount, + ) + .await + .unwrap(), + 1 + ); + + assert!(windows.next().await.unwrap().is_none()); + let snapshot = harness.registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT observation_sequence, attempt_count, next_retry_at_micros, last_error + FROM projection_queue ORDER BY observation_sequence", + (), + ) + .await + .unwrap(); + for (sequence, attempts, deadline, error) in [ + (1_i64, 3_i64, 0_i64, "recorded failure"), + (2, 4, FAR_FUTURE_MICROS, "fresh live retry"), + (3, 3, 0, "recorded failure"), + ] { + let row = rows.next().await.unwrap().unwrap(); + assert_eq!(row.get::(0).unwrap(), sequence); + assert_eq!(row.get::(1).unwrap(), attempts); + assert_eq!(row.get::(2).unwrap(), deadline); + assert_eq!(row.get::(3).unwrap(), error); + } + assert!(rows.next().await.unwrap().is_none()); +} + +#[asupersync::test] +async fn rearm_leaves_later_observations_with_reused_queue_rowids_deferred() { + let harness = RegisteredGlobalDbHarness::open("rearm-reused-queue-tail").await; + seed_queue(&harness, 4, 0).await; + let database = harness.registered.runtime_database(); + let now = 1_000_000; + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + transaction + .execute( + "DELETE FROM projection_queue WHERE observation_sequence = 4", + (), + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + let mut windows = RetryRearmWindows::new(database, 1, now).await.unwrap(); + assert_eq!( + apply_rearm_projection_retry_candidates( + database, + &windows.next().await.unwrap().expect("first retry window"), + now, + &windows.mount, + ) + .await + .unwrap(), + 1 + ); + + // A foreground pass settles the old tail. SQLite assigns its rowid to a + // newer observation, whose retry belongs to the current mount. + let transaction = harness + .registered + .runtime_database() + .begin_write_transaction("schedule projection retry") + .await + .unwrap(); + transaction + .execute( + "DELETE FROM projection_queue WHERE observation_sequence = 3", + (), + ) + .await + .unwrap(); + transaction + .execute( + "INSERT INTO projection_queue + (observation_id, observation_sequence, attempt_count, next_retry_at_micros, last_error) + SELECT observation_id, sequence, 0, 0, NULL + FROM observations WHERE sequence = 4", + (), + ) + .await + .unwrap(); + schedule_projection_retry( + &transaction, + queued_observation(3).0.observation_id(), + 1, + FAR_FUTURE_MICROS, + "new retry", + &windows.mount, + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + let mut remaining = 0; + while let Some(candidates) = windows.next().await.unwrap() { + remaining += + apply_rearm_projection_retry_candidates(database, &candidates, now, &windows.mount) + .await + .unwrap(); + } + assert_eq!( + remaining, 1, + "only the unchanged original retry survives rowid reuse" + ); + let snapshot = harness.registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT rowid, next_retry_at_micros FROM projection_queue + WHERE observation_sequence = 4", + (), + ) + .await + .unwrap(); + let new = rows.next().await.unwrap().unwrap(); + assert_eq!(new.get::(0).unwrap(), 3); + assert_eq!(new.get::(1).unwrap(), FAR_FUTURE_MICROS); +} + +#[asupersync::test] +async fn retry_recovery_releases_wal_reader_between_pages() { + let harness = RegisteredGlobalDbHarness::open("rearm-release-wal-reader").await; + seed_queue(&harness, 9, 0).await; + let database = harness.registered.runtime_database(); + let mut windows = RetryRearmWindows::new(database, 2, 1_000_000) + .await + .unwrap(); + let first = windows.next().await.unwrap().expect("first retry window"); + assert_eq!( + apply_rearm_projection_retry_candidates(database, &first, 1_000_000, &windows.mount) + .await + .unwrap(), + 2 + ); + + // Checkpoint while later pages still await their writes. A reader retained + // until the end of the pass would keep this WAL read slot pinned. + let outside = rusqlite::Connection::open(harness.registered.db_path()).unwrap(); + let (busy, log_frames, checkpointed): (i64, i64, i64) = outside + .query_row("PRAGMA wal_checkpoint(RESTART)", [], |row| { + Ok((row.get(0)?, row.get(1)?, row.get(2)?)) + }) + .unwrap(); + assert_eq!(busy, 0, "no rearm snapshot may pin the WAL after staging"); + assert!( + log_frames > 0, + "the checkpoint must observe actual writer traffic" + ); + assert_eq!(checkpointed, log_frames); + let mut remaining = Vec::new(); + while let Some(candidates) = windows.next().await.unwrap() { + remaining.push( + apply_rearm_projection_retry_candidates( + database, + &candidates, + 1_000_000, + &windows.mount, + ) + .await + .unwrap(), + ); + } + assert_eq!(remaining, [2, 2, 2, 1]); + assert_eq!(queue_state(&harness).await, (9, 0, 9)); +} + +#[asupersync::test] +async fn retry_rearm_respects_mount_ownership_across_restart_and_existing_schema() { + let harness = RegisteredGlobalDbHarness::open("rearm-mount-owner").await; + seed_queue(&harness, 2, 0).await; + let database = harness.registered.runtime_database(); + let windows = RetryRearmWindows::new(database, 1, 1_000_000) + .await + .unwrap(); + let transaction = harness + .registered + .runtime_database() + .begin_write_transaction("schedule projection retry") + .await + .unwrap(); + schedule_projection_retry( + &transaction, + queued_observation(1).0.observation_id(), + 4, + FAR_FUTURE_MICROS, + "this mount", + &windows.mount, + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + assert_eq!( + rearm_queued_projection_retries(database, 1) + .await + .unwrap() + .rows, + 1 + ); + assert_eq!( + rearm_queued_projection_retries(database, 1) + .await + .unwrap() + .rows, + 0 + ); + let (harness, _convergence) = harness.restart_for_daemon().await; + assert_eq!( + rearm_queued_projection_retries(harness.registered.runtime_database(), 1) + .await + .unwrap() + .rows, + 1 + ); + + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + transaction + .execute("DROP TABLE projection_retry_owners", ()) + .await + .unwrap(); + transaction + .execute( + "UPDATE projection_queue SET next_retry_at_micros = ?1", + params![FAR_FUTURE_MICROS], + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + let (harness, _convergence) = harness.restart_for_daemon().await; + assert_eq!( + rearm_queued_projection_retries(harness.registered.runtime_database(), 1) + .await + .unwrap() + .rows, + 2 + ); +} + +#[asupersync::test] +async fn failed_rearm_owner_write_preserves_the_previous_retry() { + let harness = RegisteredGlobalDbHarness::open("rearm-owner-rollback").await; + seed_queue(&harness, 1, 0).await; + let database = harness.registered.runtime_database(); + let windows = RetryRearmWindows::new(database, 1, 1_000_000) + .await + .unwrap(); + let setup = harness.registered.begin_write_transaction().await.unwrap(); + setup + .execute_batch( + "CREATE TRIGGER reject_retry_owner BEFORE INSERT ON projection_retry_owners + BEGIN SELECT RAISE(ABORT, 'retry owner refused'); END;", + ) + .await + .unwrap(); + setup.commit().await.unwrap(); + + { + let transaction = database + .begin_write_transaction("schedule projection retry") + .await + .unwrap(); + let error = schedule_projection_retry( + &transaction, + queued_observation(0).0.observation_id(), + 4, + FAR_FUTURE_MICROS - 1, + "new retry", + &windows.mount, + ) + .await + .unwrap_err(); + assert!(error.to_string().contains("retry owner refused")); + } + assert_eq!(queue_state(&harness).await, (1, 1, 1)); + assert_eq!( + rearm_queued_projection_retries(database, 1) + .await + .unwrap() + .rows, + 1 + ); } #[asupersync::test] @@ -215,3 +942,90 @@ async fn daemon_admission_defers_rearm_to_background_convergence() { "background convergence re-arms the fresh mount's queue" ); } + +// A rewrite for one source must not visit the entire projection queue while +// holding the SQLite writer. The real queue grows with all imported sources, +// but only records offered by this source may have their revival fence moved. +#[asupersync::test] +async fn source_rewrite_begin_writer_work_is_independent_of_unrelated_queue_rows() { + async fn writer_steps(queue_rows: i64, label: &'static str) -> u64 { + let harness = RegisteredGlobalDbHarness::open(label).await; + seed_queue(&harness, queue_rows, 0).await; + let (observation, _) = queued_observation(0); + let source = observation.source().clone(); + let scope = observation.scope().clone(); + let previous = observation.identity().generation(); + let successor = ObservationSourceGenerationV1::new(2).unwrap(); + + harness + .registered + .record_observation_source_presence( + &source, + &scope, + &[crate::ObservationSourcePresenceV1 { + observation_id: observation.observation_id().clone(), + generation: previous, + start_offset: 0, + }], + ) + .await + .unwrap(); + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + assert_eq!( + transaction + .execute( + "UPDATE projection_queue SET revival_generation = ?1, + revival_start_offset = 0 WHERE observation_id = ?2", + params![ + previous.generation_id().to_string(), + observation.observation_id().as_str() + ], + ) + .await + .unwrap(), + 1 + ); + transaction.commit().await.unwrap(); + + let before = writer_telemetry(&harness.registered).sqlite_vm.vm_steps; + harness + .registered + .begin_observation_source_rewrite(&source, &scope, previous, successor, 1) + .await + .unwrap(); + let after = writer_telemetry(&harness.registered).sqlite_vm.vm_steps; + + let snapshot = harness.registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT observation_id, revival_generation, revival_start_offset + FROM projection_queue WHERE revival_generation IS NOT NULL", + (), + ) + .await + .unwrap(); + let row = rows.next().await.unwrap().unwrap(); + assert_eq!( + row.get::(0).unwrap(), + observation.observation_id().as_str() + ); + assert_eq!( + row.get::(1).unwrap(), + successor.generation_id().to_string() + ); + assert_eq!(row.get::(2).unwrap(), 0); + assert!( + rows.next().await.unwrap().is_none(), + "unrelated queue rows must retain their original null revival state" + ); + after - before + } + + let small = writer_steps(32, "source-rewrite-sparse-32").await; + let large = writer_steps(256, "source-rewrite-sparse-256").await; + assert!(small > 0); + assert!( + large < small * 2, + "an eightfold larger unrelated queue must not amplify writer work: {small} -> {large}" + ); +} diff --git a/crates/tracedecay-global-db/src/observation_projection/rearm_tests/restart.rs b/crates/tracedecay-global-db/src/observation_projection/rearm_tests/restart.rs new file mode 100644 index 00000000000..d56aaea2fea --- /dev/null +++ b/crates/tracedecay-global-db/src/observation_projection/rearm_tests/restart.rs @@ -0,0 +1,230 @@ +use crate::observation_projection::state::{ + RetryRearmWindows, apply_rearm_projection_retry_candidates, +}; +use crate::observation_projection::{RearmedProjectionRetries, rearm_queued_projection_retries}; +use crate::tests::harness::{RegisteredGlobalDbHarness, writer_telemetry}; +use tracedecay_runtime_core::db::engine::params; + +use super::{FAR_FUTURE_MICROS, queue_state, seed_queue}; + +async fn retry_deadlines(harness: &RegisteredGlobalDbHarness) -> Vec<(i64, i64)> { + let snapshot = harness.registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT observation_sequence, next_retry_at_micros + FROM projection_queue ORDER BY observation_sequence", + (), + ) + .await + .unwrap(); + let mut deadlines = Vec::new(); + while let Some(row) = rows.next().await.unwrap() { + deadlines.push((row.get(0).unwrap(), row.get(1).unwrap())); + } + deadlines +} + +#[asupersync::test] +async fn retry_rearm_resumes_committed_windows_after_failure_and_restart() { + let harness = RegisteredGlobalDbHarness::open("rearm-interrupted-recovery").await; + seed_queue(&harness, 9, 0).await; + + // Two windows commit before the third fails. The failing window must + // roll back as a unit rather than lose its first observation's retry. + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + transaction + .execute( + "CREATE TRIGGER reject_retry_window + BEFORE UPDATE OF next_retry_at_micros ON projection_queue + WHEN OLD.observation_sequence = 6 AND NEW.next_retry_at_micros = 0 + BEGIN SELECT RAISE(ABORT, 'injected retry window failure'); END", + (), + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + assert!( + rearm_queued_projection_retries(harness.registered.runtime_database(), 2) + .await + .is_err(), + "the injected failure must interrupt recovery" + ); + + let interrupted = (1..=9_i64) + .map(|sequence| (sequence, if sequence <= 4 { 0 } else { FAR_FUTURE_MICROS })) + .collect::>(); + assert_eq!(retry_deadlines(&harness).await, interrupted); + assert_eq!(queue_state(&harness).await, (9, 5, 9)); + + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + transaction + .execute("DROP TRIGGER reject_retry_window", ()) + .await + .unwrap(); + transaction.commit().await.unwrap(); + let (harness, _convergence) = harness.restart_for_daemon().await; + assert_eq!(retry_deadlines(&harness).await, interrupted); + + // Make a replay of the committed prefix fail, even if it would write + // identical values. Row counts alone cannot detect redundant updates. + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + transaction + .execute( + "CREATE TRIGGER reject_committed_retry_replay + BEFORE UPDATE OF next_retry_at_micros ON projection_queue + WHEN OLD.observation_sequence <= 4 + BEGIN SELECT RAISE(ABORT, 'committed retry prefix replayed'); END", + (), + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + + let before = writer_telemetry(&harness.registered).transactions; + assert_eq!( + rearm_queued_projection_retries(harness.registered.runtime_database(), 2) + .await + .unwrap(), + RearmedProjectionRetries { + rows: 5, + batches: 3, + }, + "restart must resume only the five remaining rows in bounded windows" + ); + let after = writer_telemetry(&harness.registered).transactions; + assert_eq!( + after.commands - before.commands, + 3, + "the final productive window must not be followed by an empty writer lease" + ); + assert_eq!(queue_state(&harness).await, (9, 0, 9)); + assert_eq!( + retry_deadlines(&harness).await, + (1..=9_i64) + .map(|sequence| (sequence, 0)) + .collect::>() + ); + for _ in 0..2 { + assert_eq!( + rearm_queued_projection_retries(harness.registered.runtime_database(), 2) + .await + .unwrap(), + RearmedProjectionRetries::default() + ); + assert_eq!(writer_telemetry(&harness.registered).transactions, after); + } + + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + transaction + .execute("DROP TRIGGER reject_committed_retry_replay", ()) + .await + .unwrap(); + transaction.commit().await.unwrap(); +} + +/// A recovery can spend a long time between its original clock read and later +/// candidate pages. Deadlines that matured meanwhile must not be rewritten: +/// they are already eligible for ordinary projection without restart rearm. +#[asupersync::test] +async fn retry_rearm_skips_deadlines_that_mature_during_recovery() { + let harness = RegisteredGlobalDbHarness::open("rearm-midwalk-deadlines").await; + seed_queue(&harness, 4, 0).await; + let database = harness.registered.runtime_database(); + + // Model a recovery that started before deadlines at 1_500_000 micros. + // The actual runtime clock is later; this does not rely on sleeping. + let mut windows = RetryRearmWindows::new(database, 4, 1_000_000) + .await + .unwrap(); + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + assert_eq!( + transaction + .execute( + "UPDATE projection_queue SET next_retry_at_micros = 1500000 + WHERE observation_sequence IN (1, 2)", + (), + ) + .await + .unwrap(), + 2 + ); + transaction.commit().await.unwrap(); + + let before = writer_telemetry(&harness.registered).transactions; + let candidates = windows + .next() + .await + .unwrap() + .expect("two still-deferred observations remain"); + let selected: Vec<(i64, String, i64, i64)> = serde_json::from_str(&candidates).unwrap(); + assert_eq!(selected.len(), 2, "matured deadlines need no write window"); + assert!(selected.iter().all(|item| item.2 == FAR_FUTURE_MICROS)); + assert_eq!( + apply_rearm_projection_retry_candidates( + database, + &candidates, + tracedecay_contracts::clock::now_micros().0, + &windows.mount, + ) + .await + .unwrap(), + 2 + ); + assert!(windows.next().await.unwrap().is_none()); + assert_eq!( + retry_deadlines(&harness).await, + vec![(1, 1_500_000), (2, 1_500_000), (3, 0), (4, 0)] + ); + let after = writer_telemetry(&harness.registered).transactions; + assert_eq!(after.commands - before.commands, 1); + + // If *all* rows mature after the recovery cursor was established, + // the reader must finish without even acquiring an empty writer lease. + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + transaction + .execute( + "UPDATE projection_queue SET next_retry_at_micros = ?1 + WHERE observation_sequence IN (3, 4)", + params![FAR_FUTURE_MICROS], + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + let mut expired = RetryRearmWindows::new(database, 4, 1_000_000) + .await + .unwrap(); + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + transaction + .execute( + "UPDATE projection_queue SET next_retry_at_micros = 1500000 + WHERE observation_sequence IN (3, 4)", + (), + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + + let before_idle = writer_telemetry(&harness.registered).transactions; + assert!(expired.next().await.unwrap().is_none()); + assert_eq!( + writer_telemetry(&harness.registered).transactions, + before_idle + ); + assert_eq!( + rearm_queued_projection_retries(database, 4).await.unwrap(), + RearmedProjectionRetries::default() + ); + assert_eq!( + writer_telemetry(&harness.registered).transactions, + before_idle + ); + assert_eq!( + retry_deadlines(&harness).await, + vec![ + (1, 1_500_000), + (2, 1_500_000), + (3, 1_500_000), + (4, 1_500_000) + ] + ); +} diff --git a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs index beb861a9d9a..b69922daf80 100644 --- a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs +++ b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs @@ -25,7 +25,8 @@ use super::state::{ consume_projection_queue_item, decode_observation_row, decode_sequence, projected_stored_message, projection_retry_state, queued_projection, read_checkpoint, read_message, read_observation, read_session, reconcile_session_rows_detailed, - schedule_projection_retry, storage, storage_message, stored_output_digest, write_checkpoint, + retry_mount_identity, schedule_projection_retry, storage, storage_message, + stored_output_digest, write_checkpoint, }; use super::transition::{ MessageTransition, MessageTransitionState, WorkflowFactTarget, WorkflowFactTransition, @@ -220,17 +221,63 @@ pub async fn project_queued_observations( if max == 0 { return Ok(ProjectionDrainBatch::default()); } + // Sequence order, not the earliest deadline anywhere in the queue, + // determines readiness. A deferred head fences even a later due row: + // neither case should take a writer lease just to find no ready work. + // The partial sequence index excludes terminal (i64::MAX) collisions. + // The writer remains authoritative when this check finds a ready head. + { + let reader = database.read_connection(); + let mut rows = reader + .query( + // The fixed terminal sentinel matches the partial index predicate. + "SELECT next_retry_at_micros, + EXISTS(SELECT 1 FROM observation_projection_rebuilds + WHERE projector_version = ?1) + FROM projection_queue + WHERE next_retry_at_micros < 9223372036854775807 + ORDER BY observation_sequence LIMIT 1", + params![SESSION_MESSAGE_PROJECTOR_VERSION], + ) + .await + .map_err(|error| storage("probe ready projection queue", error))?; + let head = rows + .next() + .await + .map_err(|error| storage("probe ready projection queue", error))?; + let Some(head) = head else { + return Ok(ProjectionDrainBatch::default()); + }; + let deadline = head + .get::(0) + .map_err(|error| storage("decode projection head deadline", error))?; + // An active generation rebuild suppresses ordinary projection. + // Do not acquire a writer only to discover that suppression again; + // the next poll rechecks both the rebuild and the ordered queue head. + let rebuild_active = head + .get::(1) + .map_err(|error| storage("decode projection rebuild gate", error))? + != 0; + if rebuild_active || deadline > tracedecay_contracts::clock::now_micros().0 { + return Ok(ProjectionDrainBatch { + items: Vec::new(), + has_more: true, + }); + } + } let transaction = database .begin_write_transaction("begin projection window transaction") .await .map_err(|error| storage("begin projection window transaction", error))?; let now_micros = tracedecay_contracts::clock::now_micros().0; - let mut items = Vec::new(); - while items.len() < max { - let Some(observation_id) = next_ready_projection_head(&transaction, now_micros).await? - else { - break; - }; + // Read the next ordered window only once. The old query ran a fresh + // MIN(observation_sequence) scan for every projected item. Permanently + // parked collision rows remain in this queue, so every new item re-walked + // that same growing prefix even inside one 32-item write transaction. + // The first not-yet-due row still fences every later row. + let ready = ready_projection_window(&transaction, now_micros, max).await?; + let mut items = Vec::with_capacity(ready.len()); + for observation_id in ready { // Boxed for the same reason as the scalar drain: the collision-guarded // apply subtree overflows the base-opt worker stack when inlined. match Box::pin(project_observation_in_transaction_with_session( @@ -259,47 +306,57 @@ pub async fn project_queued_observations( Ok(ProjectionDrainBatch { items, has_more }) } -/// Ready queue head under the same gate as the runtime's -/// `NextQueuedProjection` read: strict minimum sequence, retry deadline -/// passed, and no active projection rebuild generation. -async fn next_ready_projection_head( +/// The strict sequence head is the first nonterminal queue row, even when +/// that row is not due yet. Read at most one write-window worth of heads +/// through the sequence index, stopping at that first deferred row. This +/// preserves head-of-line retry fences and active-rebuild suppression while +/// avoiding one full queued-prefix probe per observation. +async fn ready_projection_window( conn: &impl QueryExecutor, now_micros: i64, -) -> ProjectionStoreResult> { + max: usize, +) -> ProjectionStoreResult> { + let limit = i64::try_from(max).map_err(|_| { + storage_message( + "read projection queue window", + "projection window exceeds SQLite integer range", + ) + })?; let mut rows = conn .query( - "SELECT observation_id FROM projection_queue - WHERE next_retry_at_micros <= ?2 - AND observation_sequence = ( - SELECT MIN(observation_sequence) FROM projection_queue - WHERE next_retry_at_micros < ?3 - ) + "SELECT observation_id, next_retry_at_micros FROM projection_queue + WHERE next_retry_at_micros < 9223372036854775807 AND NOT EXISTS ( SELECT 1 FROM observation_projection_rebuilds WHERE projector_version = ?1 ) - LIMIT 1", - params![ - SESSION_MESSAGE_PROJECTOR_VERSION, - now_micros, - PROJECTION_TERMINAL_RETRY_MICROS - ], + ORDER BY observation_sequence + LIMIT ?2", + params![SESSION_MESSAGE_PROJECTOR_VERSION, limit], ) .await - .map_err(|error| storage("read projection queue head", error))?; - let Some(row) = rows + .map_err(|error| storage("read projection queue window", error))?; + let mut ready = Vec::new(); + while let Some(row) = rows .next() .await - .map_err(|error| storage("read projection queue head", error))? - else { - return Ok(None); - }; - let observation_id = row - .get::(0) - .map_err(|error| storage("decode projection queue head", error))?; - Ok(Some( - CanonicalObservationIdV1::new(observation_id).map_err(ProjectionStoreError::Contract)?, - )) + .map_err(|error| storage("read projection queue window", error))? + { + let due = row + .get::(1) + .map_err(|error| storage("decode projection queue deadline", error))?; + if due > now_micros { + break; + } + let observation_id = row + .get::(0) + .map_err(|error| storage("decode projection queue head", error))?; + ready.push( + CanonicalObservationIdV1::new(observation_id) + .map_err(ProjectionStoreError::Contract)?, + ); + } + Ok(ready) } /// Whether a later drain still has work. A row parked on a terminal collision @@ -307,8 +364,8 @@ async fn next_ready_projection_head( async fn projection_queue_has_items(conn: &impl QueryExecutor) -> ProjectionStoreResult { let mut rows = conn .query( - "SELECT 1 FROM projection_queue WHERE next_retry_at_micros < ?1 LIMIT 1", - params![PROJECTION_TERMINAL_RETRY_MICROS], + "SELECT 1 FROM projection_queue WHERE next_retry_at_micros < 9223372036854775807 LIMIT 1", + (), ) .await .map_err(|error| storage("probe projection queue", error))?; @@ -348,6 +405,7 @@ async fn persist_projection_retry_on_database( attempt_count, next_retry_at_micros, last_error, + &retry_mount_identity(database)?, ) .await?; transaction @@ -1056,9 +1114,9 @@ async fn activate_projection_rebuild_transaction( transaction .execute( "DELETE FROM projection_queue - WHERE observation_sequence <= ?1 AND next_retry_at_micros < ?2 + WHERE observation_sequence <= ?1 AND next_retry_at_micros < 9223372036854775807 AND revival_generation IS NULL", - params![job.frontier, PROJECTION_TERMINAL_RETRY_MICROS], + params![job.frontier], ) .await .map_err(|error| storage("consume rebuilt projection queue", error))?; diff --git a/crates/tracedecay-global-db/src/observation_projection/schema.rs b/crates/tracedecay-global-db/src/observation_projection/schema.rs index 849e5972d3b..90c703c22ed 100644 --- a/crates/tracedecay-global-db/src/observation_projection/schema.rs +++ b/crates/tracedecay-global-db/src/observation_projection/schema.rs @@ -19,6 +19,10 @@ const OBSERVATION_PROJECTION_SCHEMA_SQL: &str = FOREIGN KEY(observation_id) REFERENCES observations(observation_id), FOREIGN KEY(receipt_id) REFERENCES sanitization_receipts(receipt_id) ); + CREATE TABLE IF NOT EXISTS projection_retry_owners ( + observation_id TEXT PRIMARY KEY NOT NULL REFERENCES projection_queue(observation_id) ON DELETE CASCADE, + mount TEXT NOT NULL + ) WITHOUT ROWID; CREATE TABLE IF NOT EXISTS observation_projection_checkpoints ( projector_version TEXT PRIMARY KEY, last_sequence INTEGER NOT NULL CHECK(last_sequence >= 0) @@ -310,9 +314,20 @@ const OBSERVATION_PROJECTION_SCHEMA_SQL: &str = const OBSERVATION_PROJECTION_BINDING_TRIGGERS_SQL: &str = include_str!("projection_binding_triggers.sql"); +/// Queue-head reads share this partial index's terminal collision bound. +pub(crate) const ACTIVE_QUEUE_INDEX_SQL: &str = + "CREATE INDEX IF NOT EXISTS idx_projection_queue_active_sequence + ON projection_queue (observation_sequence, next_retry_at_micros) + WHERE next_retry_at_micros < 9223372036854775807;"; + /// Historical-data indexes the schema contract requires on the projection /// authority. pub(crate) const OBSERVATION_PROJECTION_PERFORMANCE_INDEX_SQL: &[&str] = &[ + ACTIVE_QUEUE_INDEX_SQL, + // Nonterminal queue probes must not scan every permanently parked collision + // on every idle pass. SQLite can also use this for retry frontier reads. + "CREATE INDEX IF NOT EXISTS idx_projection_queue_retry_deadline + ON projection_queue (next_retry_at_micros);", "CREATE INDEX IF NOT EXISTS idx_observation_projection_provenance_output ON observation_projection_provenance (projector_version, output_provider, output_message_id);", @@ -361,6 +376,12 @@ pub(crate) const OBSERVATION_PROJECTION_PERFORMANCE_INDEX_SQL: &[&str] = &[ pub(in super::super) async fn ensure_observation_projection_schema( conn: &impl Executor, ) -> Result<(), Error> { + // Existing queue shapes are immutable. A previous binary's queue cannot + // be upgraded behind the registered schema contract: refuse before DDL. + let queue_columns = projection_rebuild_column_names(conn, "projection_queue").await?; + if !queue_columns.is_empty() && !queue_columns.iter().any(|column| column == "attempt_count") { + return Err(unsupported_projection_schema("projection_queue")); + } conn.execute_batch(OBSERVATION_PROJECTION_SCHEMA_SQL) .await?; verify_final_projection_shape(conn).await?; @@ -623,6 +644,112 @@ mod tests { ); } + /// A queue missing its retry counter must remain untouched on refusal. + #[asupersync::test] + async fn incompatible_retry_queue_is_refused_without_schema_mutation() { + let temp = TempDir::new().unwrap(); + let path = temp.path().join("global.db"); + drop(open_registered_schema(&path).await.unwrap()); + + let conn = TestConnection::open(&path); + conn.execute_batch("ALTER TABLE projection_queue DROP COLUMN attempt_count") + .await + .unwrap(); + drop(conn); + + let Err(error) = open_registered_schema(&path).await else { + panic!("a prior retry queue must be refused, not upgraded in place"); + }; + let message = error.to_string(); + assert!( + message.contains("projection_queue"), + "the refusal must identify the incompatible queue: {message}" + ); + assert!( + message.contains("create a fresh one"), + "a prior schema must get the reset remedy: {message}" + ); + + let conn = TestConnection::open(&path); + let mut rows = conn + .query( + "SELECT name FROM pragma_table_xinfo('projection_queue') + WHERE name = 'attempt_count'", + (), + ) + .await + .unwrap(); + assert!( + rows.next().await.unwrap().is_none(), + "refused admission must preserve the old queue shape unchanged" + ); + } + + /// Real registered-store admission must accept the two queue indexes while + /// preserving the strict authority inventory and the partial-index contract. + #[asupersync::test] + async fn registered_projection_queue_indexes_match_authority_schema() { + let temp = TempDir::new().unwrap(); + let path = temp.path().join("global.db"); + drop(open_registered_schema(&path).await.unwrap()); + // Reopening validates the persisted index inventory, not just creation. + drop(open_registered_schema(&path).await.unwrap()); + + let conn = TestConnection::open(&path); + let mut rows = conn + .query( + "SELECT name, partial FROM pragma_index_list('projection_queue') + WHERE name IN ( + 'idx_projection_queue_active_sequence', + 'idx_projection_queue_retry_deadline' + ) ORDER BY name", + (), + ) + .await + .unwrap(); + let mut indexes = Vec::new(); + while let Some(row) = rows.next().await.unwrap() { + indexes.push((row.get::(0).unwrap(), row.get::(1).unwrap())); + } + assert_eq!( + indexes, + vec![ + ("idx_projection_queue_active_sequence".to_owned(), 1), + ("idx_projection_queue_retry_deadline".to_owned(), 0), + ], + "the registered store must retain both exact queue indexes" + ); + } + + #[asupersync::test] + async fn registered_projection_queue_rejects_wrong_partial_predicate() { + let temp = TempDir::new().unwrap(); + let path = temp.path().join("global.db"); + drop(open_registered_schema(&path).await.unwrap()); + + // IF NOT EXISTS must not make a same-named but incompatible index + // look admitted. The narrower predicate cannot serve a queue-head + // query that includes every nonterminal retry deadline. + let conn = TestConnection::open(&path); + conn.execute_batch( + "DROP INDEX idx_projection_queue_active_sequence; + CREATE INDEX idx_projection_queue_active_sequence + ON projection_queue (observation_sequence, next_retry_at_micros) + WHERE next_retry_at_micros < 0;", + ) + .await + .unwrap(); + drop(conn); + + let Err(error) = open_registered_schema(&path).await else { + panic!("a same-shaped queue index with the wrong predicate must be refused"); + }; + assert!( + error.to_string().contains("projection_queue"), + "schema refusal must identify the incompatible queue index: {error}" + ); + } + #[asupersync::test] async fn performance_indexes_install_outside_schema_transaction() { let temp = TempDir::new().unwrap(); @@ -630,7 +757,9 @@ mod tests { drop(open_registered_schema(&path).await.unwrap()); let conn = TestConnection::open(&path); conn.execute_batch( - "DROP INDEX idx_observation_projection_provenance_output; + "DROP INDEX idx_projection_queue_active_sequence; + DROP INDEX idx_projection_queue_retry_deadline; + DROP INDEX idx_observation_projection_provenance_output; DROP INDEX idx_observation_projection_provenance_global_output; DROP INDEX idx_observation_workflow_facts_query; DROP INDEX idx_observation_workflow_facts_item; @@ -656,6 +785,8 @@ mod tests { .query( "SELECT name FROM sqlite_schema WHERE name IN ( + 'idx_projection_queue_active_sequence', + 'idx_projection_queue_retry_deadline', 'idx_observation_projection_provenance_output', 'idx_observation_projection_provenance_global_output', 'idx_observation_workflow_facts_query', diff --git a/crates/tracedecay-global-db/src/observation_projection/state.rs b/crates/tracedecay-global-db/src/observation_projection/state.rs index 14763f15083..520f7cd5d1f 100644 --- a/crates/tracedecay-global-db/src/observation_projection/state.rs +++ b/crates/tracedecay-global-db/src/observation_projection/state.rs @@ -1,5 +1,6 @@ use std::borrow::Cow; use std::collections::{BTreeSet, HashMap}; + #[cfg(windows)] use std::path::{Component, Path, Prefix}; @@ -7,17 +8,16 @@ use tracedecay_domain::{ CanonicalObservationIdV1, DurableObservationV1, ObservationSourceGenerationV1, PayloadDigestV1, }; use tracedecay_store::{ - EDITED_FILES_KEY, ObservationProjection, PROJECTION_TERMINAL_RETRY_MICROS, - ProjectionCheckpoint, ProjectionStoreError, ProjectionStoreResult, - SESSION_MESSAGE_PROJECTOR_VERSION, SPAWNED_SESSIONS_KEY, SessionMessageProjection, - SessionMessageRecord, SessionRecord, message_output_digest, + EDITED_FILES_KEY, ObservationProjection, ProjectionCheckpoint, ProjectionStoreError, + ProjectionStoreResult, SESSION_MESSAGE_PROJECTOR_VERSION, SPAWNED_SESSIONS_KEY, + SessionMessageProjection, SessionMessageRecord, SessionRecord, message_output_digest, }; use tracedecay_lcm::raw::{message_body_from_record_row, stored_message_record_select_columns}; use tracedecay_lcm::retrieval_content::projected_content_hash; use tracedecay_lcm::{LcmError, LcmStorageKind}; -use tracedecay_runtime_core::db::Database; -use tracedecay_runtime_core::db::engine::{Executor, IntoParams, QueryExecutor, Row, params}; +use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor, Row, params}; +use tracedecay_runtime_core::db::{Database, DatabaseWriteTransaction}; use tracedecay_sessions::runtime::shared::durable_project_path_key; #[cfg(windows)] use tracedecay_sessions::runtime::shared::path_identity_key; @@ -257,17 +257,12 @@ pub(crate) struct RearmedProjectionRetries { } /// A projection retry deadline paces re-attempts within the store mount that -/// observed the failure. A fresh mount re-arms every queued projection for -/// immediate replay so restart recovery drains commit-before-ack work instead -/// of waiting out a dead process's backoff. Attempt counts and last errors -/// persist, so a projection that fails again resumes its escalating delay from -/// the recorded attempt history. +/// observed the failure. A fresh mount re-arms only previously deferred +/// observations. Fresh foreground retries must keep their own deadline. /// -/// The queue grows with session history, so the re-arm walks it in rowid -/// order, `batch_rows` deferred rows per short writer transaction: no single -/// statement or transaction scales with the queue, and foreground writes -/// interleave between batches. Rows queued after the walk starts belong to -/// this mount and are left alone. +/// A foreground retry carries its mount identity in the same transaction as +/// its deadline. Recovery can therefore release each snapshot before applying +/// its bounded page without clearing a retry scheduled by this mount. #[tracing::instrument( name = "global_db.observation_projection.rearm_retries", level = "trace", @@ -277,106 +272,241 @@ pub(crate) async fn rearm_queued_projection_retries( database: &Database, batch_rows: i64, ) -> ProjectionStoreResult { + if batch_rows <= 0 || batch_rows > REARM_PROJECTION_RETRY_BATCH_ROWS { + return Err(storage_message( + "rearm queued projection retries", + "projection retry batch size is outside the bounded range", + )); + } + let now = tracedecay_contracts::clock::now_micros().0; + let mut windows = RetryRearmWindows::new(database, batch_rows, now).await?; let mut rearmed = RearmedProjectionRetries::default(); - let Some(last_rowid) = query_optional_i64( - &database.read_connection(), - "SELECT MAX(rowid) FROM projection_queue", - params![], - ) - .await? - else { - return Ok(rearmed); - }; - let mut after_rowid = 0_i64; - while let Some((batch_end, rows)) = - rearm_projection_retry_batch(database, after_rowid, last_rowid, batch_rows).await? - { + while let Some(candidates) = windows.next().await? { + let rows = apply_rearm_projection_retry_candidates( + database, + &candidates, + tracedecay_contracts::clock::now_micros().0, + &windows.mount, + ) + .await?; rearmed.rows += rows; - rearmed.batches += 1; - after_rowid = batch_end; + if rows != 0 { + rearmed.batches += 1; + } } Ok(rearmed) } -/// Re-arms the next `batch_rows` deferred rows after `after_rowid` in one -/// writer transaction and returns the last rowid it covered. -#[tracing::instrument( - name = "global_db.observation_projection.rearm_retries.batch", - level = "trace", - skip_all -)] -async fn rearm_projection_retry_batch( - database: &Database, - after_rowid: i64, - last_rowid: i64, +/// The write authority distinguishes process lifetimes and the registry epoch +/// distinguishes remounts sharing the same authority. Only the digest is stored. +pub(super) fn retry_mount_identity(database: &Database) -> ProjectionStoreResult { + let authority = database + .write_authority() + .map_err(|error| storage("identify projection retry mount", error))?; + Ok(tracedecay_domain::canonical_text::canonical_framed_sha256( + b"tracedecay.projection-retry.mount", + &[ + authority.token().as_bytes(), + &database + .registered_binding() + .authority_epoch + .get() + .to_le_bytes(), + ], + )) +} + +pub(super) struct RetryRearmWindows<'a> { + database: &'a Database, batch_rows: i64, -) -> ProjectionStoreResult> { + pub(super) mount: String, + deadline: i64, + rowid: i64, + last: Option<(i64, i64)>, +} + +impl<'a> RetryRearmWindows<'a> { + pub(super) async fn new( + database: &'a Database, + batch_rows: i64, + now: i64, + ) -> ProjectionStoreResult { + const OPERATION: &str = "read projection retry frontier"; + let snapshot = database + .read_connection() + .convergence_snapshot() + .await + .map_err(|error| storage(OPERATION, error))?; + let mut rows = snapshot + .query( + // qpsg-exempt: the deadline ceiling excludes permanent collisions at the end of the index. + "SELECT next_retry_at_micros, rowid FROM projection_queue + INDEXED BY idx_projection_queue_retry_deadline + WHERE next_retry_at_micros > ?1 AND next_retry_at_micros < 9223372036854775807 + ORDER BY next_retry_at_micros DESC, rowid DESC LIMIT 1", + params![now], + ) + .await + .map_err(|error| storage(OPERATION, error))?; + let last = rows + .next() + .await + .map_err(|error| storage(OPERATION, error))? + .map(|row| Ok((row.get(0)?, row.get(1)?))) + .transpose() + .map_err(|error: tracedecay_runtime_core::db::engine::Error| { + storage(OPERATION, error) + })?; + Ok(Self { + database, + batch_rows, + mount: retry_mount_identity(database)?, + deadline: now, + rowid: i64::MAX, + last, + }) + } + + /// The deadline index includes rowid as its implicit last key. Split the + /// continuation into two index seeks so even many rows at the same + /// deadline do not repeatedly traverse the preceding pages. + pub(super) async fn next(&mut self) -> ProjectionStoreResult> { + const OPERATION: &str = "read projection retry rearm window"; + let Some(end) = self.last else { + return Ok(None); + }; + while (self.deadline, self.rowid) < end { + let snapshot = self + .database + .read_connection() + .convergence_snapshot() + .await + .map_err(|error| storage(OPERATION, error))?; + let mut rows = snapshot + .query( + // qpsg-exempt: both arms seek the retry deadline index, bounded by the captured nonterminal upper key. + "SELECT rowid AS id, next_retry_at_micros AS deadline, + CASE WHEN NOT EXISTS ( + SELECT 1 FROM projection_retry_owners AS owner + WHERE owner.observation_id = projection_queue.observation_id AND owner.mount = ?4 + ) THEN observation_id END, attempt_count + FROM projection_queue INDEXED BY idx_projection_queue_retry_deadline + WHERE next_retry_at_micros = ?1 AND rowid > ?2 + AND (next_retry_at_micros, rowid) <= (?5, ?6) + UNION ALL + SELECT rowid, next_retry_at_micros, + CASE WHEN NOT EXISTS ( + SELECT 1 FROM projection_retry_owners AS owner + WHERE owner.observation_id = projection_queue.observation_id AND owner.mount = ?4 + ) THEN observation_id END, attempt_count + FROM projection_queue INDEXED BY idx_projection_queue_retry_deadline + WHERE next_retry_at_micros > ?1 AND (next_retry_at_micros, rowid) <= (?5, ?6) + ORDER BY deadline, id LIMIT ?3", + params![ + self.deadline, + self.rowid, + self.batch_rows, + self.mount.as_str(), + end.0, + end.1 + ], + ) + .await + .map_err(|error| storage(OPERATION, error))?; + // Deadlines can become due while a large history recovery runs. + // Projection can already consume them without any rearm write. + let page_now = tracedecay_contracts::clock::now_micros().0; + let mut candidates = Vec::new(); + let mut last = None; + while let Some(row) = rows + .next() + .await + .map_err(|error| storage(OPERATION, error))? + { + let rowid: i64 = row.get(0).map_err(|error| storage(OPERATION, error))?; + let deadline: i64 = row.get(1).map_err(|error| storage(OPERATION, error))?; + let observation: Option = + row.get(2).map_err(|error| storage(OPERATION, error))?; + if deadline > page_now + && let Some(observation) = observation + { + let attempt: i64 = row.get(3).map_err(|error| storage(OPERATION, error))?; + candidates.push((rowid, observation, deadline, attempt)); + } + last = Some((deadline, rowid)); + } + let Some((deadline, rowid)) = last else { + return Ok(None); + }; + self.deadline = deadline; + self.rowid = rowid; + if !candidates.is_empty() { + return serde_json::to_string(&candidates) + .map(Some) + .map_err(|error| storage(OPERATION, error)); + } + } + Ok(None) + } +} + +/// Primary-key lookups make writer work proportional to this bounded page. +/// A newly retried, settled, terminal, or replaced observation cannot be +/// rearmed using a stale reader snapshot. A fully stale page rolls back. +pub(super) async fn apply_rearm_projection_retry_candidates( + database: &Database, + candidates: &str, + now_micros: i64, + mount: &str, +) -> ProjectionStoreResult { const OPERATION: &str = "rearm queued projection retries"; let transaction = database .begin_write_transaction(OPERATION) .await .map_err(|error| storage(OPERATION, error))?; - let Some(batch_end) = query_optional_i64( - &transaction, - "SELECT MAX(rowid) FROM ( - SELECT rowid FROM projection_queue - WHERE rowid > ?1 AND rowid <= ?2 AND next_retry_at_micros > 0 - ORDER BY rowid LIMIT ?3 - )", - params![after_rowid, last_rowid, batch_rows], - ) - .await? - else { - transaction - .rollback() - .await - .map_err(|error| storage(OPERATION, error))?; - return Ok(None); - }; let rows = transaction .execute( + // qpsg-exempt: the candidate list contains exact rowid identities, + // and the literal terminal bound protects a concurrent park. "UPDATE projection_queue SET next_retry_at_micros = 0 - WHERE rowid > ?1 AND rowid <= ?2 - AND next_retry_at_micros > 0 - AND next_retry_at_micros < ?3", - params![after_rowid, batch_end, PROJECTION_TERMINAL_RETRY_MICROS], + WHERE (rowid, observation_id, next_retry_at_micros, attempt_count) IN ( + SELECT json_extract(value, '$[0]'), + json_extract(value, '$[1]'), + json_extract(value, '$[2]'), + json_extract(value, '$[3]') + FROM json_each(?1) + ) + AND NOT EXISTS ( + SELECT 1 FROM projection_retry_owners AS owner + WHERE owner.observation_id = projection_queue.observation_id AND owner.mount = ?3 + ) + AND next_retry_at_micros > ?2 + AND next_retry_at_micros < 9223372036854775807", + params![candidates, now_micros, mount], ) .await .map_err(|error| storage(OPERATION, error))?; - transaction - .commit() - .await - .map_err(|error| storage(OPERATION, error))?; - Ok(Some((batch_end, rows))) -} - -pub(super) async fn query_optional_i64( - conn: &impl QueryExecutor, - sql: &str, - params: impl IntoParams, -) -> ProjectionStoreResult> { - const OPERATION: &str = "read projection retry rearm window"; - let mut rows = conn - .query(sql, params) - .await - .map_err(|error| storage(OPERATION, error))?; - let Some(row) = rows - .next() - .await - .map_err(|error| storage(OPERATION, error))? - else { - return Ok(None); - }; - row.get::>(0) - .map_err(|error| storage(OPERATION, error)) + if rows == 0 { + transaction + .rollback() + .await + .map_err(|error| storage(OPERATION, error))?; + } else { + transaction + .commit() + .await + .map_err(|error| storage(OPERATION, error))?; + } + Ok(rows) } pub(super) async fn schedule_projection_retry( - conn: &impl Executor, + conn: &DatabaseWriteTransaction<'_>, observation_id: &CanonicalObservationIdV1, attempt_count: u32, next_retry_at_micros: i64, last_error: &str, + mount: &str, ) -> ProjectionStoreResult<()> { let updated = conn .execute( @@ -392,11 +522,18 @@ pub(super) async fn schedule_projection_retry( ) .await .map_err(|error| storage("schedule projection retry", error))?; - if updated == 1 { - Ok(()) - } else { - Err(ProjectionStoreError::NotQueued) + if updated != 1 { + return Err(ProjectionStoreError::NotQueued); } + // Retry scheduling and ownership commit under the same queue writer. + conn.execute( + "INSERT INTO projection_retry_owners (observation_id, mount) VALUES (?1, ?2) + ON CONFLICT (observation_id) DO UPDATE SET mount = excluded.mount", + params![observation_id.as_str(), mount], + ) + .await + .map_err(|error| storage("record projection retry owner", error))?; + Ok(()) } pub(super) async fn consume_projection_queue_item( diff --git a/crates/tracedecay-global-db/src/registered_source_rewrite.rs b/crates/tracedecay-global-db/src/registered_source_rewrite.rs index c6d0cc566e5..16f6e56c187 100644 --- a/crates/tracedecay-global-db/src/registered_source_rewrite.rs +++ b/crates/tracedecay-global-db/src/registered_source_rewrite.rs @@ -14,17 +14,22 @@ use tracedecay_domain::{ ObservationSourceIdentityV1, }; use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor, params}; -use tracedecay_store::{ProjectionStoreError, ProjectionStoreResult}; +use tracedecay_store::{ + ProjectionStoreError, ProjectionStoreResult, SESSION_MESSAGE_PROJECTOR_VERSION, +}; use super::RegisteredGlobalDb; use super::observation_projection::retire_source_record; use super::registered::JobExit; +use tracedecay_session_temporal_store::request_session_temporal_resets; + +/// Each committed page settles at most this many source-presence records, +/// releasing the writer lease and making restart continuation durable. +const UNOFFERED_RECORD_PAGE_SIZE: i64 = 128; -/// 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 page is small enough for the SQLite JSON virtual table and writer +/// deadline. The entire supplied presence batch still commits atomically. +const SOURCE_PRESENCE_PAGE_ROWS: usize = 128; /// One record a source layout offered, under the identity admission derived. pub struct ObservationSourcePresenceV1 { @@ -94,21 +99,34 @@ impl RegisteredGlobalDb { .await .map_err(|error| storage("begin source presence transaction", error))?; let recorded = async { - for record in records { + for chunk in records.chunks(SOURCE_PRESENCE_PAGE_ROWS) { + let page = chunk + .iter() + .map(|record| { + Ok(( + record.observation_id.as_str(), + record.generation.generation_id().to_string(), + offset(record.start_offset)?, + )) + }) + .collect::>>()?; + let encoded = serde_json::to_string(&page) + .map_err(|error| storage("encode source presence page", error))?; transaction .execute( "INSERT INTO observation_source_presence ( source_key, observation_id, generation, start_offset - ) VALUES (?1, ?2, ?3, ?4) + ) + SELECT ?1, json_extract(value, '$[0]'), + json_extract(value, '$[1]'), json_extract(value, '$[2]') + FROM json_each(?2) AS input + WHERE true ORDER BY CAST(input.key AS INTEGER) ON CONFLICT(source_key, observation_id) DO UPDATE SET generation = excluded.generation, - start_offset = excluded.start_offset", - params![ - key.as_str(), - record.observation_id.as_str(), - record.generation.generation_id().to_string(), - offset(record.start_offset)?, - ], + start_offset = excluded.start_offset + WHERE observation_source_presence.generation IS NOT excluded.generation + OR observation_source_presence.start_offset IS NOT excluded.start_offset", + params![key.as_str(), encoded.as_str()], ) .await .map_err(|error| storage("record source presence", error))?; @@ -163,15 +181,14 @@ impl RegisteredGlobalDb { transaction .execute( "UPDATE projection_queue SET revival_generation = ?2 - WHERE EXISTS ( - SELECT 1 FROM observation_source_presence AS presence - WHERE presence.source_key = ?1 - AND presence.observation_id = projection_queue.observation_id - AND presence.start_offset < ?3 - AND presence.generation IN (?4, ?5) - AND presence.generation = projection_queue.revival_generation - AND presence.start_offset = projection_queue.revival_start_offset - )", + FROM observation_source_presence AS presence + WHERE presence.source_key = ?1 + AND presence.observation_id = projection_queue.observation_id + AND presence.start_offset < ?3 + AND presence.generation IN (?4, ?5) + AND presence.generation = projection_queue.revival_generation + AND presence.start_offset = projection_queue.revival_start_offset + AND presence.generation <> ?2", params![ key.as_str(), generation.as_str(), @@ -186,7 +203,8 @@ impl RegisteredGlobalDb { .execute( "UPDATE observation_source_presence SET generation = ?2 WHERE source_key = ?1 AND start_offset < ?3 - AND generation IN (?4, ?5)", + AND generation IN (?4, ?5) + AND generation <> ?2", params![ key.as_str(), generation.as_str(), @@ -204,7 +222,9 @@ impl RegisteredGlobalDb { ) VALUES (?1, ?2, ?3) ON CONFLICT(source_key) DO UPDATE SET generation = excluded.generation, - retained_through = excluded.retained_through", + retained_through = excluded.retained_through + WHERE observation_source_rewrites.generation IS NOT excluded.generation + OR observation_source_rewrites.retained_through IS NOT excluded.retained_through", params![key.as_str(), generation.as_str(), retained_through], ) .await @@ -239,81 +259,175 @@ impl RegisteredGlobalDb { return Ok(0); } drop(snapshot); - self.transact_or_rollback( - |error| storage("run source rewrite completion", error), - async move |transaction| { - if pending_rewrite(&transaction, &key).await?.as_deref() - != Some(generation.as_str()) - { - return Err(JobExit::Rollback(0)); - } - let mut retired = 0_u64; - // 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.as_str(), - 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 { - let retired_one: Pin< - Box> + Send + '_>, - > = Box::pin(retire_source_record(&transaction, observation_id)); - if retired_one.await? { - retired += 1; + let mut total_retired = 0_u64; + // Durably deleted pages survive restart; within this invocation seek + // past the already-visited current-generation survivor prefix. + let mut resume_from = String::new(); + loop { + // The remaining older-generation presence entries are the durable + // continuation frontier. Each page commits its retirements and + // deletes only the entries it has finished processing; a failed + // page rolls back without repeating earlier committed pages. + let key = key.clone(); + let generation = generation.clone(); + let after_id = resume_from.clone(); + let (retired, finished, next_cursor) = self + .transact_or_rollback( + |error| storage("run source rewrite completion page", error), + async move |transaction| { + if pending_rewrite(&transaction, &key).await?.as_deref() + != Some(generation.as_str()) + { + return Err(JobExit::Rollback((0_u64, true, String::new()))); } - } - if (page_len as i64) < UNOFFERED_RECORD_PAGE_SIZE { - break; - } - } - for sql in [ - "DELETE FROM observation_source_presence WHERE source_key = ?1 AND generation <> ?2", - "DELETE FROM observation_source_rewrites WHERE source_key = ?1 AND generation = ?2", - ] { - transaction - .execute(sql, params![key.as_str(), generation.as_str()]) - .await - .map_err(|error| storage("settle source rewrite", error))?; - } - Ok(retired) - }, - ) - .await + let mut rows = transaction + .query( + "SELECT presence.observation_id, + ( + ( + EXISTS ( + SELECT 1 FROM observations AS observation + WHERE observation.observation_id = presence.observation_id + ) + AND NOT EXISTS ( + SELECT 1 FROM observation_projection_dispositions AS disposition + WHERE disposition.projector_version = ?4 + AND disposition.observation_id = presence.observation_id + ) + ) + OR EXISTS ( + SELECT 1 FROM projection_queue AS queued + WHERE queued.observation_id = presence.observation_id + AND (queued.revival_generation IS NOT NULL + OR queued.revival_start_offset IS NOT NULL) + ) + ) + FROM observation_source_presence AS presence + WHERE presence.source_key = ?1 AND presence.generation <> ?2 + AND presence.observation_id >= ?5 + ORDER BY presence.observation_id + LIMIT (?3 + 1)", + params![ + key.as_str(), + generation.as_str(), + UNOFFERED_RECORD_PAGE_SIZE, + SESSION_MESSAGE_PROJECTOR_VERSION, + after_id.as_str(), + ], + ) + .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) + })?, + row.get::(1).map_err(|error| { + storage("read unoffered source records", error) + })? != 0, + )); + } + drop(rows); + // An extra read-only key establishes whether a full page + // was the final page. Never retire or delete the lookahead. + let has_next_page = page.len() > UNOFFERED_RECORD_PAGE_SIZE as usize; + if has_next_page { + page.truncate(UNOFFERED_RECORD_PAGE_SIZE as usize); + } + let mut retired = 0_u64; + let mut retired_ids = Vec::new(); + for (observation_id, needs_retirement) in &page { + if !needs_retirement { + continue; + } + let retired_one: Pin< + Box> + Send + '_>, + > = Box::pin(retire_source_record(&transaction, observation_id)); + if retired_one.await? { + retired += 1; + retired_ids.push(observation_id.as_str()); + } + } + // One source can yield many effects for the same session. + // Request one temporal reset per affected session in + // this page's transaction, rather than per observation. + request_session_temporal_resets(&transaction, &retired_ids).await?; + if let Some((last_id, _)) = page.last() { + transaction + .execute( + "DELETE FROM observation_source_presence + WHERE source_key = ?1 AND generation <> ?2 + AND observation_id >= ?4 AND observation_id <= ?3", + params![ + key.as_str(), + generation.as_str(), + last_id.as_str(), + after_id.as_str(), + ], + ) + .await + .map_err(|error| storage("settle source rewrite page", error))?; + } + // Another admitted writer can append an older-generation + // key behind this in-memory seek between commits. Recheck + // the entire source before dropping the durable marker. + let last_page = !has_next_page; + let finished = if last_page { + // With no seek this page covered the entire source + // under the writer lock, so no second scan is needed. + // A resumed page must check keys admitted behind it. + let empty = if after_id.is_empty() { + true + } else { + let mut remaining = transaction + .query( + "SELECT 1 FROM observation_source_presence + WHERE source_key = ?1 AND generation <> ?2 LIMIT 1", + params![key.as_str(), generation.as_str()], + ) + .await + .map_err(|error| storage("check remaining source records", error))?; + let empty = remaining + .next() + .await + .map_err(|error| storage("check remaining source records", error))? + .is_none(); + drop(remaining); + empty + }; + if empty { + transaction + .execute( + "DELETE FROM observation_source_rewrites + WHERE source_key = ?1 AND generation = ?2", + params![key.as_str(), generation.as_str()], + ) + .await + .map_err(|error| storage("settle source rewrite", error))?; + } + empty + } else { + false + }; + let next_cursor = if last_page { + String::new() + } else { + page.last().map(|(id, _)| id.clone()).unwrap_or_default() + }; + Ok((retired, finished, next_cursor)) + }, + ) + .await?; + total_retired += retired; + if finished { + return Ok(total_retired); + } + resume_from = next_cursor; + } } } diff --git a/crates/tracedecay-global-db/src/schema_contract/definitions.rs b/crates/tracedecay-global-db/src/schema_contract/definitions.rs index e18d02a9766..3a86b0d8083 100644 --- a/crates/tracedecay-global-db/src/schema_contract/definitions.rs +++ b/crates/tracedecay-global-db/src/schema_contract/definitions.rs @@ -511,6 +511,19 @@ pub(super) const TABLES: &[Table] = &[ "NO ACTION" )] ), + table!( + "projection_retry_owners", + [ + column("observation_id", "TEXT", true, None, 1), + column("mount", "TEXT", true, None, 0), + ], + [foreign_key( + "observation_id", + "projection_queue", + "observation_id", + "CASCADE" + )] + ), table!( "observation_projection_provenance", [ @@ -1806,6 +1819,20 @@ pub(super) const INDEXES: &[Index] = &[ origin: "u", columns: &["observation_sequence"], }, + Index { + table: "projection_queue", + name: Some("idx_projection_queue_active_sequence"), + unique: false, + origin: "c", + columns: &["observation_sequence", "next_retry_at_micros"], + }, + Index { + table: "projection_queue", + name: Some("idx_projection_queue_retry_deadline"), + unique: false, + origin: "c", + columns: &["next_retry_at_micros"], + }, Index { table: "observation_projection_provenance", name: Some("idx_observation_projection_provenance_output"), diff --git a/crates/tracedecay-global-db/src/schema_contract/invariants.rs b/crates/tracedecay-global-db/src/schema_contract/invariants.rs index f0b3aa17417..1ef9bb1191b 100644 --- a/crates/tracedecay-global-db/src/schema_contract/invariants.rs +++ b/crates/tracedecay-global-db/src/schema_contract/invariants.rs @@ -9,7 +9,7 @@ type ProjectionPage = (AuditCheckpoint, i64, i64, i64, bool); use tracedecay_store::SESSION_MESSAGE_PROJECTOR_VERSION; use super::super::global_db_operation_error; -use super::normalize_trigger_sql; +use super::normalize_schema_sql; use tracedecay_runtime_core::db::{ Database, DatabaseEngineReadSnapshot, DatabaseWriteTransaction, engine::{Executor, QueryExecutor, params}, diff --git a/crates/tracedecay-global-db/src/schema_contract/invariants/repair.rs b/crates/tracedecay-global-db/src/schema_contract/invariants/repair.rs index 0d8ec8f31a2..1d17a4aeb84 100644 --- a/crates/tracedecay-global-db/src/schema_contract/invariants/repair.rs +++ b/crates/tracedecay-global-db/src/schema_contract/invariants/repair.rs @@ -3,8 +3,8 @@ use std::collections::BTreeSet; use tracedecay_domain::{ DurableObservationV1, ObservationOrderingDomainV1, ObservationSourceCursorV1, }; +use tracedecay_store::SESSION_MESSAGE_PROJECTOR_VERSION; use tracedecay_store::observation::ObservationCoverageV1; -use tracedecay_store::{PROJECTION_TERMINAL_RETRY_MICROS, SESSION_MESSAGE_PROJECTOR_VERSION}; use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor, params}; use tracedecay_rusqlite_runtime::repository::observation_cursor_authority::PRUNE_SUPERSEDED_CURSOR_ADVANCES_SQL; @@ -151,9 +151,9 @@ pub(super) async fn repair_projection_frontier( .map_err(audit_read_error)?; conn.execute( "DELETE FROM projection_queue - WHERE observation_sequence <= ?1 AND next_retry_at_micros < ?2 + WHERE observation_sequence <= ?1 AND next_retry_at_micros < 9223372036854775807 AND revival_generation IS NULL", - params![repaired_checkpoint, PROJECTION_TERMINAL_RETRY_MICROS], + params![repaired_checkpoint], ) .await .map_err(audit_read_error)?; diff --git a/crates/tracedecay-global-db/src/schema_contract/invariants/triggers.rs b/crates/tracedecay-global-db/src/schema_contract/invariants/triggers.rs index d2a6a324c41..a7a036e61e5 100644 --- a/crates/tracedecay-global-db/src/schema_contract/invariants/triggers.rs +++ b/crates/tracedecay-global-db/src/schema_contract/invariants/triggers.rs @@ -5,7 +5,7 @@ use tracedecay_runtime_core::db::engine::{Executor, QueryExecutor, Result}; use super::rows::audit_read_error; -use super::normalize_trigger_sql; +use super::normalize_schema_sql; pub(in crate::schema_contract) struct Trigger { pub(in crate::schema_contract) name: &'static str, @@ -1684,7 +1684,7 @@ pub(in crate::schema_contract) async fn first_invalid_trigger( .get(&trigger.name.to_ascii_lowercase()) .is_some_and(|(table, sql)| { table.eq_ignore_ascii_case(trigger.table) - && normalize_trigger_sql(sql) == normalize_trigger_sql(trigger.create_sql) + && normalize_schema_sql(sql) == normalize_schema_sql(trigger.create_sql) }) })) } diff --git a/crates/tracedecay-global-db/src/schema_contract/mod.rs b/crates/tracedecay-global-db/src/schema_contract/mod.rs index 1aed76024bd..e24b9aecfed 100644 --- a/crates/tracedecay-global-db/src/schema_contract/mod.rs +++ b/crates/tracedecay-global-db/src/schema_contract/mod.rs @@ -161,7 +161,7 @@ pub(crate) fn starts_with_ignore_ascii_case(value: &str, prefix: &str) -> bool { .is_some_and(|head| head.eq_ignore_ascii_case(prefix.as_bytes())) } -fn normalize_trigger_sql(sql: &str) -> String { +fn normalize_schema_sql(sql: &str) -> String { sql.trim_end_matches(';') .split_whitespace() .collect::>() diff --git a/crates/tracedecay-global-db/src/schema_contract/pragma.rs b/crates/tracedecay-global-db/src/schema_contract/pragma.rs index 5de856441fe..765ae764efa 100644 --- a/crates/tracedecay-global-db/src/schema_contract/pragma.rs +++ b/crates/tracedecay-global-db/src/schema_contract/pragma.rs @@ -40,6 +40,7 @@ pub(super) struct ActualIndex { pub(super) unique: bool, pub(super) origin: String, pub(super) partial: bool, + pub(super) definition_sql: Option, pub(super) columns: Vec, } @@ -192,9 +193,12 @@ pub(super) async fn read_table_metadata( JOIN pragma_index_list(requested.table_name) AS p ) SELECT headers.table_name, headers.name, headers.\"unique\", - headers.origin, headers.partial, x.cid, x.name, x.desc, x.coll + headers.origin, headers.partial, x.cid, x.name, x.desc, x.coll, + declared.sql FROM headers JOIN pragma_index_xinfo(headers.name) AS x + LEFT JOIN sqlite_schema AS declared + ON declared.type = 'index' AND declared.name = headers.name WHERE x.key = 1 ORDER BY headers.table_name, headers.seq, x.seqno" ), @@ -225,6 +229,9 @@ pub(super) async fn read_table_metadata( .get::(4) .map_err(|error| global_db_operation_error(OPERATION, error))? != 0; + let definition_sql = row + .get::>(9) + .map_err(|error| global_db_operation_error(OPERATION, error))?; if current .as_ref() .is_some_and(|(current_table, index)| current_table != &table || index.name != name) @@ -249,6 +256,7 @@ pub(super) async fn read_table_metadata( unique, origin, partial, + definition_sql, columns: Vec::new(), }, ) diff --git a/crates/tracedecay-global-db/src/schema_contract/validation.rs b/crates/tracedecay-global-db/src/schema_contract/validation.rs index ae0a81e2d63..a7bded4bed7 100644 --- a/crates/tracedecay-global-db/src/schema_contract/validation.rs +++ b/crates/tracedecay-global-db/src/schema_contract/validation.rs @@ -10,11 +10,14 @@ use tracedecay_rusqlite_runtime::repository::{ GraphPublicationSchemaObjectV1, canonical_graph_publication_schema_inventory, }; +use crate::observation_projection::ACTIVE_QUEUE_INDEX_SQL; + use super::super::{global_db_operation_error, global_db_operation_message}; use super::definitions::{ Column, INDEX_DESCENDING_COLUMNS, INDEX_EXPRESSION_COLUMN, INDEXES, Index, REGISTRY_TABLE_NAMES, TABLES, Table, }; +use super::normalize_schema_sql; use super::pragma::{ ActualColumn, ActualForeignKey, ActualIndex, ActualTableMetadata, read_table_metadata, }; @@ -151,11 +154,25 @@ fn foreign_keys_match(actual: &[ActualForeignKey], contract: &Table) -> bool { }) } +// SQLite's partial-index boolean does not establish which rows the index +// covers. The queue's ordered-head plan depends on this exact terminal bound; +// a same-named index with a narrower predicate is not interchangeable. +fn queue_active_index_predicate_matches(definition_sql: Option<&str>) -> bool { + let Some(sql) = definition_sql else { + return false; + }; + // sqlite_schema stores CREATE INDEX without IF NOT EXISTS. Compare the + // complete stored definition so SQL comments cannot stand in for a predicate. + let expected = ACTIVE_QUEUE_INDEX_SQL.replacen(" IF NOT EXISTS", "", 1); + normalize_schema_sql(sql) == normalize_schema_sql(&expected) +} + fn index_matches(actual: &ActualIndex, expected: &Index) -> bool { let expected_partial = expected.name.is_some_and(|name| { name.eq_ignore_ascii_case("idx_session_temporal_generations_one_active") || name.eq_ignore_ascii_case("idx_session_refresh_operations_one_running") || name.eq_ignore_ascii_case("idx_observations_valid_session_sequence") + || name.eq_ignore_ascii_case("idx_projection_queue_active_sequence") }); if expected .name @@ -164,6 +181,8 @@ fn index_matches(actual: &ActualIndex, expected: &Index) -> bool { || !actual.origin.eq_ignore_ascii_case(expected.origin) || actual.partial != expected_partial || actual.columns.len() != expected.columns.len() + || (expected.name == Some("idx_projection_queue_active_sequence") + && !queue_active_index_predicate_matches(actual.definition_sql.as_deref())) { return false; } diff --git a/crates/tracedecay-global-db/src/tests/source_rewrite.rs b/crates/tracedecay-global-db/src/tests/source_rewrite.rs index 19672f47d8a..7fcc465491b 100644 --- a/crates/tracedecay-global-db/src/tests/source_rewrite.rs +++ b/crates/tracedecay-global-db/src/tests/source_rewrite.rs @@ -3,9 +3,8 @@ //! 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. +//! them in bounded writer pages. Each committed page must persist enough +//! progress that a restart does not repeat its already-retired records. use tempfile::TempDir; @@ -19,12 +18,12 @@ use tracedecay_store::{ProjectionSkipReason, ProjectionStoreError}; use crate::ObservationSourcePresenceV1; use crate::tests::harness::{ HostAdmissionScope, HostAdmissionTestRuntimeV1, RegisteredGlobalDbHarness, - seed_projected_messages, + seed_projected_messages, writer_telemetry, }; /// 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. +/// read and exercises repeated, independently committed writer pages. const UNOFFERED_RECORDS: i64 = 10_003; #[asupersync::test] @@ -38,21 +37,37 @@ async fn rewrite_completion_retires_unoffered_records_beyond_the_query_cap() { let key = serde_json::to_string(&(source.clone(), scope.clone())).unwrap(); let writer = harness.registered.writer_connection().unwrap(); - for index in 0..UNOFFERED_RECORDS { + // Keep the regression's 10,003 ghost records without turning fixture + // preparation into 10,003 serial writer round-trips. + for start in (0..UNOFFERED_RECORDS).step_by(128) { + let end = (start + 128).min(UNOFFERED_RECORDS); writer .execute( - "INSERT INTO observation_source_presence ( - source_key, observation_id, generation, start_offset - ) VALUES (?1, ?2, ?3, 0)", + "WITH RECURSIVE generated(i) AS ( + SELECT ?2 + UNION ALL + SELECT i + 1 FROM generated WHERE i + 1 < ?3 + ) + INSERT INTO observation_source_presence ( + source_key, observation_id, generation, start_offset + ) + SELECT ?1, printf('sha256:unoffered-%06d', i), ?4, 0 + FROM generated", params![ key.as_str(), - format!("sha256:unoffered-{index:06}"), + start, + end, previous.generation_id().to_string() ], ) .await .unwrap(); } + assert_eq!( + table_count(&writer, "SELECT COUNT(*) FROM observation_source_presence").await, + UNOFFERED_RECORDS, + "fixture must cover more rows than the query materialization cap" + ); harness .registered @@ -60,17 +75,32 @@ async fn rewrite_completion_retires_unoffered_records_beyond_the_query_cap() { .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. + // These are presence-only ghosts: no observations, dispositions, or + // queued revivals exist to retire. The final settle must still delete + // all presence rows, without issuing per-ghost UPDATE/receipt queries + // while holding the SQLite writer's fixed transaction lease. + let before = writer_telemetry(&harness.registered).sqlite_vm.statements; + let before_commits = writer_telemetry(&harness.registered) + .transactions + .committed_transactions; 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 statements = writer_telemetry(&harness.registered).sqlite_vm.statements - before; + let after = writer_telemetry(&harness.registered); + let commits = after.transactions.committed_transactions - before_commits; + assert_eq!( + commits, + (UNOFFERED_RECORDS as u64).div_ceil(128), + "each bounded page must durably commit before the next page" + ); + assert!( + statements.saturating_mul(16) < UNOFFERED_RECORDS as u64, + "presence-only ghosts must not cause per-record writer commands: {statements} statements for {UNOFFERED_RECORDS} records" + ); let mut remaining = writer .query( @@ -95,30 +125,223 @@ async fn rewrite_completion_retires_unoffered_records_beyond_the_query_cap() { assert_eq!(rewrites, 0, "completed rewrite row must be removed"); } +/// An exact multiple of the bounded page must retire its final records and +/// clear the pending marker in the same productive writer transaction. A +/// lookahead row may be read, but must never be retired in the previous page. +#[asupersync::test] +async fn rewrite_completion_exact_page_boundaries_need_no_empty_writer() { + for (unoffered, fixture) in [ + (128_i64, "source-rewrite-exact-128"), + (256, "source-rewrite-exact-256"), + (257, "source-rewrite-after-exact-256"), + ] { + let harness = RegisteredGlobalDbHarness::open(fixture).await; + let source = + ObservationSourceIdentityV1::new(SessionId::new("session.exact-pages").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(); + + writer + .execute( + "INSERT INTO observation_source_presence + (source_key, observation_id, generation, start_offset) + VALUES (?1, 'sha256:current', ?2, 0)", + params![key.as_str(), generation.generation_id().to_string()], + ) + .await + .unwrap(); + for start in (0..unoffered).step_by(128) { + let end = (start + 128).min(unoffered); + assert_eq!( + writer + .execute( + "WITH RECURSIVE generated(i) AS ( + SELECT ?2 + UNION ALL + SELECT i + 1 FROM generated WHERE i + 1 < ?3 + ) + INSERT INTO observation_source_presence + (source_key, observation_id, generation, start_offset) + SELECT ?1, printf('sha256:obsolete-%06d', i), ?4, 0 + FROM generated", + params![ + key.as_str(), + start, + end, + previous.generation_id().to_string(), + ], + ) + .await + .unwrap(), + u64::try_from(end - start).unwrap() + ); + } + + harness + .registered + .begin_observation_source_rewrite(&source, &scope, previous, generation, 0) + .await + .unwrap(); + let before = writer_telemetry(&harness.registered) + .transactions + .committed_transactions; + assert_eq!( + harness + .registered + .complete_observation_source_rewrite(&source, &scope, generation) + .await + .unwrap(), + 0, + "presence-only ghosts must not count as projected retirements" + ); + let commits = writer_telemetry(&harness.registered) + .transactions + .committed_transactions + - before; + assert_eq!( + commits, + u64::try_from(unoffered).unwrap().div_ceil(128), + "{fixture}: no empty writer transaction after an exact full page" + ); + assert_eq!( + table_count(&writer, "SELECT COUNT(*) FROM observation_source_rewrites").await, + 0, + "the durable pending rewrite marker must be cleared" + ); + assert_eq!( + table_count(&writer, "SELECT COUNT(*) FROM observation_source_presence").await, + 1, + "every obsolete identity must be retired" + ); + assert_eq!( + table_count( + &writer, + "SELECT COUNT(*) FROM observation_source_presence + WHERE observation_id = 'sha256:current' AND generation = '2'" + ) + .await, + 1, + "the current-generation survivor must remain" + ); + } +} + async fn table_count(conn: &impl QueryExecutor, sql: &str) -> i64 { let mut rows = conn.query(sql, ()).await.unwrap(); let row = rows.next().await.unwrap().unwrap(); row.get(0).unwrap() } -/// Retiring an unoffered record is real work, not a presence-row cleanup: -/// its provenance and workflow facts are removed, a `source_record_retired` -/// disposition takes their place, and a sole-owned projected output is -/// deleted. A completion that dies mid-transaction must roll every earlier -/// retirement back; the pending rewrite survives, so a retry after the -/// failure clears converges — the retry loop the paged read exists to keep -/// alive would otherwise stay wedged. -/// -/// The fixture plants an inconsistent provenance row (a surviving co-owner -/// that claims an output its durable record does not derive) on the -/// record the keyset order retires last, so the first attempt retires real -/// records across pages and then fails inside -/// `retire_source_record` with `ProvenanceCollision`. Removing the planted -/// row and completing again retires every unoffered record for real. +/// One source page can retire many effects of the same session. The +/// temporal reset is a session-level invalidation, not a per-record writer +/// upsert; an unchanged second update must be impossible in this page. +#[asupersync::test] +async fn rewrite_completion_coalesces_temporal_reset_for_one_session() { + let directory = TempDir::new().unwrap(); + let runtime = HostAdmissionTestRuntimeV1::profile(directory.path()) + .await + .unwrap(); + let seeded = seed_projected_messages(&runtime, 0..4).await; + let registered = runtime + .registered_database(HostAdmissionScope::Profile) + .expect("registered profile database"); + let source = seeded[0].0.source().clone(); + let scope = ObservationScopeV1::Profile; + let previous = ObservationSourceGenerationV1::new(1).unwrap(); + let generation = ObservationSourceGenerationV1::new(2).unwrap(); + let writer = registered.writer_connection().unwrap(); + + assert_eq!( + table_count( + &writer, + "SELECT COUNT(*) FROM session_temporal_observation_effects WHERE output_count > 0" + ) + .await, + 4, + "all projected observations must own materialized temporal effects" + ); + assert_eq!( + table_count(&writer, "SELECT COUNT(*) FROM session_temporal_resets").await, + 0, + "the fixture must not already have pending reset requests" + ); + + let offered: Vec<_> = seeded + .iter() + .map(|(observation, _)| ObservationSourcePresenceV1 { + observation_id: observation.observation_id().clone(), + generation: previous, + start_offset: observation.identity().position().start(), + }) + .collect(); + registered + .record_observation_source_presence(&source, &scope, &offered) + .await + .unwrap(); + registered + .begin_observation_source_rewrite(&source, &scope, previous, generation, 0) + .await + .unwrap(); + + // The previous per-observation upsert fails on its second update. The + // replacement must coalesce same-session effects but still retire all four. + writer + .execute( + "CREATE TRIGGER reject_redundant_temporal_reset + BEFORE UPDATE ON session_temporal_resets + BEGIN SELECT RAISE(ABORT, 'duplicate temporal reset'); END", + (), + ) + .await + .unwrap(); + + assert_eq!( + registered + .complete_observation_source_rewrite(&source, &scope, generation) + .await + .unwrap(), + 4 + ); + assert_eq!( + table_count(&writer, "SELECT COUNT(*) FROM session_temporal_resets").await, + 1, + "one session must receive exactly one durable reset request" + ); + assert_eq!( + table_count( + &writer, + "SELECT COUNT(*) FROM observation_projection_dispositions + WHERE reason = 'source_record_retired'" + ) + .await, + 4, + "coalescing must not skip retirement of any projected record" + ); + assert_eq!( + table_count(&writer, "SELECT COUNT(*) FROM observation_source_presence").await, + 0, + "all retired presence rows must be settled" + ); + assert_eq!( + table_count(&writer, "SELECT COUNT(*) FROM observation_source_rewrites").await, + 0, + "the rewrite marker must be committed with the reset" + ); +} + +/// Retiring an unoffered record also removes its projections and workflow +/// facts. A collision on a later page must roll that page back, but earlier +/// page commits must remain durable for the next daemon invocation. +/// Reopening the registered store must resume from those committed pages, +/// preserving the reoffered survivor and retiring each obsolete record once. #[asupersync::test] -async fn rewrite_completion_retires_projected_records_and_rolls_back_a_failed_attempt() { - // More real projected records than one keyset page, so a failed attempt - // leaves proven retirements to roll back on every page. +async fn rewrite_completion_resumes_committed_pages_after_failure_and_restart() { + // More real projected records than one page: prior page commits must + // remain durable when the final page's provenance collision aborts. const RECORDS: usize = 1_005; let directory = TempDir::new().unwrap(); @@ -128,7 +351,20 @@ async fn rewrite_completion_retires_projected_records_and_rolls_back_a_failed_at let registered = runtime .registered_database(HostAdmissionScope::Profile) .expect("registered profile database"); - let seeded = seed_projected_messages(&runtime, 0..RECORDS).await; + // Seed the full 1,005-observation production journey in bounded + // admission/projection windows. A single 1,005-item projection writer + // transaction in fixture setup obscures the restart behavior under test + // and can monopolize the writer before the rewrite even starts. + let mut seeded = Vec::with_capacity(RECORDS); + for start in (0..RECORDS).step_by(128) { + let end = (start + 128).min(RECORDS); + seeded.extend(seed_projected_messages(&runtime, start..end).await); + } + assert_eq!( + seeded.len(), + RECORDS, + "all source records must be projected" + ); let source = ObservationSourceIdentityV1::for_provider( ProviderId::new("codex").unwrap(), @@ -193,66 +429,100 @@ async fn rewrite_completion_retires_projected_records_and_rolls_back_a_failed_at // record's output. When the last record retires, the surviving owner is // read back and its stored rendering derives a different message, which // the retire path refuses with `ProvenanceCollision`. - let writer = registered.writer_connection().unwrap(); - writer - .execute( - "INSERT INTO observation_projection_provenance ( - projector_version, observation_id, output_ordinal, receipt_id, - output_provider, output_message_id, output_digest, message_created, - retrieval_anchor_id - ) - SELECT projector_version, observation_id, 1, receipt_id, - output_provider, ?1, output_digest, message_created, + let already_retired = { + let writer = registered.writer_connection().unwrap(); + writer + .execute( + "INSERT INTO observation_projection_provenance ( + projector_version, observation_id, output_ordinal, receipt_id, + output_provider, output_message_id, output_digest, message_created, retrieval_anchor_id - FROM observation_projection_provenance - WHERE observation_id = ?2", - params![ - collided_message_id.as_str(), - survivor.observation_id().as_str() - ], - ) - .await - .unwrap(); + ) + SELECT projector_version, observation_id, 1, receipt_id, + output_provider, ?1, output_digest, message_created, + retrieval_anchor_id + FROM observation_projection_provenance + WHERE observation_id = ?2", + params![ + collided_message_id.as_str(), + survivor.observation_id().as_str() + ], + ) + .await + .unwrap(); - let error = registered - .complete_observation_source_rewrite(&source, &scope, generation) - .await - .unwrap_err(); - assert!( - matches!(error, ProjectionStoreError::ProvenanceCollision), - "colliding output ownership must fail completion, got {error:?}" - ); + let before_commits = writer_telemetry(registered) + .transactions + .committed_transactions; + let error = registered + .complete_observation_source_rewrite(&source, &scope, generation) + .await + .unwrap_err(); + assert!( + matches!(error, ProjectionStoreError::ProvenanceCollision), + "colliding output ownership must fail completion, got {error:?}" + ); - // Every retirement the attempt already performed must be rolled back: - // provenance and projected outputs restored, no dispositions recorded, - // and the pending rewrite still waiting for a converging retry. - assert_eq!( - table_count( + // Earlier committed pages survive the failure; only the page containing + // the collision rolls back. This is the progress a new process must reuse. + let already_retired = table_count( &writer, - "SELECT COUNT(*) FROM observation_projection_dispositions" + "SELECT COUNT(*) FROM observation_projection_dispositions", ) - .await, - 0, - "retirements before the collision must roll back with the transaction" - ); + .await; + assert!( + already_retired >= 128 && already_retired < (RECORDS - 1) as i64, + "failure must preserve completed pages, but not the colliding page" + ); + let commits = writer_telemetry(registered) + .transactions + .committed_transactions + - before_commits; + assert!( + commits >= 1, + "at least one bounded retirement page must commit" + ); + assert_eq!( + table_count( + &writer, + "SELECT COUNT(*) FROM observation_projection_provenance", + ) + .await, + i64::try_from(RECORDS).unwrap() + 1 - already_retired, + "committed pages must remove their provenance" + ); + assert_eq!( + table_count(&writer, "SELECT COUNT(*) FROM lcm_raw_messages").await, + i64::try_from(RECORDS).unwrap() - already_retired, + "committed pages must remove only their projected outputs" + ); + assert_eq!( + table_count(&writer, "SELECT COUNT(*) FROM observation_source_rewrites").await, + 1, + "failed page must leave the rewrite pending for restart" + ); + already_retired + }; + + // Reopen the registered profile store as a new daemon fixture. Durable + // rewrite state, dispositions and the source-presence page frontier must + // survive rather than replay the already committed retirement work. + drop(runtime); + let runtime = HostAdmissionTestRuntimeV1::profile(directory.path()) + .await + .unwrap(); + let registered = runtime + .registered_database(HostAdmissionScope::Profile) + .expect("reopened registered profile database"); + let writer = registered.writer_connection().unwrap(); assert_eq!( table_count( &writer, - "SELECT COUNT(*) FROM observation_projection_provenance" + "SELECT COUNT(*) FROM observation_projection_dispositions", ) .await, - i64::try_from(RECORDS).unwrap() + 1, - "rolled-back retirements must restore provenance" - ); - assert_eq!( - table_count(&writer, "SELECT COUNT(*) FROM lcm_raw_messages").await, - i64::try_from(RECORDS).unwrap(), - "rolled-back retirements must restore projected outputs" - ); - assert_eq!( - table_count(&writer, "SELECT COUNT(*) FROM observation_source_rewrites").await, - 1, - "the failed attempt must leave the rewrite pending" + already_retired, + "restart must retain the completed retirement frontier" ); writer @@ -270,7 +540,11 @@ async fn rewrite_completion_retires_projected_records_and_rolls_back_a_failed_at .unwrap_or_else(|error| { panic!("retry after clearing the collision must converge: {error}") }); - assert_eq!(retired, u64::try_from(RECORDS - 1).unwrap()); + assert_eq!( + retired, + u64::try_from(RECORDS - 1).unwrap() - already_retired as u64, + "restart must retire only records that were not already settled" + ); let retired_reason = ProjectionSkipReason::SourceRecordRetired.as_str(); assert_eq!( @@ -320,3 +594,549 @@ async fn rewrite_completion_retires_projected_records_and_rolls_back_a_failed_at "retired records must request their session's temporal rebuild" ); } + +/// Beginning the same in-progress layout again must not rewrite retained +/// records, accepted revivals, or the unchanged pending rewrite marker. +/// SQL triggers reject every redundant UPDATE, while a real generation +/// transition must still advance all three authorities. +#[asupersync::test] +async fn repeated_source_rewrite_begin_preserves_already_advanced_prefix() { + let directory = TempDir::new().unwrap(); + let runtime = HostAdmissionTestRuntimeV1::profile(directory.path()) + .await + .unwrap(); + let registered = runtime + .registered_database(HostAdmissionScope::Profile) + .expect("registered profile database"); + let seeded = seed_projected_messages(&runtime, 0..2).await; + let source = + ObservationSourceIdentityV1::new(SessionId::new("session.repeat-prefix").unwrap()).unwrap(); + let scope = ObservationScopeV1::Profile; + let previous = ObservationSourceGenerationV1::new(1).unwrap(); + let generation = ObservationSourceGenerationV1::new(2).unwrap(); + let next = ObservationSourceGenerationV1::new(3).unwrap(); + let offered = seeded + .iter() + .map(|(observation, _)| ObservationSourcePresenceV1 { + observation_id: observation.observation_id().clone(), + generation: previous, + start_offset: observation.identity().position().start(), + }) + .collect::>(); + registered + .record_observation_source_presence(&source, &scope, &offered) + .await + .unwrap(); + let writer = registered.writer_connection().unwrap(); + writer + .execute( + "INSERT INTO projection_queue ( + observation_id, observation_sequence, revival_generation, revival_start_offset + ) + SELECT observation_id, sequence, ?2, ?3 + FROM observations WHERE observation_id = ?1 + ON CONFLICT(observation_id) DO UPDATE SET + revival_generation = excluded.revival_generation, + revival_start_offset = excluded.revival_start_offset", + params![ + offered[0].observation_id.as_str(), + previous.generation_id().to_string(), + i64::try_from(offered[0].start_offset).unwrap() + ], + ) + .await + .unwrap(); + + let limit = i64::MAX as u64; + registered + .begin_observation_source_rewrite(&source, &scope, previous, generation, limit) + .await + .unwrap(); + assert_eq!( + table_count( + &writer, + "SELECT COUNT(*) FROM observation_source_presence WHERE generation = '2'" + ) + .await, + 2, + "the first rewrite must advance both retained presence rows" + ); + assert_eq!( + table_count( + &writer, + "SELECT COUNT(*) FROM projection_queue WHERE revival_generation = '2'" + ) + .await, + 1, + "the first rewrite must advance the queued accepted revival" + ); + + // The second beginning is exactly what an interrupted host re-import + // repeats. Rejecting no-op updates on both tables proves the changed + // predicates actually prevent writer amplification, not just identical + // resulting values. + writer + .execute_batch( + "CREATE TRIGGER reject_unchanged_presence + BEFORE UPDATE OF generation ON observation_source_presence + WHEN NEW.generation IS OLD.generation + BEGIN SELECT RAISE(ABORT, 'repeated retained presence update'); END; + CREATE TRIGGER reject_unchanged_revival + BEFORE UPDATE OF revival_generation ON projection_queue + WHEN NEW.revival_generation IS OLD.revival_generation + BEGIN SELECT RAISE(ABORT, 'repeated accepted revival update'); END; + CREATE TRIGGER reject_unchanged_rewrite + BEFORE UPDATE ON observation_source_rewrites + WHEN NEW.generation IS OLD.generation + AND NEW.retained_through IS OLD.retained_through + BEGIN SELECT RAISE(ABORT, 'repeated pending rewrite update'); END;", + ) + .await + .unwrap(); + registered + .begin_observation_source_rewrite(&source, &scope, previous, generation, limit) + .await + .expect("restarting an identical layout must not update retained rows"); + assert_eq!( + table_count( + &writer, + "SELECT COUNT(*) FROM observation_source_presence WHERE generation = '2'" + ) + .await, + 2 + ); + assert_eq!( + table_count( + &writer, + "SELECT COUNT(*) FROM projection_queue WHERE revival_generation = '2'" + ) + .await, + 1 + ); + + registered + .begin_observation_source_rewrite(&source, &scope, generation, next, limit) + .await + .expect("a genuinely new rewrite must still advance the retained prefix"); + assert_eq!( + table_count( + &writer, + "SELECT COUNT(*) FROM observation_source_presence WHERE generation = '3'" + ) + .await, + 2 + ); + assert_eq!( + table_count( + &writer, + "SELECT COUNT(*) FROM projection_queue WHERE revival_generation = '3'" + ) + .await, + 1 + ); + assert_eq!( + table_count( + &writer, + "SELECT COUNT(*) FROM observation_source_rewrites WHERE generation = '3'" + ) + .await, + 1, + "a new source layout must still advance the pending rewrite" + ); +} + +/// Capturing one host page previously dispatched one SQLite statement per +/// record, even for reoffers identical to the already durable generation. +/// The JSON batch must preserve last-offer ordering, exact identity and +/// rollback while reducing writer commands to bounded pages. +#[asupersync::test] +async fn source_presence_pages_are_atomic_and_unchanged_reoffers_do_not_write() { + const RECORDS: usize = 257; + let harness = RegisteredGlobalDbHarness::open("source-presence-batched").await; + let source = + ObservationSourceIdentityV1::new(SessionId::new("session.presence-batched").unwrap()) + .unwrap(); + let scope = ObservationScopeV1::Profile; + let first = ObservationSourceGenerationV1::new(1).unwrap(); + let second = ObservationSourceGenerationV1::new(2).unwrap(); + let offered = (0..RECORDS) + .map(|index| ObservationSourcePresenceV1 { + observation_id: tracedecay_domain::CanonicalObservationIdV1::new(format!( + "sha256:{index:064x}" + )) + .unwrap(), + generation: first, + start_offset: index as u64, + }) + .collect::>(); + // SQLite accounts for the outer INSERT's rows when its statement ends. + // Subtract the probe's own writes so all rows from one statement share + // a marker, even though their AFTER INSERT trigger fires individually. + let setup = harness.registered.begin_write_transaction().await.unwrap(); + setup + .execute_batch( + "CREATE TABLE presence_page_probe (prior_changes INTEGER NOT NULL); + CREATE TRIGGER observe_presence_page AFTER INSERT ON observation_source_presence + BEGIN + INSERT INTO presence_page_probe VALUES ( + total_changes() - (SELECT COALESCE(MAX(rowid), 0) FROM presence_page_probe) + ); + END;", + ) + .await + .unwrap(); + setup.commit().await.unwrap(); + let statements = || writer_telemetry(&harness.registered).sqlite_vm.statements; + // Measure the registered writer's own transaction bookkeeping on this shard. + let before_empty = statements(); + let empty = harness.registered.begin_write_transaction().await.unwrap(); + empty.commit().await.unwrap(); + let transaction_overhead = statements() - before_empty; + let before = statements(); + harness + .registered + .record_observation_source_presence(&source, &scope, &offered) + .await + .unwrap(); + assert_eq!(statements() - before, transaction_overhead + 3); + let snapshot = harness.registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT COUNT(*) FROM presence_page_probe GROUP BY prior_changes ORDER BY prior_changes", + (), + ) + .await + .unwrap(); + let mut pages = Vec::new(); + while let Some(row) = rows.next().await.unwrap() { + pages.push(row.get::(0).unwrap()); + } + assert_eq!(pages, [128, 128, 1]); + drop(rows); + drop(snapshot); + + // A SQLite trigger makes a redundant UPDATE observable and fatal, even + // if SQLite would have written unchanged column values. + let transaction = harness.registered.begin_write_transaction().await.unwrap(); + transaction + .execute( + "CREATE TRIGGER forbid_identical_presence_reoffer + BEFORE UPDATE ON observation_source_presence + WHEN NEW.generation = OLD.generation + AND NEW.start_offset = OLD.start_offset + BEGIN SELECT RAISE(ABORT, 'identical presence reoffer'); END", + (), + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + let before = statements(); + harness + .registered + .record_observation_source_presence(&source, &scope, &offered) + .await + .unwrap(); + assert_eq!(statements() - before, transaction_overhead + 3); + + let mut replacement = offered + .iter() + .map(|record| ObservationSourcePresenceV1 { + observation_id: record.observation_id.clone(), + generation: second, + start_offset: record.start_offset, + }) + .collect::>(); + replacement.last_mut().unwrap().start_offset = u64::MAX; + assert!( + harness + .registered + .record_observation_source_presence(&source, &scope, &replacement) + .await + .is_err(), + "a malformed final-page offset must roll back all preceding pages" + ); + + let key = serde_json::to_string(&(source.clone(), scope.clone())).unwrap(); + let snapshot = harness.registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT COUNT(*), SUM(generation = '2') + FROM observation_source_presence WHERE source_key = ?1", + params![key.as_str()], + ) + .await + .unwrap(); + let row = rows.next().await.unwrap().unwrap(); + assert_eq!(row.get::(0).unwrap(), RECORDS as i64); + assert_eq!( + row.get::(1).unwrap(), + 0, + "the failing last page must not partially publish any generation changes" + ); + drop(rows); + drop(snapshot); + + replacement.last_mut().unwrap().start_offset = (RECORDS - 1) as u64; + // A duplicate in the final page must keep the last offer's offset rather + // than the first occurrence in an earlier page. + replacement.push(ObservationSourcePresenceV1 { + observation_id: replacement[0].observation_id.clone(), + generation: second, + start_offset: 10_000, + }); + harness + .registered + .record_observation_source_presence(&source, &scope, &replacement) + .await + .unwrap(); + let snapshot = harness.registered.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT generation, start_offset FROM observation_source_presence + WHERE source_key = ?1 AND observation_id = ?2", + params![key.as_str(), replacement[0].observation_id.as_str()], + ) + .await + .unwrap(); + let row = rows.next().await.unwrap().unwrap(); + assert_eq!(row.get::(0).unwrap(), "2"); + assert_eq!(row.get::(1).unwrap(), 10_000); + assert!(rows.next().await.unwrap().is_none()); +} + +/// Once retirement has committed, a reoffered source can still carry old +/// presence entries. Repeating their retirement must not perform an UPDATE +/// and receipt read for every already-disposed observation. This exercises +/// real projected records and the canonical disposition/queue authorities. +#[asupersync::test] +async fn rewrite_completion_skips_settled_records_without_revival() { + async fn completed_writer_statements(records: usize) -> u64 { + let directory = TempDir::new().unwrap(); + let runtime = HostAdmissionTestRuntimeV1::profile(directory.path()) + .await + .unwrap(); + let registered = runtime + .registered_database(HostAdmissionScope::Profile) + .expect("registered profile database"); + let seeded = seed_projected_messages(&runtime, 0..records).await; + let source = ObservationSourceIdentityV1::for_provider( + ProviderId::new("codex").unwrap(), + SessionId::new("session.settled-rewrite").unwrap(), + ) + .unwrap(); + let scope = ObservationScopeV1::Profile; + let first = ObservationSourceGenerationV1::new(1).unwrap(); + let second = ObservationSourceGenerationV1::new(2).unwrap(); + let third = ObservationSourceGenerationV1::new(3).unwrap(); + let offered = seeded + .iter() + .map(|(observation, _)| ObservationSourcePresenceV1 { + observation_id: observation.observation_id().clone(), + generation: first, + start_offset: observation.identity().position().start(), + }) + .collect::>(); + registered + .record_observation_source_presence(&source, &scope, &offered) + .await + .unwrap(); + registered + .begin_observation_source_rewrite(&source, &scope, first, second, 0) + .await + .unwrap(); + assert_eq!( + registered + .complete_observation_source_rewrite(&source, &scope, second) + .await + .unwrap(), + records as u64 + ); + + let replay = offered + .iter() + .map(|record| ObservationSourcePresenceV1 { + observation_id: record.observation_id.clone(), + generation: second, + start_offset: record.start_offset, + }) + .collect::>(); + registered + .record_observation_source_presence(&source, &scope, &replay) + .await + .unwrap(); + registered + .begin_observation_source_rewrite(&source, &scope, second, third, 0) + .await + .unwrap(); + + // A settled observation can still have a newly accepted revival. + // The optimization must clear this queued offer before skipping the + // retired disposition; otherwise projection can resurrect old output. + let writer = registered.writer_connection().unwrap(); + writer + .execute( + "INSERT INTO projection_queue ( + observation_id, observation_sequence, revival_generation, revival_start_offset + ) + SELECT observation_id, sequence, '2', 0 + FROM observations WHERE observation_id = ?1 + ON CONFLICT(observation_id) DO UPDATE SET + revival_generation = excluded.revival_generation, + revival_start_offset = excluded.revival_start_offset", + params![seeded[0].0.observation_id().as_str()], + ) + .await + .unwrap(); + + let statements = || writer_telemetry(registered).sqlite_vm.statements; + let before = statements(); + assert_eq!( + registered + .complete_observation_source_rewrite(&source, &scope, third) + .await + .unwrap(), + 0, + "a previously retired observation must not be retired twice" + ); + let writer_commands = statements() - before; + let snapshot = registered.read_snapshot().await.unwrap(); + let key = serde_json::to_string(&(source.clone(), scope.clone())).unwrap(); + let mut remaining = snapshot + .query( + "SELECT COUNT(*) FROM observation_source_presence WHERE source_key = ?1", + params![key.as_str()], + ) + .await + .unwrap(); + let row = remaining.next().await.unwrap().unwrap(); + assert_eq!(row.get::(0).unwrap(), 0); + drop(remaining); + let mut pending = snapshot + .query( + "SELECT COUNT(*) FROM observation_source_rewrites WHERE source_key = ?1", + params![key.as_str()], + ) + .await + .unwrap(); + let row = pending.next().await.unwrap().unwrap(); + assert_eq!(row.get::(0).unwrap(), 0); + drop(pending); + let mut revival = snapshot + .query( + "SELECT revival_generation, revival_start_offset + FROM projection_queue WHERE observation_id = ?1", + params![seeded[0].0.observation_id().as_str()], + ) + .await + .unwrap(); + let row = revival.next().await.unwrap().unwrap(); + assert_eq!(row.get::>(0).unwrap(), None); + assert_eq!(row.get::>(1).unwrap(), None); + writer_commands + } + + let small = completed_writer_statements(8).await; + let large = completed_writer_statements(64).await; + assert!(small > 0); + assert_eq!( + large, small, + "eightfold more settled records must not create per-record writer commands" + ); +} + +/// With a large current-generation prefix, each older-generation page must +/// seek beyond already checked survivors instead of walking them repeatedly. +async fn source_rewrite_steps_with_survivor_prefix(unoffered: usize) -> u64 { + const SURVIVORS: usize = 4_096; + let harness = RegisteredGlobalDbHarness::open("source-rewrite-seek-prefix").await; + let source = + ObservationSourceIdentityV1::new(SessionId::new("session.seek-prefix").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 old_id = previous.generation_id().to_string(); + let new_id = generation.generation_id().to_string(); + let writer = harness.registered.writer_connection().unwrap(); + for start in (0..(SURVIVORS + unoffered)).step_by(128) { + let end = (start + 128).min(SURVIVORS + unoffered); + let values = (start..end) + .map(|index| { + ( + format!("sha256:{index:06}"), + if index < SURVIVORS { + new_id.as_str() + } else { + old_id.as_str() + }, + ) + }) + .collect::>(); + let encoded = serde_json::to_string(&values).unwrap(); + writer + .execute( + "INSERT INTO observation_source_presence ( + source_key, observation_id, generation, start_offset + ) SELECT ?1, json_extract(value, '$[0]'), + json_extract(value, '$[1]'), 0 FROM json_each(?2)", + params![key.as_str(), encoded.as_str()], + ) + .await + .unwrap(); + } + harness + .registered + .begin_observation_source_rewrite(&source, &scope, previous, generation, 0) + .await + .unwrap(); + let before = writer_telemetry(&harness.registered); + assert_eq!( + harness + .registered + .complete_observation_source_rewrite(&source, &scope, generation) + .await + .unwrap(), + 0, + "presence-only rows have no projected observations to retire" + ); + let after = writer_telemetry(&harness.registered); + assert_eq!( + after.transactions.committed_transactions - before.transactions.committed_transactions, + unoffered.div_ceil(128) as u64, + "each bounded page must commit independently" + ); + assert_eq!( + table_count( + &writer, + &format!( + "SELECT COUNT(*) FROM observation_source_presence WHERE generation = '{old_id}'" + ), + ) + .await, + 0, + "all unoffered older records must be deleted" + ); + assert_eq!( + table_count( + &writer, + &format!( + "SELECT COUNT(*) FROM observation_source_presence WHERE generation = '{new_id}'" + ), + ) + .await, + SURVIVORS as i64, + "the current-generation survivors must remain untouched" + ); + after.sqlite_vm.vm_steps - before.sqlite_vm.vm_steps +} + +#[asupersync::test] +async fn source_rewrite_pages_do_not_rescan_the_same_survivor_prefix() { + let one_page_steps = source_rewrite_steps_with_survivor_prefix(127).await; + let eight_page_steps = source_rewrite_steps_with_survivor_prefix(1_023).await; + assert!( + eight_page_steps < one_page_steps.saturating_mul(3), + "eight retirement pages must seek beyond the 4096 survivor records: \ + one page {one_page_steps} VM steps; eight pages {eight_page_steps}" + ); +} diff --git a/crates/tracedecay-rusqlite-runtime/src/repository/observation/mod.rs b/crates/tracedecay-rusqlite-runtime/src/repository/observation/mod.rs index fd152032664..9dcc1586d54 100644 --- a/crates/tracedecay-rusqlite-runtime/src/repository/observation/mod.rs +++ b/crates/tracedecay-rusqlite-runtime/src/repository/observation/mod.rs @@ -16,9 +16,9 @@ use tracedecay_store::{ AnchoredObservationWrite, CanonicalBodyError, CursorAdvanceLedgerDisagreementV1, CursorAdvanceLedgerIdentityV1, LOAD_CANONICAL_BODY_SQL, ObservationCoverageReason, ObservationCursorAdvance, ObservationReadOperationV1, ObservationReadResultV1, - PROJECTION_TERMINAL_RETRY_MICROS, ProjectionRebuildProgressV1, ProjectionRebuildStateV1, - ProjectionSkipReason, SESSION_MESSAGE_PROJECTOR_VERSION, UPSERT_CANONICAL_BODY_SQL, - parse_stored_observation, slim_stored_json, + ProjectionRebuildProgressV1, ProjectionRebuildStateV1, ProjectionSkipReason, + SESSION_MESSAGE_PROJECTOR_VERSION, UPSERT_CANONICAL_BODY_SQL, parse_stored_observation, + slim_stored_json, }; use crate::operation::StorageOperationError; @@ -333,27 +333,22 @@ impl ObservationExecutor { Ok(ObservationReadResultV1::Replay(observations)) } ObservationReadOperationV1::NextQueuedProjection { now_micros } => { + // Read the first nonterminal head through the stable + // partial index. A deferred head fences later due rows. let observation_id = snapshot .query_row_cached( - "SELECT observation_id FROM projection_queue - WHERE next_retry_at_micros <= ?2 - AND observation_sequence = ( - SELECT MIN(observation_sequence) FROM projection_queue - WHERE next_retry_at_micros < ?3 - ) + "SELECT observation_id, next_retry_at_micros FROM projection_queue + WHERE next_retry_at_micros < 9223372036854775807 AND NOT EXISTS ( - SELECT 1 FROM observation_projection_rebuilds - WHERE projector_version = ?1 - ) - LIMIT 1", - ( - SESSION_MESSAGE_PROJECTOR_VERSION, - now_micros, - PROJECTION_TERMINAL_RETRY_MICROS, - ), - |row| row.get::<_, String>(0), + SELECT 1 FROM observation_projection_rebuilds + WHERE projector_version = ?1 + ) + ORDER BY observation_sequence LIMIT 1", + [SESSION_MESSAGE_PROJECTOR_VERSION], + |row| Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?)), ) .optional()? + .and_then(|(id, deadline)| (deadline <= *now_micros).then_some(id)) .map(CanonicalObservationIdV1::new) .transpose() .map_err(invalid)?; diff --git a/crates/tracedecay-rusqlite-runtime/src/telemetry.rs b/crates/tracedecay-rusqlite-runtime/src/telemetry.rs index ef445b5eb5d..0e54152b167 100644 --- a/crates/tracedecay-rusqlite-runtime/src/telemetry.rs +++ b/crates/tracedecay-rusqlite-runtime/src/telemetry.rs @@ -134,6 +134,8 @@ pub struct WriterTransactionMetrics { #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] pub struct SqliteVmSnapshot { + /// Completed prepared executions observed by the runtime, including reads. + pub statements: u64, pub fullscan_steps: u64, pub sort_steps: u64, pub vm_steps: u64, @@ -142,6 +144,7 @@ pub struct SqliteVmSnapshot { impl SqliteVmSnapshot { pub(crate) fn saturating_add(self, other: Self) -> Self { Self { + statements: self.statements.saturating_add(other.statements), fullscan_steps: self.fullscan_steps.saturating_add(other.fullscan_steps), sort_steps: self.sort_steps.saturating_add(other.sort_steps), vm_steps: self.vm_steps.saturating_add(other.vm_steps), diff --git a/crates/tracedecay-rusqlite-runtime/src/telemetry/reader.rs b/crates/tracedecay-rusqlite-runtime/src/telemetry/reader.rs index 6d3efba5ebc..a210d1a95cd 100644 --- a/crates/tracedecay-rusqlite-runtime/src/telemetry/reader.rs +++ b/crates/tracedecay-rusqlite-runtime/src/telemetry/reader.rs @@ -16,6 +16,7 @@ pub(crate) struct ReaderAdmissionRecorder { release_events: AtomicU64, wait_micros: AtomicU64, execution_micros: AtomicU64, + statements: AtomicU64, fullscan_steps: AtomicU64, sort_steps: AtomicU64, vm_steps: AtomicU64, @@ -51,6 +52,8 @@ impl ReaderAdmissionRecorder { pub(crate) fn executed(&self, elapsed: Duration, sqlite_vm: SqliteVmSnapshot) { self.execution_micros .fetch_add(duration_micros(elapsed), Ordering::Relaxed); + self.statements + .fetch_add(sqlite_vm.statements, Ordering::Relaxed); self.fullscan_steps .fetch_add(sqlite_vm.fullscan_steps, Ordering::Relaxed); self.sort_steps @@ -69,6 +72,7 @@ impl ReaderAdmissionRecorder { wait_micros: self.wait_micros.load(Ordering::Relaxed), execution_micros: self.execution_micros.load(Ordering::Relaxed), sqlite_vm: SqliteVmSnapshot { + statements: self.statements.load(Ordering::Relaxed), fullscan_steps: self.fullscan_steps.load(Ordering::Relaxed), sort_steps: self.sort_steps.load(Ordering::Relaxed), vm_steps: self.vm_steps.load(Ordering::Relaxed), @@ -91,6 +95,7 @@ mod tests { recorder.executed( Duration::from_micros(11), SqliteVmSnapshot { + statements: 1, fullscan_steps: 2, sort_steps: 1, vm_steps: 9, @@ -104,6 +109,7 @@ mod tests { assert_eq!(snapshot.release_events, 1); assert_eq!(snapshot.wait_micros, 10); assert_eq!(snapshot.execution_micros, 11); + assert_eq!(snapshot.sqlite_vm.statements, 1); assert_eq!(snapshot.sqlite_vm.fullscan_steps, 2); assert_eq!(snapshot.sqlite_vm.sort_steps, 1); assert_eq!(snapshot.sqlite_vm.vm_steps, 9); diff --git a/crates/tracedecay-rusqlite-runtime/src/telemetry/sqlite_vm.rs b/crates/tracedecay-rusqlite-runtime/src/telemetry/sqlite_vm.rs index 5600f0d0890..90fc2641d3c 100644 --- a/crates/tracedecay-rusqlite-runtime/src/telemetry/sqlite_vm.rs +++ b/crates/tracedecay-rusqlite-runtime/src/telemetry/sqlite_vm.rs @@ -10,6 +10,7 @@ use super::SqliteVmSnapshot; thread_local! { static OBSERVED: Cell = const { Cell::new(SqliteVmSnapshot { + statements: 0, fullscan_steps: 0, sort_steps: 0, vm_steps: 0, @@ -18,6 +19,9 @@ thread_local! { pub(crate) fn observe_statement(statement: &Statement<'_>) { let delta = SqliteVmSnapshot { + // SQLITE_STMTSTATUS_RUN counts trigger subprograms as additional runs. + // This boundary observes the caller's prepared statement once. + statements: 1, fullscan_steps: take_status(statement, StatementStatus::FullscanStep), sort_steps: take_status(statement, StatementStatus::Sort), vm_steps: take_status(statement, StatementStatus::VmStep), @@ -64,6 +68,7 @@ mod tests { assert_eq!(count, 2); observe_statement(&statement); let snapshot = take_observed_vm(); + assert_eq!(snapshot.statements, 1); assert!( snapshot.fullscan_steps > 0, "unindexed table scan must increment fullscan" diff --git a/crates/tracedecay-rusqlite-runtime/src/telemetry/tests.rs b/crates/tracedecay-rusqlite-runtime/src/telemetry/tests.rs index 190ff459462..cbdb1de3424 100644 --- a/crates/tracedecay-rusqlite-runtime/src/telemetry/tests.rs +++ b/crates/tracedecay-rusqlite-runtime/src/telemetry/tests.rs @@ -72,6 +72,7 @@ fn transaction_close_records_commands_rows_lock_and_outcomes() { lock_held_micros: 4, transaction_micros: 9, sqlite_vm: SqliteVmSnapshot { + statements: 2, fullscan_steps: 1, sort_steps: 2, vm_steps: 6, @@ -111,6 +112,7 @@ fn transaction_close_records_commands_rows_lock_and_outcomes() { snapshot.transactions.lock_held_micros <= snapshot.transactions.transaction_micros, "lock-held time cannot exceed total transaction time" ); + assert_eq!(snapshot.sqlite_vm.statements, 2); assert_eq!(snapshot.sqlite_vm.fullscan_steps, 1); assert_eq!(snapshot.sqlite_vm.sort_steps, 2); assert_eq!(snapshot.sqlite_vm.vm_steps, 6); diff --git a/crates/tracedecay-session-temporal-store/src/lib.rs b/crates/tracedecay-session-temporal-store/src/lib.rs index 59fe925f254..5f8f5d27272 100644 --- a/crates/tracedecay-session-temporal-store/src/lib.rs +++ b/crates/tracedecay-session-temporal-store/src/lib.rs @@ -116,7 +116,7 @@ pub use doctor_health::{ pub use projection::{ record_canonical_observation_effect, record_live_collision_preserving_canonical_observation_effect, request_session_temporal_reset, - request_session_temporal_reset_if_materialized, + request_session_temporal_reset_if_materialized, request_session_temporal_resets, }; pub use refresh::{ SessionRefreshBeginBatchItemV1, SessionRefreshBeginBatchOutcomeV1, SessionRefreshBeginPlanV1, diff --git a/crates/tracedecay-session-temporal-store/src/projection.rs b/crates/tracedecay-session-temporal-store/src/projection.rs index 8477f47d373..6981f8b8208 100644 --- a/crates/tracedecay-session-temporal-store/src/projection.rs +++ b/crates/tracedecay-session-temporal-store/src/projection.rs @@ -30,7 +30,7 @@ pub(crate) use receipts::{base_source_frontier, digest_bytes}; pub use receipts::{ record_canonical_observation_effect, record_live_collision_preserving_canonical_observation_effect, request_session_temporal_reset, - request_session_temporal_reset_if_materialized, + request_session_temporal_reset_if_materialized, request_session_temporal_resets, }; const DISCOVER_REFRESH: &str = "discover session temporal refresh"; diff --git a/crates/tracedecay-session-temporal-store/src/projection/receipts.rs b/crates/tracedecay-session-temporal-store/src/projection/receipts.rs index 99d2478555a..39614b51b7e 100644 --- a/crates/tracedecay-session-temporal-store/src/projection/receipts.rs +++ b/crates/tracedecay-session-temporal-store/src/projection/receipts.rs @@ -384,6 +384,38 @@ pub async fn request_session_temporal_reset( Ok(()) } +/// Coalesces temporal resets for a bounded source-rewrite retirement page. +/// The caller atomically commits these requests with retired projections and +/// the durable page frontier. Different sessions retain distinct requests. +#[tracing::instrument( + name = "session_temporal.persist.reset_request_batch", + level = "trace", + skip_all +)] +pub async fn request_session_temporal_resets( + conn: &impl Executor, + observation_ids: &[&str], +) -> ProjectionStoreResult<()> { + if observation_ids.is_empty() { + return Ok(()); + } + let encoded = serde_json::to_string(observation_ids) + .map_err(|error| ProjectionStoreError::storage("encode temporal reset page", error))?; + conn.execute( + "INSERT INTO session_temporal_resets (session_id, requested_at) + SELECT DISTINCT effect.session_id, unixepoch() * 1000000 + FROM json_each(?1) AS retired + JOIN session_temporal_observation_effects AS effect + ON effect.observation_id = retired.value + WHERE effect.output_count > 0 + ON CONFLICT(session_id) DO UPDATE SET requested_at = excluded.requested_at", + params![encoded.as_str()], + ) + .await + .map_err(|error| ProjectionStoreError::storage("request temporal reset page", error))?; + Ok(()) +} + /// [`request_session_temporal_reset`] for an observation whose output was /// retired, limited to a session where a live or pending generation already /// covers it. A refresh past that frontier skips the retired effect on its diff --git a/crates/tracedecay/tests/session_suite/observation_projection/queue.rs b/crates/tracedecay/tests/session_suite/observation_projection/queue.rs index 8308e448ad9..bd51932b777 100644 --- a/crates/tracedecay/tests/session_suite/observation_projection/queue.rs +++ b/crates/tracedecay/tests/session_suite/observation_projection/queue.rs @@ -568,3 +568,452 @@ async fn stale_exact_duplicate_queue_item_is_consumed_before_later_observation() ); assert_eq!(projection_counts(&tmp).await, (1, 2, 2, 1, 0, 0)); } + +#[tokio::test] +async fn batched_projection_rearms_the_deferred_head_after_the_durable_checkpoint() { + let tmp = TempDir::new().unwrap(); + let database_path; + let records = (0_u64..5) + .map(|ordinal| { + observation( + "session-windowed-restart", + ordinal * 100, + (ordinal + 1) * 100, + &format!("receipt.windowed.{ordinal}"), + conversational_payload( + &format!("message-windowed-{ordinal}"), + &format!("windowed restart canary {ordinal}"), + ), + ) + }) + .collect::>(); + + { + let runtime = profile_runtime(&tmp).await; + let store = runtime + .observation_store(HostAdmissionScope::Profile) + .unwrap(); + database_path = runtime + .database_path(HostAdmissionScope::Profile) + .unwrap() + .to_path_buf(); + + for (index, record) in records.iter().enumerate() { + let expected = + (index > 0).then(|| cursor("session-windowed-restart", index as u64 * 100)); + persist(&store, record.clone(), expected).await; + } + store + .project_observation(records[0].observation_id()) + .await + .unwrap(); + assert_eq!( + store.projection_checkpoint().await.unwrap().last_sequence(), + 1 + ); + + // A terminal collision remains in the queue behind the committed + // checkpoint, while the next real head is initially ready. + let conn = rusqlite::Connection::open(&database_path).unwrap(); + conn.execute( + "INSERT INTO projection_queue (observation_id, observation_sequence) + VALUES (?1, 1)", + rusqlite::params![records[0].observation_id().as_str()], + ) + .unwrap(); + conn.execute( + "UPDATE projection_queue SET next_retry_at_micros = ?2 + WHERE observation_id = ?1", + rusqlite::params![records[0].observation_id().as_str(), i64::MAX], + ) + .unwrap(); + // A nonterminal future retry at sequence 4 must fence sequence 5 + // even though sequence 5 is already due. + conn.execute( + "UPDATE projection_queue SET next_retry_at_micros = ?2 + WHERE observation_id = ?1", + rusqlite::params![records[3].observation_id().as_str(), i64::MAX - 1], + ) + .unwrap(); + drop(conn); + + let batch = store + .project_queued_observations(32) + .await + .unwrap() + .unwrap(); + assert_eq!(batch.items.len(), 2, "only sequences 2 and 3 are ready"); + assert!( + batch.has_more, + "the deferred sequence 4 must remain pending" + ); + assert_eq!( + store.projection_checkpoint().await.unwrap().last_sequence(), + 3 + ); + assert_eq!(table_count(&tmp, "lcm_raw_messages").await, 3); + } + + { + let runtime = profile_runtime(&tmp).await; + let store = runtime + .observation_store(HostAdmissionScope::Profile) + .unwrap(); + // Fresh admission re-arms the previous mount's retry. Its first ready + // item must be sequence four, after the durable checkpoint. + assert_eq!( + store.next_queued_observation().await.unwrap().as_ref(), + Some(records[3].observation_id()) + ); + assert_eq!( + store.projection_checkpoint().await.unwrap().last_sequence(), + 3 + ); + assert_eq!(table_count(&tmp, "lcm_raw_messages").await, 3); + + let first = store.project_queued_observations(1).await.unwrap().unwrap(); + assert_eq!( + first.items.len(), + 1, + "one-item window must not overrun its bound" + ); + assert!(matches!( + first.items[0].outcome, + ProjectionPersistOutcome::Projected(_) + )); + assert!(first.has_more); + assert_eq!( + store.projection_checkpoint().await.unwrap().last_sequence(), + 4 + ); + + let tail = store + .project_queued_observations(32) + .await + .unwrap() + .unwrap(); + assert_eq!(tail.items.len(), 1); + assert!( + !tail.has_more, + "the terminal collision is not runnable backlog" + ); + assert_eq!( + store.projection_checkpoint().await.unwrap().last_sequence(), + 5 + ); + assert_eq!(table_count(&tmp, "lcm_raw_messages").await, 5); + assert_eq!(table_count(&tmp, "projection_queue").await, 1); + + // A terminal-only queue may survive indefinitely; it must not take + // a writer lease while another service holds the same database writer. + let writer = rusqlite::Connection::open(&database_path).unwrap(); + writer.execute_batch("BEGIN IMMEDIATE").unwrap(); + let idle = tokio::time::timeout( + std::time::Duration::from_secs(2), + store.project_queued_observations(32), + ) + .await + .expect("a terminal-only queue cannot acquire the writer") + .unwrap() + .unwrap(); + writer.execute_batch("ROLLBACK").unwrap(); + assert!(idle.items.is_empty()); + assert!(!idle.has_more); + assert_eq!( + store.projection_checkpoint().await.unwrap().last_sequence(), + 5 + ); + + // The skip is not cached: a later admission still advances the + // durable checkpoint past the parked terminal collision. + persist( + &store, + observation( + "session-windowed-restart", + 500, + 600, + "receipt.windowed.5", + conversational_payload("message-windowed-5", "post-terminal admission"), + ), + Some(cursor("session-windowed-restart", 500)), + ) + .await; + let resumed = store + .project_queued_observations(32) + .await + .unwrap() + .unwrap(); + assert_eq!(resumed.items.len(), 1); + assert!(!resumed.has_more); + assert_eq!( + store.projection_checkpoint().await.unwrap().last_sequence(), + 6 + ); + assert_eq!(table_count(&tmp, "lcm_raw_messages").await, 6); + } +} + +#[tokio::test] +async fn reopened_queue_skips_writer_for_a_newly_deferred_head() { + let tmp = TempDir::new().unwrap(); + let database_path; + let first = observation( + "session-deferred-only", + 0, + 100, + "receipt.deferred-only-first", + conversational_payload("message-deferred-only-first", "first deferred-only canary"), + ); + let second = observation( + "session-deferred-only", + 100, + 200, + "receipt.deferred-only-second", + conversational_payload( + "message-deferred-only-second", + "second deferred-only canary", + ), + ); + let third = observation( + "session-deferred-only", + 200, + 300, + "receipt.deferred-only-third", + conversational_payload("message-deferred-only-third", "due behind deferred head"), + ); + { + let runtime = profile_runtime(&tmp).await; + let store = runtime + .observation_store(HostAdmissionScope::Profile) + .unwrap(); + database_path = runtime + .database_path(HostAdmissionScope::Profile) + .unwrap() + .to_path_buf(); + persist(&store, first.clone(), None).await; + persist( + &store, + second.clone(), + Some(cursor("session-deferred-only", 100)), + ) + .await; + persist( + &store, + third.clone(), + Some(cursor("session-deferred-only", 200)), + ) + .await; + store + .project_observation(first.observation_id()) + .await + .unwrap(); + + // Terminal collisions persist across fresh admission. + let connection = rusqlite::Connection::open(&database_path).unwrap(); + connection + .execute( + "INSERT INTO projection_queue (observation_id, observation_sequence) + VALUES (?1, 1)", + rusqlite::params![first.observation_id().as_str()], + ) + .unwrap(); + connection + .execute( + "UPDATE projection_queue SET next_retry_at_micros = ?2 + WHERE observation_id = ?1", + rusqlite::params![first.observation_id().as_str(), i64::MAX], + ) + .unwrap(); + } + + let runtime = profile_runtime(&tmp).await; + let store = runtime + .observation_store(HostAdmissionScope::Profile) + .unwrap(); + let writer = rusqlite::Connection::open(&database_path).unwrap(); + // A retry deferred by this mount fences the already-due third row. + assert_eq!( + writer + .execute( + "UPDATE projection_queue SET next_retry_at_micros = ?2 + WHERE observation_id = ?1", + rusqlite::params![second.observation_id().as_str(), i64::MAX - 1], + ) + .unwrap(), + 1 + ); + writer.execute_batch("BEGIN IMMEDIATE").unwrap(); + for _ in 0..2 { + let deferred = tokio::time::timeout( + std::time::Duration::from_secs(2), + store.project_queued_observations(32), + ) + .await + .expect("deferred-only projection must not acquire the writer") + .unwrap() + .unwrap(); + assert!(deferred.items.is_empty()); + assert!(deferred.has_more, "the deferred head remains pending"); + } + assert!( + store.next_queued_observation().await.unwrap().is_none(), + "the later due row must not bypass a deferred sequence head" + ); + writer.execute_batch("ROLLBACK").unwrap(); + assert_eq!( + store.projection_checkpoint().await.unwrap().last_sequence(), + 1 + ); + assert_eq!(table_count(&tmp, "lcm_raw_messages").await, 1); + + // Skipping the writer must never cache a negative readiness decision. + rusqlite::Connection::open(&database_path) + .unwrap() + .execute( + "UPDATE projection_queue SET next_retry_at_micros = 0 + WHERE observation_id = ?1", + rusqlite::params![second.observation_id().as_str()], + ) + .unwrap(); + assert_eq!( + store.next_queued_observation().await.unwrap(), + Some(second.observation_id().clone()), + "the released sequence head must be the next queued observation" + ); + let resumed = store.project_queued_observations(1).await.unwrap().unwrap(); + assert_eq!( + resumed.items.len(), + 1, + "the one-item window cannot skip ahead" + ); + assert!( + resumed.has_more, + "the later ready sequence still needs projection" + ); + assert_eq!( + store.projection_checkpoint().await.unwrap().last_sequence(), + 2 + ); + assert_eq!(table_count(&tmp, "lcm_raw_messages").await, 2); + + let tail = store.project_queued_observations(1).await.unwrap().unwrap(); + assert_eq!(tail.items.len(), 1); + assert!(!tail.has_more, "only the terminal collision remains"); + assert_eq!( + store.projection_checkpoint().await.unwrap().last_sequence(), + 3 + ); + assert_eq!(table_count(&tmp, "lcm_raw_messages").await, 3); +} + +#[tokio::test] +async fn empty_projection_drain_does_not_wait_for_an_unrelated_writer() { + let tmp = TempDir::new().unwrap(); + let runtime = profile_runtime(&tmp).await; + let store = runtime + .observation_store(HostAdmissionScope::Profile) + .unwrap(); + let database_path = runtime.database_path(HostAdmissionScope::Profile).unwrap(); + + // A separate connection holds SQLite's writer reservation. An idle + // projection pass must still finish using only the read connection. + let writer = rusqlite::Connection::open(database_path).unwrap(); + writer.execute_batch("BEGIN IMMEDIATE").unwrap(); + let empty = tokio::time::timeout( + std::time::Duration::from_secs(2), + store.project_queued_observations(32), + ) + .await + .expect("an empty projection pass must not acquire the writer") + .unwrap() + .unwrap(); + assert!(empty.items.is_empty()); + assert!(!empty.has_more); + writer.execute_batch("ROLLBACK").unwrap(); + + // Skipping the empty writer transaction must never suppress subsequent + // admissions after the external writer releases its reservation. + let new_record = observation( + "session-empty-drain", + 0, + 100, + "receipt.empty-drain", + conversational_payload("message-empty-drain", "post-idle admission"), + ); + persist(&store, new_record, None).await; + let after = store + .project_queued_observations(32) + .await + .unwrap() + .unwrap(); + assert_eq!(after.items.len(), 1); + assert!(!after.has_more); + assert_eq!(table_count(&tmp, "lcm_raw_messages").await, 1); +} + +#[tokio::test] +async fn active_rebuild_keeps_projection_pending_without_acquiring_writer() { + let tmp = TempDir::new().unwrap(); + let runtime = profile_runtime(&tmp).await; + let store = runtime + .observation_store(HostAdmissionScope::Profile) + .unwrap(); + let database_path = runtime.database_path(HostAdmissionScope::Profile).unwrap(); + persist( + &store, + observation( + "session-rebuild-gate", + 0, + 100, + "receipt.rebuild-gate", + conversational_payload("message-rebuild-gate", "project after rebuild"), + ), + None, + ) + .await; + + let writer = rusqlite::Connection::open(database_path).unwrap(); + writer + .execute( + "INSERT INTO observation_projection_rebuilds + (projector_version, generation, frontier_sequence, state) + VALUES (?1, 'rebuild-gate-generation', 1, 'building')", + rusqlite::params![tracedecay_store::SESSION_MESSAGE_PROJECTOR_VERSION], + ) + .unwrap(); + + // A live rebuild fences ordinary projection, so an empty drain should + // return pending without waiting for a separate SQLite writer reservation. + writer.execute_batch("BEGIN IMMEDIATE").unwrap(); + for _ in 0..2 { + let deferred = tokio::time::timeout( + std::time::Duration::from_secs(2), + store.project_queued_observations(32), + ) + .await + .expect("an active rebuild must not acquire the projection writer") + .unwrap() + .unwrap(); + assert!(deferred.items.is_empty()); + assert!(deferred.has_more, "the observation remains queued"); + } + writer.execute_batch("ROLLBACK").unwrap(); + assert_eq!(table_count(&tmp, "lcm_raw_messages").await, 0); + + // Clearing the durable rebuild fence must expose the original queued + // observation on the next poll without a cached negative gate. + writer + .execute( + "DELETE FROM observation_projection_rebuilds WHERE projector_version = ?1", + rusqlite::params![tracedecay_store::SESSION_MESSAGE_PROJECTOR_VERSION], + ) + .unwrap(); + let resumed = store.project_queued_observations(1).await.unwrap().unwrap(); + assert_eq!(resumed.items.len(), 1); + assert!(!resumed.has_more); + assert_eq!( + store.projection_checkpoint().await.unwrap().last_sequence(), + 1 + ); + assert_eq!(table_count(&tmp, "lcm_raw_messages").await, 1); +}