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 ee3ab998e87..c268d65ea18 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(), 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 5bc41596610..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 @@ -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,30 @@ 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, + BackgroundStage::History, + ) + .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 +394,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 +409,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; } @@ -465,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; @@ -498,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; @@ -937,23 +965,39 @@ 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> { +#[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> { database: &'a RegisteredGlobalDbLeaseV1, state: &'a Arc, projector: &'a dyn SessionTemporalRefreshProjector, policy: SessionTemporalRefreshPolicy, } -impl LiveProjectionDuringSummary<'_> { +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, @@ -971,19 +1015,20 @@ 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 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 } }; @@ -996,10 +1041,10 @@ impl LiveProjectionDuringSummary<'_> { if !woke { return None; } - if !self.state.take_dirty() { + if !self.projection_due(background, joins) || !self.state.take_dirty() { continue; } - summary_pass_owed |= !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); @@ -1059,15 +1104,14 @@ impl LiveProjectionDuringSummary<'_> { 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 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, recheck_summary)) } } @@ -2256,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()); @@ -2277,7 +2354,7 @@ mod tests { entered.notify_one(); finish.notified().await; }; - let live_projection = LiveProjectionDuringSummary { + let live_projection = LiveProjection { database: &database, state: &state, projector: &CanonicalSessionTemporalProjector, @@ -2286,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 } }); @@ -2360,7 +2437,7 @@ mod tests { finish.notified().await; 17usize }; - let live_projection = LiveProjectionDuringSummary { + let live_projection = LiveProjection { database: &database, state: &state, projector: &CanonicalSessionTemporalProjector, @@ -2369,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-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. diff --git a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs index 28a1f8d0d5c..a7bc3b04760 100644 --- a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs +++ b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs @@ -1,4 +1,3 @@ -use std::future::Future; use std::sync::Arc; use std::time::Duration; @@ -325,17 +324,31 @@ async fn activation_is_pinned_to_the_generation_the_refresh_extends() { ); } -#[derive(Default)] struct GatedHistory { entered: tokio::sync::Notify, - release: tokio::sync::Notify, + 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(); - self.release.notified().await; + // 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() }) } @@ -345,12 +358,26 @@ impl SessionHistoricalIngestor for GatedHistory { } fn cancel(&self) { - self.release.notify_one(); + let _ = self.release.send_replace(true); } } +#[derive(Clone, Copy, Debug)] +enum PublicationScope { + Session, + Profile, +} + #[tokio::test] async fn reset_requested_during_a_refresh_supersedes_its_candidate() { + // Each request wakes the worker immediately. A second request may already + // belong to the replacement pass, so exercise each scope on its own reset. + for scope in [PublicationScope::Session, PublicationScope::Profile] { + assert_superseded_publication(scope).await; + } +} + +async fn assert_superseded_publication(scope: PublicationScope) { let tmp = TempDir::new().unwrap(); let (runtime, path, stale) = frontier_digest_fixture(&tmp, true).await; let session_id = stale.session_id().clone(); @@ -398,23 +425,28 @@ async fn reset_requested_during_a_refresh_supersedes_its_candidate() { ) .await; history.entered.notified().await; - let named_join = wake.wake_and_wait_for_session_projection(&session_id, Duration::from_secs(5)); - let aggregate_join = wake.wake_and_wait_for_projection(Duration::from_secs(5)); - tokio::pin!(named_join, aggregate_join); - assert_eq!( - std::future::poll_fn(|cx| std::task::Poll::Ready(( - named_join.as_mut().poll(cx).is_pending(), - aggregate_join.as_mut().poll(cx).is_pending(), - ))) - .await, - (true, true) + let timeout = Duration::from_secs(5); + let published = match scope { + PublicationScope::Session => { + wake.wake_and_wait_for_session_projection(&session_id, timeout) + .await + } + PublicationScope::Profile => wake.wake_and_wait_for_projection(timeout).await, + }; + assert!( + !published, + "the superseded operation published for {scope:?}" ); - history.release.notify_one(); assert_eq!( - tokio::join!(named_join, aggregate_join), - (false, false), - "the superseded operation must refuse both waiting publication joins" + rows( + &path, + "SELECT operation_id FROM session_refresh_operations WHERE state = 'failed'" + ) + .await, + vec![stale.operation_id().as_str()], + "{scope:?} must observe the failed operation while history remains parked" ); + assert!(!history.release.send_replace(true)); registry.wait_profile_idle(&path).await; assert_eq!(