diff --git a/crates/tracedecay-cli/tests/core_cli_suite/main.rs b/crates/tracedecay-cli/tests/core_cli_suite/main.rs index 4ac196eee9..f319aa39e6 100644 --- a/crates/tracedecay-cli/tests/core_cli_suite/main.rs +++ b/crates/tracedecay-cli/tests/core_cli_suite/main.rs @@ -26,6 +26,8 @@ mod monitor_test; mod provision_host_cli_fixture; #[cfg(unix)] mod remote_cli_test; +#[cfg(unix)] +mod session_store_reopen_test; mod source_provenance_test; mod sync_test; mod test_profile_isolation_test; diff --git a/crates/tracedecay-cli/tests/core_cli_suite/session_store_reopen_test.rs b/crates/tracedecay-cli/tests/core_cli_suite/session_store_reopen_test.rs new file mode 100644 index 0000000000..86a600a8eb --- /dev/null +++ b/crates/tracedecay-cli/tests/core_cli_suite/session_store_reopen_test.rs @@ -0,0 +1,170 @@ +//! A daemon that restarts without its authority record starts at authority +//! epoch 1 again and so reuses the previous generation's store incarnation. +//! The project sessions store then already holds a writer authority above the +//! epoch the new daemon allocates for it. The restarted daemon must converge +//! on that persisted authority and keep projecting; it must not retry a write +//! the store fences on every pass. + +use std::fs; +use std::path::{Path, PathBuf}; +use std::process::Command; +use std::time::{Duration, Instant}; + +use tempfile::TempDir; + +use crate::common::{self, canonical_existing_path, tracedecay_command_with_home}; + +const THREAD_ID: &str = "codex-reopen-thread"; +const FIRST_MARKER: &str = "ledgerquartz"; +const RESTART_MARKER: &str = "restartquartz"; +/// A projection pass runs in the background; this is how long one import may +/// take to become searchable on a loaded runner. +const PROJECTION_BUDGET: Duration = Duration::from_secs(120); + +fn git(project: &Path, args: &[&str]) { + let output = Command::new(common::git_program()) + .args(args) + .current_dir(project) + .output() + .expect("git should run"); + assert!( + output.status.success(), + "git {args:?} failed: {}", + String::from_utf8_lossy(&output.stderr) + ); +} + +fn rollout_line(timestamp: &str, kind: &str, payload: serde_json::Value) -> String { + serde_json::json!({ "timestamp": timestamp, "type": kind, "payload": payload }).to_string() +} + +fn write_rollout(home: &Path, project: &Path, messages: &[&str]) -> PathBuf { + let dir = home.join(".codex/sessions/2026/01/01"); + fs::create_dir_all(&dir).unwrap(); + let path = dir.join(format!("rollout-2026-01-01T00-00-00-{THREAD_ID}.jsonl")); + let mut lines = vec![rollout_line( + "2026-01-01T00:00:00.000Z", + "session_meta", + serde_json::json!({ + "id": THREAD_ID, + "cwd": project.to_string_lossy(), + "model": "gpt-5.5" + }), + )]; + for (index, message) in messages.iter().enumerate() { + lines.push(rollout_line( + &format!("2026-01-01T00:00:{:02}.000Z", index + 1), + "event_msg", + serde_json::json!({ "type": "agent_message", "message": message }), + )); + } + fs::write(&path, format!("{}\n", lines.join("\n"))).unwrap(); + path +} + +fn import_sessions(home: &Path, project: &Path) { + let output = tracedecay_command_with_home(home) + .current_dir(project) + .args(["sessions", "import", "--project-path"]) + .arg(project) + .output() + .expect("tracedecay sessions import should run"); + assert!( + output.status.success(), + "sessions import failed\nstdout:\n{}\nstderr:\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); +} + +fn wait_until_searchable(home: &Path, project: &Path, marker: &str, context: &str) { + common::poll_until( + Instant::now() + PROJECTION_BUDGET, + Duration::from_millis(500), + || { + let output = tracedecay_command_with_home(home) + .current_dir(project) + .args([ + "sessions", + "search", + marker, + "--provider", + "codex", + "--project-path", + ]) + .arg(project) + .args(["--limit", "20", "--json"]) + .output() + .expect("tracedecay sessions search should run"); + (output.status.success() && String::from_utf8_lossy(&output.stdout).contains(marker)) + .then_some(()) + }, + || format!("{context}: {marker} never became searchable"), + ); +} + +#[test] +fn a_daemon_restarted_on_a_reused_incarnation_converges_on_the_persisted_writer_authority() { + let home = TempDir::new().unwrap(); + let project = TempDir::new().unwrap(); + let home_path = canonical_existing_path(home.path()); + let project_path = canonical_existing_path(project.path()); + fs::write(project_path.join("main.rs"), "fn main() {}\n").unwrap(); + git(&project_path, &["init", "--quiet", "--initial-branch=main"]); + git(&project_path, &["add", "."]); + git( + &project_path, + &[ + "-c", + "user.email=reopen@tracedecay.invalid", + "-c", + "user.name=Reopen", + "commit", + "--quiet", + "-m", + "fixture", + ], + ); + + let first_log = home_path.join("daemon-first.log"); + let mut first = common::spawn_tracedecay_daemon_logged(&home_path, &first_log, |_| {}); + let init = tracedecay_command_with_home(&home_path) + .current_dir(&project_path) + .arg("init") + .output() + .expect("tracedecay init should run"); + assert!( + init.status.success(), + "tracedecay init failed: {}", + String::from_utf8_lossy(&init.stderr) + ); + write_rollout(&home_path, &project_path, &[FIRST_MARKER]); + import_sessions(&home_path, &project_path); + wait_until_searchable(&home_path, &project_path, FIRST_MARKER, "first daemon"); + first.kill_and_wait().expect("stop the first daemon"); + + // The managed-daemon fixture and the work-loop journey both drop this + // record between restarts, as does an operator clearing stale daemon + // state. The next daemon then takes authority epoch 1 again and the + // same store incarnation as the generation that just wrote the store. + fs::remove_file(common::daemon_authority_path( + &home_path.join(".tracedecay"), + )) + .expect("remove the first daemon's authority record"); + let second_log = home_path.join("daemon-second.log"); + let _second = common::spawn_tracedecay_daemon_logged(&home_path, &second_log, |_| {}); + write_rollout(&home_path, &project_path, &[FIRST_MARKER, RESTART_MARKER]); + import_sessions(&home_path, &project_path); + wait_until_searchable( + &home_path, + &project_path, + RESTART_MARKER, + "restarted daemon", + ); + + let second_stderr = fs::read_to_string(&second_log).expect("read the second daemon's log"); + assert!( + !second_stderr.contains("stale writer authority epoch"), + "the restarted daemon retried writes its store had fenced:\n{second_stderr}" + ); +} diff --git a/crates/tracedecay-contracts/src/session_temporal_refresh.rs b/crates/tracedecay-contracts/src/session_temporal_refresh.rs index b1cc77c845..9a2b1cc10a 100644 --- a/crates/tracedecay-contracts/src/session_temporal_refresh.rs +++ b/crates/tracedecay-contracts/src/session_temporal_refresh.rs @@ -2,13 +2,30 @@ use std::future::Future; use std::pin::Pin; use std::time::Duration; +use tracedecay_domain::SessionId; + pub type SessionTemporalRefreshWakeFuture<'a> = Pin + Send + 'a>>; /// Wake and observe the session-temporal refresh worker bound to one store. pub trait SessionTemporalRefreshWakePort: Send + Sync { fn wake(&self) -> bool; fn is_unavailable(&self) -> bool; - fn wake_and_wait_until_idle(&self, timeout: Duration) -> SessionTemporalRefreshWakeFuture<'_>; + /// Join temporal publication caused by this wake. Background history and + /// summary convergence may continue after the projection has committed. + fn wake_and_wait_for_projection( + &self, + timeout: Duration, + ) -> SessionTemporalRefreshWakeFuture<'_>; + + /// Join the named session's committed temporal publication. Owners that + /// cannot prove a session frontier retain the stricter store-wide join. + fn wake_and_wait_for_session_projection<'a>( + &'a self, + _session_id: &'a SessionId, + timeout: Duration, + ) -> SessionTemporalRefreshWakeFuture<'a> { + self.wake_and_wait_for_projection(timeout) + } } /// Typed missing-worker implementation for unmounted servers and tests. @@ -24,7 +41,10 @@ impl SessionTemporalRefreshWakePort for UnavailableSessionTemporalRefreshWake { true } - fn wake_and_wait_until_idle(&self, _timeout: Duration) -> SessionTemporalRefreshWakeFuture<'_> { + fn wake_and_wait_for_projection( + &self, + _timeout: Duration, + ) -> SessionTemporalRefreshWakeFuture<'_> { Box::pin(async { false }) } } diff --git a/crates/tracedecay-daemon-service/src/session_refresh.rs b/crates/tracedecay-daemon-service/src/session_refresh.rs index b878505890..a49649a40b 100644 --- a/crates/tracedecay-daemon-service/src/session_refresh.rs +++ b/crates/tracedecay-daemon-service/src/session_refresh.rs @@ -295,7 +295,10 @@ impl SessionTemporalRefreshWakePort for FixedSessionTemporalRefreshWake { !self.0 } - fn wake_and_wait_until_idle(&self, _timeout: Duration) -> SessionTemporalRefreshWakeFuture<'_> { + fn wake_and_wait_for_projection( + &self, + _timeout: Duration, + ) -> SessionTemporalRefreshWakeFuture<'_> { let accepted = self.0; Box::pin(async move { accepted }) } diff --git a/crates/tracedecay-global-db/src/observation_adapter.rs b/crates/tracedecay-global-db/src/observation_adapter.rs index 648f47944a..b62a03a30e 100644 --- a/crates/tracedecay-global-db/src/observation_adapter.rs +++ b/crates/tracedecay-global-db/src/observation_adapter.rs @@ -2517,6 +2517,13 @@ fn rejected_submit_error( RuntimeSubmitOutcomeV1::Saturated { .. } => { ObservationStoreError::WriterSaturated { operation } } + RuntimeSubmitOutcomeV1::Fenced { expected, actual } => ObservationStoreError::Storage { + operation, + source: Box::new(tracedecay_store::StoreWriterFencedV1 { + requested: *expected, + current: *actual, + }), + }, outcome => runtime_storage_error(operation, format!("{rejection}: {outcome:?}")), } } @@ -2609,6 +2616,24 @@ mod tests { use super::*; + #[test] + fn writer_fence_survives_observation_and_projection_error_wrapping() { + let requested = tracedecay_store::StoreAuthorityEpochV1::new(2).unwrap(); + let current = tracedecay_store::StoreAuthorityEpochV1::new(3).unwrap(); + let error = rejected_submit_error( + "submit anchored observation", + "runtime rejected observation write", + &RuntimeSubmitOutcomeV1::Fenced { + expected: requested, + actual: current, + }, + ); + assert_eq!( + projection_runtime_error(error).writer_fence(), + Some(tracedecay_store::StoreWriterFencedV1 { requested, current }), + ); + } + #[test] fn writer_saturation_is_backpressure_not_a_storage_failure() { for scope in [ diff --git a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs index e225c63ca3..e1fc137277 100644 --- a/crates/tracedecay-global-db/src/observation_projection/rebuild.rs +++ b/crates/tracedecay-global-db/src/observation_projection/rebuild.rs @@ -103,7 +103,9 @@ fn json_extract_neq_predicates(left_alias: &str, json_column: &str, fields: &[&s } fn records_transient_projection_retry(error: &ProjectionStoreError) -> bool { - matches!(error, ProjectionStoreError::Storage { .. }) && !error.is_sqlite_disk_image_corrupt() + matches!(error, ProjectionStoreError::Storage { .. }) + && !error.is_sqlite_disk_image_corrupt() + && error.writer_fence().is_none() } /// Projects one queued observation through the guarded registered database diff --git a/crates/tracedecay-host-admission/src/lib.rs b/crates/tracedecay-host-admission/src/lib.rs index d17fcd9ccd..b4cb09dc0a 100644 --- a/crates/tracedecay-host-admission/src/lib.rs +++ b/crates/tracedecay-host-admission/src/lib.rs @@ -1140,6 +1140,9 @@ fn projection_error_outcome(error: &ProjectionStoreError) -> HostAdmissionOutcom ProjectionStoreError::Storage { .. } if error.is_sqlite_disk_image_corrupt() => { HostAdmissionOutcome::terminal_unavailable("projection_store_corrupt") } + ProjectionStoreError::Storage { .. } if error.writer_fence().is_some() => { + HostAdmissionOutcome::terminal_unavailable("projection_store_writer_fenced") + } ProjectionStoreError::Storage { .. } => { HostAdmissionOutcome::retained_unavailable("projection_storage_retry_scheduled") } @@ -1244,6 +1247,29 @@ mod projection_error_outcome_tests { ); assert!(outcome.retryable); } + + /// A lease whose authority epoch is behind the store's can never commit + /// again; the drain must stop instead of rescheduling the same write. + #[test] + fn a_fenced_writer_classifies_as_blocked_not_retryable() { + let fence = tracedecay_store::StoreWriterFencedV1 { + requested: tracedecay_store::StoreAuthorityEpochV1::new(2).unwrap(), + current: tracedecay_store::StoreAuthorityEpochV1::new(3).unwrap(), + }; + let error = ProjectionStoreError::Storage { + operation: "dispatch observation projection runtime operation", + source: Box::new(ObservationStoreError::Storage { + operation: "submit anchored observation", + source: Box::new(fence), + }), + }; + + let outcome = projection_error_outcome(&error); + + assert_eq!(outcome.reason_code, Some("projection_store_writer_fenced")); + assert!(!outcome.retryable, "a fenced writer must not be retried"); + assert_eq!(outcome.status, HostAdmissionStatus::Unavailable); + } } fn host_scope(scope: &ObservationScopeV1) -> HostAdmissionScope { @@ -1607,6 +1633,23 @@ fn classify_error(error: &ObservationApplicationError) -> HostAdmissionOutcome { ObservationApplicationError::Store(ObservationStoreError::WriterSaturated { .. }) => { HostAdmissionOutcome::retained_backpressured("writer_lane_saturated") } + // The store fenced this lease: its authority epoch is behind the one + // the store holds, so re-driving the same batch through it can never + // commit. The owner must reopen the store, not retry. + ObservationApplicationError::Store(ObservationStoreError::Storage { + operation, + source, + }) if tracedecay_store::error_chain_writer_fence(source.as_ref()).is_some() => { + tracing::warn!( + operation, + error = %source, + reason_code = "authority_writer_fenced", + "observation store fenced this writer; its lease must be reopened" + ); + let mut outcome = HostAdmissionOutcome::terminal_unavailable("authority_writer_fenced"); + outcome.cause = Some(format!("{operation}: {source}")); + outcome + } ObservationApplicationError::Store(ObservationStoreError::Storage { operation, source, diff --git a/crates/tracedecay-mcp/src/hook_runtime/errors.rs b/crates/tracedecay-mcp/src/hook_runtime/errors.rs index 46ec42ef98..b6c71afc83 100644 --- a/crates/tracedecay-mcp/src/hook_runtime/errors.rs +++ b/crates/tracedecay-mcp/src/hook_runtime/errors.rs @@ -27,6 +27,13 @@ pub fn map_transcript_ingest_error( detail: Some(cause), .. } => format!("transcript ingest failed: authority_write_failed: {cause}"), + tracedecay_sessions::runtime::source::TranscriptIngestError::BackgroundResourceUnavailable { + provider, + resource, + } => format!( + "transcript ingest failed: {}: {provider}: {resource}", + disposition.reason_code + ), _ => format!("transcript ingest failed: {}", disposition.reason_code), }; hook_admission_error( diff --git a/crates/tracedecay-mcp/src/hook_runtime/errors/tests.rs b/crates/tracedecay-mcp/src/hook_runtime/errors/tests.rs index 3588e47dd8..db6cd6a255 100644 --- a/crates/tracedecay-mcp/src/hook_runtime/errors/tests.rs +++ b/crates/tracedecay-mcp/src/hook_runtime/errors/tests.rs @@ -251,6 +251,35 @@ fn authority_write_failed_includes_cause_in_hook_detail() { ); } +#[test] +fn transcript_resource_refusal_keeps_the_resource_and_backpressure_status() { + let mapped = map_transcript_ingest_error( + &tracedecay_sessions::runtime::source::TranscriptIngestError::BackgroundResourceUnavailable { + provider: "codex", + resource: "shared JSONL page resident-memory capacity", + }, + ); + let data = structured_hook_error_data(&mapped).unwrap(); + assert_eq!( + data["reason_code"], + "transcript_background_resource_unavailable" + ); + assert_eq!(data["status"], "backpressured"); + assert_eq!(data["retryable"], true); + let detail = data["detail"].as_str().expect("hook detail"); + assert!(detail.contains("codex"), "{detail}"); + assert!( + detail.contains("shared JSONL page resident-memory capacity"), + "{detail}" + ); + assert!( + mapped + .to_string() + .contains("shared JSONL page resident-memory capacity"), + "{mapped}" + ); +} + #[test] fn transcript_hook_errors_keep_bounded_retry_data_without_source_detail() { let error = diff --git a/crates/tracedecay-mcp/src/server/live_transcript_refresh.rs b/crates/tracedecay-mcp/src/server/live_transcript_refresh.rs index 8b7388b13d..6797a0b32c 100644 --- a/crates/tracedecay-mcp/src/server/live_transcript_refresh.rs +++ b/crates/tracedecay-mcp/src/server/live_transcript_refresh.rs @@ -1,7 +1,8 @@ use std::time::Duration; -use serde_json::Value; +use serde_json::{Map, Value}; +use tracedecay_domain::SessionId; use tracedecay_domain::errors::{Result, TraceDecayError}; use tracedecay_sessions::serving::SessionRefreshWorkerPort; @@ -54,8 +55,10 @@ pub async fn join_required_live_transcript_refresh( let Some(scope) = required_refresh_scope(tool_name, arguments) else { return Ok(LiveTranscriptRefreshJoin::NotRequired); }; - join_scope(scope, project_wake, user_wake, || TraceDecayError::Config { - message: REFRESH_UNAVAILABLE.to_owned(), + join_scope(scope, None, project_wake, user_wake, || { + TraceDecayError::Config { + message: REFRESH_UNAVAILABLE.to_owned(), + } }) .await } @@ -65,6 +68,7 @@ pub async fn join_required_live_transcript_refresh( /// otherwise. pub async fn join_hook_ingest_refresh( user_scope: bool, + session_id: Option<&SessionId>, project_wake: Option<&dyn SessionRefreshWorkerPort>, user_wake: Option<&dyn SessionRefreshWorkerPort>, ) -> Result<()> { @@ -73,7 +77,7 @@ pub async fn join_hook_ingest_refresh( } else { LiveTranscriptRefreshScope::Project }; - join_scope(scope, project_wake, user_wake, || { + join_scope(scope, session_id, project_wake, user_wake, || { TraceDecayError::hook_runtime_with_status( "temporal_refresh_unavailable", true, @@ -85,8 +89,33 @@ pub async fn join_hook_ingest_refresh( .map(|_| ()) } +/// Hermes inline capture preserves this exact public session identity. Source +/// sweeps may write several sessions, and other providers can normalize their +/// native IDs, so those calls keep the aggregate publication contract. +/// Capture still owns request validation and admission before the join runs. +pub fn hook_ingest_refresh_session(arguments: &Map) -> Result> { + if arguments.get("action").and_then(Value::as_str) != Some("ingest_transcript") + || arguments.get("provider").and_then(Value::as_str) != Some("hermes") + || !arguments.get("messages").is_some_and(Value::is_array) + { + return Ok(None); + } + let session_id = arguments + .get("session_id") + .and_then(Value::as_str) + .ok_or_else(|| TraceDecayError::Config { + message: "Hermes turn callback requires a session_id".to_owned(), + })?; + SessionId::new(session_id) + .map(Some) + .map_err(|error| TraceDecayError::Config { + message: format!("invalid Hermes turn session: {error}"), + }) +} + async fn join_scope( scope: LiveTranscriptRefreshScope, + session_id: Option<&SessionId>, project_wake: Option<&dyn SessionRefreshWorkerPort>, user_wake: Option<&dyn SessionRefreshWorkerPort>, unavailable: impl Fn() -> TraceDecayError, @@ -96,10 +125,17 @@ async fn join_scope( LiveTranscriptRefreshScope::User => user_wake, } .ok_or_else(&unavailable)?; - if wake - .wake_and_wait_until_idle(LIVE_TRANSCRIPT_REFRESH_DEADLINE) - .await - { + let published = match session_id { + Some(session_id) => { + wake.wake_and_wait_for_session_projection(session_id, LIVE_TRANSCRIPT_REFRESH_DEADLINE) + .await + } + None => { + wake.wake_and_wait_for_projection(LIVE_TRANSCRIPT_REFRESH_DEADLINE) + .await + } + }; + if published { Ok(LiveTranscriptRefreshJoin::PublicationJoined) } else { Err(unavailable()) @@ -108,11 +144,13 @@ async fn join_scope( #[cfg(test)] mod tests { + use std::sync::Mutex; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use serde_json::json; use tracedecay_contracts::{SessionTemporalRefreshWakeFuture, SessionTemporalRefreshWakePort}; + use tracedecay_domain::SessionId; use tracedecay_sessions::serving::{ SessionConvergenceStatus, SessionProjectionServingState, SessionProjectionServingStatus, SessionProjectionServingStatusPort, @@ -125,12 +163,14 @@ mod tests { /// Refresh owner whose publication is the state after `wake_and_wait`. struct PublishingRefresh { published: AtomicBool, + session: Mutex>, } impl PublishingRefresh { fn idle() -> Self { Self { published: AtomicBool::new(false), + session: Mutex::new(None), } } @@ -149,13 +189,23 @@ mod tests { false } - fn wake_and_wait_until_idle( + fn wake_and_wait_for_projection( &self, _timeout: Duration, ) -> SessionTemporalRefreshWakeFuture<'_> { let published = self.wake(); Box::pin(async move { published }) } + + fn wake_and_wait_for_session_projection<'a>( + &'a self, + session_id: &'a SessionId, + _timeout: Duration, + ) -> SessionTemporalRefreshWakeFuture<'a> { + *self.session.lock().expect("record session target") = Some(session_id.clone()); + let published = self.wake(); + Box::pin(async move { published }) + } } impl SessionProjectionServingStatusPort for PublishingRefresh { @@ -176,6 +226,7 @@ mod tests { async fn completed_hook_ingest_fails_when_its_refresh_owner_is_unavailable() { let error = super::join_hook_ingest_refresh( false, + None, Some(&UnavailableSessionTemporalRefreshWake), None, ) @@ -198,7 +249,7 @@ mod tests { #[tokio::test] async fn user_scope_never_falls_back_to_the_project_refresh_owner() { let project = PublishingRefresh::idle(); - let error = super::join_hook_ingest_refresh(true, Some(&project), None) + let error = super::join_hook_ingest_refresh(true, None, Some(&project), None) .await .expect_err("user ingest must require the user refresh owner"); @@ -215,7 +266,7 @@ mod tests { #[tokio::test] async fn project_ingest_does_not_publish_through_the_user_refresh_owner() { let user = PublishingRefresh::idle(); - let error = super::join_hook_ingest_refresh(false, None, Some(&user)) + let error = super::join_hook_ingest_refresh(false, None, None, Some(&user)) .await .expect_err("project ingest must require the project refresh owner"); @@ -233,7 +284,7 @@ mod tests { async fn hook_ingest_joins_the_project_refresh_owner() { let project = PublishingRefresh::idle(); let user = PublishingRefresh::idle(); - super::join_hook_ingest_refresh(false, Some(&project), Some(&user)) + super::join_hook_ingest_refresh(false, None, Some(&project), Some(&user)) .await .expect("project hook ingest must join its refresh owner"); @@ -247,6 +298,40 @@ mod tests { ); } + #[tokio::test] + async fn hermes_inline_join_names_its_session_but_source_scans_remain_aggregate() { + for (arguments, expected) in [ + ( + json!({"action": "ingest_transcript", "provider": "hermes", "session_id": "hermes-live", "messages": [{"role": "user", "content": "live turn"}]}), + Some(SessionId::new("hermes-live").expect("session id")), + ), + ( + json!({"action": "ingest_transcript", "provider": "hermes", "messages": null}), + None, + ), + ( + json!({"action": "ingest_transcript", "provider": "hermes"}), + None, + ), + ( + json!({"action": "ingest_transcript", "provider": "codex", "session_id": "native-provider-id", "messages": []}), + None, + ), + ] { + let target = super::hook_ingest_refresh_session(arguments.as_object().unwrap()) + .expect("derive the canonical target without changing source-scan validation"); + assert_eq!(target, expected); + let project = PublishingRefresh::idle(); + let user = PublishingRefresh::idle(); + super::join_hook_ingest_refresh(true, target.as_ref(), Some(&project), Some(&user)) + .await + .expect("join the owning user store"); + assert!(user.published()); + assert_eq!(*user.session.lock().unwrap(), expected); + assert!(!project.published()); + } + } + #[tokio::test] async fn projected_lcm_preflight_joins_its_user_refresh_owner() { let project = PublishingRefresh::idle(); diff --git a/crates/tracedecay-mcp/src/server/mod.rs b/crates/tracedecay-mcp/src/server/mod.rs index e46070ad4c..615d88ab00 100644 --- a/crates/tracedecay-mcp/src/server/mod.rs +++ b/crates/tracedecay-mcp/src/server/mod.rs @@ -21,7 +21,8 @@ pub use dispatch::{ needs_lazy_sync_before_dispatch, }; pub use live_transcript_refresh::{ - LiveTranscriptRefreshJoin, join_hook_ingest_refresh, join_required_live_transcript_refresh, + LiveTranscriptRefreshJoin, hook_ingest_refresh_session, join_hook_ingest_refresh, + join_required_live_transcript_refresh, }; pub use project_host_admission_replay::{ ProjectHostAdmissionReplayTask, ProjectHostAdmissionReplayWorker, diff --git a/crates/tracedecay-runtime-core/src/shard_runtime/registry.rs b/crates/tracedecay-runtime-core/src/shard_runtime/registry.rs index 495177b8c3..da1e161500 100644 --- a/crates/tracedecay-runtime-core/src/shard_runtime/registry.rs +++ b/crates/tracedecay-runtime-core/src/shard_runtime/registry.rs @@ -1189,6 +1189,15 @@ pub enum StoreRuntimeRegistryFailure { operation_leases: u32, }, AuthorityEpochExhausted, + /// The store's ledger already holds a newer writer authority for this + /// binding's shard and incarnation, so a writer published at the binding + /// could never commit. `open` retains `persisted` as the process epoch + /// floor and retries once unless a live graph publication still owns the + /// stale binding; its owners must release that binding before reopening. + WriterAuthorityBehindPersisted { + binding: Box, + persisted: StoreAuthorityEpochV1, + }, OpenAttemptExhausted, EvictionAttemptExhausted, EvictionReservationLost { diff --git a/crates/tracedecay-runtime-core/src/shard_runtime/registry/open.rs b/crates/tracedecay-runtime-core/src/shard_runtime/registry/open.rs index a725ed23f4..28b718ebf2 100644 --- a/crates/tracedecay-runtime-core/src/shard_runtime/registry/open.rs +++ b/crates/tracedecay-runtime-core/src/shard_runtime/registry/open.rs @@ -415,6 +415,7 @@ impl StoreRuntimeRegistry { }, ); }; + let mut converged_on_persisted_authority = false; loop { if let Some(path) = request .database_authority @@ -443,7 +444,50 @@ impl StoreRuntimeRegistry { StoreRuntimeOpenBegin::Rejected( StoreRuntimeRegistryFailure::DestructiveMaintenanceInProgress { .. }, ) => {} - begin => return begin.wait().await, + begin => match begin.wait().await { + // A store the same incarnation already wrote under a newer + // authority fences every writer below it. Allocating above + // the persisted epoch converges this process on the store's + // authority; a second fence means the store moved again + // while this open ran, which the caller must see. + StoreRuntimeOpenResult::Failed( + StoreRuntimeRegistryFailure::WriterAuthorityBehindPersisted { + binding, + persisted, + }, + ) if !converged_on_persisted_authority => { + retain_authority_epoch_floor(persisted); + let retained_graph_is_stale = self + .lock_state() + .graph_publications + .get(request.key()) + .is_some_and(|graph| graph.binding.authority_epoch < persisted); + if retained_graph_is_stale { + // Graph owner and operation leases retain this + // exact immutable binding. A retry would reuse the + // same stale epoch, and rebinding here would bypass + // their retirement authority. Fail once; after the + // owners release, the retained floor admits a new + // shared graph/relational binding above the ledger. + return StoreRuntimeOpenResult::Failed( + StoreRuntimeRegistryFailure::WriterAuthorityBehindPersisted { + binding, + persisted, + }, + ); + } + converged_on_persisted_authority = true; + tracing::warn!( + event = "store_writer_authority_floor_retained", + shard = ?binding.shard_id, + incarnation = binding.incarnation.get(), + requested_epoch = binding.authority_epoch.get(), + persisted_epoch = persisted.get(), + "store ledger holds a newer writer authority; retrying above its retained epoch floor" + ); + } + result => return result, + }, } } } diff --git a/crates/tracedecay-runtime-core/src/shard_runtime/registry/ports.rs b/crates/tracedecay-runtime-core/src/shard_runtime/registry/ports.rs index 9253d0c629..033c2b8e59 100644 --- a/crates/tracedecay-runtime-core/src/shard_runtime/registry/ports.rs +++ b/crates/tracedecay-runtime-core/src/shard_runtime/registry/ports.rs @@ -174,6 +174,33 @@ async fn publish_lifecycle_runtime( attachment.abort(request.locator.is_prospective()); return Err(error); } + if request.mode == StoreRuntimeOpenMode::Existing + && request.access == StoreRuntimeAccessMode::ReadWrite + && let Err(error) = + verify_writer_authority_before_existing_publication(&request, attachment.as_physical()) + .await + { + // This failed attachment must finish closing before the registry can + // retry at a new epoch. Joining its physical workers is blocking work. + let created = request.locator.is_prospective(); + crate::native_runtime::spawn_blocking(move || { + if created { + attachment.0.abort_initialization() + } else { + attachment.0.drain()?; + attachment.0.close_and_join() + } + }) + .await + .and_then(|closed| closed) + .map_err( + |message| StoreRuntimeRegistryFailure::PhysicalRuntimeFailed { + operation: "close runtime after writer authority refusal", + message, + }, + )?; + return Err(error); + } if let Err(error) = runtime.transition(RuntimeMaintenanceStateV1::Ready) { attachment.abort(request.locator.is_prospective()); return Err(runtime_lifecycle_failure(error)); @@ -190,6 +217,44 @@ async fn publish_lifecycle_runtime( Ok(PublishedShardRuntime::new(runtime, attachment.into_box())) } +/// A shard whose ledger already holds a newer writer authority for this +/// incarnation must not be published at the stale binding: every submit +/// through it would be fenced. The registry retains the persisted epoch floor +/// and retries only if no live graph publication still owns the stale binding. +async fn verify_writer_authority_before_existing_publication( + request: &ShardRuntimeBuildRequest, + attachment: &dyn PhysicalRuntimeAttachment, +) -> Result<(), StoreRuntimeRegistryFailure> { + const OPERATION: &str = "read persisted writer authority for existing-runtime admission"; + let handle = attachment.exact_sql_handle().map_err(|message| { + StoreRuntimeRegistryFailure::PhysicalRuntimeFailed { + operation: OPERATION, + message, + } + })?; + let persisted = tracedecay_rusqlite_runtime::runtime_ledger::persisted_writer_authority_epoch( + &handle.read_only_clone(), + &request.binding, + // Publication has not exposed this attachment's fresh reader pool to + // any clients, so this admission cannot contend with user snapshots. + Duration::ZERO, + ) + .await + .map_err(|error| StoreRuntimeRegistryFailure::PhysicalRuntimeFailed { + operation: OPERATION, + message: error.to_string(), + })?; + match persisted { + Some(persisted) if persisted > request.binding.authority_epoch => Err( + StoreRuntimeRegistryFailure::WriterAuthorityBehindPersisted { + binding: Box::new(request.binding.clone()), + persisted, + }, + ), + _ => Ok(()), + } +} + fn runtime_core_final_schema_applies(scope: &StoreShardScopeV1) -> bool { matches!( scope, diff --git a/crates/tracedecay-runtime-core/src/shard_runtime/registry/tests/production_routes.rs b/crates/tracedecay-runtime-core/src/shard_runtime/registry/tests/production_routes.rs index eecb8233ef..93c943b690 100644 --- a/crates/tracedecay-runtime-core/src/shard_runtime/registry/tests/production_routes.rs +++ b/crates/tracedecay-runtime-core/src/shard_runtime/registry/tests/production_routes.rs @@ -5,6 +5,7 @@ use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; use rusqlite::Connection; use tempfile::TempDir; @@ -12,11 +13,16 @@ use tracedecay_domain::{ BrainId, BrainNodeId, LocatorDigest, ProjectId, RepositoryId, UserProfileId, UtcMicros, WorktreeId, }; +use tracedecay_graph_db::{ + GraphDbOwnerRegistrationV1, GraphDbRegistration, GraphDbRegistry, GraphDbRegistryConfig, + NeverCancelled, +}; use tracedecay_store::{ - CodeShardScopeV1, ConsistencyModeV1, OperationPriorityV1, RuntimeCancellationIdV1, - RuntimeCancellationIdentityV1, RuntimeDeadlineIdV1, RuntimeDeadlineV1, RuntimeReadOperationV1, - RuntimeReadRequestV1, RuntimeReadResultV1, RuntimeRequestControlV1, RuntimeRequestProbeV1, - StoreShardIdV1, StoreShardScopeV1, VerifiedStoreLocatorV1, + CodeShardScopeV1, ConsistencyModeV1, OperationPriorityV1, RetainedGraphStoreLeaseV1, + RetainedGraphStoreOwnerAttachmentV1, RuntimeCancellationIdV1, RuntimeCancellationIdentityV1, + RuntimeDeadlineIdV1, RuntimeDeadlineV1, RuntimeReadOperationV1, RuntimeReadRequestV1, + RuntimeReadResultV1, RuntimeRequestControlV1, RuntimeRequestProbeV1, StoreShardIdV1, + StoreShardScopeV1, VerifiedStoreLocatorV1, canonical_store_locator_digest, }; use super::super::*; @@ -134,6 +140,26 @@ impl StoreRuntimeResolver for FileResolver { ); Box::pin(async move { Ok(ResolvedStoreLocator::new(locator, path)) }) } + + fn resolve_graph<'a>( + &'a self, + key: &'a StoreRuntimeKey, + ) -> StoreRuntimeRegistryFuture<'a, Result> + { + Box::pin(async move { + let resolved = self + .resolve(key, StoreRuntimeOpenMode::Existing, None) + .await?; + Ok(ResolvedStoreLocator::new( + VerifiedStoreLocatorV1::new( + key.shard_id.clone(), + key.incarnation, + canonical_store_locator_digest(resolved.path()).unwrap(), + ), + resolved.path().to_path_buf(), + )) + }) + } } fn health_request( @@ -649,3 +675,329 @@ async fn existing_store_at_an_older_schema_is_refused_even_with_write_authority( "a refused open must leave the stamp alone" ); } + +struct SubmitProbe { + cancellation: RuntimeCancellationIdentityV1, + deadline: RuntimeDeadlineV1, +} + +impl RuntimeRequestProbeV1 for SubmitProbe { + fn cancellation_identity(&self) -> &RuntimeCancellationIdentityV1 { + &self.cancellation + } + + fn deadline_identity(&self) -> &RuntimeDeadlineV1 { + &self.deadline + } + + fn interruption(&self) -> Option { + None + } + + fn try_begin_commit(&self) -> bool { + true + } +} + +fn diagnostics_submit( + binding: &tracedecay_store::StoreRuntimeBindingV1, + name: &str, +) -> (tracedecay_store::RuntimeSubmitRequestV1, SubmitProbe) { + let metadata: tracedecay_store::StoreOperationMetadataV1 = + serde_json::from_value(serde_json::json!({ + "operation_id": format!("operation.{name}"), + "client_id": "client.reopen", + "shard_id": binding.shard_id, + "incarnation": binding.incarnation.get(), + "authority_epoch": binding.authority_epoch.get(), + "idempotency": { + "key": format!("key.{name}"), + "command_digest": format!("sha256:{}", "e".repeat(64)), + }, + "durability": "full", + "priority": "foreground", + "admission_bytes": 128, + "admitted_at": 1 + })) + .unwrap(); + let scope = tracedecay_store::RuntimeTransactionScopeV1 { + transaction_id: tracedecay_store::RuntimeTransactionIdV1::new(format!( + "transaction.{name}" + )) + .unwrap(), + compatibility: tracedecay_store::RuntimeBatchCompatibilityV1::from_operation(&metadata) + .unwrap(), + opened_at: metadata.admitted_at, + }; + let cancellation = RuntimeCancellationIdentityV1 { + cancellation_id: RuntimeCancellationIdV1::new(format!("cancel.{name}")).unwrap(), + generation: 1, + }; + let deadline = RuntimeDeadlineV1 { + deadline_id: RuntimeDeadlineIdV1::new(format!("deadline.{name}")).unwrap(), + }; + let control = RuntimeRequestControlV1 { + requested_at: UtcMicros(1), + deadline: deadline.clone(), + cancellation: cancellation.clone(), + }; + let request = tracedecay_store::RuntimeSubmitRequestV1::new( + tracedecay_store::RepositoryOperationEnvelopeV1 { + metadata, + payload: tracedecay_store::RepositoryWritePayloadV1::Diagnostics(Box::new( + tracedecay_store::SanitizedCleanDiagnosticSnapshotV1::new( + tracedecay_domain::CodeGenerationId::new(format!("generation.{name}")).unwrap(), + Vec::new(), + ) + .unwrap(), + )), + }, + scope, + control, + ) + .unwrap(); + ( + request, + SubmitProbe { + cancellation, + deadline, + }, + ) +} + +async fn commit_diagnostics_with_writer( + path: &Path, + binding: &tracedecay_store::StoreRuntimeBindingV1, + name: &str, +) { + let writer = tracedecay_rusqlite_runtime::PersistentWriter::start( + &crate::native_runtime::handle().unwrap(), + tracedecay_rusqlite_runtime::ExistingWriterLocator::new( + binding.clone(), + VerifiedStoreLocatorV1::new( + binding.shard_id.clone(), + binding.incarnation, + LocatorDigest::new(format!("sha256:{}", "c".repeat(64))).unwrap(), + ), + path.to_path_buf(), + ) + .unwrap(), + tracedecay_store::AdmissionConfigV1::default(), + tracedecay_rusqlite_runtime::repository::ConcreteRepositoryWriteExecutor::default(), + ) + .unwrap(); + let (request, probe) = diagnostics_submit(binding, name); + let outcome = writer + .submit( + &crate::native_runtime::cx().unwrap(), + request, + Arc::new(probe), + ) + .await + .unwrap(); + assert!( + matches!( + outcome, + tracedecay_store::RuntimeSubmitOutcomeV1::Committed { .. } + ), + "the writer must durably establish its authority: {outcome:?}" + ); + crate::native_runtime::spawn_blocking(move || writer.shutdown_and_join()) + .await + .unwrap() + .unwrap(); +} + +/// A store this incarnation already wrote under a newer authority (a previous +/// daemon of the same incarnation, or a sibling runtime on the file) fences +/// every writer below that authority. The reopen converges above it instead of +/// publishing a runtime whose every submit would be fenced. +#[asupersync::test] +async fn reopening_a_store_behind_its_persisted_writer_authority_converges_above_it() { + let root = TempDir::new().unwrap(); + let resolver = Arc::new(FileResolver::default()); + resolver.push(seed_db(&root, "profile.db")); + let project_path = seed_final_graph_db(&root, "project.db").await; + Connection::open(&project_path) + .unwrap() + .execute_batch(tracedecay_store::GENERATION_DIAGNOSTICS_SCHEMA_DDL) + .unwrap(); + resolver.push(project_path.clone()); + resolver.push(project_path.clone()); + let shard = StoreShardIdV1::project( + id::("brain.registry"), + id::("profile.registry"), + id::("project.reopen"), + ); + let ahead = tracedecay_store::StoreRuntimeBindingV1::new( + shard.clone(), + incarnation(), + tracedecay_store::StoreAuthorityEpochV1::new(1 << 40).unwrap(), + ); + let runtime = asupersync::runtime::Runtime::current_handle().unwrap(); + commit_diagnostics_with_writer(&project_path, &ahead, "previous-authority").await; + + let registry = + StoreRuntimeRegistry::new(runtime, resolver, Arc::new(LifecycleShardRuntimePublisher)) + .unwrap(); + let _profile = open_published( + ®istry, + StoreRuntimeOpenRequest::new(profile_shard(), incarnation(), None), + ) + .await; + let pin = match registry.profile_authority_pin(&profile_shard()) { + ProfileAuthorityPinResult::Pinned(pin) => pin, + other => panic!("profile was not pinned: {other:?}"), + }; + let authority = crate::db::DatabaseAuthority::for_runtime(&project_path, "reopen project") + .expect("project database authority"); + let project_runtime = open_published( + ®istry, + StoreRuntimeOpenRequest::new_authorized(shard, incarnation(), Some(pin), authority.clone()), + ) + .await; + + assert!( + project_runtime.binding().authority_epoch > ahead.authority_epoch, + "the reopened runtime publishes above the persisted authority: {:?} vs {:?}", + project_runtime.binding().authority_epoch, + ahead.authority_epoch + ); + let (request, probe) = diagnostics_submit(project_runtime.binding(), "converged-authority"); + let outcome = project_runtime + .dispatch_submit_authorized(request, Arc::new(probe), authority) + .await + .expect("converged runtime accepts writes"); + assert!( + matches!( + outcome, + tracedecay_store::RuntimeSubmitOutcomeV1::Committed { .. } + ), + "a write through the converged runtime commits: {outcome:?}" + ); + assert!( + tracedecay_rusqlite_runtime::exact_sql::native_thread_storage_wait_sites() + .iter() + .all(|site| !site.call_site.contains("/ledger/checkpoint.rs:")), + "persisted writer authority reads must not park native runtime workers" + ); +} + +/// A graph-first owner cannot be rebound while any of its graph leases remain +/// live. Reopening refuses once with the exact writer fence, then succeeds +/// above the ledger after the ordinary graph close releases that authority. +#[asupersync::test] +async fn a_stale_graph_writer_authority_refuses_once_then_reopens_after_graph_close() { + let root = TempDir::new().unwrap(); + let resolver = Arc::new(FileResolver::default()); + resolver.push(seed_db(&root, "profile.db")); + let graph_path = root.path().join("project.grafeo"); + resolver.push(graph_path.clone()); + let project_path = seed_final_graph_db(&root, "project.db").await; + Connection::open(&project_path) + .unwrap() + .execute_batch(tracedecay_store::GENERATION_DIAGNOSTICS_SCHEMA_DDL) + .unwrap(); + resolver.push(project_path.clone()); + resolver.push(project_path.clone()); + resolver.push(graph_path); + let registry = StoreRuntimeRegistry::new( + asupersync::runtime::Runtime::current_handle().unwrap(), + resolver.clone(), + Arc::new(LifecycleShardRuntimePublisher), + ) + .unwrap(); + let _profile = open_published( + ®istry, + StoreRuntimeOpenRequest::new(profile_shard(), incarnation(), None), + ) + .await; + let pin = match registry.profile_authority_pin(&profile_shard()) { + ProfileAuthorityPinResult::Pinned(pin) => pin, + other => panic!("profile was not pinned: {other:?}"), + }; + let shard = project_shard("project.graph-first-reopen"); + let key = StoreRuntimeKey::new(shard.clone(), incarnation()); + let (store_attachment, store_target) = registry + .attach_graph_store_owner(key.clone()) + .await + .unwrap(); + let stale = store_attachment.binding().clone(); + let graph_locator = store_attachment.verified_locator().clone(); + let operation = store_attachment.issue_operation_lease().unwrap(); + let graphs = GraphDbRegistry::new(GraphDbRegistryConfig { max_open: 1 }).unwrap(); + let graph_owner = graphs + .resolve_owner_attachment(GraphDbOwnerRegistrationV1 { + operation: GraphDbRegistration { + authority_lease: operation, + cancellation: Arc::new(NeverCancelled), + lifecycle_cancellation: Arc::new(NeverCancelled), + deadline: Instant::now() + Duration::from_secs(30), + }, + authority_attachment: Box::new(store_attachment), + }) + .unwrap(); + let client = graph_owner.issue_lease().unwrap(); + + let mut ahead = stale.clone(); + ahead.authority_epoch = tracedecay_store::StoreAuthorityEpochV1::new( + stale.authority_epoch.get().checked_add(1).unwrap(), + ) + .unwrap(); + commit_diagnostics_with_writer(&project_path, &ahead, "graph-newer-writer").await; + let authority = + crate::db::DatabaseAuthority::for_runtime(&project_path, "reopen project").unwrap(); + let request = + StoreRuntimeOpenRequest::new_authorized(shard, incarnation(), Some(pin), authority.clone()); + let calls = resolver.calls.load(Ordering::SeqCst); + assert!(matches!( + registry.open(request.clone()).await, + StoreRuntimeOpenResult::Failed( + StoreRuntimeRegistryFailure::WriterAuthorityBehindPersisted { binding, persisted } + ) if *binding == stale && persisted == ahead.authority_epoch + )); + assert_eq!( + resolver.calls.load(Ordering::SeqCst), + calls + 1, + "a live stale graph binding cannot converge through another identical open" + ); + assert_eq!(graph_owner.binding(), &stale); + assert!(graph_owner.shares_runtime_with(&client)); + drop( + client + .snapshot() + .expect("the old graph client stays readable"), + ); + assert!( + graphs.close_retained(&stale, &graph_locator).is_err(), + "the refused writer must preserve the graph owner's live close blockers" + ); + drop( + graph_owner + .issue_lease() + .expect("the old exact graph owner stays usable"), + ); + + drop(client); + drop(graph_owner); + assert!(graphs.close_retained(&stale, &graph_locator).unwrap()); + drop(store_target); + assert_eq!(registry.retained_graph_publications_for_test(), 0); + + let project = open_published(®istry, request).await; + assert!(project.binding().authority_epoch > ahead.authority_epoch); + let graph = registry.retain_graph_store(key).await.unwrap(); + assert_eq!(graph.binding(), project.binding()); + let (request, probe) = diagnostics_submit(project.binding(), "graph-reopened-writer"); + let outcome = project + .dispatch_submit_authorized(request, Arc::new(probe), authority) + .await + .unwrap(); + assert!( + matches!( + outcome, + tracedecay_store::RuntimeSubmitOutcomeV1::Committed { .. } + ), + "the reopened shared authority accepts writes: {outcome:?}" + ); +} diff --git a/crates/tracedecay-rusqlite-runtime/src/ledger.rs b/crates/tracedecay-rusqlite-runtime/src/ledger.rs index bc9e79291e..8b03992db6 100644 --- a/crates/tracedecay-rusqlite-runtime/src/ledger.rs +++ b/crates/tracedecay-rusqlite-runtime/src/ledger.rs @@ -12,6 +12,7 @@ mod schema; mod sqlite; pub(crate) use checkpoint::current_watermark; +pub use checkpoint::{PersistedWriterAuthorityError, persisted_writer_authority_epoch}; #[cfg(test)] pub(crate) use commit::record_commit; pub(crate) use commit::record_runtime_commit; diff --git a/crates/tracedecay-rusqlite-runtime/src/ledger/checkpoint.rs b/crates/tracedecay-rusqlite-runtime/src/ledger/checkpoint.rs index d1539565df..e4b876d505 100644 --- a/crates/tracedecay-rusqlite-runtime/src/ledger/checkpoint.rs +++ b/crates/tracedecay-rusqlite-runtime/src/ledger/checkpoint.rs @@ -1,3 +1,6 @@ +use std::fmt; +use std::time::Duration; + use rusqlite::{Row, params}; use tracedecay_store::{ CommitSequenceV1, DurabilityClassV1, RuntimeTransactionScopeV1, ShardWatermarkV1, @@ -8,8 +11,110 @@ use super::{ LedgerError, sqlite::{BindingKey, LedgerTransaction, Submission, decode_json, encode_json, sqlite_u64}, }; +use crate::exact_sql::{ExactSqlError, ExactSqlHandle, ExactSqlStatement, ExactSqlValue}; const CHECKPOINT_TABLE: &str = "td_runtime_writer_checkpoint_v1"; +const SELECT_PERSISTED_AUTHORITY_EPOCH: &str = r#" +SELECT authority_epoch +FROM td_runtime_writer_checkpoint_v1 +WHERE shard_json = ?1 AND incarnation = ?2 +"#; + +/// Why the persisted writer authority of a shard incarnation could not be read. +#[derive(Debug)] +pub enum PersistedWriterAuthorityError { + Sql(ExactSqlError), + Corrupt { field: &'static str }, +} + +impl fmt::Display for PersistedWriterAuthorityError { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Sql(error) => { + write!(formatter, "persisted writer authority read failed: {error}") + } + Self::Corrupt { field } => write!( + formatter, + "persisted writer authority row is corrupt in {CHECKPOINT_TABLE}.{field}" + ), + } + } +} + +impl std::error::Error for PersistedWriterAuthorityError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Sql(error) => Some(error), + Self::Corrupt { .. } => None, + } + } +} + +impl From for PersistedWriterAuthorityError { + fn from(error: ExactSqlError) -> Self { + Self::Sql(error) + } +} + +/// The authority epoch the ledger last checkpointed for `binding`'s shard and +/// incarnation, or `None` for a store no writer of that incarnation has +/// committed to. Read through the attached runtime's own readers, before the +/// runtime is published, so a reopen can learn whether its allocated epoch +/// is already behind the store. Reader admission and execution are awaited +/// off the native runtime worker through the existing exact-SQL async port. +pub async fn persisted_writer_authority_epoch( + handle: &ExactSqlHandle, + binding: &StoreRuntimeBindingV1, + max_wait: Duration, +) -> Result, PersistedWriterAuthorityError> { + if handle.binding() != binding { + return Err(ExactSqlError::AuthorityMismatch.into()); + } + let key = + BindingKey::from_binding(binding).map_err(|_| PersistedWriterAuthorityError::Corrupt { + field: "shard_json", + })?; + let schema = handle + .query_async( + ExactSqlStatement::new( + "SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE type = 'table' AND name = ?1)" + .to_owned(), + vec![ExactSqlValue::Text(CHECKPOINT_TABLE.to_owned())], + )?, + max_wait, + ) + .await?; + match schema.rows.first().and_then(|row| row.values.first()) { + Some(ExactSqlValue::Integer(0)) => return Ok(None), + Some(ExactSqlValue::Integer(1)) => {} + _ => return Err(PersistedWriterAuthorityError::Corrupt { field: "schema" }), + } + let rows = handle + .query_async( + ExactSqlStatement::new( + SELECT_PERSISTED_AUTHORITY_EPOCH.to_owned(), + vec![ + ExactSqlValue::Text(key.shard_json), + ExactSqlValue::Integer(key.incarnation_sql), + ], + )?, + max_wait, + ) + .await?; + let Some(row) = rows.rows.first() else { + return Ok(None); + }; + match row.values.first() { + Some(ExactSqlValue::Integer(raw)) => decode_authority_epoch(*raw, "authority_epoch") + .map(Some) + .map_err(|_| PersistedWriterAuthorityError::Corrupt { + field: "authority_epoch", + }), + _ => Err(PersistedWriterAuthorityError::Corrupt { + field: "authority_epoch", + }), + } +} const SELECT_CHECKPOINT: &str = r#" SELECT authority_epoch, commit_sequence, watermark_json, transaction_scope_json, original_receipt_json, operation_id, durability_json, committed_at_micros diff --git a/crates/tracedecay-rusqlite-runtime/src/lib.rs b/crates/tracedecay-rusqlite-runtime/src/lib.rs index b22ccf9ec3..4749a19e4f 100644 --- a/crates/tracedecay-rusqlite-runtime/src/lib.rs +++ b/crates/tracedecay-rusqlite-runtime/src/lib.rs @@ -18,7 +18,10 @@ mod ledger; /// Canonical schema for the runtime writer ledger installed in registered /// SQLite stores. pub mod runtime_ledger { - pub use crate::ledger::{RETIRE_RUNTIME_EFFECTS_V1, RUNTIME_LEDGER_SCHEMA}; + pub use crate::ledger::{ + PersistedWriterAuthorityError, RETIRE_RUNTIME_EFFECTS_V1, RUNTIME_LEDGER_SCHEMA, + persisted_writer_authority_epoch, + }; } pub mod native_runtime; #[cfg(test)] diff --git a/crates/tracedecay-rusqlite-runtime/src/persistence.rs b/crates/tracedecay-rusqlite-runtime/src/persistence.rs index 09761c19cf..b19e7c2fdb 100644 --- a/crates/tracedecay-rusqlite-runtime/src/persistence.rs +++ b/crates/tracedecay-rusqlite-runtime/src/persistence.rs @@ -4,7 +4,7 @@ use rusqlite::{Savepoint, Transaction}; use tracedecay_store::{ CorruptionClassV1, IdempotencyIdentityV1, RuntimeSubmitRequestV1, StorageRuntimeErrorV1, - StoreCommitReceiptV1, StoreRuntimeBindingV1, + StoreAuthorityEpochV1, StoreCommitReceiptV1, StoreRuntimeBindingV1, }; use crate::{ @@ -64,33 +64,65 @@ impl RuntimeWriterPersistence { savepoint: &mut Savepoint<'_>, binding: &StoreRuntimeBindingV1, request: &RuntimeSubmitRequestV1, - ) -> Result { + ) -> Result { if binding != request.binding() { - return Err(infrastructure("apply-and-record binding mismatch")); + return Err(infrastructure("apply-and-record binding mismatch").into()); } operation::execute(savepoint, request, &mut self.executor).map_err(map_operation_error)?; - match ledger::record_runtime_commit( + let disposition = match ledger::record_runtime_commit( savepoint, &request.envelope().metadata, request.transaction_scope(), &request.envelope().payload, - ) - .map_err(map_ledger_error)? - { + ) { + Ok(disposition) => disposition, + Err(LedgerError::StaleAuthority { + persisted, + requested, + }) => { + return Err(ApplyRecordError::Fenced { + requested, + persisted, + }); + } + Err(error) => return Err(map_ledger_error(error).into()), + }; + match disposition { LedgerDisposition::Committed(receipt) => Ok(receipt), LedgerDisposition::Replay(receipt) | LedgerDisposition::Conflict(receipt) => { drop(receipt); - Err(infrastructure( - "idempotency disposition changed after the writer lookup", - )) + Err( + infrastructure("idempotency disposition changed after the writer lookup") + .into(), + ) + } + LedgerDisposition::New => { + Err(infrastructure("runtime ledger did not record the applied operation").into()) } - LedgerDisposition::New => Err(infrastructure( - "runtime ledger did not record the applied operation", - )), } } } +/// Why one request could not be applied and recorded in its savepoint. +/// +/// A fence is a submit decision, not a storage failure: the ledger already +/// holds a newer writer authority for this shard and incarnation, so the same +/// request can never commit through this writer. +#[derive(Debug)] +pub(crate) enum ApplyRecordError { + Fenced { + requested: StoreAuthorityEpochV1, + persisted: StoreAuthorityEpochV1, + }, + Storage(StorageRuntimeErrorV1), +} + +impl From for ApplyRecordError { + fn from(error: StorageRuntimeErrorV1) -> Self { + Self::Storage(error) + } +} + fn map_ledger_error(error: LedgerError) -> StorageRuntimeErrorV1 { match error { LedgerError::Corrupt { .. } => StorageRuntimeErrorV1::Corrupt { @@ -292,4 +324,48 @@ mod tests { ); transaction.commit().unwrap(); } + + #[test] + fn a_writer_behind_the_persisted_authority_is_fenced_not_failed() { + let mut connection = Connection::open_in_memory().unwrap(); + let newer = request(metadata("operation.newer", "key.newer", 'a')); + let mut stale_metadata = metadata("operation.stale", "key.stale", 'b'); + stale_metadata.authority_epoch = StoreAuthorityEpochV1::new(2).unwrap(); + let stale = request(stale_metadata); + let mut persistence = RuntimeWriterPersistence::new(MarkerExecutor); + let mut transaction = connection.transaction().unwrap(); + assert!( + persistence + .lookup_idempotency( + &transaction, + newer.binding(), + &newer.envelope().metadata.idempotency, + ) + .unwrap() + .is_none() + ); + { + let mut savepoint = transaction.savepoint().unwrap(); + persistence + .apply_and_record(&mut savepoint, newer.binding(), &newer) + .unwrap(); + savepoint.commit().unwrap(); + } + + let mut savepoint = transaction.savepoint().unwrap(); + let fenced = persistence + .apply_and_record(&mut savepoint, stale.binding(), &stale) + .unwrap_err(); + + assert!( + matches!( + fenced, + ApplyRecordError::Fenced { + requested, + persisted, + } if requested.get() == 2 && persisted.get() == 7 + ), + "a stale writer is fenced, not failed: {fenced:?}" + ); + } } diff --git a/crates/tracedecay-rusqlite-runtime/src/writer/transaction.rs b/crates/tracedecay-rusqlite-runtime/src/writer/transaction.rs index d3391acf42..3a62990511 100644 --- a/crates/tracedecay-rusqlite-runtime/src/writer/transaction.rs +++ b/crates/tracedecay-rusqlite-runtime/src/writer/transaction.rs @@ -16,7 +16,7 @@ use crate::{ RuntimeWriteAuthorityStage, StorageOperationExecutor, admission::QueueItem, connection, - persistence::RuntimeWriterPersistence, + persistence::{ApplyRecordError, RuntimeWriterPersistence}, telemetry::{ LockWorkScope, WriterBatchMetrics, WriterLockWorkSnapshot, WriterTelemetry, WriterTransactionMetrics, WriterTransactionOutcome, take_observed_vm, @@ -364,7 +364,27 @@ fn apply_new( // One operation-and-ledger boundary: the receipt is the one this savepoint recorded. let receipt = match persistence.apply_and_record(&mut savepoint, binding, &item.request) { Ok(receipt) => receipt, - Err(error) => { + Err(ApplyRecordError::Fenced { + requested, + persisted, + }) => { + return match savepoint.rollback() { + Ok(()) => processed( + item, + Ok(RuntimeSubmitOutcomeV1::Fenced { + expected: requested, + actual: persisted, + }), + false, + ), + Err(error) => { + let result = + driver_failure(error, "rollback fenced request").result(&item.request); + processed(item, result, false) + } + }; + } + Err(ApplyRecordError::Storage(error)) => { if let Some(outcome) = interruption_outcome( &item.request, item.probe.as_ref(), diff --git a/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/storage_contention.rs b/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/storage_contention.rs index 1567455f40..cb2ed1e38b 100644 --- a/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/storage_contention.rs +++ b/crates/tracedecay-rusqlite-runtime/tests/rusqlite_suite/storage_contention.rs @@ -26,6 +26,7 @@ use tracedecay_rusqlite_runtime::{ ConcreteRepositoryReadExecutor, ConcreteRepositoryWriteExecutor, RepositoryPhysicalAttachmentFactory, }, + runtime_ledger::{PersistedWriterAuthorityError, persisted_writer_authority_epoch}, }; use tracedecay_store::{ AdmissionConfigV1, CommitSequenceV1, ConsistencyModeV1, DiagnosticReadOperationV1, @@ -1166,3 +1167,117 @@ fn at_least_reads_under_a_superseded_authority_epoch_are_typed_unavailable() { attachment.drain().unwrap(); attachment.close_and_join().unwrap(); } + +/// Two live writers on one file under one incarnation. Once the newer authority +/// commits, a submit from the older writer is the typed fence every caller can +/// classify as terminal, not a storage failure they would retry forever. +#[test] +fn a_writer_behind_the_persisted_authority_epoch_is_fenced_on_submit() { + let database = TestDatabase::new("write-superseded-epoch.db"); + seed(&database); + let superseded = binding(0); + let mut current = superseded.clone(); + current.authority_epoch = + StoreAuthorityEpochV1::new(superseded.authority_epoch.get() + 1).unwrap(); + let stale_writer = start_writer( + &database, + &superseded, + ConcreteRepositoryWriteExecutor::default(), + ); + let current_writer = start_writer( + &database, + ¤t, + ConcreteRepositoryWriteExecutor::default(), + ); + commit( + ¤t_writer, + publication(¤t, "generation.current", 1), + ); + + let request = publication(&superseded, "generation.stale", 1); + let probe = Probe::for_submit(&request); + let outcome = run(stale_writer.submit(&native_runtime::cx().unwrap(), request, probe)).unwrap(); + + assert!( + matches!( + outcome, + RuntimeSubmitOutcomeV1::Fenced { expected, actual } + if expected == superseded.authority_epoch && actual == current.authority_epoch + ), + "stale writer submit is a typed fence: {outcome:?}" + ); + assert_eq!( + persisted_rows(&database.connect(), "generation_diagnostics"), + 1, + "the fenced write rolled back and only the current authority's rows remain" + ); + stale_writer.shutdown_and_join().unwrap(); + current_writer.shutdown_and_join().unwrap(); +} + +#[test] +fn persisted_writer_authority_is_scoped_to_the_shard_and_incarnation() { + let database = TestDatabase::new("persisted-authority.db"); + seed(&database); + let current = binding(0); + let writer = start_writer( + &database, + ¤t, + ConcreteRepositoryWriteExecutor::default(), + ); + commit(&writer, publication(¤t, "generation.persisted", 1)); + writer.shutdown_and_join().unwrap(); + let mut other_incarnation = current.clone(); + other_incarnation.incarnation = + tracedecay_store::StoreIncarnationV1::new(current.incarnation.get() + 1).unwrap(); + + for (binding, expected) in [ + (current.clone(), Some(current.authority_epoch)), + (binding(1), None), + (other_incarnation, None), + ] { + let attachment = RepositoryPhysicalAttachmentFactory + .attach_read_only( + binding.clone(), + verified_locator(&binding), + database.path.canonicalize().unwrap(), + AdmissionConfigV1::default(), + ) + .unwrap(); + let handle = attachment.exact_sql_handle().unwrap(); + assert_eq!( + run(persisted_writer_authority_epoch( + &handle, + &binding, + Duration::ZERO + )) + .unwrap(), + expected + ); + attachment.drain().unwrap(); + attachment.close_and_join().unwrap(); + } + + database.connect().execute_batch("PRAGMA ignore_check_constraints = ON; UPDATE td_runtime_writer_checkpoint_v1 SET authority_epoch = -1;").unwrap(); + let attachment = RepositoryPhysicalAttachmentFactory + .attach_read_only( + current.clone(), + verified_locator(¤t), + database.path.canonicalize().unwrap(), + AdmissionConfigV1::default(), + ) + .unwrap(); + let handle = attachment.exact_sql_handle().unwrap(); + assert!(matches!( + run(persisted_writer_authority_epoch( + &handle, + ¤t, + Duration::ZERO + )), + Err(PersistedWriterAuthorityError::Corrupt { + field: "authority_epoch" + }) + )); + attachment.drain().unwrap(); + attachment.close_and_join().unwrap(); +} diff --git a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry.rs b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry.rs index 7a58d40ece..68840e7cca 100644 --- a/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry.rs +++ b/crates/tracedecay-session-runtime/src/session_temporal_refresh_scheduler/registry.rs @@ -17,6 +17,7 @@ use super::wake::{ use super::wake::{SessionTemporalRefreshWorkerStatus, enabled_idle_notification}; use super::worker::run_session_temporal_refresh_scheduler; use crate::StoreOwnerKey; +use tracedecay_domain::SessionId; use tracedecay_global_db::RegisteredGlobalDbLeaseV1; use tracedecay_runtime_core::background_cpu::ProcessBackgroundCpuV1; use tracedecay_runtime_core::native_runtime::{self, NativeTask, NativeTaskOutcome}; @@ -35,12 +36,14 @@ macro_rules! define_pass_report { $visibility projected_batches: usize, $visibility completed: usize, $visibility failed: usize, + $visibility failed_sessions: HashSet, $visibility cancelled: usize, $visibility deferred: usize, $visibility retryable_errors: usize, $visibility terminal_errors: usize, $visibility deadline_errors: usize, $visibility saturated: bool, + $visibility discovery_advanced: bool, $visibility backlog: Option, $visibility retry_class: Option, $visibility last_error: Option, 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 7e5bf6190a..c11b7f5689 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 @@ -2,7 +2,9 @@ use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::Arc; use std::sync::PoisonError; use std::sync::atomic::{AtomicBool, Ordering}; +#[cfg(test)] use std::time::Instant; +use tracedecay_domain::SessionId; use tracedecay_runtime_core::native_runtime; use tracedecay_store::SessionRefreshBeginOrJoinRequestV1; use tracedecay_temporal_query::execution::ExecutionControl; @@ -19,6 +21,15 @@ use tracedecay_sessions::serving::{ SessionProjectionWorkerRetryClass, }; +/// One caller's publication obligation. Named sessions capture their source +/// frontier before the worker starts the pass; unnamed sweeps retain the +/// aggregate drain contract. +pub(super) struct SessionProjectionJoin { + pub(super) session_id: Option, + pub(super) observed_through: Option, + pub(super) completion: tokio::sync::oneshot::Sender, +} + #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub enum SessionTemporalRefreshRetryClass { Storage, @@ -141,6 +152,7 @@ macro_rules! define_wake_state { /// is still unpublished. The next history window waits. pub(super) history_after_projection: AtomicBool, pub(super) pass_count: std::sync::atomic::AtomicUsize, + projection_joins: std::sync::Mutex>, pub(super) wake: tokio::sync::Notify, pub(super) idle: tokio::sync::Notify, pub(super) cancelled: AtomicBool, @@ -173,6 +185,7 @@ impl Default for SessionTemporalRefreshWakeState { history_retry_pending: AtomicBool::new(false), history_after_projection: AtomicBool::new(false), pass_count: std::sync::atomic::AtomicUsize::new(0), + projection_joins: std::sync::Mutex::new(Vec::new()), wake: tokio::sync::Notify::new(), idle: tokio::sync::Notify::new(), cancelled: AtomicBool::new(false), @@ -343,6 +356,41 @@ impl SessionTemporalRefreshWakeState { self.wake.notify_one(); } + fn request_projection_join( + &self, + session_id: Option, + ) -> Option> { + let mut joins = self + .projection_joins + .lock() + .unwrap_or_else(PoisonError::into_inner); + if self.cancelled.load(Ordering::Acquire) { + return None; + } + // Expired callers retain no obligation. The worker owns only live + // joins, and each result is delivered once even if a later pass fails. + joins.retain(|join| !join.completion.is_closed()); + let (completion, receiver) = tokio::sync::oneshot::channel(); + joins.push(SessionProjectionJoin { + session_id, + observed_through: None, + completion, + }); + self.wake(); + Some(receiver) + } + + /// 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 { + std::mem::take( + &mut *self + .projection_joins + .lock() + .unwrap_or_else(PoisonError::into_inner), + ) + } + pub fn requeue_projection(&self) { self.dirty.swap(true, Ordering::AcqRel); self.mark_converging(); @@ -692,6 +740,9 @@ impl SessionTemporalRefreshWakeState { self.completion_control.cancel(); self.clear_worker_instrumentation(); self.mark_stopped(); + for join in self.take_projection_joins() { + let _ = join.completion.send(false); + } self.cancellation.notify_waiters(); self.wake.notify_waiters(); } @@ -872,6 +923,7 @@ struct SessionTemporalRefreshWakeRoute { target: std::sync::RwLock>, } +#[cfg(any(test, feature = "test-helpers"))] pub(super) fn enabled_idle_notification( state: &SessionTemporalRefreshWakeState, ) -> std::pin::Pin>> { @@ -966,13 +1018,55 @@ impl SessionTemporalRefreshWake { true } - /// Wake the refresh worker and wait until it finishes the resulting pass. + /// Wake the refresh worker and wait until the resulting projection drains. /// /// Live LCM transcript projection writes must become temporally searchable /// before the mutating tool returns so Hermes `sync_turn` → `lcm_grep` /// stays available on the same route without inventing stores or weakening - /// empty/unavailable semantics. - pub async fn wake_and_wait_until_idle(&self, timeout: std::time::Duration) -> bool { + /// empty/unavailable semantics. Historical discovery and derived summaries + /// may continue after that publication; their idle/admission state is not + /// evidence that a temporal generation has committed. + pub async fn wake_and_wait_for_projection(&self, timeout: std::time::Duration) -> bool { + self.join_projection(None, timeout).await + } + + /// Join only this session's captured source frontier. Other sessions may + /// remain blocked without invalidating its committed publication. + pub async fn wake_and_wait_for_session_projection( + &self, + session_id: &SessionId, + timeout: std::time::Duration, + ) -> bool { + self.join_projection(Some(session_id.clone()), timeout) + .await + } + + async fn join_projection( + &self, + session_id: Option, + timeout: std::time::Duration, + ) -> bool { + let Some(state) = self.target() else { + return false; + }; + let Some(completion) = state.request_projection_join(session_id) else { + return false; + }; + matches!( + native_runtime::timeout( + timeout, + tracing::Instrument::instrument( + completion, + tracing::trace_span!("daemon.scheduler.session_temporal.publication_wait"), + ), + ) + .await, + Ok(Ok(true)) + ) + } + + #[cfg(test)] + pub(super) async fn wake_and_wait_until_idle(&self, timeout: std::time::Duration) -> bool { let Some(state) = self.target() else { return false; }; @@ -1038,14 +1132,26 @@ impl tracedecay_contracts::SessionTemporalRefreshWakePort for SessionTemporalRef self.status().unavailable_reason.is_some() } - fn wake_and_wait_until_idle( + fn wake_and_wait_for_projection( &self, timeout: std::time::Duration, ) -> tracedecay_contracts::SessionTemporalRefreshWakeFuture<'_> { - Box::pin(SessionTemporalRefreshWake::wake_and_wait_until_idle( + Box::pin(SessionTemporalRefreshWake::wake_and_wait_for_projection( self, timeout, )) } + + fn wake_and_wait_for_session_projection<'a>( + &'a self, + session_id: &'a SessionId, + timeout: std::time::Duration, + ) -> tracedecay_contracts::SessionTemporalRefreshWakeFuture<'a> { + Box::pin( + SessionTemporalRefreshWake::wake_and_wait_for_session_projection( + self, session_id, timeout, + ), + ) + } } impl SessionProjectionServingStatusPort for SessionTemporalRefreshWake { 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 e94b21337e..d9056c8f24 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 @@ -25,8 +25,8 @@ use super::projector::{ }; use super::registry::{SessionTemporalRefreshPassReport, session_refresh_retry_delay}; use super::wake::{ - PendingBeginRequestGuard, RecoverySelectionGuard, SessionTemporalRefreshRetryClass, - SessionTemporalRefreshWakeState, TerminalAttemptGuard, + PendingBeginRequestGuard, RecoverySelectionGuard, SessionProjectionJoin, + SessionTemporalRefreshRetryClass, SessionTemporalRefreshWakeState, TerminalAttemptGuard, }; use tracedecay_global_db::{RegisteredGlobalDb, RegisteredGlobalDbLeaseV1}; use tracedecay_runtime_core::db::engine::Error as EngineError; @@ -238,6 +238,13 @@ pub(super) async fn run_session_temporal_refresh_scheduler( let mut history_priority_passes = 0u32; let mut history_release = None; let mut granted_history_admission = None; + let mut projection_joins: Vec = Vec::new(); + let live_projection = LiveProjectionDuringSummary { + database: &database, + state: &state, + projector: projector.as_ref(), + policy, + }; let _instrumentation = SessionTemporalRefreshWorkerInstrumentation::new(&state); state.mark_running(); loop { @@ -287,8 +294,7 @@ pub(super) async fn run_session_temporal_refresh_scheduler( None }; // A wake that lands after this pass started is served by its - // projection instead of a pass of its own, so it counts as a - // pass: `wake_and_wait_until_idle` joins the pass after its wake. + // projection instead of a pass of its own. if state.take_dirty() { projection_requested = true; state.pass_count.fetch_add(1, Ordering::AcqRel); @@ -326,7 +332,7 @@ pub(super) async fn run_session_temporal_refresh_scheduler( ) || history_outcome .as_ref() .is_some_and(SessionHistoricalIngestOutcome::made_progress); - let report = if projection_requested + let mut report = if projection_requested || state.has_requests() || history_requires_projection { @@ -337,6 +343,7 @@ pub(super) async fn run_session_temporal_refresh_scheduler( &state, projector.as_ref(), policy, + &mut projection_joins, )); let pass = tracing::Instrument::instrument( projection, @@ -366,24 +373,35 @@ pub(super) async fn run_session_temporal_refresh_scheduler( } else { history_outcome.clone() }; - let convergence_admission = + let mut convergence_admission = lcm_convergence_admission(convergence_outcome, &mut history_priority_passes); - // Derived from the admission so the pass report can never disagree - // with what this pass actually ran. let history_needs_another_pass = !matches!( convergence_admission, LcmConvergenceAdmission::Admitted(LcmConvergencePage::Full) ); + // Finish the temporal generation before spending the next pass on + // derived summaries. In particular, waiting for summary admission + // must not strand a live hook's partially published projection. + // The bounded predecessor rewrite retains its history fairness + // turn; it performs no model call. + if matches!( + convergence_admission, + LcmConvergenceAdmission::Admitted(LcmConvergencePage::Full) + ) && projection_still_unpublished(&report) + && (projection_published_work(&report) + || (!projection_joins.is_empty() && report.discovery_advanced)) + { + convergence_admission = LcmConvergenceAdmission::Deferred; + } let ( summary_convergence_made_progress, summary_convergence_has_more, summary_retry_delay, ) = match convergence_admission { LcmConvergenceAdmission::Deferred => { - // Historical continuation owns the next bounded pass. LCM - // summaries are independent derived work and can run after - // the raw frontier is terminal; placing a model call between - // source windows delays both project and profile readiness. + // Historical continuation or temporal publication owns the + // next bounded pass. Summaries can run after those frontiers + // settle without delaying live projection joins. (false, false, None) } LcmConvergenceAdmission::Admitted(convergence_page) => { @@ -408,10 +426,11 @@ pub(super) async fn run_session_temporal_refresh_scheduler( // registered to receive one. state.mark_worker_idle(); state.idle.notify_waiters(); - let permit = tokio::select! { - biased; - () = tracing::Instrument::instrument(state.wait_for_cancellation(), tracing::trace_span!("daemon.scheduler.lcm_summary.admission_cancel")) => return, - permit = &mut admission => permit, + let Some(permit) = live_projection + .wait_for(&mut admission, &mut projection_joins, &mut report, true) + .await + else { + return; }; state.mark_worker_busy(); permit @@ -447,10 +466,11 @@ pub(super) async fn run_session_temporal_refresh_scheduler( ), }; tokio::pin!(page); - let result = tokio::select! { - biased; - () = tracing::Instrument::instrument(state.wait_for_cancellation(), tracing::trace_span!("daemon.scheduler.lcm_summary.cancel")) => return, - result = &mut page => result, + let Some(result) = live_projection + .wait_for(&mut page, &mut projection_joins, &mut report, false) + .await + else { + return; }; drop(permit); result @@ -541,7 +561,8 @@ pub(super) async fn run_session_temporal_refresh_scheduler( match window_follow_up( history_continuation(history_outcome), projection_still_unpublished(&report), - projection_published_work(&report), + projection_published_work(&report) + || (!projection_joins.is_empty() && report.discovery_advanced), holding_publication, ) { // The admitted window is not searchable until this backlog @@ -770,8 +791,236 @@ async fn session_projection_refresh( state: &Arc, projector: &dyn SessionTemporalRefreshProjector, policy: SessionTemporalRefreshPolicy, + joins: &mut Vec, ) -> SessionTemporalRefreshPassReport { - run_session_temporal_refresh_pass(database, state, projector, policy).await + joins.retain(|join| !join.completion.is_closed()); + joins.extend(state.take_projection_joins()); + let access = SessionTemporalAccess::new(database.as_ref()); + capture_projection_targets(&access, joins).await; + let report = run_session_temporal_refresh_pass(database, state, projector, policy).await; + if state.cancelled.load(Ordering::Acquire) { + return report; + } + let result = if report.failed > 0 || report.cancelled > 0 || report.terminal_errors > 0 { + Some(false) + } else if report.backlog == Some(0) + && !report.saturated + && report.retry_class.is_none() + && report.retryable_errors == 0 + && report.deadline_errors == 0 + && report.deferred == 0 + && !state.has_requests() + { + Some(true) + } else { + None + }; + settle_projection_joins(&access, joins, &report, result).await; + report +} + +async fn capture_projection_targets( + access: &SessionTemporalAccess<'_, RegisteredGlobalDb>, + joins: &mut Vec, +) { + let mut frontiers = HashMap::new(); + for mut join in std::mem::take(joins) { + if let Some(session_id) = &join.session_id + && join.observed_through.is_none() + { + if let std::collections::hash_map::Entry::Vacant(entry) = + frontiers.entry(session_id.clone()) + { + let frontier = access + .session_temporal_publication_result(session_id) + .await + .map(|publication| publication.observed_through()) + .map_err(|error| { + tracing::warn!(%session_id, %error, "session publication target unavailable"); + }) + .ok(); + entry.insert(frontier); + } + let Some(frontier) = frontiers.get(session_id).copied().flatten() else { + let _ = join.completion.send(false); + continue; + }; + join.observed_through = Some(frontier); + } + joins.push(join); + } +} + +async fn settle_projection_joins( + access: &SessionTemporalAccess<'_, RegisteredGlobalDb>, + joins: &mut Vec, + report: &SessionTemporalRefreshPassReport, + aggregate: Option, +) { + let mut publications = HashMap::new(); + for join in std::mem::take(joins) { + if join.completion.is_closed() { + continue; + } + let result = if let Some(session_id) = &join.session_id { + if let std::collections::hash_map::Entry::Vacant(entry) = + publications.entry(session_id.clone()) + { + let publication = access + .session_temporal_publication_result(session_id) + .await + .map_err(|error| { + tracing::warn!(%session_id, %error, "session publication proof unavailable"); + }) + .ok(); + entry.insert(publication); + } + match publications.get(session_id).and_then(Option::as_ref) { + Some(publication) + if join + .observed_through + .is_some_and(|target| publication.covers(target)) => + { + Some(true) + } + Some(_) + if report.failed_sessions.contains(session_id) || aggregate == Some(true) => + { + // The named session must prove its own frontier. A drained + // scope or another session's successful receipt is not proof. + Some(false) + } + Some(_) => None, + None => Some(false), + } + } else { + aggregate + }; + if let Some(published) = result { + let _ = join.completion.send(published); + } else { + joins.push(join); + } + } +} + +/// 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> { + database: &'a RegisteredGlobalDbLeaseV1, + state: &'a Arc, + projector: &'a dyn SessionTemporalRefreshProjector, + policy: SessionTemporalRefreshPolicy, +} + +impl LiveProjectionDuringSummary<'_> { + async fn wait_for( + &self, + stage: F, + joins: &mut Vec, + report: &mut SessionTemporalRefreshPassReport, + idle_between_projections: bool, + ) -> Option + where + F: Future + Send, + F::Output: Send, + { + tokio::pin!(stage); + let mut retry_attempt = 0u32; + let mut retry_at = None; + loop { + let projection_wake = async { + let wake = self.state.wake.notified(); + 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 + } + }; + let woke = tokio::select! { + biased; + () = self.state.wait_for_cancellation() => return None, + result = &mut stage => return Some(result), + woke = projection_wake => woke, + }; + if !woke { + return None; + } + if !self.state.take_dirty() { + continue; + } + self.state.mark_worker_busy(); + self.state.begin_pass(); + self.state.pass_count.fetch_add(1, Ordering::AcqRel); + let (next_report, completed_stage) = { + let projection: Pin< + Box + Send + '_>, + > = Box::pin(session_projection_refresh( + self.database, + self.state, + self.projector, + self.policy, + joins, + )); + let projection = tracing::Instrument::instrument( + projection, + tracing::trace_span!("daemon.scheduler.session_temporal.projection"), + ); + tokio::pin!(projection); + tokio::select! { + biased; + () = self.state.wait_for_cancellation() => return None, + next = &mut projection => (next, None), + result = &mut stage => { + // Finish the admitted projection rather than cancelling + // it when the summary stage happens to complete first. + let next = tokio::select! { + biased; + () = self.state.wait_for_cancellation() => return None, + next = &mut projection => next, + }; + (next, Some(result)) + } + } + }; + *report = next_report; + if let Some(backlog) = report.backlog { + self.state + .record_pass(backlog, projection_published_work(report)); + } + if let Some(class) = report.retry_class { + retry_attempt = retry_attempt.saturating_add(1); + self.state.mark_recovering(class.into(), class); + self.state.requeue_projection(); + retry_at = Some(Instant::now() + session_refresh_retry_delay(class, retry_attempt)); + } else { + retry_attempt = 0; + retry_at = None; + self.state.mark_running(); + if self.state.has_requests() + || report.begun > 0 + || report.saturated + || report.projected_batches > 0 + { + self.state.requeue_projection(); + } + } + if let Some(result) = completed_stage { + return Some(result); + } + if idle_between_projections { + self.state.mark_worker_idle(); + self.state.idle.notify_waiters(); + } + } + } } /// True when replaying this store failure unchanged could still succeed. @@ -798,9 +1047,12 @@ fn is_deterministic_refusal(error: &SessionStoreError) -> bool { SessionStoreError::Cancelled | SessionStoreError::DeadlineExceeded | SessionStoreError::BudgetExceeded { .. } => false, - SessionStoreError::Storage { source, .. } => source - .downcast_ref::() - .is_some_and(EngineError::is_deterministic_refusal), + SessionStoreError::Storage { source, .. } => { + source + .downcast_ref::() + .is_some_and(EngineError::is_deterministic_refusal) + || tracedecay_store::error_chain_writer_fence(source.as_ref()).is_some() + } _ => true, } } @@ -948,6 +1200,7 @@ pub async fn begin_admitted_session_refreshes<'a>( } }; let (requests, next_cursor, has_more) = page.into_parts(); + report.discovery_advanced |= next_cursor != cursor; for request in requests.into_iter().rev() { if !state.suppresses_discovered_request(&request) { state.requeue_request(request); @@ -1609,6 +1862,7 @@ async fn apply_fail_effect( "session temporal refresh failed durably" ); report.failed += 1; + report.failed_sessions.insert(recovery.session_id().clone()); if matches!(disposition, RefreshFailureDisposition::SuppressDiscovery) { state.record_terminal_discovery_failure(recovery); } @@ -1948,10 +2202,110 @@ pub async fn run_session_temporal_refresh_pass( #[cfg(test)] mod tests { use super::*; + use crate::session_temporal_refresh_scheduler::projector::CanonicalSessionTemporalProjector; use tracedecay_domain::SessionId; use tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness; use tracedecay_runtime_core::db::engine::params; use tracedecay_sessions::runtime::{SessionMessageRecord, SessionRecord}; + use tracedecay_store::{SessionRefreshReceiptRequestV1, SessionRefreshTerminalStateV1}; + + #[tokio::test] + async fn an_in_flight_summary_stage_keeps_polling_while_a_live_wake_projects() { + let harness = RegisteredGlobalDbHarness::open("refresh-during-summary-writer").await; + let store = SessionTemporalStore::new(harness.registered.as_ref()); + let begun = store + .begin_or_join_session_refresh(SessionRefreshBeginOrJoinRequestV1::new( + SessionId::new("refresh-during-summary-writer").expect("session id"), + SessionRefreshFrontierV1::new(0, 0).expect("empty frontier"), + )) + .await + .expect("running refresh"); + let state = Arc::new(SessionTemporalRefreshWakeState::default()); + let entered = Arc::new(tokio::sync::Notify::new()); + let release_writer = Arc::new(tokio::sync::Notify::new()); + let finish = Arc::new(tokio::sync::Notify::new()); + let worker = tokio::spawn({ + let database = harness.registered.clone(); + let state = Arc::clone(&state); + let entered = Arc::clone(&entered); + let release_writer = Arc::clone(&release_writer); + let finish = Arc::clone(&finish); + async move { + let stage = async { + let transaction = database.begin_write_transaction().await.expect("writer"); + entered.notify_one(); + release_writer.notified().await; + transaction.commit().await.expect("release summary writer"); + finish.notified().await; + 17usize + }; + let live_projection = LiveProjectionDuringSummary { + database: &database, + state: &state, + projector: &CanonicalSessionTemporalProjector, + policy: SessionTemporalRefreshPolicy::default(), + }; + let mut joins = Vec::new(); + let mut report = SessionTemporalRefreshPassReport::default(); + live_projection + .wait_for(stage, &mut joins, &mut report, false) + .await + } + }); + entered.notified().await; + + let handle = state.handle(); + let join = handle.wake_and_wait_for_projection(Duration::from_secs(5)); + tokio::pin!(join); + assert!( + std::future::poll_fn(|cx| std::task::Poll::Ready(join.as_mut().poll(cx))) + .await + .is_pending() + ); + tokio::time::timeout(Duration::from_secs(5), async { + while state.pass_count.load(Ordering::Acquire) == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("live projection starts while the summary stage owns the writer"); + release_writer.notify_one(); + assert!( + join.await, + "summary must keep polling to release the writer needed by projection" + ); + let receipt = store + .session_refresh_receipt(SessionRefreshReceiptRequestV1::new( + begun.operation_id().clone(), + begun.session_id().clone(), + )) + .await + .expect("read terminal receipt") + .expect("projection publishes before summary finishes"); + assert_eq!(receipt.state(), SessionRefreshTerminalStateV1::Complete); + assert!(!worker.is_finished(), "the summary stage remains in flight"); + finish.notify_one(); + assert_eq!(worker.await.expect("summary stage completes"), Some(17)); + } + + /// A lease the store has fenced resubmits the identical request on every + /// pass and can never commit through it, so the pass retires the refresh + /// instead of backing off forever. + #[test] + fn a_fenced_writer_is_a_deterministic_refusal_not_a_retry() { + let fence = tracedecay_store::StoreWriterFencedV1 { + requested: tracedecay_store::StoreAuthorityEpochV1::new(2).unwrap(), + current: tracedecay_store::StoreAuthorityEpochV1::new(3).unwrap(), + }; + let wrapped = tracedecay_store::ObservationStoreError::Storage { + operation: "submit anchored observation", + source: Box::new(fence), + }; + let error = SessionStoreError::storage("persist session history batch", wrapped); + + assert!(is_deterministic_refusal(&error)); + assert!(!is_retryable_storage(&error)); + } /// Commits `item_count` items. A group of several items never commits, and /// neither does a `slow` item alone: its transaction ends in the 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 f2f48b5d36..0d24a5f947 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 @@ -1,4 +1,5 @@ use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use tracedecay_domain::{SessionId, UtcMicros}; @@ -7,7 +8,8 @@ use tracedecay_global_db::tests::harness::RegisteredGlobalDbHarness; use tracedecay_session_temporal_store::{SessionRefreshRecoveryV1, SessionTemporalStore}; use tracedecay_store::{ SessionRefreshBeginOrJoinRequestV1, SessionRefreshFrontierV1, SessionRefreshProgressV1, - SessionRefreshStore, SessionTemporalProjectionBatchV1, + SessionRefreshReceiptRequestV1, SessionRefreshStore, SessionRefreshTerminalStateV1, + SessionTemporalProjectionBatchV1, }; use super::history::{ @@ -16,10 +18,13 @@ use super::history::{ }; use super::projector::{ CanonicalSessionTemporalProjector, SessionTemporalRefreshEffect, SessionTemporalRefreshPolicy, - SessionTemporalRefreshProjectionFuture, SessionTemporalRefreshProjector, zero_refresh_coverage, + SessionTemporalRefreshProjectionFuture, SessionTemporalRefreshProjector, + SessionTemporalRefreshProjectorError, zero_refresh_coverage, }; use super::registry::SessionTemporalRefreshPassReport; -use super::wake::{SessionTemporalRefreshRetryClass, SessionTemporalRefreshWakeState}; +use super::wake::{ + SessionTemporalRefreshRetryClass, SessionTemporalRefreshWakeState, enabled_idle_notification, +}; use super::worker::{ apply_refresh_effect, run_session_temporal_refresh_pass, run_session_temporal_refresh_scheduler, }; @@ -220,9 +225,9 @@ async fn a_wake_served_by_the_running_pass_joins_that_pass() { history.entered.notified().await; let handle = state.handle(); - let join = handle.wake_and_wait_until_idle(Duration::from_secs(5)); + let join = handle.wake_and_wait_for_projection(Duration::from_secs(5)); tokio::pin!(join); - // The first poll records the pass count and delivers the wake. + // 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 @@ -238,6 +243,375 @@ async fn a_wake_served_by_the_running_pass_joins_that_pass() { worker.await.expect("refresh worker exits on cancellation"); } +/// Every history window advances but waits for the test before returning. +/// This leaves the next window running after the preceding projection drains. +struct ContinuingHistory { + entered: tokio::sync::Notify, + release: tokio::sync::Notify, +} + +impl SessionHistoricalIngestor for ContinuingHistory { + fn run_pass(&self) -> SessionHistoricalIngestPass<'_> { + Box::pin(async move { + self.entered.notify_one(); + self.release.notified().await; + SessionHistoricalIngestOutcome::Pending { + made_progress: true, + } + .into() + }) + } + + fn capacity_release(&self) -> SessionHistoricalCapacityRelease { + SessionHistoricalCapacityRelease::new(Vec::new()) + } + + fn cancel(&self) { + self.release.notify_one(); + } +} + +#[tokio::test] +async fn a_published_wake_does_not_wait_for_the_next_history_window() { + let harness = RegisteredGlobalDbHarness::open("refresh-join-before-next-history").await; + let state = Arc::new(SessionTemporalRefreshWakeState::default()); + let history = Arc::new(ContinuingHistory { + entered: tokio::sync::Notify::new(), + release: tokio::sync::Notify::new(), + }); + let ingestor: SharedSessionHistoricalIngestor = history.clone(); + state.wake_history(); + let worker = tokio::spawn(run_session_temporal_refresh_scheduler( + harness.registered.clone(), + Arc::clone(&state), + Arc::new(CanonicalSessionTemporalProjector), + Arc::new(std::sync::RwLock::new(Some(ingestor))), + Arc::new(tokio::sync::Semaphore::new(1)), + SessionTemporalRefreshPolicy::default(), + )); + history.entered.notified().await; + + let handle = state.handle(); + let join = handle.wake_and_wait_for_projection(Duration::from_secs(1)); + tokio::pin!(join); + assert!( + std::future::poll_fn(|cx| std::task::Poll::Ready(join.as_mut().poll(cx))) + .await + .is_pending() + ); + history.release.notify_one(); + history.entered.notified().await; + + assert!( + join.await, + "the published wake must join while the next history window is still running" + ); + state.cancel(); + history.release.notify_one(); + worker.await.expect("refresh worker exits on cancellation"); +} + +#[tokio::test] +async fn a_live_wake_publishes_before_waiting_for_summary_admission() { + let harness = RegisteredGlobalDbHarness::open("refresh-join-before-summary-admission").await; + let store = SessionTemporalStore::new(harness.registered.as_ref()); + let begun = store + .begin_or_join_session_refresh(SessionRefreshBeginOrJoinRequestV1::new( + SessionId::new("refresh-before-summary").expect("session id"), + SessionRefreshFrontierV1::new(0, 0).expect("empty frontier"), + )) + .await + .expect("running refresh"); + let state = Arc::new(SessionTemporalRefreshWakeState::default()); + let handle = state.handle(); + let join = handle.wake_and_wait_for_projection(Duration::from_secs(5)); + tokio::pin!(join); + assert!( + std::future::poll_fn(|cx| std::task::Poll::Ready(join.as_mut().poll(cx))) + .await + .is_pending() + ); + let worker = tokio::spawn(run_session_temporal_refresh_scheduler( + harness.registered.clone(), + Arc::clone(&state), + Arc::new(CanonicalSessionTemporalProjector), + Arc::new(std::sync::RwLock::new(None)), + Arc::new(tokio::sync::Semaphore::new(0)), + SessionTemporalRefreshPolicy::default(), + )); + + assert!( + join.await, + "summary admission must not hold a live projection" + ); + let receipt = store + .session_refresh_receipt(SessionRefreshReceiptRequestV1::new( + begun.operation_id().clone(), + begun.session_id().clone(), + )) + .await + .expect("read terminal receipt") + .expect("joining the wake must publish a terminal receipt"); + assert_eq!(receipt.state(), SessionRefreshTerminalStateV1::Complete); + + state.cancel(); + worker.await.expect("refresh worker exits on cancellation"); +} + +#[tokio::test] +async fn a_live_wake_during_summary_admission_preserves_the_queued_turn() { + let harness = RegisteredGlobalDbHarness::open("refresh-join-during-summary-admission").await; + let state = Arc::new(SessionTemporalRefreshWakeState::default()); + let admission = Arc::new(tokio::sync::Semaphore::new(0)); + state.wake(); + let worker = tokio::spawn(run_session_temporal_refresh_scheduler( + harness.registered.clone(), + Arc::clone(&state), + Arc::new(CanonicalSessionTemporalProjector), + Arc::new(std::sync::RwLock::new(None)), + Arc::clone(&admission), + SessionTemporalRefreshPolicy::default(), + )); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let idle = enabled_idle_notification(&state); + if state.is_idle() { + break; + } + idle.await; + } + }) + .await + .expect("worker queues for summary admission before advertising idle"); + + // Register another profile behind this worker before the live wake. A + // projection must not surrender the existing FIFO turn and queue anew. + let later = admission.acquire(); + tokio::pin!(later); + assert!( + std::future::poll_fn(|cx| std::task::Poll::Ready(later.as_mut().poll(cx))) + .await + .is_pending() + ); + let store = SessionTemporalStore::new(harness.registered.as_ref()); + let begun = store + .begin_or_join_session_refresh(SessionRefreshBeginOrJoinRequestV1::new( + SessionId::new("refresh-during-summary").expect("session id"), + SessionRefreshFrontierV1::new(0, 0).expect("empty frontier"), + )) + .await + .expect("running refresh"); + assert!( + state + .handle() + .wake_and_wait_for_projection(Duration::from_secs(5)) + .await, + "a new wake must publish while summary admission has no permits" + ); + let receipt = store + .session_refresh_receipt(SessionRefreshReceiptRequestV1::new( + begun.operation_id().clone(), + begun.session_id().clone(), + )) + .await + .expect("read terminal receipt") + .expect("joining the wake must publish a terminal receipt"); + assert_eq!(receipt.state(), SessionRefreshTerminalStateV1::Complete); + assert_eq!(admission.available_permits(), 0); + + admission.add_permits(1); + assert!( + std::future::poll_fn(|cx| std::task::Poll::Ready(later.as_mut().poll(cx))) + .await + .is_pending(), + "the worker's original summary admission must retain the first turn" + ); + state.cancel(); + worker.await.expect("refresh worker exits on cancellation"); +} + +struct ControlledProjector { + reject: AtomicBool, +} + +/// Fails one session while every other session uses the production projector. +struct SessionRefusingProjector { + refused: SessionId, +} + +impl SessionTemporalRefreshProjector for SessionRefusingProjector { + fn project<'a>( + &'a self, + database: &'a RegisteredGlobalDbLeaseV1, + recovery: SessionRefreshRecoveryV1, + ) -> SessionTemporalRefreshProjectionFuture<'a> { + Box::pin(async move { + if recovery.session_id() == &self.refused { + Err(SessionTemporalRefreshProjectorError::terminal( + "fixture_session_projection_refused", + )) + } else { + CanonicalSessionTemporalProjector + .project(database, recovery) + .await + } + }) + } +} + +#[tokio::test] +async fn named_projection_join_survives_an_unrelated_session_failure() { + let harness = RegisteredGlobalDbHarness::open("refresh-join-session-isolation").await; + let store = SessionTemporalStore::new(harness.registered.as_ref()); + let successful = SessionId::new("projection-accepted").expect("successful session"); + let refused = SessionId::new("projection-refused").expect("refused session"); + begin_empty_refreshes(&store, ["projection-accepted", "projection-refused"]).await; + let recoveries = store.running_session_refreshes().await.expect("recoveries"); + assert_eq!(recoveries.len(), 2); + let state = Arc::new(SessionTemporalRefreshWakeState::default()); + let handle = state.handle(); + let successful_join = + handle.wake_and_wait_for_session_projection(&successful, Duration::from_secs(5)); + let refused_join = + handle.wake_and_wait_for_session_projection(&refused, Duration::from_secs(5)); + let aggregate_join = handle.wake_and_wait_for_projection(Duration::from_secs(5)); + tokio::pin!(successful_join, refused_join, aggregate_join); + // Put all three obligations in the same pass before starting the worker. + // The unnamed caller still owns the aggregate failure contract. + let pending = std::future::poll_fn(|cx| { + std::task::Poll::Ready(( + successful_join.as_mut().poll(cx).is_pending(), + refused_join.as_mut().poll(cx).is_pending(), + aggregate_join.as_mut().poll(cx).is_pending(), + )) + }) + .await; + assert_eq!(pending, (true, true, true)); + let worker = tokio::spawn(run_session_temporal_refresh_scheduler( + harness.registered.clone(), + Arc::clone(&state), + Arc::new(SessionRefusingProjector { + refused: refused.clone(), + }), + Arc::new(std::sync::RwLock::new(None)), + Arc::new(tokio::sync::Semaphore::new(0)), + SessionTemporalRefreshPolicy::default(), + )); + + let (successful_result, refused_result, aggregate_result) = + tokio::join!(successful_join, refused_join, aggregate_join); + state.cancel(); + worker.await.expect("worker exits on cancellation"); + assert!( + successful_result, + "another session cannot veto a committed publication" + ); + assert!( + !refused_result, + "the failed session must refuse its own join" + ); + assert!( + !aggregate_result, + "unnamed sweeps must retain aggregate failure" + ); + for recovery in recoveries { + let receipt = store + .session_refresh_receipt(SessionRefreshReceiptRequestV1::new( + recovery.operation_id().clone(), + recovery.session_id().clone(), + )) + .await + .expect("read durable receipt") + .expect("both sessions must settle durably"); + let expected = if recovery.session_id() == &successful { + SessionRefreshTerminalStateV1::Complete + } else { + SessionRefreshTerminalStateV1::Failed + }; + assert_eq!(receipt.state(), expected); + } +} + +impl SessionTemporalRefreshProjector for ControlledProjector { + fn project<'a>( + &'a self, + database: &'a RegisteredGlobalDbLeaseV1, + recovery: SessionRefreshRecoveryV1, + ) -> SessionTemporalRefreshProjectionFuture<'a> { + Box::pin(async move { + if self.reject.load(Ordering::Acquire) { + Err(SessionTemporalRefreshProjectorError::terminal( + "fixture_projection_refused", + )) + } else { + CanonicalSessionTemporalProjector + .project(database, recovery) + .await + } + }) + } +} + +#[tokio::test] +async fn projection_joins_keep_their_result_when_later_work_settles_differently() { + for reject_first in [false, true] { + let label = format!("refresh-join-result-reject-first-{reject_first}"); + let harness = RegisteredGlobalDbHarness::open(&label).await; + let store = SessionTemporalStore::new(harness.registered.as_ref()); + begin_empty_refreshes(&store, ["first-projection"]).await; + let state = Arc::new(SessionTemporalRefreshWakeState::default()); + let history = Arc::new(ContinuingHistory { + entered: tokio::sync::Notify::new(), + release: tokio::sync::Notify::new(), + }); + let ingestor: SharedSessionHistoricalIngestor = history.clone(); + let projector = Arc::new(ControlledProjector { + reject: AtomicBool::new(reject_first), + }); + state.wake_history(); + let worker = tokio::spawn(run_session_temporal_refresh_scheduler( + harness.registered.clone(), + Arc::clone(&state), + projector.clone(), + Arc::new(std::sync::RwLock::new(Some(ingestor))), + Arc::new(tokio::sync::Semaphore::new(1)), + SessionTemporalRefreshPolicy::default(), + )); + history.entered.notified().await; + + let handle = state.handle(); + let first = handle.wake_and_wait_for_projection(Duration::from_secs(5)); + tokio::pin!(first); + assert!( + std::future::poll_fn(|cx| std::task::Poll::Ready(first.as_mut().poll(cx))) + .await + .is_pending() + ); + history.release.notify_one(); + history.entered.notified().await; + + // Do not poll the first join again until a different projection has + // settled with the opposite result. Neither result may overwrite the + // one already delivered to the earlier request. + projector.reject.store(!reject_first, Ordering::Release); + begin_empty_refreshes(&store, ["second-projection"]).await; + let second = handle.wake_and_wait_for_projection(Duration::from_secs(5)); + tokio::pin!(second); + assert!( + std::future::poll_fn(|cx| std::task::Poll::Ready(second.as_mut().poll(cx))) + .await + .is_pending() + ); + history.release.notify_one(); + assert_eq!(second.await, reject_first); + assert_eq!(first.await, !reject_first); + + state.cancel(); + history.release.notify_one(); + worker.await.expect("refresh worker exits on cancellation"); + } +} + /// Replays scripted passes, each with its own provider unit tallies. struct TalliedHistory { passes: std::sync::Mutex>, diff --git a/crates/tracedecay-session-temporal-store/src/lib.rs b/crates/tracedecay-session-temporal-store/src/lib.rs index bf57ec23e5..24a1b74b6c 100644 --- a/crates/tracedecay-session-temporal-store/src/lib.rs +++ b/crates/tracedecay-session-temporal-store/src/lib.rs @@ -30,7 +30,10 @@ mod hydration; pub mod operations; mod participant_freeze; mod projection; -pub use projection::{SessionTemporalRefreshDiscoveryCursor, SessionTemporalRefreshDiscoveryPage}; +pub use projection::{ + SessionTemporalPublication, SessionTemporalRefreshDiscoveryCursor, + SessionTemporalRefreshDiscoveryPage, +}; mod query; mod rebuild; mod refresh; diff --git a/crates/tracedecay-session-temporal-store/src/projection.rs b/crates/tracedecay-session-temporal-store/src/projection.rs index 53165af2aa..87863770b8 100644 --- a/crates/tracedecay-session-temporal-store/src/projection.rs +++ b/crates/tracedecay-session-temporal-store/src/projection.rs @@ -75,6 +75,26 @@ pub struct SessionTemporalRefreshDiscoveryPage { has_more: bool, } +/// The canonical source frontier and the active generation's publication. +/// Only an applied relation receipt belonging to that active generation and +/// matching its expected graph watermark proves it is searchable. A pending +/// receipt or retired generation cannot satisfy a new join. +pub struct SessionTemporalPublication { + observed_through: u64, + published_through: Option, +} + +impl SessionTemporalPublication { + pub const fn observed_through(&self) -> u64 { + self.observed_through + } + + pub fn covers(&self, target: u64) -> bool { + self.published_through + .is_some_and(|published| target <= published && published <= self.observed_through) + } +} + impl SessionTemporalRefreshDiscoveryPage { pub fn active_rows_scanned(&self) -> usize { self.active_rows_scanned @@ -402,6 +422,68 @@ async fn skip_trailing_history_effects( } impl SessionTemporalAccess<'_, D> { + /// Read one session's source and publication from the same snapshot. + /// A pending reset invalidates the old publication even when its numeric + /// frontier still covers the source; a future frontier is never proof. + pub async fn session_temporal_publication_result( + &self, + session_id: &SessionId, + ) -> SessionStoreResult { + let snapshot = self + .read_snapshot() + .await + .map_err(|error| storage(DISCOVER_REFRESH, error))?; + let mut rows = snapshot + .query( + "SELECT + COALESCE(( + SELECT effect.observation_sequence + FROM session_temporal_observation_effects AS effect + WHERE effect.session_id = ?1 AND effect.output_count > 0 + ORDER BY effect.observation_sequence DESC LIMIT 1 + ), 0), + (SELECT CAST(json_extract( + active.frozen_watermarks_json, '$.projection_frontier' + ) AS INTEGER) + FROM session_temporal_generations AS active + JOIN session_relation_receipts AS receipt + ON receipt.session_id = active.session_id + AND receipt.generation = active.generation + AND receipt.state = 'applied' + AND receipt.graph_watermark = receipt.expected_graph_watermark + WHERE active.session_id = ?1 AND active.state = 'active' + AND NOT EXISTS ( + SELECT 1 FROM session_temporal_resets AS reset + WHERE reset.session_id = ?1 AND reset.requested_at IS NOT NULL + ))", + params![session_id.as_str()], + ) + .await + .map_err(|error| storage(DISCOVER_REFRESH, error))?; + let row = rows + .next() + .await + .map_err(|error| storage(DISCOVER_REFRESH, error))? + .ok_or_else(|| { + storage_message(DISCOVER_REFRESH, "session publication returned no row") + })?; + let observed_through = u64::try_from( + row.get::(0) + .map_err(|error| storage(DISCOVER_REFRESH, error))?, + ) + .map_err(|error| storage(DISCOVER_REFRESH, error))?; + let published_through = row + .get::>(1) + .map_err(|error| storage(DISCOVER_REFRESH, error))? + .map(u64::try_from) + .transpose() + .map_err(|error| storage(DISCOVER_REFRESH, error))?; + Ok(SessionTemporalPublication { + observed_through, + published_through, + }) + } + /// Discovers sessions that need temporal projection: sessions whose /// output-producing effects committed past `cursor` left them behind their /// projection frontier, plus active generations that lack an applied diff --git a/crates/tracedecay-session-temporal-store/src/projection/tests.rs b/crates/tracedecay-session-temporal-store/src/projection/tests.rs index 3908b37ebb..1ea25043a5 100644 --- a/crates/tracedecay-session-temporal-store/src/projection/tests.rs +++ b/crates/tracedecay-session-temporal-store/src/projection/tests.rs @@ -2505,6 +2505,163 @@ async fn refresh_through( } } +#[asupersync::test] +async fn publication_requires_applied_current_receipt_and_source_coverage() { + let tmp = TempDir::new().unwrap(); + let runtime = HostAdmissionTestRuntimeV1::profile(tmp.path()) + .await + .unwrap(); + let store = temporal_store(&runtime); + let database = runtime + .registered_database(HostAdmissionScope::Profile) + .unwrap(); + let access = SessionTemporalAccess::new(database); + let session_id = fixture_session("session.publication-proof"); + let (first, first_write) = fixture_observation(&session_id, 0, None, false); + Box::pin(persist_fixture(&runtime, first, first_write)).await; + let first_generation = refresh_through(&store, &session_id, 1, 0).await; + let publication = access + .session_temporal_publication_result(&session_id) + .await + .unwrap(); + assert_eq!(publication.observed_through(), 1); + assert!(publication.covers(1)); + assert!( + !publication.covers(2), + "a future source target has no publication proof" + ); + + let (second, second_write) = fixture_observation(&session_id, 1, None, true); + let second_observation = second.observation_id().clone(); + Box::pin(persist_fixture(&runtime, second, second_write)).await; + let publication = access + .session_temporal_publication_result(&session_id) + .await + .unwrap(); + assert_eq!(publication.observed_through(), 2); + assert!(publication.covers(1)); + assert!( + !publication.covers(2), + "the old receipt cannot cover the new observation" + ); + let active_generation = refresh_through(&store, &session_id, 2, 1).await; + assert_ne!(first_generation, active_generation); + let snapshot = database.read_snapshot().await.unwrap(); + let mut older_receipt = snapshot.query( + "SELECT COUNT(*) FROM session_relation_receipts WHERE session_id = ?1 AND generation = ?2 AND state = 'applied'", + params![session_id.as_str(), i64::try_from(first_generation.value()).unwrap()], + ).await.unwrap(); + assert_eq!( + older_receipt + .next() + .await + .unwrap() + .unwrap() + .get::(0) + .unwrap(), + 1 + ); + drop(older_receipt); + drop(snapshot); + assert!( + access + .session_temporal_publication_result(&session_id) + .await + .unwrap() + .covers(2) + ); + + // A pending graph application cannot serve this generation, even though + // an older generation still has its applied receipt. + let transaction = database.begin_write_transaction().await.unwrap(); + assert_eq!(transaction.execute( + "UPDATE session_relation_receipts SET state = 'pending', graph_watermark = NULL, applied_at = NULL WHERE session_id = ?1 AND generation = ?2", + params![session_id.as_str(), i64::try_from(active_generation.value()).unwrap()], + ).await.unwrap(), 1); + transaction.commit().await.unwrap(); + assert!( + !access + .session_temporal_publication_result(&session_id) + .await + .unwrap() + .covers(2), + "a pending graph receipt must not publish" + ); + + let transaction = database.begin_write_transaction().await.unwrap(); + transaction.execute( + "UPDATE session_relation_receipts SET state = 'applied', graph_watermark = expected_graph_watermark || '.mismatch', applied_at = 1 WHERE session_id = ?1 AND generation = ?2", + params![session_id.as_str(), i64::try_from(active_generation.value()).unwrap()], + ).await.expect_err("schema must refuse a mismatched graph publication"); + transaction.execute( + "UPDATE session_relation_receipts SET state = 'applied', graph_watermark = expected_graph_watermark, applied_at = 1 WHERE session_id = ?1 AND generation = ?2", + params![session_id.as_str(), i64::try_from(active_generation.value()).unwrap()], + ).await.unwrap(); + transaction.commit().await.unwrap(); + assert!( + access + .session_temporal_publication_result(&session_id) + .await + .unwrap() + .covers(2) + ); + + let transaction = database.begin_write_transaction().await.unwrap(); + super::request_session_temporal_reset(&transaction, second_observation.as_str()) + .await + .unwrap(); + transaction.commit().await.unwrap(); + assert!( + !access + .session_temporal_publication_result(&session_id) + .await + .unwrap() + .covers(2), + "a pending reset invalidates the old publication" + ); + + let transaction = database.begin_write_transaction().await.unwrap(); + transaction + .execute( + "DELETE FROM session_temporal_resets WHERE session_id = ?1", + params![session_id.as_str()], + ) + .await + .unwrap(); + assert_eq!( + transaction + .execute( + "DELETE FROM session_relation_receipts WHERE session_id = ?1 AND generation = ?2", + params![ + session_id.as_str(), + i64::try_from(active_generation.value()).unwrap() + ] + ) + .await + .unwrap(), + 1 + ); + transaction.commit().await.unwrap(); + assert!( + !access + .session_temporal_publication_result(&session_id) + .await + .unwrap() + .covers(1), + "a retired generation's receipt cannot stand in for the active generation" + ); +} + +#[test] +fn publication_refuses_a_frontier_beyond_the_observed_source() { + let publication = super::SessionTemporalPublication { + observed_through: 2, + published_through: Some(3), + }; + assert!(!publication.covers(2)); + assert!(!publication.covers(3)); +} + /// One observation of a live session: the thread changes every third /// message, each message replies to the previous one, and every fourth /// message supersedes the one before it. diff --git a/crates/tracedecay-sessions/src/runtime/ingest/failure.rs b/crates/tracedecay-sessions/src/runtime/ingest/failure.rs index c5ae81d0d8..c5a3134eac 100644 --- a/crates/tracedecay-sessions/src/runtime/ingest/failure.rs +++ b/crates/tracedecay-sessions/src/runtime/ingest/failure.rs @@ -343,6 +343,11 @@ pub fn classify_transcript_ingest_disposition( source::TranscriptIngestError::Store(TranscriptStoreError::Conflict { .. }) => { ("transcript_cursor_conflict", true, Backpressured) } + source::TranscriptIngestError::Store(TranscriptStoreError::Storage { source, .. }) + if tracedecay_store::error_chain_writer_fence(source.as_ref()).is_some() => + { + ("transcript_writer_fenced", false, Unavailable) + } source::TranscriptIngestError::Store(TranscriptStoreError::Storage { .. }) => { ("transcript_storage_failed", true, Unavailable) } @@ -500,6 +505,17 @@ pub fn classify_claude_observation_failure( Store::CursorCoverageMismatch => permanent("observation_cursor_coverage_gap"), Store::InvalidReplayLimit { .. } => permanent("observation_replay_limit_invalid"), Store::Contract(_) => permanent("observation_contract_invalid"), + // The lease's authority is behind the store's: no replay through this + // lease can commit, so the pass stops instead of spinning on it. + Store::Storage { source, .. } + if tracedecay_store::error_chain_writer_fence(source.as_ref()).is_some() => + { + ClaudeObservationFailureClass { + reason_code: "observation_writer_fenced", + retryable: false, + status: Unavailable, + } + } _ => unavailable("observation_storage_failed"), }; let projection = |error: &Projection| { @@ -510,6 +526,13 @@ pub fn classify_claude_observation_failure( status: Unavailable, }; } + if error.writer_fence().is_some() { + return ClaudeObservationFailureClass { + reason_code: "projection_store_writer_fenced", + retryable: false, + status: Unavailable, + }; + } match error { Projection::Corrupt { .. } => ClaudeObservationFailureClass { reason_code: "projection_store_corrupt", @@ -707,4 +730,35 @@ mod cancellation_tests { ); assert_eq!(disposition.status, HostAdmissionStatus::Degraded); } + + /// A store that fenced the ingest lease rejects every replay through it; + /// the pass must not keep retrying the same write. + #[test] + fn a_fenced_writer_is_not_retryable() { + let fence = tracedecay_store::StoreWriterFencedV1 { + requested: tracedecay_store::StoreAuthorityEpochV1::new(2).unwrap(), + current: tracedecay_store::StoreAuthorityEpochV1::new(3).unwrap(), + }; + let fenced = + source::TranscriptIngestError::Store(tracedecay_store::TranscriptStoreError::Storage { + operation: "persist transcript observations", + source: Box::new(fence), + }); + + let disposition = classify_transcript_ingest_disposition(&fenced); + + assert_eq!(disposition.reason_code, "transcript_writer_fenced"); + assert!(!disposition.retryable, "a fenced lease never commits again"); + assert_eq!(disposition.status, HostAdmissionStatus::Unavailable); + + let transient = + source::TranscriptIngestError::Store(tracedecay_store::TranscriptStoreError::Storage { + operation: "persist transcript observations", + source: Box::new(std::io::Error::other("database is locked")), + }); + assert!( + classify_transcript_ingest_disposition(&transient).retryable, + "ordinary storage failures stay retryable" + ); + } } diff --git a/crates/tracedecay-store/src/lib.rs b/crates/tracedecay-store/src/lib.rs index 1dd1f488b2..1c2cfffeb5 100644 --- a/crates/tracedecay-store/src/lib.rs +++ b/crates/tracedecay-store/src/lib.rs @@ -220,10 +220,11 @@ pub use runtime::{ StorageRuntimeErrorV1, StoreAuthorityEpochV1, StoreClientIdV1, StoreCommitReceiptV1, StoreIdempotencyKeyV1, StoreIncarnationV1, StoreOperationIdV1, StoreOperationMetadataV1, StoreRuntimeBindingV1, StoreRuntimeRegistryPublicationV1, StoreShardIdV1, StoreShardScopeV1, - StoreSnapshotIdV1, StoredObservationRowV1, UnavailableReasonV1, UserProfileId, - VerifiedStoreLocatorV1, WAL_HARD_LIMIT_BYTES, WAL_SOFT_LIMIT_BYTES, + StoreSnapshotIdV1, StoreWriterFencedV1, StoredObservationRowV1, UnavailableReasonV1, + UserProfileId, VerifiedStoreLocatorV1, WAL_HARD_LIMIT_BYTES, WAL_SOFT_LIMIT_BYTES, WORKSTATION_GLOBAL_QUEUE_BYTES, WalBudgetV1, WatermarkCoverageStatusV1, WorktreeId, - canonical_store_locator_digest, graph_store_locator_path, single_shard_required_coverage_v1, + canonical_store_locator_digest, error_chain_writer_fence, graph_store_locator_path, + single_shard_required_coverage_v1, }; pub use schema::{ CLEAR_CURRENT_DIAGNOSTIC_PUBLICATION_SQL, CLEAR_CURRENT_DIAGNOSTICS_SQL, diff --git a/crates/tracedecay-store/src/projection.rs b/crates/tracedecay-store/src/projection.rs index 02db331c26..aacfdfe6cb 100644 --- a/crates/tracedecay-store/src/projection.rs +++ b/crates/tracedecay-store/src/projection.rs @@ -665,6 +665,18 @@ impl ProjectionStoreError { } } + /// The store's writer fence when this failure is one: the lease's + /// authority epoch is behind the authority the store holds, so replaying + /// the request on that lease can never commit. + pub fn writer_fence(&self) -> Option { + match self { + Self::Storage { source, .. } => { + crate::runtime::error_chain_writer_fence(source.as_ref()) + } + _ => None, + } + } + pub fn durable_detail(&self) -> String { durable_error_detail(self) } diff --git a/crates/tracedecay-store/src/runtime/error.rs b/crates/tracedecay-store/src/runtime/error.rs index 83c1c69e7a..84d5830db4 100644 --- a/crates/tracedecay-store/src/runtime/error.rs +++ b/crates/tracedecay-store/src/runtime/error.rs @@ -2,9 +2,34 @@ use serde::{Deserialize, Serialize}; use thiserror::Error; use tracedecay_domain::ObservationSourceCursorV1; -use super::{DurabilityClassV1, StoreIncarnationV1}; +use super::{DurabilityClassV1, StoreAuthorityEpochV1, StoreIncarnationV1}; use crate::observation::CursorAdvanceLedgerDisagreementV1; +/// A writer whose authority epoch is behind the authority its store already +/// holds for the same shard and incarnation. The fence is permanent for that +/// writer: replaying the same request on the same lease can never commit, so +/// every retry loop must treat it as terminal and reopen the store instead. +#[derive(Clone, Copy, Debug, Error, PartialEq, Eq)] +#[error("store writer authority epoch {requested:?} is fenced by authority epoch {current:?}")] +pub struct StoreWriterFencedV1 { + pub requested: StoreAuthorityEpochV1, + pub current: StoreAuthorityEpochV1, +} + +/// The writer fence carried anywhere in `error`'s source chain. +pub fn error_chain_writer_fence( + error: &(dyn std::error::Error + 'static), +) -> Option { + let mut current: Option<&(dyn std::error::Error + 'static)> = Some(error); + while let Some(err) = current { + if let Some(fence) = err.downcast_ref::() { + return Some(*fence); + } + current = err.source(); + } + None +} + /// Validation failures for pure runtime contract DTOs. #[derive(Clone, Debug, Error, PartialEq, Eq)] pub enum StorageRuntimeContractErrorV1 { diff --git a/crates/tracedecay-store/src/runtime/mod.rs b/crates/tracedecay-store/src/runtime/mod.rs index a79fd358be..0ed0560236 100644 --- a/crates/tracedecay-store/src/runtime/mod.rs +++ b/crates/tracedecay-store/src/runtime/mod.rs @@ -26,7 +26,8 @@ pub use consistency::{ }; pub use error::{ CorruptionClassV1, RuntimeCancellationStageV1, SaturationScopeV1, - StorageRuntimeContractErrorV1, StorageRuntimeErrorV1, UnavailableReasonV1, + StorageRuntimeContractErrorV1, StorageRuntimeErrorV1, StoreWriterFencedV1, UnavailableReasonV1, + error_chain_writer_fence, }; pub use graph_publication::{ GraphCanonicalReplaySourceDigestV1, GraphDependencyGenerationClosureDigestV1, diff --git a/crates/tracedecay/src/daemon/profile_owner.rs b/crates/tracedecay/src/daemon/profile_owner.rs index f625653e85..26ea964257 100644 --- a/crates/tracedecay/src/daemon/profile_owner.rs +++ b/crates/tracedecay/src/daemon/profile_owner.rs @@ -40,7 +40,7 @@ use tracedecay_global_db::RegisteredGlobalDbLeaseV1; use tracedecay_mcp::handlers::{ SessionAuthorities, decode_primitive_request, hook_runtime, unknown_tool_error, }; -use tracedecay_mcp::server::join_hook_ingest_refresh; +use tracedecay_mcp::server::{hook_ingest_refresh_session, join_hook_ingest_refresh}; use tracedecay_mcp::tools::dispatch_ceiling::{tool_dispatch_budget, tool_dispatch_deadline_error}; use tracedecay_runtime_core::cancellation::CancellationToken; use tracedecay_runtime_core::native_runtime::{self, TimeoutError}; @@ -185,6 +185,7 @@ async fn profile_hook_runtime( profile_identity: &LocalProfileIdentityAuthorityV1, arguments: Map, ) -> Result { + let publication_session = hook_ingest_refresh_session(&arguments)?; let request = hook_runtime::decode_hook_runtime_request(&Value::Object(arguments))?; let global_db = Box::pin(store_administration.registered_profile_database()).await?; let user_session_db = @@ -208,7 +209,13 @@ async fn profile_hook_runtime( )) .await?; if let HookRuntimeResultV1::IngestTranscript(ingest) = &result { - join_hook_ingest_refresh(ingest.user_scope, None, Some(&refresh_wake)).await?; + join_hook_ingest_refresh( + ingest.user_scope, + publication_session.as_ref(), + None, + Some(&refresh_wake), + ) + .await?; } else { refresh_wake.wake(); } diff --git a/crates/tracedecay/src/mcp/server/graph_tool_owner.rs b/crates/tracedecay/src/mcp/server/graph_tool_owner.rs index 912e89af5a..be4bd8e18f 100644 --- a/crates/tracedecay/src/mcp/server/graph_tool_owner.rs +++ b/crates/tracedecay/src/mcp/server/graph_tool_owner.rs @@ -15,7 +15,7 @@ use tracedecay_daemon_service::{ }; use tracedecay_domain::errors::{Result, TraceDecayError}; use tracedecay_mcp::handlers::graph_tool::graph_tool_owner_stores; -use tracedecay_mcp::server::join_hook_ingest_refresh; +use tracedecay_mcp::server::{hook_ingest_refresh_session, join_hook_ingest_refresh}; use tracedecay_mcp::tools::binding::{BranchSensitivity, tool_branch_sensitivity}; use tracedecay_project::project::TraceDecay; use tracedecay_runtime_core::branch::BranchMemo; @@ -152,6 +152,12 @@ impl McpServer { &self, invocation: GraphToolInvocationV1, ) -> Result { + let publication_session = + if invocation.operation == ApplicationSurfaceOperation::HookRuntime { + hook_ingest_refresh_session(&invocation.arguments)? + } else { + None + }; let (cg, live_branch) = self.reopen_if_branch_drifted_memoized().await; self.probe_before_graph_read(invocation.operation, &cg, &live_branch); let server_stats = match invocation.operation { @@ -191,6 +197,7 @@ impl McpServer { // only once their refresh owner has published it. join_hook_ingest_refresh( ingest.user_scope, + publication_session.as_ref(), self.project_session_refresh_wake.as_deref(), self.user_session_refresh_wake.as_deref(), ) diff --git a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_refresh_cancel_test.rs b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_refresh_cancel_test.rs index 9785b6eb53..b9bd80e133 100644 --- a/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_refresh_cancel_test.rs +++ b/crates/tracedecay/tests/mcp_suite/mcp_handler_test/session_refresh_cancel_test.rs @@ -46,7 +46,7 @@ impl SessionTemporalRefreshWakePort for AcceptedIdleSessionRefreshWake { false } - fn wake_and_wait_until_idle( + fn wake_and_wait_for_projection( &self, _timeout: Duration, ) -> tracedecay_contracts::SessionTemporalRefreshWakeFuture<'_> { diff --git a/crates/tracedecay/tests/runtime_acceptance_suite/host_event_fixture_test.rs b/crates/tracedecay/tests/runtime_acceptance_suite/host_event_fixture_test.rs index fc73003cf5..dbc9b4e425 100644 --- a/crates/tracedecay/tests/runtime_acceptance_suite/host_event_fixture_test.rs +++ b/crates/tracedecay/tests/runtime_acceptance_suite/host_event_fixture_test.rs @@ -2004,7 +2004,8 @@ fn assert_legal_host_response(provider: &str, state: &str, expected: &Value, out // zero-finding notice as plain text. assert_eq!( expected["stdout"].as_str().unwrap(), - if provider == "kiro" { "{}" } else { "" } + if provider == "kiro" { "{}" } else { "" }, + "{provider}/{state} fixture baseline" ); let encoded = stdout .strip_prefix("TraceDecay feedback ready for authorized lookup: ") diff --git a/crates/tracedecay/tests/session_suite/session_runtime/temporal_refresh.rs b/crates/tracedecay/tests/session_suite/session_runtime/temporal_refresh.rs index aade4e52f4..9fb34b48f3 100644 --- a/crates/tracedecay/tests/session_suite/session_runtime/temporal_refresh.rs +++ b/crates/tracedecay/tests/session_suite/session_runtime/temporal_refresh.rs @@ -974,9 +974,15 @@ async fn failure_effects_use_typed_terminal_store_operations() { ); } +enum BlockingProjectorOutcome { + CompleteEmptyProjection, + Deferred, +} + struct BlockingProjector { started: Arc, release: Arc, + outcome: BlockingProjectorOutcome, } impl SessionTemporalRefreshProjector for BlockingProjector { @@ -987,10 +993,16 @@ impl SessionTemporalRefreshProjector for BlockingProjector { ) -> SessionTemporalRefreshProjectionFuture<'a> { let started = Arc::clone(&self.started); let release = Arc::clone(&self.release); + let outcome = &self.outcome; Box::pin(async move { started.notify_one(); release.notified().await; - Ok(empty_projection_effect(&recovery)) + match outcome { + BlockingProjectorOutcome::Deferred => Ok(SessionTemporalRefreshEffect::Deferred), + BlockingProjectorOutcome::CompleteEmptyProjection => { + Ok(empty_projection_effect(&recovery)) + } + } }) } } @@ -1011,6 +1023,7 @@ async fn stale_owner_cannot_persist_after_cancellation() { let projector = BlockingProjector { started: Arc::clone(&started), release: Arc::clone(&release), + outcome: BlockingProjectorOutcome::CompleteEmptyProjection, }; let pass = tokio::spawn({ let authority = Arc::clone(&authority); @@ -1081,7 +1094,7 @@ async fn hook_ingest_join_runs_the_bound_refresh_worker() { let wake = authority.ensure_profile(®istry).await; let before = registry.profile_pass_count(db.db_path()).await; - tracedecay_mcp::server::join_hook_ingest_refresh(false, Some(&wake), None) + tracedecay_mcp::server::join_hook_ingest_refresh(false, None, Some(&wake), None) .await .expect("hook ingest must join the refresh worker that owns the written store"); @@ -1142,6 +1155,7 @@ async fn project_retirement_cancels_and_awaits_an_inflight_projector() { Arc::new(BlockingProjector { started: Arc::clone(&started), release: Arc::new(tokio::sync::Notify::new()), + outcome: BlockingProjectorOutcome::CompleteEmptyProjection, }), SessionTemporalRefreshPolicy::default(), ); @@ -1445,9 +1459,30 @@ async fn profile_database_has_one_scheduler_and_equivalent_kicks_coalesce() { let authority = registered_test_database(&temp, "profile-coalescing", HostAdmissionScope::Profile).await; let db = authority.database(); - let registry = SessionTemporalRefreshSchedulerRegistry::default(); + let started = Arc::new(tokio::sync::Notify::new()); + let release = Arc::new(tokio::sync::Notify::new()); + let mut registry = SessionTemporalRefreshSchedulerRegistry::default(); + configure_scheduler( + &mut registry, + Arc::new(BlockingProjector { + started: Arc::clone(&started), + release: Arc::clone(&release), + outcome: BlockingProjectorOutcome::Deferred, + }), + SessionTemporalRefreshPolicy::default(), + ); + let store = tracedecay_session_temporal_store::SessionTemporalStore::new(db); + let started_refresh = store + .begin_or_join_session_refresh(request("session.profile.coalescing", 0)) + .await + .unwrap(); let first = authority.ensure_profile(®istry).await; + tokio::time::timeout(Duration::from_secs(1), started.notified()) + .await + .unwrap(); + assert_eq!(registry.profile_pass_count(db.db_path()).await, 1); + let second = authority.ensure_profile(®istry).await; for _ in 0..32 { second.wake(); @@ -1455,8 +1490,27 @@ async fn profile_database_has_one_scheduler_and_equivalent_kicks_coalesce() { assert!(first.same_route(&second)); assert_eq!(registry.profile_worker_count().await, 1); - registry.wait_profile_idle(db.db_path()).await; - assert!(registry.profile_pass_count(db.db_path()).await <= 2); + assert_eq!(registry.profile_pass_count(db.db_path()).await, 1); + release.notify_one(); + tokio::time::timeout(Duration::from_secs(1), started.notified()) + .await + .unwrap(); + assert_eq!(registry.profile_pass_count(db.db_path()).await, 2); + release.notify_one(); + tokio::time::timeout( + Duration::from_secs(1), + registry.wait_profile_idle(db.db_path()), + ) + .await + .unwrap(); + let pending = store.running_session_refreshes().await.unwrap(); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].session_id(), started_refresh.session_id()); + let pass_count = registry.profile_pass_count(db.db_path()).await; + assert!( + pass_count <= 2, + "profile scheduler used {pass_count} passes" + ); registry.shutdown().await; assert_eq!(registry.profile_worker_count().await, 0); } @@ -1490,7 +1544,10 @@ async fn project_rekey_retires_old_owner_before_rebinding_wake() { .rekey_project(®istry, &old_owner, new_owner.clone()) .await; // The replacement worker owns the dirty flag; join its actual pass. - assert!(wake.wake_and_wait_until_idle(Duration::from_secs(1)).await); + assert!( + wake.wake_and_wait_for_projection(Duration::from_secs(1)) + .await + ); assert!(old_state.is_cancelled()); assert!(registry.project_state(&old_owner).await.is_none()); diff --git a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs index 4dd2d7b146..28a1f8d0d5 100644 --- a/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs +++ b/crates/tracedecay/tests/session_suite/temporal_projection/activation.rs @@ -1,4 +1,12 @@ +use std::future::Future; +use std::sync::Arc; +use std::time::Duration; + use tracedecay_session_runtime::session_temporal_refresh_scheduler::SessionTemporalRefreshSchedulerRegistry; +use tracedecay_session_runtime::session_temporal_refresh_scheduler::history::{ + SessionHistoricalCapacityRelease, SessionHistoricalIngestOutcome, SessionHistoricalIngestPass, + SessionHistoricalIngestor, +}; use super::*; @@ -317,6 +325,30 @@ async fn activation_is_pinned_to_the_generation_the_refresh_extends() { ); } +#[derive(Default)] +struct GatedHistory { + entered: tokio::sync::Notify, + release: tokio::sync::Notify, +} + +impl SessionHistoricalIngestor for GatedHistory { + fn run_pass(&self) -> SessionHistoricalIngestPass<'_> { + Box::pin(async move { + self.entered.notify_one(); + self.release.notified().await; + SessionHistoricalIngestOutcome::Complete.into() + }) + } + + fn capacity_release(&self) -> SessionHistoricalCapacityRelease { + SessionHistoricalCapacityRelease::new(Vec::new()) + } + + fn cancel(&self) { + self.release.notify_one(); + } +} + #[tokio::test] async fn reset_requested_during_a_refresh_supersedes_its_candidate() { let tmp = TempDir::new().unwrap(); @@ -355,16 +387,35 @@ async fn reset_requested_during_a_refresh_supersedes_its_candidate() { assert_eq!(requested, 1, "the retired observation requests one reset"); let registry = SessionTemporalRefreshSchedulerRegistry::default(); - registry - .ensure_profile( + let history = Arc::new(GatedHistory::default()); + let wake = registry + .ensure_profile_with_history( path.clone(), runtime .registered_database_arc(HostAdmissionScope::Profile) .unwrap(), + history.clone(), ) .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) + ); + history.release.notify_one(); + assert_eq!( + tokio::join!(named_join, aggregate_join), + (false, false), + "the superseded operation must refuse both waiting publication joins" + ); registry.wait_profile_idle(&path).await; - registry.shutdown().await; assert_eq!( rows( @@ -412,6 +463,18 @@ async fn reset_requested_during_a_refresh_supersedes_its_candidate() { .await, vec![format!("message.temporal.{}.0", session_id.as_str())] ); + + assert!( + wake.wake_and_wait_for_session_projection(&session_id, Duration::from_secs(5)) + .await, + "a new join must observe the automatically rebuilt session publication" + ); + assert!( + wake.wake_and_wait_for_projection(Duration::from_secs(5)) + .await, + "a new aggregate join must observe the completed replacement" + ); + registry.shutdown().await; } #[tokio::test]