From 39339ac30a6f373385f0fcf86d9cfdff577073d5 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 21:09:50 -0800 Subject: [PATCH 01/17] perf(sessions): avoid idle deferred-receipt scans Probe the indexed next relation recovery deadline before reading the creation-order journal, and detect one extra due receipt so the bounded recovery worker continues without missing backlog. Extend database-backed recovery regressions for deferred idle reads, due-page continuation, and settled quiescence. Refs #4481 --- .../src/lcm_effects/tests.rs | 57 +++++++++++++++++++ .../src/relation_projection.rs | 40 ++++++++++++- 2 files changed, 96 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs index ee3ab998e87..a1663317965 100644 --- a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs +++ b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs @@ -2440,6 +2440,63 @@ async fn malformed_relation_receipt_is_permanent_without_starving_summary_work() .is_some_and(|delay| !delay.is_zero() && delay <= std::time::Duration::from_secs(60)), "the scheduler must retain the isolated relation retry deadline" ); + let deferred = SessionTemporalAccess::new(&*db) + .recover_pending_session_relation_projection_page( + 1, + std::sync::Arc::new(tracedecay_graph_db::NeverCancelled), + ) + .await + .unwrap(); + assert_eq!(deferred.processed, 0, "a future retry is not due"); + assert!(!deferred.has_more); + assert!(deferred.next_retry_at_unix_seconds.is_some()); + + // Two durable receipts becoming due at once must be offered in distinct + // bounded pages. A one-row LIMIT used to hide the second due receipt + // even though the scheduler depended on has_more to resume promptly. + let transaction = db.begin_write_transaction().await.unwrap(); + let changed = transaction + .execute( + "UPDATE session_relation_receipts + SET state = 'pending', recovery_state = 'pending', + recovery_failure_code = NULL, recovery_failure_count = 0, + recovery_next_attempt_at = 0 + WHERE session_id IN (?1, ?2)", + params![poison_session, healthy_session], + ) + .await + .unwrap(); + assert_eq!(changed, 2, "both durable receipts must be present"); + transaction.commit().await.unwrap(); + + let recovered_first = SessionTemporalAccess::new(&*db) + .recover_pending_session_relation_projection_page( + 1, + std::sync::Arc::new(tracedecay_graph_db::NeverCancelled), + ) + .await + .unwrap(); + assert_eq!(recovered_first.processed, 1); + assert!(recovered_first.has_more, "the second due receipt is owed"); + + let recovered_second = SessionTemporalAccess::new(&*db) + .recover_pending_session_relation_projection_page( + 1, + std::sync::Arc::new(tracedecay_graph_db::NeverCancelled), + ) + .await + .unwrap(); + assert_eq!(recovered_second.processed, 1); + assert!(!recovered_second.has_more); + let idle = SessionTemporalAccess::new(&*db) + .recover_pending_session_relation_projection_page( + 1, + std::sync::Arc::new(tracedecay_graph_db::NeverCancelled), + ) + .await + .unwrap(); + assert_eq!(idle.processed, 0, "settled receipts stay parked"); + assert!(!idle.has_more); } #[test] diff --git a/crates/tracedecay-session-temporal-store/src/relation_projection.rs b/crates/tracedecay-session-temporal-store/src/relation_projection.rs index 8b5058dcecd..4f098fe74b5 100644 --- a/crates/tracedecay-session-temporal-store/src/relation_projection.rs +++ b/crates/tracedecay-session-temporal-store/src/relation_projection.rs @@ -297,6 +297,44 @@ impl SessionTemporalAccess<'_, D> { let (scope, _) = self .session_relation_store() .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; + // An idle pass with only sleeping retries must not scan all pending + // receipts in creation order. The recovery-due index answers this + // probe without visiting each deferred journal row. + let mut due_rows = snapshot + .query( + "SELECT MIN(recovery_next_attempt_at), unixepoch() + FROM session_relation_receipts + WHERE state = 'pending' + AND recovery_state IN ('pending', 'retryable')", + (), + ) + .await + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; + let earliest = due_rows + .next() + .await + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))? + .ok_or_else(|| { + storage_message( + RECONSTRUCT_OPERATION, + "relation retry probe returned no aggregate row", + ) + })?; + let retry_at: Option = earliest + .get(0) + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; + let now: i64 = earliest + .get(1) + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; + drop(due_rows); + if retry_at.is_none_or(|deadline| deadline > now) { + return Ok(SessionRelationRecoveryPage { + recovered: 0, + processed: 0, + has_more: false, + next_retry_at_unix_seconds: retry_at, + }); + } let mut rows = snapshot .query( "SELECT receipt.session_id, receipt.generation, receipt.scope_kind, @@ -311,7 +349,7 @@ impl SessionTemporalAccess<'_, D> { AND receipt.recovery_next_attempt_at <= unixepoch() ORDER BY receipt.created_at, receipt.session_id, receipt.generation LIMIT ?1", - params![bounded_query_limit(limit)?], + params![bounded_query_limit(limit.saturating_add(1))?], ) .await .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; From b1be997b7a09473153ff4174517c925689cd2e98 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 21:52:47 -0800 Subject: [PATCH 02/17] fix(sessions): avoid double-padding relation recovery pages The shared bounded_query_limit helper already requests one sentinel row for has_more. Do not pad the caller's limit a second time. Refs #4481 --- .../src/relation_projection.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tracedecay-session-temporal-store/src/relation_projection.rs b/crates/tracedecay-session-temporal-store/src/relation_projection.rs index 4f098fe74b5..c8b43c3a520 100644 --- a/crates/tracedecay-session-temporal-store/src/relation_projection.rs +++ b/crates/tracedecay-session-temporal-store/src/relation_projection.rs @@ -349,7 +349,7 @@ impl SessionTemporalAccess<'_, D> { AND receipt.recovery_next_attempt_at <= unixepoch() ORDER BY receipt.created_at, receipt.session_id, receipt.generation LIMIT ?1", - params![bounded_query_limit(limit.saturating_add(1))?], + params![bounded_query_limit(limit)?], ) .await .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; From 36ace66f121980c883192efa691e569a1dba60d4 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 21:53:08 -0800 Subject: [PATCH 03/17] test(sessions): describe the existing recovery page sentinel The bounded query helper already fetched limit plus one before this PR. Keep the behavior regression and describe what it actually proves. Refs #4481 --- crates/tracedecay-session-runtime/src/lcm_effects/tests.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs index a1663317965..3b21424a37f 100644 --- a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs +++ b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs @@ -2451,9 +2451,9 @@ async fn malformed_relation_receipt_is_permanent_without_starving_summary_work() assert!(!deferred.has_more); assert!(deferred.next_retry_at_unix_seconds.is_some()); - // Two durable receipts becoming due at once must be offered in distinct - // bounded pages. A one-row LIMIT used to hide the second due receipt - // even though the scheduler depended on has_more to resume promptly. + // Preserve one-sentinel-row pagination after the idle eligibility probe: + // two durable due receipts must be recovered in distinct bounded pages, + // with has_more advertising the unprocessed second receipt. let transaction = db.begin_write_transaction().await.unwrap(); let changed = transaction .execute( From f085c0f03194759295c900c539f63f8663f5d822 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 00:25:21 -0800 Subject: [PATCH 04/17] test(sessions): reset relation receipt state consistently The one-row recovery pagination regression reused an applied receipt while setting its state to pending. The actual SQLite schema refuses that mixed state unless graph_watermark and applied_at are cleared in the same update. Restore the valid pending transition without weakening the receipt constraint or the two-page recovery and idle-queue assertions. Refs #4481 --- crates/tracedecay-session-runtime/src/lcm_effects/tests.rs | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs index 3b21424a37f..9449053eca4 100644 --- a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs +++ b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs @@ -2458,8 +2458,9 @@ async fn malformed_relation_receipt_is_permanent_without_starving_summary_work() let changed = transaction .execute( "UPDATE session_relation_receipts - SET state = 'pending', recovery_state = 'pending', - recovery_failure_code = NULL, recovery_failure_count = 0, + SET state = 'pending', graph_watermark = NULL, applied_at = NULL, + recovery_state = 'pending', recovery_failure_code = NULL, + recovery_failure_count = 0, recovery_next_attempt_at = 0 WHERE session_id IN (?1, ?2)", params![poison_session, healthy_session], From 1430ba1c48176cbd1c0949e89ddb503befe1d669 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 01:52:34 -0800 Subject: [PATCH 05/17] test(automation): bind scheduler fixtures to backend identities Repair three hosted automation_runner_test failures by supplying the configured executable identity required by the production prefilter. Use the same identity for the prefilter and retained runner, preserving manual-schedule, exact-ledger, live-lock and lock-race assertions. Add a production-prefilter regression for unconfigured refusal, deduplication and same-occurrence recovery after binding, with zero backend calls and no effectful terminal. Do not relax backend admission, change timeouts, or alter the graph publication changes. Refs #4532 --- .../tests/automation_runner_test/jobs.rs | 124 ++++++++++++++++-- 1 file changed, 110 insertions(+), 14 deletions(-) diff --git a/crates/tracedecay/tests/automation_runner_test/jobs.rs b/crates/tracedecay/tests/automation_runner_test/jobs.rs index 740283d56e8..7342786381b 100644 --- a/crates/tracedecay/tests/automation_runner_test/jobs.rs +++ b/crates/tracedecay/tests/automation_runner_test/jobs.rs @@ -86,6 +86,7 @@ fn enabled_job_config() -> AutomationSettingsV1 { struct ContentBackend { calls: AtomicUsize, content: &'static str, + executable: Option, } impl ContentBackend { @@ -93,9 +94,15 @@ impl ContentBackend { Self { calls: AtomicUsize::new(0), content, + executable: None, } } + fn with_executable(mut self, executable: &Path) -> Self { + self.executable = Some(executable.to_path_buf()); + self + } + fn calls(&self) -> usize { self.calls.load(Ordering::SeqCst) } @@ -126,7 +133,7 @@ impl AgentTaskBackend for ContentBackend { } fn executable(&self) -> Option<&std::path::Path> { - None + self.executable.as_deref() } } @@ -988,6 +995,10 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let mut job = sample_job("sched-job"); job.schedule = Some("manual".to_string()); let config = enabled_job_config(); + // Reach the schedule gate with a bound identity, not the earlier + // backend-unconfigured refusal. This gate never invokes the backend. + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = ContentBackend::new("unused").with_executable(executable.path()); // Manual schedule: every scheduler tick of the same occurrence reaches // the prefilter, which records the skip once. @@ -996,7 +1007,7 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let run = evaluate_and_record_scheduler_skip( &dashboard_root, &config, - None, + backend.executable(), &job, "sched-occurrence", None, @@ -1019,7 +1030,7 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let repeated = evaluate_and_record_scheduler_skip( &dashboard_root, &config, - None, + backend.executable(), &job, "sched-occurrence", None, @@ -1031,6 +1042,76 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { repeated.ledger_record, records[0], "an immediate repeat skip must return the exact durable prior row" ); + assert_eq!(backend.calls(), 0); +} + +#[tokio::test] +async fn scheduler_prefilter_rejects_unconfigured_backend_and_recovers_without_effect() { + let temp = tempdir().unwrap(); + let dashboard_root = temp.path().join("dashboard"); + let job = sample_job("backend-recovery"); + let config = enabled_job_config(); + let occurrence = "backend-recovery-occurrence"; + let backend = ContentBackend::new("unused"); + + let first = evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .expect("a due job still requires a configured backend"); + assert_eq!(first.report["reason"], json!("backend_unconfigured")); + assert_eq!(first.ledger_record.status, AutomationRunStatus::Skipped); + assert_ne!(first.run_id, occurrence); + let repeated = evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .unwrap(); + assert_eq!(repeated.ledger_record, first.ledger_record); + + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = backend.with_executable(executable.path()); + assert!( + evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .is_none(), + "binding the backend must admit the same due occurrence without cooldown" + ); + assert_eq!(backend.calls(), 0); + assert_eq!( + load_run_records(&dashboard_root, 10).await.unwrap(), + vec![first.ledger_record], + "advisory admission must not append an effectful terminal" + ); + assert!( + tracedecay_automation_runtime::automation::run_ledger::find_run_record_exact_bounded( + &dashboard_root, + occurrence, + ) + .await + .unwrap() + .is_none() + ); } #[tokio::test] @@ -1206,15 +1287,22 @@ async fn scheduler_prefilter_live_lock_wins_over_not_due_summary() { let config = enabled_job_config(); let occurrence = "locked-prefilter-occurrence"; - let prefilter = - evaluate_and_record_scheduler_skip(&dashboard_root, &config, None, &job, occurrence, None) - .await - .unwrap() - .expect("the live job lock must produce a diagnostic"); + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = ContentBackend::new("unused").with_executable(executable.path()); + let prefilter = evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .expect("the live job lock must produce a diagnostic"); assert_eq!(prefilter.report["reason"], json!("scheduler_lock_active")); assert_ne!(prefilter.run_id, occurrence); - let backend = ContentBackend::new("unused"); let admitted_job = tracedecay_automation_runtime::automation::jobs::admit_job(&job).unwrap(); let retained = run_user_job_with_backend_for_retained_settlement( &dashboard_root, @@ -1262,12 +1350,21 @@ async fn retained_scheduler_runner_reacquires_after_due_prefilter() { job.schedule = Some("every 1h".to_owned()); let config = enabled_job_config(); let occurrence = "prefilter-gap-occurrence"; + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = ContentBackend::new("unused").with_executable(executable.path()); assert!( - evaluate_and_record_scheduler_skip(&dashboard_root, &config, None, &job, occurrence, None) - .await - .unwrap() - .is_none(), + evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .is_none(), "a new interval job is advisory-due before effect admission" ); let _raced_in = @@ -1276,7 +1373,6 @@ async fn retained_scheduler_runner_reacquires_after_due_prefilter() { .unwrap() .expect("the advisory prefilter must release its reservation lock"); - let backend = ContentBackend::new("unused"); let admitted_job = tracedecay_automation_runtime::automation::jobs::admit_job(&job).unwrap(); let retained = run_user_job_with_backend_for_retained_settlement( &dashboard_root, From c67d2db91424687e29e6e99249805e10e10b7a78 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 13:55:25 -0800 Subject: [PATCH 06/17] fix(sessions): serve live projection during historical ingest --- .../worker.rs | 73 +++++++++++-------- .../worker_tests.rs | 44 ++++++----- 2 files changed, 69 insertions(+), 48 deletions(-) diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs index 5bc41596610..87611566be5 100644 --- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs +++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs @@ -7,6 +7,7 @@ use std::sync::PoisonError; use std::sync::atomic::Ordering; use std::time::{Duration, Instant}; +use futures_util::future::{FusedFuture, FutureExt}; use tracedecay_lcm::LcmError; use tracedecay_store::{ SessionRefreshBeginOrJoinRequestV1, SessionRefreshCompletionRequestV1, @@ -246,7 +247,7 @@ pub(super) async fn run_session_temporal_refresh_scheduler( let mut history_release = None; let mut granted_history_admission = None; let mut projection_joins: Vec = Vec::new(); - let live_projection = LiveProjectionDuringSummary { + let live_projection = LiveProjection { database: &database, state: &state, projector: projector.as_ref(), @@ -264,16 +265,18 @@ pub(super) async fn run_session_temporal_refresh_scheduler( // Busy before any wake is consumed: a waiter must never see its // wake taken while the worker still reads idle. state.mark_worker_busy(); - let mut projection_requested = state.take_dirty(); let mut history_requested = state.take_historical_dirty(); // A wake that arrives while the admitted window is still - // unpublished must not start the next history window. That window - // is the out-of-scope cursor sweep, and it holds this worker until - // it returns, so the newest day never leaves its building generation. - if state.history_held_for_projection() { + // unpublished must not start the next history window. Publish + // the newest day before admitting the out-of-scope cursor sweep. + let holding_projection = state.history_held_for_projection(); + if holding_projection { history_requested = false; - projection_requested = true; } + // A history window polls live projection alongside ingest, so leave + // its dirty flag available until that work can actually start. + let mut projection_requested = + !history_requested && (state.take_dirty() || holding_projection); if !projection_requested && !history_requested && !summary_recheck { break; } @@ -289,6 +292,7 @@ pub(super) async fn run_session_temporal_refresh_scheduler( let granted = granted_history_admission .take() .filter(|_| history_requested && !history_suppressed); + let mut report = SessionTemporalRefreshPassReport::default(); let history_outcome = if history_requested && !history_suppressed { let history_pass: Pin + Send + '_>> = Box::pin(session_history_refresh( @@ -297,11 +301,25 @@ pub(super) async fn run_session_temporal_refresh_scheduler( granted, &state, )); - let (outcome, release) = tracing::Instrument::instrument( + let history_pass = tracing::Instrument::instrument( history_pass, tracing::trace_span!("daemon.scheduler.session_temporal.history"), ) - .await; + .fuse(); + tokio::pin!(history_pass); + // A hook can arrive before or during this source window. Keep the + // admitted ingestor running while publishing its already-durable input. + let Some(((outcome, release), _)) = live_projection + .wait_for(&mut history_pass, &mut projection_joins, &mut report, false) + .await + else { + // Retirement owns this join. Cancellation has signalled the + // ingestor, but its store and admission stay leased until it exits. + if !history_pass.is_terminated() { + history_pass.await; + } + return; + }; history_release = release; Some(outcome) } else { @@ -371,10 +389,7 @@ pub(super) async fn run_session_temporal_refresh_scheduler( blocked_history_attempts = 0; blocked_history_until = None; } - let mut report = if projection_requested - || state.has_requests() - || history_requires_projection - { + if projection_requested || state.has_requests() || history_requires_projection { let projection: Pin< Box + Send + '_>, > = Box::pin(session_projection_refresh( @@ -389,14 +404,12 @@ pub(super) async fn run_session_temporal_refresh_scheduler( tracing::trace_span!("daemon.scheduler.session_temporal.projection"), ); tokio::pin!(pass); - tokio::select! { + report = tokio::select! { biased; () = tracing::Instrument::instrument(state.wait_for_cancellation(), tracing::trace_span!("daemon.scheduler.session_temporal.projection_cancel")) => return, report = &mut pass => report, - } - } else { - SessionTemporalRefreshPassReport::default() - }; + }; + } if state.cancelled.load(Ordering::Acquire) { return; } @@ -937,17 +950,16 @@ async fn settle_projection_joins( } } -/// Keeps one fair summary admission or in-flight summary page alive while the -/// same worker services live temporal projection. Both futures keep polling: -/// suspending a summary while it owns a writer could block the projection. -struct LiveProjectionDuringSummary<'a> { +/// Keeps history, summary admission, or a summary page in flight while serving +/// live projection. A stage holding a writer must keep polling so it can release it. +struct LiveProjection<'a> { database: &'a RegisteredGlobalDbLeaseV1, state: &'a Arc, projector: &'a dyn SessionTemporalRefreshProjector, policy: SessionTemporalRefreshPolicy, } -impl LiveProjectionDuringSummary<'_> { +impl LiveProjection<'_> { async fn wait_for( &self, stage: F, @@ -971,7 +983,7 @@ impl LiveProjectionDuringSummary<'_> { self.state.requeue_projection(); retry_at = Some(Instant::now() + session_refresh_retry_delay(class, retry_attempt)); } - let mut summary_pass_owed = false; + let mut projected_during_stage = false; let result = loop { let projection_wake = async { let wake = self.state.wake.notified(); @@ -999,7 +1011,7 @@ impl LiveProjectionDuringSummary<'_> { if !self.state.take_dirty() { continue; } - summary_pass_owed |= !waiting_for_admission; + projected_during_stage |= !waiting_for_admission; self.state.mark_worker_busy(); self.state.begin_pass(); self.state.pass_count.fetch_add(1, Ordering::AcqRel); @@ -1064,10 +1076,9 @@ impl LiveProjectionDuringSummary<'_> { self.state.idle.notify_waiters(); } }; - // A running page may have read its inputs before the consumed wake. - // Its follow-up needs summary convergence, not another projection of - // the frontier this wake already served. Admission has no page yet. - Some((result, summary_pass_owed)) + // A summary page may have read its inputs before this projection, + // while a queued admission still has its first page ahead of it. + Some((result, projected_during_stage)) } } @@ -2277,7 +2288,7 @@ mod tests { entered.notify_one(); finish.notified().await; }; - let live_projection = LiveProjectionDuringSummary { + let live_projection = LiveProjection { database: &database, state: &state, projector: &CanonicalSessionTemporalProjector, @@ -2360,7 +2371,7 @@ mod tests { finish.notified().await; 17usize }; - let live_projection = LiveProjectionDuringSummary { + let live_projection = LiveProjection { database: &database, state: &state, projector: &CanonicalSessionTemporalProjector, diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker_tests.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker_tests.rs index bf9b23cf40e..ad619330ce0 100644 --- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker_tests.rs +++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker_tests.rs @@ -333,12 +333,9 @@ impl SessionHistoricalIngestor for GatedHistory { fn cancel(&self) {} } -/// A hook ingest joins the refresh worker with a wake. When that wake lands -/// while a history pass is running, the pass's projection serves it and no -/// further pass starts, so the join must end with that pass rather than wait -/// out its deadline. +/// An already admitted history scan must not hold a live hook behind its window. #[tokio::test] -async fn a_wake_served_by_the_running_pass_joins_that_pass() { +async fn a_live_session_publishes_while_a_history_window_is_running() { let harness = RegisteredGlobalDbHarness::open("refresh-wake-served-mid-pass").await; let state = Arc::new(SessionTemporalRefreshWakeState::default()); let history = Arc::new(GatedHistory { @@ -357,23 +354,36 @@ async fn a_wake_served_by_the_running_pass_joins_that_pass() { )); history.entered.notified().await; + let session_id = SessionId::new("live-during-history").expect("session id"); + let store = SessionTemporalStore::new(harness.registered.as_ref()); + let begun = store + .begin_or_join_session_refresh(SessionRefreshBeginOrJoinRequestV1::new( + session_id.clone(), + SessionRefreshFrontierV1::new(0, 0).expect("empty frontier"), + )) + .await + .expect("running refresh"); let handle = state.handle(); - let join = handle.wake_and_wait_for_projection(Duration::from_secs(5)); - tokio::pin!(join); - // The first poll registers the join and delivers the wake. - assert!( - std::future::poll_fn(|cx| std::task::Poll::Ready(join.as_mut().poll(cx))) - .await - .is_pending() - ); + let published = handle + .wake_and_wait_for_session_projection(&session_id, Duration::from_secs(5)) + .await; history.release.notify_one(); + state.cancel(); + worker.await.expect("refresh worker exits on cancellation"); assert!( - join.await, - "the pass that served the wake must end the join" + published, + "the live session must publish while history is held" ); - state.cancel(); - worker.await.expect("refresh worker exits on cancellation"); + let receipt = store + .session_refresh_receipt(SessionRefreshReceiptRequestV1::new( + begun.operation_id().clone(), + begun.session_id().clone(), + )) + .await + .expect("read publication receipt") + .expect("the live generation must be durable"); + assert_eq!(receipt.state(), SessionRefreshTerminalStateV1::Complete); } /// Every history window advances but waits for the test before returning. From 548db008c2d137e28a5d4e4583aeccf41b88ef05 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 13:55:27 -0800 Subject: [PATCH 07/17] fix(lcm): serialize summary publication per store and session --- .../src/lcm_effects.rs | 45 ++++++++++++++++++- .../src/lcm_effects/tests.rs | 20 +++++++++ 2 files changed, 64 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay-session-runtime/src/lcm_effects.rs b/crates/tracedecay-session-runtime/src/lcm_effects.rs index 678df440765..b58bade23c9 100644 --- a/crates/tracedecay-session-runtime/src/lcm_effects.rs +++ b/crates/tracedecay-session-runtime/src/lcm_effects.rs @@ -1,9 +1,13 @@ use std::future::Future; +use std::path::PathBuf; use std::pin::Pin; -use std::sync::Arc; +use std::sync::{Arc, LazyLock}; use std::time::Duration; +use tokio::sync::{Mutex, OwnedMutexGuard}; use tracedecay_contracts::{CancellationSignal, Deadline}; +use tracedecay_runtime_core::native_runtime::{self, TimeoutError}; +use tracedecay_runtime_core::weak_registry::WeakRegistry; use tracedecay_temporal_query::execution::ExecutionControl; use tracedecay_domain::SessionId; @@ -110,10 +114,48 @@ impl DaemonLcmEffectService { } } + /// Planning and publishing use separate transactions around the provider + /// call. A foreground hook and retained convergence for the same session + /// must re-plan after the preceding publication before launching a model. + async fn compression_turn( + &self, + request: &LcmCompressionRequest, + ) -> Result, LcmError> { + type SessionTurns = WeakRegistry<(PathBuf, String, String), Mutex<()>>; + static TURNS: LazyLock = LazyLock::new(WeakRegistry::new); + let key = ( + self.db.db_path().to_path_buf(), + request.provider.clone(), + request.session_id.clone(), + ); + let (turn, _) = TURNS.get_or_insert_with(key, || Arc::new(Mutex::new(()))); + let remaining = self.control.remaining()?; + if let Ok(guard) = Arc::clone(&turn).try_lock_owned() { + return Ok(guard); + } + let cancelled = async { + match &self.control.cancellation { + Some(signal) => signal.cancelled().await, + None => std::future::pending().await, + } + }; + tokio::select! { + biased; + () = cancelled => Err(LcmError::Cancelled), + result = native_runtime::timeout(remaining, turn.lock_owned()) => result.map_err(|error| { + match error { + TimeoutError::Elapsed => LcmError::DeadlineExceeded, + TimeoutError::TimerUnavailable(reason) => LcmError::Io(reason), + } + }), + } + } + pub(super) async fn compress( &self, mut request: LcmCompressionRequest, ) -> Result { + let _turn = self.compression_turn(&request).await?; // Observation-projected sessions land raw rows without ingest // protection; the verified raw loads inside compression reject them. // Hydrate the canonical sanitization receipts first so the daemon @@ -197,6 +239,7 @@ impl DaemonLcmEffectService { mut request: LcmCompressionRequest, convergence_candidate: &tracedecay_lcm::summary_convergence::LcmSummaryConvergenceCandidate, ) -> Result { + let _turn = self.compression_turn(&request).await?; if request.summarizer.has_nonempty_provided_summary() { return self .commit_retained_compression(&request, Some(convergence_candidate), None) diff --git a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs index 9449053eca4..e4f14a706aa 100644 --- a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs +++ b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs @@ -91,6 +91,26 @@ fn compression_request(session_id: &str) -> LcmCompressionRequest { } } +#[tokio::test] +async fn a_queued_compression_honors_cancellation_without_blocking_other_sessions() { + let harness = RegisteredGlobalDbHarness::open("lcm-session-compression-turn").await; + let signal = CancellationSignal::active("queued-compression").unwrap(); + let owner = DaemonLcmEffectService::new(harness.registered.clone(), None, None, None); + let caller = DaemonLcmEffectService::new(harness.registered.clone(), None, Some(&signal), None); + let first = compression_request("first-session"); + let _held = owner.compression_turn(&first).await.unwrap(); + let waiting = caller.compress(first); + tokio::pin!(waiting); + assert!(futures_util::poll!(&mut waiting).is_pending()); + + let _independent = owner + .compression_turn(&compression_request("other-session")) + .await + .unwrap(); + signal.cancel(UtcMicros(1)); + assert!(matches!(waiting.await, Err(LcmError::Cancelled))); +} + fn provided_summary(summary_text: impl Into) -> LcmSummarizerMode { LcmSummarizerMode::Provided { summary_text: summary_text.into(), From dbf3b0d8f19bb38acff6e93acddf9e9185edc50a Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 14:42:22 -0800 Subject: [PATCH 08/17] test(storage): synchronize owner startup with daemon readiness --- .../multi_connection_test/fail_closed.rs | 5 ++--- .../multi_connection_test/harness.rs | 18 ++++++------------ 2 files changed, 8 insertions(+), 15 deletions(-) diff --git a/crates/tracedecay/tests/storage_suite/multi_connection_test/fail_closed.rs b/crates/tracedecay/tests/storage_suite/multi_connection_test/fail_closed.rs index cfe16b33c9c..e1fcc3f8cab 100644 --- a/crates/tracedecay/tests/storage_suite/multi_connection_test/fail_closed.rs +++ b/crates/tracedecay/tests/storage_suite/multi_connection_test/fail_closed.rs @@ -59,7 +59,7 @@ fn split_brain_is_rejected_and_unavailable_daemon_fails_closed_until_restart() { "rejected contender is still running after fail-closed rejection" ); - stop_child(&mut owner); + drop(owner); install_unavailable_socket_sentinel(&socket_path); let before = storage_snapshot(&profile_db_path); for (label, mut command) in [ @@ -142,7 +142,7 @@ fn split_brain_is_rejected_and_unavailable_daemon_fails_closed_until_restart() { &profile_db_path, ); - let mut restarted = spawn_daemon(&home_path, &socket_path); + let restarted = spawn_daemon(&home_path, &socket_path); assert_command_success( "restarted daemon status", &tool_status(&home_path, &project_path, &socket_path), @@ -161,5 +161,4 @@ fn split_brain_is_rejected_and_unavailable_daemon_fails_closed_until_restart() { authority_after["epoch"].as_u64() > authority_before["epoch"].as_u64(), "restart must advance daemon authority epoch" ); - stop_child(&mut restarted); } diff --git a/crates/tracedecay/tests/storage_suite/multi_connection_test/harness.rs b/crates/tracedecay/tests/storage_suite/multi_connection_test/harness.rs index 8fb0c7d7af5..3e05bc4da24 100644 --- a/crates/tracedecay/tests/storage_suite/multi_connection_test/harness.rs +++ b/crates/tracedecay/tests/storage_suite/multi_connection_test/harness.rs @@ -136,19 +136,13 @@ fn wait_for_socket(socket_path: &Path, child: &mut Child) { ); } -pub(super) fn spawn_daemon(home: &Path, socket_path: &Path) -> ChildGuard { - let mut child = ChildGuard::new( - common::tracedecay_command_with_home(home) - .args(["daemon", "run", "--socket"]) +pub(super) fn spawn_daemon(home: &Path, socket_path: &Path) -> common::DaemonProcess { + common::spawn_tracedecay_daemon_with(home, |command| { + command + .arg("--socket") .arg(socket_path) - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .expect("spawn daemon"), - ); - wait_for_socket(socket_path, &mut child); - child + .env_remove("TRACEDECAY_TEST_ALLOW_INCOMPLETE_HOLDER_SCAN"); + }) } /// Where the spawned daemon's stderr was redirected, when the caller captured From a7da7bcd42cbbe793e7c01a3c40f77c279e0047a Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 14:47:10 -0800 Subject: [PATCH 09/17] fix(sessions): reserve history preemption for waiting readers --- .../wake.rs | 8 ++ .../worker.rs | 102 ++++++++++++++---- .../temporal_projection/activation.rs | 12 ++- 3 files changed, 99 insertions(+), 23 deletions(-) diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/wake.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/wake.rs index d4eea3369fb..95bc1c443f5 100644 --- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/wake.rs +++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/wake.rs @@ -378,6 +378,14 @@ impl SessionTemporalRefreshWakeState { Some(receiver) } + pub(super) fn has_pending_projection_join(&self) -> bool { + self.projection_joins + .lock() + .unwrap_or_else(PoisonError::into_inner) + .iter() + .any(|join| !join.completion.is_closed()) + } + /// Take the joins covered by the projection about to start. A join that /// arrives during that projection remains queued for the next pass. pub(super) fn take_projection_joins(&self) -> Vec { diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs index 87611566be5..57a7a987f49 100644 --- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs +++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/worker.rs @@ -310,7 +310,12 @@ pub(super) async fn run_session_temporal_refresh_scheduler( // A hook can arrive before or during this source window. Keep the // admitted ingestor running while publishing its already-durable input. let Some(((outcome, release), _)) = live_projection - .wait_for(&mut history_pass, &mut projection_joins, &mut report, false) + .wait_for( + &mut history_pass, + &mut projection_joins, + &mut report, + BackgroundStage::History, + ) .await else { // Retirement owns this join. Cancellation has signalled the @@ -478,7 +483,12 @@ pub(super) async fn run_session_temporal_refresh_scheduler( state.mark_worker_idle(); state.idle.notify_waiters(); let Some((permit, _)) = live_projection - .wait_for(&mut admission, &mut projection_joins, &mut report, true) + .wait_for( + &mut admission, + &mut projection_joins, + &mut report, + BackgroundStage::SummaryAdmission, + ) .await else { return; @@ -511,7 +521,12 @@ pub(super) async fn run_session_temporal_refresh_scheduler( ); tokio::pin!(page); let Some((result, recheck)) = live_projection - .wait_for(&mut page, &mut projection_joins, &mut report, false) + .wait_for( + &mut page, + &mut projection_joins, + &mut report, + BackgroundStage::SummaryPage, + ) .await else { return; @@ -950,6 +965,13 @@ async fn settle_projection_joins( } } +#[derive(Clone, Copy, PartialEq, Eq)] +enum BackgroundStage { + History, + SummaryAdmission, + SummaryPage, +} + /// Keeps history, summary admission, or a summary page in flight while serving /// live projection. A stage holding a writer must keep polling so it can release it. struct LiveProjection<'a> { @@ -960,12 +982,22 @@ struct LiveProjection<'a> { } impl LiveProjection<'_> { + fn projection_due(&self, stage: BackgroundStage, joins: &[SessionProjectionJoin]) -> bool { + // An admitted history window owns background recovery until its SQL + // source frontier settles. Only a waiting reader needs to publish + // ahead of that window. + self.state.dirty.load(Ordering::Acquire) + && (stage != BackgroundStage::History + || joins.iter().any(|join| !join.completion.is_closed()) + || self.state.has_pending_projection_join()) + } + async fn wait_for( &self, stage: F, joins: &mut Vec, report: &mut SessionTemporalRefreshPassReport, - waiting_for_admission: bool, + background: BackgroundStage, ) -> Option<(F::Output, bool)> where F: Future + Send, @@ -983,19 +1015,20 @@ impl LiveProjection<'_> { self.state.requeue_projection(); retry_at = Some(Instant::now() + session_refresh_retry_delay(class, retry_attempt)); } - let mut projected_during_stage = false; + let mut recheck_summary = false; let result = loop { let projection_wake = async { let wake = self.state.wake.notified(); + if !self.projection_due(background, joins) { + wake.await; + return true; + } if let Some(deadline) = retry_at { tokio::select! { () = wake => true, fired = native_runtime::sleep_until(deadline) => !stop_without_timer(fired), } } else { - if !self.state.dirty.load(Ordering::Acquire) { - wake.await; - } true } }; @@ -1008,10 +1041,10 @@ impl LiveProjection<'_> { if !woke { return None; } - if !self.state.take_dirty() { + if !self.projection_due(background, joins) || !self.state.take_dirty() { continue; } - projected_during_stage |= !waiting_for_admission; + recheck_summary |= background == BackgroundStage::SummaryPage; self.state.mark_worker_busy(); self.state.begin_pass(); self.state.pass_count.fetch_add(1, Ordering::AcqRel); @@ -1071,14 +1104,14 @@ impl LiveProjection<'_> { if let Some(result) = completed_stage { break result; } - if waiting_for_admission { + if background == BackgroundStage::SummaryAdmission { self.state.mark_worker_idle(); self.state.idle.notify_waiters(); } }; // A summary page may have read its inputs before this projection, // while a queued admission still has its first page ahead of it. - Some((result, projected_during_stage)) + Some((result, recheck_summary)) } } @@ -2267,12 +2300,45 @@ mod tests { use tracedecay_runtime_core::db::engine::params; use tracedecay_store::{SessionRefreshReceiptRequestV1, SessionRefreshTerminalStateV1}; + #[tokio::test] + async fn a_history_window_leaves_background_projection_queued() { + let harness = RegisteredGlobalDbHarness::open("history-retains-background-work").await; + let state = Arc::new(SessionTemporalRefreshWakeState::default()); + state.wake(); + let live = LiveProjection { + database: &harness.registered, + state: &state, + projector: &CanonicalSessionTemporalProjector, + policy: SessionTemporalRefreshPolicy::default(), + }; + let mut joins = Vec::new(); + let mut report = SessionTemporalRefreshPassReport::default(); + live.wait_for( + tokio::task::yield_now(), + &mut joins, + &mut report, + BackgroundStage::History, + ) + .await + .expect("the history window finishes"); + assert_eq!(state.pass_count.load(Ordering::Acquire), 0); + assert!( + state.take_dirty(), + "background recovery still owns its turn" + ); + } + #[tokio::test] async fn a_live_wake_during_summary_retains_one_convergence_pass() { - for (name, waiting_for_admission, cancel, owes_convergence) in [ - ("completed-page", false, false, true), - ("completed-admission", true, false, false), - ("cancelled-page", false, true, false), + for (name, background, cancel, owes_convergence) in [ + ("completed-page", BackgroundStage::SummaryPage, false, true), + ( + "completed-admission", + BackgroundStage::SummaryAdmission, + false, + false, + ), + ("cancelled-page", BackgroundStage::SummaryPage, true, false), ] { let harness = RegisteredGlobalDbHarness::open(name).await; let state = Arc::new(SessionTemporalRefreshWakeState::default()); @@ -2297,7 +2363,7 @@ mod tests { let mut joins = Vec::new(); let mut report = SessionTemporalRefreshPassReport::default(); live_projection - .wait_for(stage, &mut joins, &mut report, waiting_for_admission) + .wait_for(stage, &mut joins, &mut report, background) .await } }); @@ -2380,7 +2446,7 @@ mod tests { let mut joins = Vec::new(); let mut report = SessionTemporalRefreshPassReport::default(); live_projection - .wait_for(stage, &mut joins, &mut report, false) + .wait_for(stage, &mut joins, &mut report, BackgroundStage::SummaryPage) .await } }); diff --git a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs index 28a1f8d0d5c..340c44db1f5 100644 --- a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs +++ b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs @@ -1,5 +1,5 @@ use std::future::Future; -use std::sync::Arc; +use std::sync::{Arc, OnceLock}; use std::time::Duration; use tracedecay_session_runtime::session_temporal_refresh_scheduler::SessionTemporalRefreshSchedulerRegistry; @@ -328,14 +328,16 @@ async fn activation_is_pinned_to_the_generation_the_refresh_extends() { #[derive(Default)] struct GatedHistory { entered: tokio::sync::Notify, - release: tokio::sync::Notify, + release: OnceLock<()>, } impl SessionHistoricalIngestor for GatedHistory { fn run_pass(&self) -> SessionHistoricalIngestPass<'_> { Box::pin(async move { self.entered.notify_one(); - self.release.notified().await; + // Both readers must register before the native worker can capture + // the superseded candidate. A yielding history pass serves live joins. + self.release.wait(); SessionHistoricalIngestOutcome::Complete.into() }) } @@ -345,7 +347,7 @@ impl SessionHistoricalIngestor for GatedHistory { } fn cancel(&self) { - self.release.notify_one(); + let _ = self.release.set(()); } } @@ -409,7 +411,7 @@ async fn reset_requested_during_a_refresh_supersedes_its_candidate() { .await, (true, true) ); - history.release.notify_one(); + history.release.set(()).unwrap(); assert_eq!( tokio::join!(named_join, aggregate_join), (false, false), From ddd13c72d284642ed7ca6749504d530fd7e31d06 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 15:09:06 -0800 Subject: [PATCH 10/17] test(dashboard): skip unused facts in retained mutation journeys --- .../tests/dashboard_api_test/dashboard_api_support.rs | 4 ---- .../tracedecay/tests/dashboard_api_test/memory_curation.rs | 6 +++--- 2 files changed, 3 insertions(+), 7 deletions(-) diff --git a/crates/tracedecay/tests/dashboard_api_test/dashboard_api_support.rs b/crates/tracedecay/tests/dashboard_api_test/dashboard_api_support.rs index 35a94248875..22402def73a 100644 --- a/crates/tracedecay/tests/dashboard_api_test/dashboard_api_support.rs +++ b/crates/tracedecay/tests/dashboard_api_test/dashboard_api_support.rs @@ -1105,10 +1105,6 @@ pub(crate) async fn start_dashboard_configuration_fixture() -> DashboardFixture start_dashboard_fixture_with_options(false, false, true).await } -pub(crate) async fn start_dashboard_retained_memory_fixture() -> DashboardFixture { - start_dashboard_fixture_with_options(false, true, true).await -} - /// Starts a dashboard fixture with a fake `DashboardDeliveryReadPortV1` and /// code-index freshness reader mounted, proving the admitted-PR HTTP path /// (`GET /api/delivery/inbox`) joins the provider read with an indexed head diff --git a/crates/tracedecay/tests/dashboard_api_test/memory_curation.rs b/crates/tracedecay/tests/dashboard_api_test/memory_curation.rs index 878afd77d04..fa9b596d1de 100644 --- a/crates/tracedecay/tests/dashboard_api_test/memory_curation.rs +++ b/crates/tracedecay/tests/dashboard_api_test/memory_curation.rs @@ -57,7 +57,7 @@ fn retained_effect_payload<'a>(response: &'a Value, operation: &str) -> &'a Valu fn retained_admin_journey_commits_add_update_feedback_and_remove() { let runtime = create_runtime(); runtime.block_on(async { - let fixture = start_dashboard_retained_memory_fixture().await; + let fixture = start_dashboard_configuration_fixture().await; let agent = http_agent(); let (status, added) = post_json_body( @@ -214,7 +214,7 @@ fn retained_admin_journey_commits_add_update_feedback_and_remove() { fn fact_detail_keeps_a_superseded_fact_readable_by_id() { let runtime = create_runtime(); runtime.block_on(async { - let fixture = start_dashboard_retained_memory_fixture().await; + let fixture = start_dashboard_configuration_fixture().await; let agent = http_agent(); let add = |content: &str| { let (status, added) = post_json_body( @@ -276,7 +276,7 @@ fn fact_detail_keeps_a_superseded_fact_readable_by_id() { fn retained_mutations_deny_foreign_project_scope_without_a_receipt() { let runtime = create_runtime(); runtime.block_on(async { - let fixture = start_dashboard_retained_memory_fixture().await; + let fixture = start_dashboard_configuration_fixture().await; let agent = http_agent(); let (_target_root, target_cg) = crate::projects::setup_target_project(&fixture).await; let target_project_id = target_cg From b0f6277da2bee07ec9be6d0a68d54ac4520f63a1 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 15:48:35 -0800 Subject: [PATCH 11/17] test(sessions): yield refresh history gate instead of blocking worker A synchronous OnceLock wait inside the asynchronous historical-ingest fixture blocked the worker that must poll live projection concurrently. Use a durable async watch latch to preserve release ordering while exercising the real interleaving and supersession journey. Refs #4481 --- .../temporal_projection/activation.rs | 28 +++++++++++++------ 1 file changed, 20 insertions(+), 8 deletions(-) diff --git a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs index 340c44db1f5..e4562af19a8 100644 --- a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs +++ b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs @@ -1,5 +1,5 @@ use std::future::Future; -use std::sync::{Arc, OnceLock}; +use std::sync::Arc; use std::time::Duration; use tracedecay_session_runtime::session_temporal_refresh_scheduler::SessionTemporalRefreshSchedulerRegistry; @@ -325,19 +325,31 @@ async fn activation_is_pinned_to_the_generation_the_refresh_extends() { ); } -#[derive(Default)] struct GatedHistory { entered: tokio::sync::Notify, - release: OnceLock<()>, + release: tokio::sync::watch::Sender, +} + +impl Default for GatedHistory { + fn default() -> Self { + Self { + entered: tokio::sync::Notify::new(), + release: tokio::sync::watch::Sender::new(false), + } + } } impl SessionHistoricalIngestor for GatedHistory { fn run_pass(&self) -> SessionHistoricalIngestPass<'_> { + let mut release = self.release.subscribe(); Box::pin(async move { self.entered.notify_one(); - // Both readers must register before the native worker can capture - // the superseded candidate. A yielding history pass serves live joins. - self.release.wait(); + // The watch latch preserves release even when it precedes this poll, + // while yielding the native worker to concurrent live projection. + release + .wait_for(|released| *released) + .await + .expect("history release sender remains open"); SessionHistoricalIngestOutcome::Complete.into() }) } @@ -347,7 +359,7 @@ impl SessionHistoricalIngestor for GatedHistory { } fn cancel(&self) { - let _ = self.release.set(()); + let _ = self.release.send_replace(true); } } @@ -411,7 +423,7 @@ async fn reset_requested_during_a_refresh_supersedes_its_candidate() { .await, (true, true) ); - history.release.set(()).unwrap(); + assert!(!history.release.send_replace(true)); assert_eq!( tokio::join!(named_join, aggregate_join), (false, false), From 05fe61fa5aee484f664ec9f7bc5a74ad8cdd1efb Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 15:09:06 -0800 Subject: [PATCH 12/17] test(dashboard): skip unused facts in retained mutation journeys --- .../tests/dashboard_api_test/dashboard_api_support.rs | 4 ---- .../tracedecay/tests/dashboard_api_test/memory_curation.rs | 6 +++--- 2 files changed, 3 insertions(+), 7 deletions(-) diff --git a/crates/tracedecay/tests/dashboard_api_test/dashboard_api_support.rs b/crates/tracedecay/tests/dashboard_api_test/dashboard_api_support.rs index 35a94248875..22402def73a 100644 --- a/crates/tracedecay/tests/dashboard_api_test/dashboard_api_support.rs +++ b/crates/tracedecay/tests/dashboard_api_test/dashboard_api_support.rs @@ -1105,10 +1105,6 @@ pub(crate) async fn start_dashboard_configuration_fixture() -> DashboardFixture start_dashboard_fixture_with_options(false, false, true).await } -pub(crate) async fn start_dashboard_retained_memory_fixture() -> DashboardFixture { - start_dashboard_fixture_with_options(false, true, true).await -} - /// Starts a dashboard fixture with a fake `DashboardDeliveryReadPortV1` and /// code-index freshness reader mounted, proving the admitted-PR HTTP path /// (`GET /api/delivery/inbox`) joins the provider read with an indexed head diff --git a/crates/tracedecay/tests/dashboard_api_test/memory_curation.rs b/crates/tracedecay/tests/dashboard_api_test/memory_curation.rs index 878afd77d04..fa9b596d1de 100644 --- a/crates/tracedecay/tests/dashboard_api_test/memory_curation.rs +++ b/crates/tracedecay/tests/dashboard_api_test/memory_curation.rs @@ -57,7 +57,7 @@ fn retained_effect_payload<'a>(response: &'a Value, operation: &str) -> &'a Valu fn retained_admin_journey_commits_add_update_feedback_and_remove() { let runtime = create_runtime(); runtime.block_on(async { - let fixture = start_dashboard_retained_memory_fixture().await; + let fixture = start_dashboard_configuration_fixture().await; let agent = http_agent(); let (status, added) = post_json_body( @@ -214,7 +214,7 @@ fn retained_admin_journey_commits_add_update_feedback_and_remove() { fn fact_detail_keeps_a_superseded_fact_readable_by_id() { let runtime = create_runtime(); runtime.block_on(async { - let fixture = start_dashboard_retained_memory_fixture().await; + let fixture = start_dashboard_configuration_fixture().await; let agent = http_agent(); let add = |content: &str| { let (status, added) = post_json_body( @@ -276,7 +276,7 @@ fn fact_detail_keeps_a_superseded_fact_readable_by_id() { fn retained_mutations_deny_foreign_project_scope_without_a_receipt() { let runtime = create_runtime(); runtime.block_on(async { - let fixture = start_dashboard_retained_memory_fixture().await; + let fixture = start_dashboard_configuration_fixture().await; let agent = http_agent(); let (_target_root, target_cg) = crate::projects::setup_target_project(&fixture).await; let target_project_id = target_cg From 586b0945173b917ddf466dae77cee66d8a5d71d6 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 18:58:10 -0800 Subject: [PATCH 13/17] perf(sessions): select bounded relation recovery query plans Use a bounded creation-order prefix for dense due work and indexed cardinality to choose between deadline and creation order when mixed. Retain due-time consistency and one-row pagination; exercise deferred older receipts with real registered-store recovery. Refs #4481 --- .../src/lcm_effects/tests.rs | 72 ++++++++++++ .../src/relation_projection.rs | 109 +++++++++++++++--- 2 files changed, 167 insertions(+), 14 deletions(-) diff --git a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs index e4f14a706aa..d6ae5bdba4d 100644 --- a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs +++ b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs @@ -2488,6 +2488,59 @@ async fn malformed_relation_receipt_is_permanent_without_starving_summary_work() .await .unwrap(); assert_eq!(changed, 2, "both durable receipts must be present"); + // The older receipts are valid rows in the registered temporal schema, + // but their deadlines are in the future. Only the two real publication + // receipts should be recovered, in creation order, on bounded pages. + let generations = transaction + .execute( + "WITH RECURSIVE seq(n) AS ( + SELECT 1 UNION ALL SELECT n + 1 FROM seq WHERE n < 256 + ) + INSERT INTO session_temporal_generations ( + session_id, generation, state, frozen_watermarks_json, + created_at, ready_at, activated_at + ) + SELECT printf('deferred-relation-%04d', n), 1, 'active', '{}', + n, n, n FROM seq", + (), + ) + .await + .unwrap(); + assert_eq!(generations, 256); + let deferred_count = transaction + .execute( + "WITH RECURSIVE seq(n) AS ( + SELECT 1 UNION ALL SELECT n + 1 FROM seq WHERE n < 256 + ) + INSERT INTO session_relation_receipts ( + session_id, generation, scope_kind, scope_id, + expected_graph_watermark, state, created_at, + recovery_state, recovery_failure_count, + recovery_next_attempt_at + ) + SELECT printf('deferred-relation-%04d', n), 1, + scope.scope_kind, scope.scope_id, 'deferred-test', + 'pending', n, 'retryable', 1, unixepoch() + 3600 + FROM seq CROSS JOIN ( + SELECT scope_kind, scope_id + FROM session_relation_receipts WHERE session_id = ?1 + LIMIT 1 + ) AS scope", + params![poison_session], + ) + .await + .unwrap(); + assert_eq!(deferred_count, 256); + let ordered = transaction + .execute( + "UPDATE session_relation_receipts + SET created_at = CASE session_id WHEN ?1 THEN 1000 ELSE 2000 END + WHERE session_id IN (?1, ?2)", + params![poison_session, healthy_session], + ) + .await + .unwrap(); + assert_eq!(ordered, 2); transaction.commit().await.unwrap(); let recovered_first = SessionTemporalAccess::new(&*db) @@ -2499,6 +2552,21 @@ async fn malformed_relation_receipt_is_permanent_without_starving_summary_work() .unwrap(); assert_eq!(recovered_first.processed, 1); assert!(recovered_first.has_more, "the second due receipt is owed"); + let snapshot = db.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT recovery_state FROM session_relation_receipts WHERE session_id = ?1", + params![poison_session], + ) + .await + .unwrap(); + assert_eq!( + rows.next().await.unwrap().unwrap().get::(0).unwrap(), + "permanent", + "the earlier due receipt must settle on the first page" + ); + drop(rows); + drop(snapshot); let recovered_second = SessionTemporalAccess::new(&*db) .recover_pending_session_relation_projection_page( @@ -2518,6 +2586,10 @@ async fn malformed_relation_receipt_is_permanent_without_starving_summary_work() .unwrap(); assert_eq!(idle.processed, 0, "settled receipts stay parked"); assert!(!idle.has_more); + assert!( + idle.next_retry_at_unix_seconds.is_some(), + "sleeping receipts must keep their future retry deadline" + ); } #[test] diff --git a/crates/tracedecay-session-temporal-store/src/relation_projection.rs b/crates/tracedecay-session-temporal-store/src/relation_projection.rs index c8b43c3a520..1656ec9c77f 100644 --- a/crates/tracedecay-session-temporal-store/src/relation_projection.rs +++ b/crates/tracedecay-session-temporal-store/src/relation_projection.rs @@ -28,6 +28,8 @@ const DEFAULT_MAX_RELATIONS: usize = 100_000; // reconstruction and its receipt acknowledgement; each retry re-reads the // refreshed generation, so the bound only limits back-to-back supersessions. const APPLY_PUBLICATION_RACE_ATTEMPTS: usize = 3; +// A bounded creation-order prefix avoids counting a dense due backlog. +const RECOVERY_CREATION_PROBE_ROWS: i64 = 64; #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub struct SessionRelationRecoveryPage { @@ -335,24 +337,103 @@ impl SessionTemporalAccess<'_, D> { next_retry_at_unix_seconds: retry_at, }); } - let mut rows = snapshot + let query_limit = bounded_query_limit(limit)?; + // Dense ready queues should retain the creation-order plan. A bounded + // prefix detects that case without counting every receipt. When the + // prefix sleeps, indexed cardinalities distinguish sparse due work + // from bulk catch-up: the deadline index avoids reading all older + // deferred receipts, while the creation index avoids sorting all due + // receipts. Neither plan changes the ordered pagination contract. + let mut prefix_rows = snapshot .query( - "SELECT receipt.session_id, receipt.generation, receipt.scope_kind, - receipt.scope_id, journal.projection_json, - receipt.recovery_failure_count - FROM session_relation_receipts AS receipt - LEFT JOIN session_relation_effect_journal AS journal - ON journal.session_id = receipt.session_id - AND journal.generation = receipt.generation - WHERE receipt.state = 'pending' - AND receipt.recovery_state IN ('pending', 'retryable') - AND receipt.recovery_next_attempt_at <= unixepoch() - ORDER BY receipt.created_at, receipt.session_id, receipt.generation - LIMIT ?1", - params![bounded_query_limit(limit)?], + "SELECT COUNT(*) FROM ( + SELECT recovery_state, recovery_next_attempt_at + FROM session_relation_receipts + INDEXED BY idx_session_relation_receipts_pending + WHERE state = 'pending' + ORDER BY created_at, session_id, generation + LIMIT ?1 + ) WHERE recovery_state IN ('pending', 'retryable') + AND recovery_next_attempt_at <= ?2", + params![RECOVERY_CREATION_PROBE_ROWS, now], ) .await .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; + let prefix_due: i64 = prefix_rows + .next() + .await + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))? + .ok_or_else(|| { + storage_message( + RECONSTRUCT_OPERATION, + "relation recovery prefix probe returned no aggregate row", + ) + })? + .get(0) + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; + drop(prefix_rows); + let prefer_creation_order = if prefix_due >= query_limit { + true + } else { + let mut count_rows = snapshot + .query( + "SELECT + (SELECT COUNT(*) FROM session_relation_receipts + INDEXED BY idx_session_relation_receipts_recovery_due + WHERE state = 'pending' + AND recovery_state IN ('pending', 'retryable')), + (SELECT COUNT(*) FROM session_relation_receipts + INDEXED BY idx_session_relation_receipts_recovery_due + WHERE state = 'pending' + AND recovery_state IN ('pending', 'retryable') + AND recovery_next_attempt_at <= ?1)", + params![now], + ) + .await + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; + let counts = count_rows + .next() + .await + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))? + .ok_or_else(|| { + storage_message( + RECONSTRUCT_OPERATION, + "relation recovery count probe returned no aggregate row", + ) + })?; + let eligible: i64 = counts + .get(0) + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; + let due: i64 = counts + .get(1) + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; + drop(count_rows); + due.saturating_mul(3) >= eligible + }; + let index_name = if prefer_creation_order { + "idx_session_relation_receipts_pending" + } else { + "idx_session_relation_receipts_recovery_due" + }; + // The index is one of two literal schema names, not external input. + let candidate_sql = format!( + "SELECT receipt.session_id, receipt.generation, receipt.scope_kind, + receipt.scope_id, journal.projection_json, + receipt.recovery_failure_count + FROM session_relation_receipts AS receipt INDEXED BY {index_name} + LEFT JOIN session_relation_effect_journal AS journal + ON journal.session_id = receipt.session_id + AND journal.generation = receipt.generation + WHERE receipt.state = 'pending' + AND receipt.recovery_state IN ('pending', 'retryable') + AND receipt.recovery_next_attempt_at <= ?2 + ORDER BY receipt.created_at, receipt.session_id, receipt.generation + LIMIT ?1" + ); + let mut rows = snapshot + .query(candidate_sql.as_str(), params![query_limit, now]) + .await + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; let mut pending = Vec::new(); while let Some(row) = rows .next() From ca06fd7dd651a7c0cb1d1aba5c9f7225cdb6e2e5 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 19:04:29 -0800 Subject: [PATCH 14/17] perf(sessions): bound relation recovery plan probing Replace full-cardinality reads with a deadline-index probe capped at 16,384 entries. Preserve sparse/dense plan selection and cover waking an entire deferred backlog through registered-store recovery. Refs #4481 --- .../src/lcm_effects/tests.rs | 40 +++++++++++++++++++ .../src/relation_projection.rs | 34 ++++++++-------- 2 files changed, 57 insertions(+), 17 deletions(-) diff --git a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs index d6ae5bdba4d..02bad741678 100644 --- a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs +++ b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs @@ -2590,6 +2590,46 @@ async fn malformed_relation_receipt_is_permanent_without_starving_summary_work() idle.next_retry_at_unix_seconds.is_some(), "sleeping receipts must keep their future retry deadline" ); + // Waking the entire deferred backlog switches to the creation-ordered + // catch-up path. A one-item page must still surface a sentinel and settle + // the oldest durable receipt, even when its journal is unavailable. + let transaction = db.begin_write_transaction().await.unwrap(); + let woken = transaction + .execute( + "UPDATE session_relation_receipts + SET recovery_next_attempt_at = 0 + WHERE session_id LIKE 'deferred-relation-%'", + (), + ) + .await + .unwrap(); + assert_eq!(woken, 256); + transaction.commit().await.unwrap(); + let dense = SessionTemporalAccess::new(&*db) + .recover_pending_session_relation_projection_page( + 1, + std::sync::Arc::new(tracedecay_graph_db::NeverCancelled), + ) + .await + .unwrap(); + assert_eq!(dense.processed, 1); + assert!(dense.has_more, "a dense due queue keeps the page sentinel"); + let snapshot = db.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT recovery_state, recovery_failure_code + FROM session_relation_receipts + WHERE session_id = 'deferred-relation-0001'", + (), + ) + .await + .unwrap(); + let row = rows.next().await.unwrap().unwrap(); + assert_eq!(row.get::(0).unwrap(), "permanent"); + assert_eq!( + row.get::>(1).unwrap().as_deref(), + Some("journal_missing") + ); } #[test] diff --git a/crates/tracedecay-session-temporal-store/src/relation_projection.rs b/crates/tracedecay-session-temporal-store/src/relation_projection.rs index 1656ec9c77f..6d75ba22d74 100644 --- a/crates/tracedecay-session-temporal-store/src/relation_projection.rs +++ b/crates/tracedecay-session-temporal-store/src/relation_projection.rs @@ -30,6 +30,8 @@ const DEFAULT_MAX_RELATIONS: usize = 100_000; const APPLY_PUBLICATION_RACE_ATTEMPTS: usize = 3; // A bounded creation-order prefix avoids counting a dense due backlog. const RECOVERY_CREATION_PROBE_ROWS: i64 = 64; +// Limit the deadline-probe work before falling back to the ordered index. +const RECOVERY_DUE_PROBE_ROWS: i64 = 16_384; #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub struct SessionRelationRecoveryPage { @@ -375,23 +377,25 @@ impl SessionTemporalAccess<'_, D> { let prefer_creation_order = if prefix_due >= query_limit { true } else { + // Count at most a fixed number of deadline-index entries, not the + // entire pending queue. Sparse due work then sorts only those + // eligible receipts; a large due backlog instead streams in + // creation order without sorting the whole catch-up batch. let mut count_rows = snapshot .query( - "SELECT - (SELECT COUNT(*) FROM session_relation_receipts + "SELECT COUNT(*) FROM ( + SELECT 1 FROM session_relation_receipts INDEXED BY idx_session_relation_receipts_recovery_due - WHERE state = 'pending' - AND recovery_state IN ('pending', 'retryable')), - (SELECT COUNT(*) FROM session_relation_receipts - INDEXED BY idx_session_relation_receipts_recovery_due - WHERE state = 'pending' - AND recovery_state IN ('pending', 'retryable') - AND recovery_next_attempt_at <= ?1)", - params![now], + WHERE state = 'pending' + AND recovery_state IN ('pending', 'retryable') + AND recovery_next_attempt_at <= ?1 + LIMIT ?2 + )", + params![now, RECOVERY_DUE_PROBE_ROWS], ) .await .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; - let counts = count_rows + let due_in_probe: i64 = count_rows .next() .await .map_err(|error| storage(RECONSTRUCT_OPERATION, error))? @@ -400,15 +404,11 @@ impl SessionTemporalAccess<'_, D> { RECONSTRUCT_OPERATION, "relation recovery count probe returned no aggregate row", ) - })?; - let eligible: i64 = counts + })? .get(0) .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; - let due: i64 = counts - .get(1) - .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; drop(count_rows); - due.saturating_mul(3) >= eligible + due_in_probe >= RECOVERY_DUE_PROBE_ROWS }; let index_name = if prefer_creation_order { "idx_session_relation_receipts_pending" From 7a1e6b11390ac5566bc438c5bbf9a62e691c3f5f Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 19:41:50 -0800 Subject: [PATCH 15/17] perf(sessions): bound mixed receipt recovery across deadlines --- .../src/lcm_effects/tests.rs | 94 +++++++- .../src/relation_projection.rs | 228 ++++++++++-------- 2 files changed, 222 insertions(+), 100 deletions(-) diff --git a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs index 02bad741678..1ef310c4266 100644 --- a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs +++ b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs @@ -2277,6 +2277,87 @@ async fn partial_revision_invalidation_hides_replay_and_yields_to_a_due_peer() { ); } +#[tokio::test] +async fn relation_recovery_orders_tied_keys_and_preserves_the_deferred_deadline() { + let harness = RegisteredGlobalDbHarness::open("relation-recovery-page-order").await; + let db = &harness.registered; + let transaction = db.begin_write_transaction().await.unwrap(); + transaction + .execute_batch( + "WITH RECURSIVE seq(n) AS ( + VALUES(0) UNION ALL SELECT n + 1 FROM seq WHERE n < 70 + ) + INSERT INTO session_temporal_generations ( + session_id, generation, state, frozen_watermarks_json, created_at + ) SELECT printf('relation-page-%02d', n / 2), 1 + n % 2, 'building', '{}', n + FROM seq; + INSERT INTO session_relation_receipts ( + session_id, generation, scope_kind, scope_id, expected_graph_watermark, + state, created_at, recovery_state, recovery_next_attempt_at + ) SELECT session_id, generation, 'project_sessions', 'unrelated-scope', 'expected', + 'pending', 42, CASE WHEN generation = 1 THEN 'retryable' ELSE 'pending' END, + CASE WHEN created_at = 70 THEN unixepoch() + 86400 + ELSE (70 - created_at) % 3 END + FROM session_temporal_generations;", + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + let access = SessionTemporalAccess::new(&**db); + let cancellation = std::sync::Arc::new(tracedecay_graph_db::NeverCancelled); + let first = access + .recover_pending_session_relation_projection_page(65, cancellation.clone()) + .await + .unwrap(); + assert_eq!( + (first.processed, first.recovered, first.has_more), + (65, 0, true) + ); + let snapshot = db.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT generation, recovery_state, recovery_failure_code + FROM session_relation_receipts WHERE session_id = 'relation-page-32' + ORDER BY generation", + (), + ) + .await + .unwrap(); + let settled = rows.next().await.unwrap().unwrap(); + assert_eq!(settled.get::(0).unwrap(), 1); + assert_eq!(settled.get::(2).unwrap(), "scope_mismatch"); + let sentinel = rows.next().await.unwrap().unwrap(); + assert_eq!(sentinel.get::(0).unwrap(), 2); + assert_eq!(sentinel.get::(1).unwrap(), "pending"); + assert_eq!(sentinel.get::>(2).unwrap(), None); + drop(rows); + let mut rows = snapshot + .query( + "SELECT recovery_next_attempt_at FROM session_relation_receipts + WHERE session_id = 'relation-page-35' AND generation = 1", + (), + ) + .await + .unwrap(); + let deferred_deadline = rows.next().await.unwrap().unwrap().get::(0).unwrap(); + drop(rows); + drop(snapshot); + let rest = access + .recover_pending_session_relation_projection_page(65, cancellation.clone()) + .await + .unwrap(); + assert_eq!( + (rest.processed, rest.recovered, rest.has_more), + (5, 0, false) + ); + let idle = access + .recover_pending_session_relation_projection_page(65, cancellation) + .await + .unwrap(); + assert_eq!((idle.processed, idle.has_more), (0, false)); + assert_eq!(idle.next_retry_at_unix_seconds, Some(deferred_deadline)); +} + #[tokio::test] async fn malformed_relation_receipt_is_permanent_without_starving_summary_work() { let harness = RegisteredGlobalDbHarness::open("lcm-relation-recovery-poison").await; @@ -2497,11 +2578,9 @@ async fn malformed_relation_receipt_is_permanent_without_starving_summary_work() SELECT 1 UNION ALL SELECT n + 1 FROM seq WHERE n < 256 ) INSERT INTO session_temporal_generations ( - session_id, generation, state, frozen_watermarks_json, - created_at, ready_at, activated_at + session_id, generation, state, frozen_watermarks_json, created_at ) - SELECT printf('deferred-relation-%04d', n), 1, 'active', '{}', - n, n, n FROM seq", + SELECT printf('deferred-relation-%04d', n), 1, 'building', '{}', n FROM seq", (), ) .await @@ -2561,7 +2640,12 @@ async fn malformed_relation_receipt_is_permanent_without_starving_summary_work() .await .unwrap(); assert_eq!( - rows.next().await.unwrap().unwrap().get::(0).unwrap(), + rows.next() + .await + .unwrap() + .unwrap() + .get::(0) + .unwrap(), "permanent", "the earlier due receipt must settle on the first page" ); diff --git a/crates/tracedecay-session-temporal-store/src/relation_projection.rs b/crates/tracedecay-session-temporal-store/src/relation_projection.rs index 6d75ba22d74..86b879c18e8 100644 --- a/crates/tracedecay-session-temporal-store/src/relation_projection.rs +++ b/crates/tracedecay-session-temporal-store/src/relation_projection.rs @@ -6,7 +6,7 @@ use tracedecay_domain::{ SessionProjectionGenerationV1, TemporalValidityV1, ThreadId, UtcMicros, }; use tracedecay_graph_db::{GraphCancellation, GraphWatermark}; -use tracedecay_runtime_core::db::engine::params; +use tracedecay_runtime_core::db::engine::{IntoParams, params}; use tracedecay_store::{SessionStoreError, SessionStoreResult}; use super::operations::CanonicalPublicationManifest; @@ -17,7 +17,7 @@ use super::relations::{ ThreadHierarchyRelation, WorkflowAgentMembership, }; use crate::handle::{ - SessionTemporalAccess, SessionTemporalExec, SessionTemporalRegisteredDb, + SessionTemporalAccess, SessionTemporalExec, SessionTemporalQuery, SessionTemporalRegisteredDb, SessionTemporalWriteTxn, }; @@ -28,10 +28,25 @@ const DEFAULT_MAX_RELATIONS: usize = 100_000; // reconstruction and its receipt acknowledgement; each retry re-reads the // refreshed generation, so the bound only limits back-to-back supersessions. const APPLY_PUBLICATION_RACE_ATTEMPTS: usize = 3; -// A bounded creation-order prefix avoids counting a dense due backlog. -const RECOVERY_CREATION_PROBE_ROWS: i64 = 64; -// Limit the deadline-probe work before falling back to the ordered index. -const RECOVERY_DUE_PROBE_ROWS: i64 = 16_384; +const INITIAL_RECOVERY_PROBE_ROWS: i64 = 64; + +// Seek the next distinct deadline in each eligible state. A fifth live +// deadline means the four equality ranges for that state are insufficient. +const RECOVERY_DEADLINES_SQL: &str = " + WITH RECURSIVE states(kind) AS (VALUES('pending'), ('retryable')), + bands(kind, deadline, remaining) AS ( + SELECT kind, ( + SELECT MIN(recovery_next_attempt_at) FROM session_relation_receipts + WHERE state = 'pending' AND recovery_state = states.kind + AND recovery_next_attempt_at <= ?2 + ), 4 FROM states + UNION ALL + SELECT kind, ( + SELECT MIN(recovery_next_attempt_at) FROM session_relation_receipts + WHERE state = 'pending' AND recovery_state = bands.kind + AND recovery_next_attempt_at > bands.deadline AND recovery_next_attempt_at <= ?2 + ), remaining - 1 FROM bands WHERE deadline IS NOT NULL AND remaining > 0 + )"; #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] pub struct SessionRelationRecoveryPage { @@ -340,107 +355,38 @@ impl SessionTemporalAccess<'_, D> { }); } let query_limit = bounded_query_limit(limit)?; - // Dense ready queues should retain the creation-order plan. A bounded - // prefix detects that case without counting every receipt. When the - // prefix sleeps, indexed cardinalities distinguish sparse due work - // from bulk catch-up: the deadline index avoids reading all older - // deferred receipts, while the creation index avoids sorting all due - // receipts. Neither plan changes the ordered pagination contract. - let mut prefix_rows = snapshot - .query( - "SELECT COUNT(*) FROM ( - SELECT recovery_state, recovery_next_attempt_at - FROM session_relation_receipts - INDEXED BY idx_session_relation_receipts_pending - WHERE state = 'pending' - ORDER BY created_at, session_id, generation - LIMIT ?1 - ) WHERE recovery_state IN ('pending', 'retryable') - AND recovery_next_attempt_at <= ?2", - params![RECOVERY_CREATION_PROBE_ROWS, now], - ) - .await - .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; - let prefix_due: i64 = prefix_rows - .next() - .await - .map_err(|error| storage(RECONSTRUCT_OPERATION, error))? - .ok_or_else(|| { - storage_message( - RECONSTRUCT_OPERATION, - "relation recovery prefix probe returned no aggregate row", - ) - })? - .get(0) - .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; - drop(prefix_rows); - let prefer_creation_order = if prefix_due >= query_limit { - true - } else { - // Count at most a fixed number of deadline-index entries, not the - // entire pending queue. Sparse due work then sorts only those - // eligible receipts; a large due backlog instead streams in - // creation order without sorting the whole catch-up batch. - let mut count_rows = snapshot - .query( - "SELECT COUNT(*) FROM ( - SELECT 1 FROM session_relation_receipts - INDEXED BY idx_session_relation_receipts_recovery_due - WHERE state = 'pending' - AND recovery_state IN ('pending', 'retryable') - AND recovery_next_attempt_at <= ?1 - LIMIT ?2 - )", - params![now, RECOVERY_DUE_PROBE_ROWS], - ) - .await - .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; - let due_in_probe: i64 = count_rows - .next() - .await - .map_err(|error| storage(RECONSTRUCT_OPERATION, error))? - .ok_or_else(|| { - storage_message( - RECONSTRUCT_OPERATION, - "relation recovery count probe returned no aggregate row", - ) - })? - .get(0) - .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; - drop(count_rows); - due_in_probe >= RECOVERY_DUE_PROBE_ROWS - }; - let index_name = if prefer_creation_order { - "idx_session_relation_receipts_pending" - } else { - "idx_session_relation_receipts_recovery_due" - }; - // The index is one of two literal schema names, not external input. + let candidates = + recovery_candidate_query(&snapshot, now, query_limit, &cancellation).await?; let candidate_sql = format!( - "SELECT receipt.session_id, receipt.generation, receipt.scope_kind, - receipt.scope_id, journal.projection_json, - receipt.recovery_failure_count - FROM session_relation_receipts AS receipt INDEXED BY {index_name} + "WITH candidates AS MATERIALIZED ({candidates}), page AS MATERIALIZED ( + SELECT created_at, session_id, generation FROM candidates + ORDER BY created_at, session_id, generation LIMIT (?1 - 1) + ) + SELECT page.session_id, page.generation, receipt.scope_kind, receipt.scope_id, + journal.projection_json, receipt.recovery_failure_count, + (SELECT COUNT(*) FROM candidates) = ?1 + FROM page JOIN session_relation_receipts AS receipt + ON receipt.session_id = page.session_id AND receipt.generation = page.generation LEFT JOIN session_relation_effect_journal AS journal - ON journal.session_id = receipt.session_id - AND journal.generation = receipt.generation - WHERE receipt.state = 'pending' - AND receipt.recovery_state IN ('pending', 'retryable') - AND receipt.recovery_next_attempt_at <= ?2 - ORDER BY receipt.created_at, receipt.session_id, receipt.generation - LIMIT ?1" + ON journal.session_id = page.session_id AND journal.generation = page.generation + ORDER BY page.created_at, page.session_id, page.generation" ); let mut rows = snapshot .query(candidate_sql.as_str(), params![query_limit, now]) .await .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; let mut pending = Vec::new(); + let mut has_more = false; while let Some(row) = rows .next() .await .map_err(|error| storage(RECONSTRUCT_OPERATION, error))? { require_not_cancelled(&cancellation)?; + has_more = row + .get::(6) + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))? + != 0; pending.push(PendingRelationRecovery { session_id: row .get(0) @@ -468,8 +414,6 @@ impl SessionTemporalAccess<'_, D> { SessionRelationScope::ProjectSessions { .. } => "project_sessions", SessionRelationScope::ProfileSessions { .. } => "profile_sessions", }; - let has_more = pending.len() > limit; - pending.truncate(limit); let mut recovered = 0_usize; let mut processed = 0_usize; for candidate in &pending { @@ -1719,6 +1663,100 @@ async fn active_generation( }) } +async fn recovery_probe( + snapshot: &impl SessionTemporalQuery, + sql: &str, + values: impl IntoParams + Send, +) -> SessionStoreResult { + let mut rows = snapshot + .query(sql, values) + .await + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))?; + rows.next() + .await + .map_err(|error| storage(RECONSTRUCT_OPERATION, error))? + .ok_or_else(|| storage_message(RECONSTRUCT_OPERATION, "missing relation recovery probe"))? + .get::(0) + .map(|value| value != 0) + .map_err(|error| storage(RECONSTRUCT_OPERATION, error)) +} + +async fn recovery_candidate_query( + snapshot: &impl SessionTemporalQuery, + now: i64, + query_limit: i64, + cancellation: &Arc, +) -> SessionStoreResult { + let mut window = INITIAL_RECOVERY_PROBE_ROWS; + let index = loop { + require_not_cancelled(cancellation)?; + // Stop the creation scan as soon as the requested page and sentinel + // are present, even if the current probe window is much larger. + let page_in_prefix = "SELECT EXISTS ( + SELECT 1 FROM ( + SELECT recovery_state, recovery_next_attempt_at FROM session_relation_receipts + INDEXED BY idx_session_relation_receipts_pending + WHERE state = 'pending' + ORDER BY created_at, session_id, generation LIMIT ?1 + ) WHERE recovery_state IN ('pending', 'retryable') AND recovery_next_attempt_at <= ?2 + LIMIT 1 OFFSET (?3 - 1) + )"; + if recovery_probe(snapshot, page_in_prefix, params![window, now, query_limit]).await? { + break "idx_session_relation_receipts_pending"; + } + + if window == INITIAL_RECOVERY_PROBE_ROWS { + let complete_bands = format!( + "{RECOVERY_DEADLINES_SQL} SELECT NOT EXISTS ( + SELECT 1 FROM bands WHERE deadline IS NOT NULL AND remaining = 0)" + ); + if recovery_probe(snapshot, &complete_bands, params![query_limit, now]).await? { + // Each exact state/deadline range is already in creation order. + // Only its first page can contribute to the global first page. + return Ok(format!( + "{RECOVERY_DEADLINES_SQL} + SELECT r.created_at, r.session_id, r.generation FROM bands + CROSS JOIN session_relation_receipts AS r + WHERE bands.deadline IS NOT NULL AND r.rowid IN ( + SELECT rowid FROM session_relation_receipts + INDEXED BY idx_session_relation_receipts_recovery_due + WHERE state = 'pending' AND recovery_state = bands.kind + AND recovery_next_attempt_at = bands.deadline + ORDER BY created_at, session_id, generation LIMIT ?1 + ) + ORDER BY r.created_at, r.session_id, r.generation LIMIT ?1" + )); + } + } + + // For fragmented deadlines, compare the distance to the ordered page + // against the complete due population. Neither index is scanned past + // the growing window until it proves which bounded result to read. + if !recovery_probe( + snapshot, + "SELECT EXISTS ( + SELECT 1 FROM session_relation_receipts + INDEXED BY idx_session_relation_receipts_recovery_due + WHERE state = 'pending' AND recovery_state IN ('pending', 'retryable') + AND recovery_next_attempt_at <= ?2 LIMIT 1 OFFSET ?1 + )", + // The covering deadline index can inspect four keys for roughly + // the work of fetching one receipt through the creation index. + params![window.saturating_mul(4), now], + ) + .await? + { + break "idx_session_relation_receipts_recovery_due"; + } + window = window.saturating_mul(2); + }; + Ok(format!( + "SELECT created_at, session_id, generation FROM session_relation_receipts INDEXED BY {index} + WHERE state = 'pending' AND recovery_state IN ('pending', 'retryable') + AND recovery_next_attempt_at <= ?2 ORDER BY created_at, session_id, generation LIMIT ?1" + )) +} + fn bounded_query_limit(limit: usize) -> SessionStoreResult { let bounded = limit .checked_add(1) From a5ddc7e69f9412db0ac9e28e942bbd117aed4a5b Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 20:08:51 -0800 Subject: [PATCH 16/17] perf(sessions): hydrate relation receipts from the bounded page --- .../src/relation_projection.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay-session-temporal-store/src/relation_projection.rs b/crates/tracedecay-session-temporal-store/src/relation_projection.rs index 86b879c18e8..33403ba3d3e 100644 --- a/crates/tracedecay-session-temporal-store/src/relation_projection.rs +++ b/crates/tracedecay-session-temporal-store/src/relation_projection.rs @@ -357,6 +357,8 @@ impl SessionTemporalAccess<'_, D> { let query_limit = bounded_query_limit(limit)?; let candidates = recovery_candidate_query(&snapshot, now, query_limit, &cancellation).await?; + // Drive hydration from the bounded page: a planner-selected receipt + // outer loop would revisit the sleeping backlog on each small page. let candidate_sql = format!( "WITH candidates AS MATERIALIZED ({candidates}), page AS MATERIALIZED ( SELECT created_at, session_id, generation FROM candidates @@ -365,7 +367,7 @@ impl SessionTemporalAccess<'_, D> { SELECT page.session_id, page.generation, receipt.scope_kind, receipt.scope_id, journal.projection_json, receipt.recovery_failure_count, (SELECT COUNT(*) FROM candidates) = ?1 - FROM page JOIN session_relation_receipts AS receipt + FROM page CROSS JOIN session_relation_receipts AS receipt ON receipt.session_id = page.session_id AND receipt.generation = page.generation LEFT JOIN session_relation_effect_journal AS journal ON journal.session_id = page.session_id AND journal.generation = page.generation From f306a33a1be7967eacab8c4a1f31f4d22de0fff7 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Mon, 9 Nov 2026 04:03:54 -0800 Subject: [PATCH 17/17] test(sessions): cover fragmented relation recovery pagination Exercise 24 distinct due retry deadlines behind 512 sleeping receipts through successive one-item registered-store recovery pages. Assert each page settles the oldest due receipt and preserves the future retry wakeup. Refs #4481 --- .../src/lcm_effects/tests.rs | 70 +++++++++++++++++++ 1 file changed, 70 insertions(+) diff --git a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs index 1ef310c4266..6b51b5c8caa 100644 --- a/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs +++ b/crates/tracedecay-session-runtime/src/lcm_effects/tests.rs @@ -2358,6 +2358,76 @@ async fn relation_recovery_orders_tied_keys_and_preserves_the_deferred_deadline( assert_eq!(idle.next_retry_at_unix_seconds, Some(deferred_deadline)); } +#[tokio::test] +async fn relation_recovery_fragmented_deadlines_preserve_order_across_single_item_pages() { + let harness = RegisteredGlobalDbHarness::open("relation-fragmented-deadlines").await; + let db = &harness.registered; + let transaction = db.begin_write_transaction().await.unwrap(); + transaction + .execute_batch( + "WITH RECURSIVE seq(n) AS ( + VALUES(0) UNION ALL SELECT n + 1 FROM seq WHERE n < 535 + ) + INSERT INTO session_temporal_generations ( + session_id, generation, state, frozen_watermarks_json, created_at + ) SELECT printf('fragmented-relation-%04d', n), 1, 'building', '{}', n + FROM seq; + INSERT INTO session_relation_receipts ( + session_id, generation, scope_kind, scope_id, expected_graph_watermark, + state, created_at, recovery_state, recovery_next_attempt_at + ) SELECT session_id, generation, 'project_sessions', 'unrelated-scope', + 'expected', 'pending', created_at, + CASE WHEN created_at < 512 THEN 'retryable' ELSE 'pending' END, + CASE WHEN created_at < 512 THEN unixepoch() + 86400 + ELSE created_at - 511 END + FROM session_temporal_generations;", + ) + .await + .unwrap(); + transaction.commit().await.unwrap(); + + let access = SessionTemporalAccess::new(&**db); + let cancellation = std::sync::Arc::new(tracedecay_graph_db::NeverCancelled); + // The 512 older receipts are sleeping, while 24 newer due receipts have + // 24 distinct deadlines. No four-deadline band can cover the due set. + // This forces the fragmented fallback repeatedly, not just on page one. + for index in 0..24 { + let page = access + .recover_pending_session_relation_projection_page(1, cancellation.clone()) + .await + .unwrap(); + assert_eq!((page.processed, page.recovered), (1, 0)); + assert_eq!(page.has_more, index < 23); + + let snapshot = db.read_snapshot().await.unwrap(); + let mut rows = snapshot + .query( + "SELECT session_id FROM session_relation_receipts + WHERE recovery_state = 'permanent' ORDER BY created_at", + (), + ) + .await + .unwrap(); + let mut settled = Vec::new(); + while let Some(row) = rows.next().await.unwrap() { + settled.push(row.get::(0).unwrap()); + } + assert_eq!(settled.len(), index + 1); + let expected = format!("fragmented-relation-{:04}", index + 512); + assert_eq!(settled.last().map(String::as_str), Some(expected.as_str())); + } + + let idle = access + .recover_pending_session_relation_projection_page(1, cancellation) + .await + .unwrap(); + assert_eq!((idle.processed, idle.has_more), (0, false)); + assert!( + idle.next_retry_at_unix_seconds.is_some(), + "the sleeping prefix must retain its future wakeup" + ); +} + #[tokio::test] async fn malformed_relation_receipt_is_permanent_without_starving_summary_work() { let harness = RegisteredGlobalDbHarness::open("lcm-relation-recovery-poison").await;