From 11c76cdf57878891ffcdb2fcfb3508106e002d5d Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Wed, 9 Sep 2026 03:13:15 +0000 Subject: [PATCH] refactor(session-refresh): call wake authority directly --- .../src/session/mod.rs | 4 +- .../src/session/refresh.rs | 19 +-- .../src/mcp/server/session_refresh.rs | 114 +++++++++++------- .../src/session_temporal_benchmark.rs | 15 +-- .../root_relation_fixture.rs | 4 +- .../temporal_refresh_application.rs | 45 +++---- 6 files changed, 102 insertions(+), 99 deletions(-) diff --git a/crates/tracedecay-session-memory/src/session/mod.rs b/crates/tracedecay-session-memory/src/session/mod.rs index 5bb8437884..ed091ec30c 100644 --- a/crates/tracedecay-session-memory/src/session/mod.rs +++ b/crates/tracedecay-session-memory/src/session/mod.rs @@ -14,8 +14,8 @@ pub use ports::{ }; pub use refresh::{ SessionRefreshConfiguration, SessionRefreshDigest, SessionRefreshHandle, SessionRefreshOutcome, - SessionRefreshRequestError, SessionRefreshSchedulerError, SessionRefreshSchedulerPort, - SessionRefreshService, SessionRefreshTarget, + SessionRefreshRequestError, SessionRefreshSchedulerError, SessionRefreshService, + SessionRefreshTarget, }; pub use refresh_service::{ SessionRefreshAction, SessionRefreshCommand, SessionRefreshCoverageView, diff --git a/crates/tracedecay-session-memory/src/session/refresh.rs b/crates/tracedecay-session-memory/src/session/refresh.rs index 07b184d8c8..0f4b1c83ab 100644 --- a/crates/tracedecay-session-memory/src/session/refresh.rs +++ b/crates/tracedecay-session-memory/src/session/refresh.rs @@ -199,19 +199,6 @@ impl fmt::Display for SessionRefreshSchedulerError { impl std::error::Error for SessionRefreshSchedulerError {} -pub trait SessionRefreshSchedulerPort { - fn wake(&self) -> Result<(), SessionRefreshSchedulerError>; -} - -impl SessionRefreshSchedulerPort for &T -where - T: SessionRefreshSchedulerPort + ?Sized, -{ - fn wake(&self) -> Result<(), SessionRefreshSchedulerError> { - (*self).wake() - } -} - #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum SessionRefreshRequestError { InvalidProjectorVersion, @@ -258,7 +245,7 @@ impl SessionRefreshService where A: SessionScopeAuthorizer, S: SessionRefreshStore, - W: SessionRefreshSchedulerPort, + W: Fn() -> Result<(), SessionRefreshSchedulerError>, { #[hotpath::measure(label = "usecases.session.refresh.begin", future = true)] pub async fn begin_or_join( @@ -336,7 +323,7 @@ where // The durable operation is authoritative once the store call returns. // Delivery failure must preserve that commit and require reconciliation // through the persisted recovery row rather than report plain acceptance. - match (receipt.disposition(), self.scheduler.wake()) { + match (receipt.disposition(), (self.scheduler)()) { (SessionRefreshDispositionV1::Started, Ok(())) => { SessionRefreshOutcome::Started(handle) } @@ -455,7 +442,7 @@ where .await { Ok(Ok(receipt)) => { - if self.scheduler.wake().is_err() { + if (self.scheduler)().is_err() { SessionRefreshOutcome::CancelledReconciliationRequired(receipt) } else { terminal_outcome(receipt) diff --git a/crates/tracedecay/src/mcp/server/session_refresh.rs b/crates/tracedecay/src/mcp/server/session_refresh.rs index f5cb45af3d..c4c0d693e5 100644 --- a/crates/tracedecay/src/mcp/server/session_refresh.rs +++ b/crates/tracedecay/src/mcp/server/session_refresh.rs @@ -6,9 +6,13 @@ use std::collections::HashMap; use std::future::Future; use std::pin::Pin; use std::sync::PoisonError; +#[cfg(test)] +use std::time::Duration; use sha2::{Digest, Sha256}; -use tracedecay_contracts::RequestContext; +#[cfg(test)] +use tracedecay_contracts::SessionTemporalRefreshWakeFuture; +use tracedecay_contracts::{RequestContext, SessionTemporalRefreshWakePort}; use tracedecay_domain::ProjectId; use tracedecay_global_db::RegisteredGlobalDbLeaseV1; @@ -17,9 +21,9 @@ use tracedecay_session_memory::session::{ SessionRefreshAction, SessionRefreshCommand, SessionRefreshConfiguration, SessionRefreshCoverageView, SessionRefreshFrontierView, SessionRefreshHandle, SessionRefreshOutcome, SessionRefreshProgressView, SessionRefreshReceiptView, - SessionRefreshSchedulerError, SessionRefreshSchedulerPort, SessionRefreshService, - SessionRefreshServiceOutcome, SessionRefreshServicePort, SessionRequestBinding, - SessionScopeAuthorizationRequest, SessionScopeAuthorizer, utc_micros_value, + SessionRefreshSchedulerError, SessionRefreshService, SessionRefreshServiceOutcome, + SessionRefreshServicePort, SessionRequestBinding, SessionScopeAuthorizationRequest, + SessionScopeAuthorizer, utc_micros_value, }; use tracedecay_session_temporal_store::GlobalDbSessionTemporalStore; @@ -51,23 +55,9 @@ impl SessionScopeAuthorizer for DaemonSessionRefreshAuthorizer<'_> { } } -#[derive(Clone)] -struct DaemonSessionRefreshWake( - std::sync::Arc, -); - -impl SessionRefreshSchedulerPort for DaemonSessionRefreshWake { - fn wake(&self) -> std::result::Result<(), SessionRefreshSchedulerError> { - self.0 - .wake() - .then_some(()) - .ok_or(SessionRefreshSchedulerError) - } -} - pub(crate) struct DaemonSessionRefreshService { database: RegisteredGlobalDbLeaseV1, - wake: DaemonSessionRefreshWake, + wake: std::sync::Arc, expected_project_id: Option, handles: std::sync::Mutex>, } @@ -81,40 +71,17 @@ enum SessionRefreshHandleLookup { impl DaemonSessionRefreshService { pub(crate) fn new( database: RegisteredGlobalDbLeaseV1, - wake: std::sync::Arc, + wake: std::sync::Arc, expected_project_id: Option, ) -> Self { Self { database, - wake: DaemonSessionRefreshWake(wake), + wake, expected_project_id, handles: std::sync::Mutex::new(HashMap::new()), } } - fn service( - &self, - ) -> Option< - SessionRefreshService< - DaemonSessionRefreshAuthorizer<'_>, - GlobalDbSessionTemporalStore<'_, tracedecay_global_db::RegisteredGlobalDb>, - &DaemonSessionRefreshWake, - >, - > { - Some(SessionRefreshService::new( - DaemonSessionRefreshAuthorizer { - expected_project_id: self.expected_project_id.as_deref(), - }, - GlobalDbSessionTemporalStore::new(self.database.as_ref()), - &self.wake, - SessionRefreshConfiguration::new( - SESSION_REFRESH_PROJECTOR_VERSION, - SESSION_REFRESH_CONFIG_VERSION, - ) - .ok()?, - )) - } - fn handle(&self, token: &str) -> SessionRefreshHandleLookup { if !is_session_refresh_handle_token(token) { return missing_session_refresh_handle_lookup(token); @@ -154,9 +121,20 @@ impl DaemonSessionRefreshService { &self, command: SessionRefreshCommand, ) -> SessionRefreshServiceOutcome { - let Some(service) = self.service() else { + let Ok(configuration) = SessionRefreshConfiguration::new( + SESSION_REFRESH_PROJECTOR_VERSION, + SESSION_REFRESH_CONFIG_VERSION, + ) else { return SessionRefreshServiceOutcome::Unavailable; }; + let service = SessionRefreshService::new( + DaemonSessionRefreshAuthorizer { + expected_project_id: self.expected_project_id.as_deref(), + }, + GlobalDbSessionTemporalStore::new(self.database.as_ref()), + || wake_session_refresh_scheduler(self.wake.as_ref()), + configuration, + ); let outcome = match command.action { SessionRefreshAction::Begin => { service @@ -260,6 +238,14 @@ impl DaemonSessionRefreshService { } } +fn wake_session_refresh_scheduler( + wake: &dyn SessionTemporalRefreshWakePort, +) -> std::result::Result<(), SessionRefreshSchedulerError> { + wake.wake() + .then_some(()) + .ok_or(SessionRefreshSchedulerError) +} + fn is_session_refresh_handle_token(token: &str) -> bool { token.strip_prefix("srh_").is_some_and(|digest| { digest.len() == 64 && digest.bytes().all(|byte| byte.is_ascii_hexdigit()) @@ -274,6 +260,44 @@ fn missing_session_refresh_handle_lookup(token: &str) -> SessionRefreshHandleLoo } } +#[cfg(test)] +#[derive(Clone, Copy)] +struct FixedSessionTemporalRefreshWake(bool); + +#[cfg(test)] +impl SessionTemporalRefreshWakePort for FixedSessionTemporalRefreshWake { + fn wake(&self) -> bool { + self.0 + } + + fn is_unavailable(&self) -> bool { + !self.0 + } + + fn wake_and_wait_until_idle(&self, _timeout: Duration) -> SessionTemporalRefreshWakeFuture<'_> { + let accepted = self.0; + Box::pin(async move { accepted }) + } +} + +#[cfg(test)] +#[test] +fn accepted_session_refresh_wake_maps_to_typed_success() { + assert_eq!( + wake_session_refresh_scheduler(&FixedSessionTemporalRefreshWake(true)), + Ok(()) + ); +} + +#[cfg(test)] +#[test] +fn refused_session_refresh_wake_maps_to_scheduler_error() { + assert_eq!( + wake_session_refresh_scheduler(&FixedSessionTemporalRefreshWake(false)), + Err(SessionRefreshSchedulerError) + ); +} + #[cfg(test)] #[test] fn session_refresh_handle_tokens_are_closed_and_non_leaking() { diff --git a/crates/tracedecay/src/session_temporal_benchmark.rs b/crates/tracedecay/src/session_temporal_benchmark.rs index 5960f73772..64ef685745 100644 --- a/crates/tracedecay/src/session_temporal_benchmark.rs +++ b/crates/tracedecay/src/session_temporal_benchmark.rs @@ -42,9 +42,9 @@ use tracedecay_session_memory::context::{ }; use tracedecay_session_memory::session::{ AuthorizationGrantId, SessionAuthorizationError, SessionAuthorizationGrant, - SessionRefreshSchedulerError, SessionRefreshSchedulerPort, SessionRequestBinding, - SessionRetrievalConfiguration, SessionRetrievalOutcome, SessionRetrievalService, - SessionScopeAuthorizationRequest, SessionScopeAuthorizer, SessionTemporalQuery, + SessionRequestBinding, SessionRetrievalConfiguration, SessionRetrievalOutcome, + SessionRetrievalService, SessionScopeAuthorizationRequest, SessionScopeAuthorizer, + SessionTemporalQuery, }; use tracedecay_session_temporal_store::RegisteredGlobalDbSessionTemporalExecution; use tracedecay_sessions::observation::ObservationCancellation; @@ -257,15 +257,6 @@ impl SessionScopeAuthorizer for AllowAuthorizer { } } -#[derive(Clone, Copy, Default)] -struct NoopWake; - -impl SessionRefreshSchedulerPort for NoopWake { - fn wake(&self) -> Result<(), SessionRefreshSchedulerError> { - Ok(()) - } -} - struct Words(&'static str); impl VersionedTokenEstimator for Words { diff --git a/crates/tracedecay/src/session_temporal_benchmark/root_relation_fixture.rs b/crates/tracedecay/src/session_temporal_benchmark/root_relation_fixture.rs index 66741d2cd7..55fdf9fb3b 100644 --- a/crates/tracedecay/src/session_temporal_benchmark/root_relation_fixture.rs +++ b/crates/tracedecay/src/session_temporal_benchmark/root_relation_fixture.rs @@ -17,7 +17,7 @@ use tracedecay_temporal_query::context::ContextBudget; use tracedecay_temporal_query::ports::ExecutionControl; use tracedecay_temporal_query::ranking::DiversityLimits; -use super::{AllowAuthorizer, BenchResult, CONFIG_VERSION, NoopWake, PROJECTOR_VERSION}; +use super::{AllowAuthorizer, BenchResult, CONFIG_VERSION, PROJECTOR_VERSION}; use tracedecay_session_temporal_store::GlobalDbSessionTemporalStore; pub(super) const ROOT_RELATION_PARTICIPANT_COUNT: usize = 64; @@ -57,7 +57,7 @@ pub(super) async fn refresh_sessions( let refresh = SessionRefreshService::new( AllowAuthorizer, GlobalDbSessionTemporalStore::new(db), - NoopWake, + || Ok(()), SessionRefreshConfiguration::new(PROJECTOR_VERSION, CONFIG_VERSION) .map_err(|error| format!("root refresh configuration: {error}"))?, ); diff --git a/crates/tracedecay/tests/session_suite/temporal_refresh_application.rs b/crates/tracedecay/tests/session_suite/temporal_refresh_application.rs index 9825fca2de..dfa17f36a1 100644 --- a/crates/tracedecay/tests/session_suite/temporal_refresh_application.rs +++ b/crates/tracedecay/tests/session_suite/temporal_refresh_application.rs @@ -20,9 +20,8 @@ use tracedecay_session_memory::context::{ use tracedecay_session_memory::session::{ AuthorizationGrantId, SessionAuthorizationError, SessionAuthorizationGrant, SessionRefreshConfiguration, SessionRefreshHandle, SessionRefreshOutcome, - SessionRefreshSchedulerError, SessionRefreshSchedulerPort, SessionRefreshService, - SessionRefreshTarget, SessionRequestBinding, SessionScopeAuthorizationRequest, - SessionScopeAuthorizer, + SessionRefreshSchedulerError, SessionRefreshService, SessionRefreshTarget, + SessionRequestBinding, SessionScopeAuthorizationRequest, SessionScopeAuthorizer, }; use tracedecay_session_temporal_store::GlobalDbSessionTemporalStore; use tracedecay_store::{ @@ -95,9 +94,7 @@ impl RecordingWake { fn calls(&self) -> usize { self.calls.load(Ordering::Acquire) } -} -impl SessionRefreshSchedulerPort for RecordingWake { fn wake(&self) -> Result<(), SessionRefreshSchedulerError> { self.calls.fetch_add(1, Ordering::AcqRel); if self.fail.load(Ordering::Acquire) { @@ -108,6 +105,10 @@ impl SessionRefreshSchedulerPort for RecordingWake { } } +fn recording_wake(wake: RecordingWake) -> impl Fn() -> Result<(), SessionRefreshSchedulerError> { + move || wake.wake() +} + fn configuration() -> SessionRefreshConfiguration { SessionRefreshConfiguration::new(PROJECTOR_VERSION, CONFIG_VERSION).unwrap() } @@ -405,7 +406,7 @@ async fn equivalent_requests_join_with_stable_digests_excluding_request_id() { let service = SessionRefreshService::new( AllowAuthorizer, session_temporal_store(&db), - wake.clone(), + recording_wake(wake.clone()), configuration(), ); let first_context = project_context( @@ -459,7 +460,7 @@ async fn query_only_mode_and_grain_share_one_projection_refresh() { let service = SessionRefreshService::new( AllowAuthorizer, session_temporal_store(&db), - wake.clone(), + recording_wake(wake.clone()), configuration(), ); let context = project_context( @@ -543,7 +544,7 @@ async fn conflicting_target_is_busy_and_does_not_wake_scheduler() { let service = SessionRefreshService::new( AllowAuthorizer, session_temporal_store(&db), - wake.clone(), + recording_wake(wake.clone()), configuration(), ); let context = project_context( @@ -593,7 +594,7 @@ async fn wake_failure_leaves_recoverable_operation_that_joins_after_restart() { let first = match SessionRefreshService::new( AllowAuthorizer, session_temporal_store(&db), - failing_wake.clone(), + recording_wake(failing_wake.clone()), configuration(), ) .begin_or_join(&context, context.binding(), target.clone()) @@ -615,7 +616,7 @@ async fn wake_failure_leaves_recoverable_operation_that_joins_after_restart() { let restarted = SessionRefreshService::new( AllowAuthorizer, session_temporal_store(&db), - healthy_wake.clone(), + recording_wake(healthy_wake.clone()), configuration(), ); let joined = handle( @@ -646,13 +647,13 @@ async fn status_and_cancel_reauthorize_and_preserve_terminal_coverage() { let denied = SessionRefreshService::new( DenyAuthorizer, session_temporal_store(&db), - wake.clone(), + recording_wake(wake.clone()), configuration(), ); let allowed = SessionRefreshService::new( AllowAuthorizer, session_temporal_store(&db), - wake.clone(), + recording_wake(wake.clone()), configuration(), ); let target = target("session.refresh.cancel", 0); @@ -716,13 +717,13 @@ async fn project_and_profile_scopes_are_isolated_without_root_fallback() { let project_service = SessionRefreshService::new( AllowAuthorizer, session_temporal_store(&project_db), - RecordingWake::default(), + recording_wake(RecordingWake::default()), configuration(), ); let profile_service = SessionRefreshService::new( AllowAuthorizer, session_temporal_store(&profile_db), - RecordingWake::default(), + recording_wake(RecordingWake::default()), configuration(), ); let target = target("session.refresh.scope", 0); @@ -771,7 +772,7 @@ async fn status_maps_complete_and_failed_receipts_without_error_details() { let complete_service = SessionRefreshService::new( AllowAuthorizer, session_temporal_store(&complete_db), - RecordingWake::default(), + recording_wake(RecordingWake::default()), configuration(), ); let complete_target = target("session.refresh.complete", 0); @@ -807,7 +808,7 @@ async fn status_maps_complete_and_failed_receipts_without_error_details() { let failed_service = SessionRefreshService::new( AllowAuthorizer, session_temporal_store(&failed_db), - RecordingWake::default(), + recording_wake(RecordingWake::default()), configuration(), ); let failed_target = target("session.refresh.failed", 0); @@ -848,7 +849,7 @@ async fn concurrent_callers_share_one_operation_and_keep_caller_idempotency() { let service = SessionRefreshService::new( AllowAuthorizer, session_temporal_store(&db), - wake, + recording_wake(wake), configuration(), ); let first_context = project_context( @@ -908,7 +909,7 @@ async fn cancel_before_first_progress_returns_durable_zero_coverage_receipt() { let service = SessionRefreshService::new( AllowAuthorizer, session_temporal_store(&db), - wake.clone(), + recording_wake(wake.clone()), configuration(), ); let context = project_context( @@ -960,7 +961,7 @@ async fn application_preserves_each_temporal_mode_in_terminal_source_coverage() let service = SessionRefreshService::new( AllowAuthorizer, session_temporal_store(&db), - RecordingWake::default(), + recording_wake(RecordingWake::default()), configuration(), ); let context = project_context( @@ -1002,7 +1003,7 @@ async fn expired_or_cancelled_requests_do_not_create_refresh_operations() { let service = SessionRefreshService::new( AllowAuthorizer, session_temporal_store(&db), - wake.clone(), + recording_wake(wake.clone()), configuration(), ); let template = project_context( @@ -1079,7 +1080,7 @@ async fn request_abort_and_deadline_do_not_claim_durable_operation_cancellation( let service = SessionRefreshService::new( AllowAuthorizer, session_temporal_store(&db), - wake, + recording_wake(wake), configuration(), ); let context = project_context( @@ -1143,7 +1144,7 @@ async fn status_is_read_only_and_does_not_wake_the_daemon() { let service = SessionRefreshService::new( AllowAuthorizer, session_temporal_store(&db), - wake.clone(), + recording_wake(wake.clone()), configuration(), ); let context = project_context(