From 15971cf0371bfc503025c064db008b7577c0719f Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 20:42:57 +0000 Subject: [PATCH 01/10] test(daemon): drive replay tests by signals, not wall clocks Refs #4149 --- .../src/profile_host_admission_replay.rs | 158 +++++++++++------- 1 file changed, 97 insertions(+), 61 deletions(-) diff --git a/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs b/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs index 84045634b1..6b724e0866 100644 --- a/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs +++ b/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs @@ -984,8 +984,8 @@ mod tests { let profile_root = temp.path().join("profile"); std::fs::create_dir_all(&profile_root).unwrap(); let registry = ProfileHostAdmissionReplayRegistry::with_bootstrap_cache_for( - Duration::from_millis(20), - Duration::from_millis(20), + Duration::ZERO, + Duration::ZERO, ); let attempts = Arc::new(AtomicUsize::new(0)); let operation_attempts = Arc::clone(&attempts); @@ -1006,7 +1006,6 @@ mod tests { .await, BootstrapCompletion::Completed, ); - tokio::time::sleep(Duration::from_millis(30)).await; registry.ensure_bootstrap(&profile_root, operation).await; assert_eq!( registry @@ -1024,9 +1023,9 @@ mod tests { let temp = tempfile::TempDir::new().unwrap(); let profile_root = temp.path().join("profile"); std::fs::create_dir_all(&profile_root).unwrap(); - let registry = ProfileHostAdmissionReplayRegistry::with_bootstrap_cache_for( - Duration::from_secs(1), - Duration::from_millis(20), + let mut registry = ProfileHostAdmissionReplayRegistry::with_bootstrap_cache_for( + Duration::MAX, + Duration::MAX, ); let attempts = Arc::new(AtomicUsize::new(0)); let operation_attempts = Arc::clone(&attempts); @@ -1056,7 +1055,7 @@ mod tests { .await; assert_eq!(attempts.load(Ordering::Acquire), 1); - tokio::time::sleep(Duration::from_millis(30)).await; + registry.bootstrap_terminal_cache_for = Duration::ZERO; registry.ensure_bootstrap(&profile_root, operation).await; assert_eq!( registry @@ -1240,7 +1239,17 @@ mod tests { ); let observed = attempts.load(Ordering::Acquire); assert!(observed >= 2, "the budget must allow real retries first"); - tokio::time::sleep(Duration::from_millis(120)).await; + while !registry + .bootstrap_workers + .lock() + .await + .get(&bootstrap_key(&profile_root)) + .expect("bootstrap worker") + .task + .is_finished() + { + tokio::task::yield_now().await; + } assert_eq!( attempts.load(Ordering::Acquire), observed, @@ -1318,10 +1327,7 @@ mod tests { }); } while kick_tasks.join_next().await.is_some() {} - assert!( - registry.wait_idle(&db_path, Duration::from_secs(2)).await, - "coalesced worker must become idle" - ); + settled(®istry, &db_path).await; let observed = registry.pass_count(&db_path).await; assert!( observed <= 3, @@ -1353,10 +1359,7 @@ mod tests { )); registry.ensure(&db_path, &profile_root, &broker).await; - assert!( - registry.wait_idle(&db_path, Duration::from_secs(2)).await, - "installed pass must complete a replay worker" - ); + settled(®istry, &db_path).await; assert_eq!(passes.load(Ordering::Acquire), 1); registry.shutdown().await; } @@ -1394,10 +1397,7 @@ mod tests { registry .ensure_with_pass_override(&db_path, &profile_root, &broker, pass_override) .await; - assert!( - registry.wait_idle(&db_path, Duration::from_secs(2)).await, - "retryable worker must become idle after success" - ); + settled(®istry, &db_path).await; let elapsed = started.elapsed(); assert!( registry.backoff_count(&db_path).await >= 2, @@ -1429,14 +1429,19 @@ mod tests { registry .ensure_with_pass_override(&db_path, &profile_root, &broker, pass_override) .await; - tokio::time::timeout(Duration::from_secs(1), async { - while registry.backoff_count(&db_path).await < 2 { - tokio::task::yield_now().await; - } - }) - .await - .expect("pending no-progress replay must back off instead of spinning"); - assert!(registry.pass_count(&db_path).await <= 3); + let worker = worker(®istry, &db_path).await; + while worker.backoff_count.load(Ordering::Acquire) < 2 + && worker.pass_count.load(Ordering::Acquire) <= 3 + { + tokio::task::yield_now().await; + } + let passes = worker.pass_count.load(Ordering::Acquire); + let backoffs = worker.backoff_count.load(Ordering::Acquire); + assert!( + backoffs >= 2 && passes <= backoffs + 1, + "pending no-progress replay must back off before every retry instead of spinning; \ + passes={passes} backoffs={backoffs}" + ); registry.shutdown().await; } @@ -1451,29 +1456,46 @@ mod tests { let broker = Arc::new(tracedecay_host_admission::HostAdmissionBroker::new(runtime)); broker.admit("test:pending", b"pending").await.unwrap(); let registry = ProfileHostAdmissionReplayRegistry::default(); - let pass_override = Arc::new(|| { - Box::pin(async { HostAdmissionOutcome::not_applicable("code_index_not_applicable") }) - as std::pin::Pin + Send>> - }); + let started = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let pass_override: ReplayPassOverride = { + let started = Arc::clone(&started); + let release = Arc::clone(&release); + Arc::new(move || { + let started = Arc::clone(&started); + let release = Arc::clone(&release); + Box::pin(async move { + started.notify_one(); + release.notified().await; + HostAdmissionOutcome::not_applicable("code_index_not_applicable") + }) + }) + }; registry .ensure_with_pass_override(&db_path, &profile_root, &broker, pass_override) .await; - tokio::time::timeout(Duration::from_secs(1), async { - while registry.pass_count(&db_path).await == 0 { - tokio::task::yield_now().await; - } - }) - .await - .expect("profile replay must attempt the not-applicable record"); - tokio::time::sleep(Duration::from_millis(150)).await; - let passes = registry.pass_count(&db_path).await; - assert!(passes >= 1); - assert_eq!(registry.backoff_count(&db_path).await, 0); - tokio::time::sleep(Duration::from_millis(150)).await; - - assert_eq!(registry.pass_count(&db_path).await, passes); - assert_eq!(registry.backoff_count(&db_path).await, 0); + started.notified().await; + // The admission and the ensure each raised a replay request before + // this pass began. One left pending wakes the idle worker into a + // second pass over the same not-applicable record. + let stale_request = std::pin::pin!(broker.wait_for_replay_request()) + .poll(&mut std::task::Context::from_waker(std::task::Waker::noop())) + .is_ready(); + assert!( + !stale_request, + "a running pass must answer every replay request raised before it began" + ); + release.notify_one(); + + let worker = worker(®istry, &db_path).await; + while worker.busy.load(Ordering::Acquire) + && worker.backoff_count.load(Ordering::Acquire) == 0 + { + tokio::task::yield_now().await; + } + assert_eq!(worker.backoff_count.load(Ordering::Acquire), 0); + assert_eq!(worker.pass_count.load(Ordering::Acquire), 1); registry.shutdown().await; } @@ -1602,20 +1624,15 @@ mod tests { ); probe_started.notified().await; - let started = Instant::now(); + // The probe never answers and nothing cancels before shutdown, so + // only the caller's grace can end this wait. assert!( !registry .wait_idle(&db_path, Duration::from_millis(20)) .await, "blocked pending probe must respect the caller's replay grace" ); - assert!( - started.elapsed() < Duration::from_millis(250), - "blocked pending probe exceeded its bounded grace" - ); - tokio::time::timeout(Duration::from_secs(1), registry.shutdown()) - .await - .expect("shutdown must cancel and join a blocked pending probe"); + registry.shutdown().await; assert_eq!(registry.worker_count().await, 0); } @@ -1774,13 +1791,32 @@ mod tests { registry .ensure_with_pass_override(&db_path, &profile_root, &broker, pass_override) .await; - tokio::time::timeout(Duration::from_secs(1), async { - while registry.worker_count().await != 0 { - tokio::task::yield_now().await; + while registry.worker_count().await != 0 { + tokio::task::yield_now().await; + } + } + + async fn worker( + registry: &ProfileHostAdmissionReplayRegistry, + broker_path: &Path, + ) -> Arc { + let workers = registry.workers.lock().await; + Arc::clone(&workers.get(broker_path).expect("replay worker").worker) + } + + /// `wait_idle` without its grace: the worker's idle transition is the + /// signal, and a worker that never settles fails at the harness timeout. + async fn settled(registry: &ProfileHostAdmissionReplayRegistry, broker_path: &Path) { + let worker = worker(registry, broker_path).await; + loop { + let idle = worker.idle.notified(); + tokio::pin!(idle); + idle.as_mut().enable(); + if worker.is_idle().await { + return; } - }) - .await - .expect("idle replay worker must be evicted"); + idle.await; + } } fn open_test_broker(profile_root: &Path) -> (PathBuf, SharedHostAdmissionBroker) { From 84e58aa8f710cc962d73154a04ec2878ae4a5da9 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 20:42:59 +0000 Subject: [PATCH 02/10] fix(host-admission): answer replay requests when a pass starts The profile replay worker kept two wake signals for one kick (its own Notify permit and the broker's replay request) and cleared neither when a pass began. A pass that served both left them pending, so the idle worker ran one or two redundant passes over the same spool. Refs #4149 --- .../src/profile_host_admission_replay.rs | 6 ++---- crates/tracedecay-host-admission/src/lib.rs | 8 ++++++++ 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs b/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs index 6b724e0866..9a0bc60bbd 100644 --- a/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs +++ b/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs @@ -166,7 +166,6 @@ struct ProfileHostAdmissionReplayWorker { pass_count: AtomicUsize, backoff_count: AtomicUsize, idle: Notify, - wake: Notify, cancellation: Arc, replay_pass: Option, #[cfg(test)] @@ -750,7 +749,6 @@ impl ProfileHostAdmissionReplayWorker { pass_count: AtomicUsize::new(0), backoff_count: AtomicUsize::new(0), idle: Notify::new(), - wake: Notify::new(), cancellation, replay_pass, #[cfg(test)] @@ -762,7 +760,7 @@ impl ProfileHostAdmissionReplayWorker { fn kick(&self) { self.dirty.store(true, Ordering::Release); - self.wake.notify_one(); + self.broker.request_replay(); } async fn is_idle(&self) -> bool { @@ -826,6 +824,7 @@ impl ProfileHostAdmissionReplayWorker { } self.busy.store(true, Ordering::Release); let _ = self.dirty.swap(false, Ordering::AcqRel); + self.broker.clear_replay_request(); self.pass_count.fetch_add(1, Ordering::AcqRel); let Some(pending_before) = self.pending_replay_count_or_cancelled().await else { self.mark_idle(); @@ -889,7 +888,6 @@ impl ProfileHostAdmissionReplayWorker { self.mark_idle(); tokio::select! { () = self.wait_for_cancellation() => return, - () = self.wake.notified() => {} () = self.broker.wait_for_replay_request() => {} slept = native_runtime::sleep_until(Instant::now() + idle_eviction_after) => { // Without a timer the idle wait cannot be bounded; exit so diff --git a/crates/tracedecay-host-admission/src/lib.rs b/crates/tracedecay-host-admission/src/lib.rs index 9a494e3421..497800c23d 100644 --- a/crates/tracedecay-host-admission/src/lib.rs +++ b/crates/tracedecay-host-admission/src/lib.rs @@ -174,6 +174,14 @@ impl HostAdmissionBroker { self.replay_wake.notify_waiters(); } + /// Consume the pending replay request, if any. A worker calls this as a + /// pass starts: the pass reads the spool after this point, so it answers + /// every request raised before it, and a stale request would only buy a + /// redundant pass. + pub fn clear_replay_request(&self) { + self.replay_requested.store(false, Ordering::Release); + } + pub async fn wait_for_replay_request(&self) { self.replay_wake .wait_until(|| self.replay_requested.swap(false, Ordering::AcqRel)) From a30d9e390ae6fff25195639693e636e98bb7cd33 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 20:43:00 +0000 Subject: [PATCH 03/10] test(daemon): drop prepare guard with the future, not first poll Refs #4149 --- .../src/shutdown/orchestration.rs | 28 ++++++------------- 1 file changed, 9 insertions(+), 19 deletions(-) diff --git a/crates/tracedecay-daemon-service/src/shutdown/orchestration.rs b/crates/tracedecay-daemon-service/src/shutdown/orchestration.rs index b0d35f6ffa..47ef1272d8 100644 --- a/crates/tracedecay-daemon-service/src/shutdown/orchestration.rs +++ b/crates/tracedecay-daemon-service/src/shutdown/orchestration.rs @@ -1277,29 +1277,19 @@ mod tests { let lifecycle = DaemonLifecycle::default(); let prepare_lock = Arc::new(tokio::sync::Mutex::new(())); let held_lock = prepare_lock.lock().await; - let prepare_started = Arc::new(tokio::sync::Notify::new()); let dropped = Arc::new(std::sync::atomic::AtomicBool::new(false)); let deadline = Instant::now() + Duration::from_millis(200); - let first_lifecycle = lifecycle.clone(); + // The deadline may pass before the runner first polls the prepare, so + // the guard lives in the prepare future itself, not in its first poll. + let prepare_guard = Dropped(Arc::clone(&dropped)); let first_lock = Arc::clone(&prepare_lock); - let first_started = Arc::clone(&prepare_started); - let first_dropped = Arc::clone(&dropped); - let first = tokio::spawn(async move { - coordinate_daemon_shutdown(&first_lifecycle, deadline, async move { - let _dropped = Dropped(first_dropped); - first_started.notify_one(); - let _lock = first_lock.lock().await; - DaemonShutdownPlan::new(Vec::new(), |_| async { ShutdownTaskReceipt::default() }) - }) - .await - }); - prepare_started.notified().await; - - let receipt = tokio::time::timeout(Duration::from_secs(10), first) - .await - .expect("the shutdown deadline bounds a contended prepare") - .expect("shutdown prepare timeout receipt"); + let receipt = coordinate_daemon_shutdown(&lifecycle, deadline, async move { + let _dropped = prepare_guard; + let _lock = first_lock.lock().await; + DaemonShutdownPlan::new(Vec::new(), |_| async { ShutdownTaskReceipt::default() }) + }) + .await; assert!( Instant::now() >= deadline, "the prepare is cut off at the shutdown deadline, not before it" From 159c22be155e17715bc7644079c7b5574aa4178e Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 20:52:53 +0000 Subject: [PATCH 04/10] test(daemon): hold the TLS client's reply until the first read Refs #4149 --- .../src/remote_http_transport.rs | 22 +++++++++++++++---- 1 file changed, 18 insertions(+), 4 deletions(-) diff --git a/crates/tracedecay-daemon-service/src/remote_http_transport.rs b/crates/tracedecay-daemon-service/src/remote_http_transport.rs index eb95fa81b6..9ad88d4ec5 100644 --- a/crates/tracedecay-daemon-service/src/remote_http_transport.rs +++ b/crates/tracedecay-daemon-service/src/remote_http_transport.rs @@ -1140,8 +1140,15 @@ mod tests { /// Completes the handshake, sends `REQUEST` in the same flight as the /// client Finished, then holds the connection open until released, so the /// server sees no readiness edge after that flight. Reports on `sent` once - /// the `ClientHello` and once the final flight are on the wire. - fn hold_open_client(address: SocketAddr, sent: mpsc::Sender<()>, release: mpsc::Receiver<()>) { + /// the `ClientHello` and once the final flight are on the wire. Answers + /// the server's flight only once `answer` fires, so the listener's first + /// read cannot finish the handshake however its thread is scheduled. + fn hold_open_client( + address: SocketAddr, + sent: mpsc::Sender<()>, + answer: mpsc::Receiver<()>, + release: mpsc::Receiver<()>, + ) { let mut roots = rustls::RootCertStore::empty(); roots .add(CertificateDer::from_pem_slice(TRUST_ROOT).unwrap()) @@ -1164,6 +1171,7 @@ mod tests { tls.write_tls(&mut socket).unwrap(); } let _ = sent.send(()); + let _ = answer.recv(); while tls.is_handshaking() || tls.wants_write() { tls.complete_io(&mut socket).unwrap(); } @@ -1177,8 +1185,10 @@ mod tests { let listener = tls_listener(directory.path()); let address = listener.bound_addr().unwrap(); let (sent, _) = mpsc::channel(); + let (answer, await_answer) = mpsc::channel(); let (release, held) = mpsc::channel(); - let client = std::thread::spawn(move || hold_open_client(address, sent, held)); + let client = + std::thread::spawn(move || hold_open_client(address, sent, await_answer, held)); let (mut io, _) = native_runtime::block_on(listener.accept()) .unwrap() .expect("accepted connection"); @@ -1194,6 +1204,7 @@ mod tests { .is_pending(), "the reader parks on the handshake" ); + answer.send(()).unwrap(); let mut seen = 0; while matches!(io.transport, RemoteBrainTlsTransport::Handshaking(_)) { seen = waker.wait_past(seen); @@ -1229,8 +1240,10 @@ mod tests { let listener = tls_listener(directory.path()).with_deadlines(deadlines); let address = listener.bound_addr().unwrap(); let (sent, client_sent) = mpsc::channel(); + let (answer, await_answer) = mpsc::channel(); let (release, held) = mpsc::channel(); - let client = std::thread::spawn(move || hold_open_client(address, sent, held)); + let client = + std::thread::spawn(move || hold_open_client(address, sent, await_answer, held)); let (mut io, _) = native_runtime::block_on(listener.accept()) .unwrap() .expect("accepted connection"); @@ -1251,6 +1264,7 @@ mod tests { first_read.is_pending(), "reading the waiting ClientHello restarts both read deadlines: {first_read:?}" ); + answer.send(()).unwrap(); client_sent.recv().expect("client sends its final flight"); poll_late(deadlines.request_read * 3); let read = Pin::new(&mut io).poll_read(&mut context, &mut buffer); From 3c411cfce6aa9d1ced439c9c1ec4797663523b0b Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Fri, 16 Oct 2026 20:46:00 +0000 Subject: [PATCH 05/10] fix(daemon): wait for restarted refresh worker idle, not a 2 s wall retained_begin_and_join_report_partial_effect_and_restart_recovers_same_operation bounded the restarted scheduler's recovery pass with a 2 s wake wait. Under CPU starvation the pass was still running (backlog 1, no blocker) at 2 s and settled at 4.4-5.4 s, so the test failed 27 of 40 in the loaded recipe. The test now waits on the worker's idle notification through a new wait_project_idle test helper, which mirrors wait_profile_idle. The receipt read still proves the restart settled the persisted operation. --- .../session_retained_effect_tests.rs | 12 ++++----- .../registry.rs | 27 +++++++++++++------ 2 files changed, 24 insertions(+), 15 deletions(-) diff --git a/crates/tracedecay-daemon-service/src/retained_owner/session_retained_effect_tests.rs b/crates/tracedecay-daemon-service/src/retained_owner/session_retained_effect_tests.rs index d90964eebf..9445ee8f4b 100644 --- a/crates/tracedecay-daemon-service/src/retained_owner/session_retained_effect_tests.rs +++ b/crates/tracedecay-daemon-service/src/retained_owner/session_retained_effect_tests.rs @@ -1,6 +1,5 @@ use std::collections::BTreeSet; use std::sync::Arc; -use std::time::Duration; use tempfile::TempDir; use tracedecay_contracts::retained_surfaces::{ @@ -320,13 +319,12 @@ async fn reopen_and_settle( .registered_database_arc(HostAdmissionScope::Project) .expect("reopened project database"); let registry = SessionTemporalRefreshSchedulerRegistry::default(); - let wake = registry - .ensure_project(owner_for(&database, project_id), database.clone()) + let owner = owner_for(&database, project_id); + registry + .ensure_project(owner.clone(), database.clone()) .await; - assert!( - wake.wake_and_wait_until_idle(Duration::from_secs(2)).await, - "restarted scheduler must settle persisted recovery" - ); + // The mount wakes the worker; idle means its recovery pass finished. + registry.wait_project_idle(&owner).await; let receipt = SessionTemporalStore::new(database.as_ref()) .session_refresh_receipt(SessionRefreshReceiptRequestV1::new( SessionRefreshOperationIdV1::new(operation_id.to_owned()) 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 f736bde79a..a1f3bbc8e7 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 @@ -972,16 +972,27 @@ impl SessionTemporalRefreshSchedulerRegistry { .await .get(database_path) .map(|entry| Arc::clone(&entry.state)); - let Some(state) = state else { + if let Some(state) = state { + wait_idle(&state).await; + } + } + + #[cfg(any(test, feature = "test-helpers"))] + pub async fn wait_project_idle(&self, owner: &StoreOwnerKey) { + if let Some(state) = self.project_state(owner).await { + wait_idle(&state).await; + } + } +} + +#[cfg(any(test, feature = "test-helpers"))] +async fn wait_idle(state: &SessionTemporalRefreshWakeState) { + loop { + let idle = enabled_idle_notification(state); + if state.is_idle() { return; - }; - loop { - let idle = enabled_idle_notification(&state); - if state.is_idle() { - return; - } - idle.await; } + idle.await; } } From 3f78d3afd009f3359a913377a8d9de11701a39c2 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 17 Oct 2026 00:18:03 +0000 Subject: [PATCH 06/10] test(daemon): run attempt tests off native threads like production The attempt harness polled attempts on a native runtime thread, which production never does. Their synchronous storage writes raised the process-wide native writer-wait counter that off-Tokio tests in the same binary assert stays flat. Poll them on the blocking pool, as spawn_for_worktree does, and drop the serializing probe mutex. --- .../src/invocation.rs | 5 ---- .../src/invocation/work.rs | 3 --- .../src/invocation/work_attempt_exec/tests.rs | 26 +++++++++++++------ .../work_blocked_interval_recovery.rs | 3 --- 4 files changed, 18 insertions(+), 19 deletions(-) diff --git a/crates/tracedecay-daemon-service/src/invocation.rs b/crates/tracedecay-daemon-service/src/invocation.rs index 0119748912..ec3ff502e5 100644 --- a/crates/tracedecay-daemon-service/src/invocation.rs +++ b/crates/tracedecay-daemon-service/src/invocation.rs @@ -197,11 +197,6 @@ mod work_attempt_exec; mod work_blocked_interval_recovery; mod work_routing; -/// Held by tests that read the process-wide native writer-wait counter, and -/// by the off-Tokio attempt tests whose synchronous Work storage writes raise it. -#[cfg(test)] -pub(crate) static NATIVE_WRITER_WAITS_PROBE: std::sync::Mutex<()> = std::sync::Mutex::new(()); - pub use clock::now_millis; use configuration::*; pub use configuration::{ProfileConfigurationAuthorityV1, execute_profile_configuration}; diff --git a/crates/tracedecay-daemon-service/src/invocation/work.rs b/crates/tracedecay-daemon-service/src/invocation/work.rs index 23176d8317..657b46ed53 100644 --- a/crates/tracedecay-daemon-service/src/invocation/work.rs +++ b/crates/tracedecay-daemon-service/src/invocation/work.rs @@ -532,9 +532,6 @@ mod native_activity_tests { fn background_task_activity_publishes_without_a_tokio_runtime() { let profile = tempfile::tempdir().expect("profile root"); let project = tempfile::tempdir().expect("project root"); - let _probe = crate::invocation::NATIVE_WRITER_WAITS_PROBE - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); let writer_waits = tracedecay_rusqlite_runtime::exact_sql::native_thread_writer_waits(); native_runtime::block_on(async { assert!(tokio::runtime::Handle::try_current().is_err()); diff --git a/crates/tracedecay-daemon-service/src/invocation/work_attempt_exec/tests.rs b/crates/tracedecay-daemon-service/src/invocation/work_attempt_exec/tests.rs index a706409b38..d167bfbec5 100644 --- a/crates/tracedecay-daemon-service/src/invocation/work_attempt_exec/tests.rs +++ b/crates/tracedecay-daemon-service/src/invocation/work_attempt_exec/tests.rs @@ -81,16 +81,26 @@ use tracedecay_domain::test_fixtures::id; use tracedecay_domain::test_fixtures::digest; /// Runs `future` with no Tokio runtime anywhere on the stack, so any Tokio -/// reactor, timer or blocking-pool call on the attempt path panics. -fn without_tokio(future: F) -> F::Output { - let _probe = crate::invocation::NATIVE_WRITER_WAITS_PROBE - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); - native_runtime::block_on(async { +/// reactor, timer or blocking-pool call on the attempt path panics. Like +/// `spawn_for_worktree`, it polls the attempt on a blocking-pool thread, where +/// synchronous storage ports may park without holding a native worker. +fn without_tokio(future: F) +where + F: std::future::Future + Send + 'static, +{ + let outcome = native_runtime::block_on(async { assert!(tokio::runtime::Handle::try_current().is_err()); - future.await + native_runtime::NativeTask::spawn_off_native_thread(future) + .expect("native blocking pool available") + .wait() + .await }) - .expect("native runtime available") + .expect("native runtime available"); + match outcome { + native_runtime::NativeTaskOutcome::Completed(()) => {} + native_runtime::NativeTaskOutcome::Panicked(payload) => panic!("{payload}"), + native_runtime::NativeTaskOutcome::Cancelled => panic!("attempt task was cancelled"), + } } fn sha256_digest(bytes: &[u8]) -> ManifestDigest { diff --git a/crates/tracedecay-daemon-service/src/invocation/work_blocked_interval_recovery.rs b/crates/tracedecay-daemon-service/src/invocation/work_blocked_interval_recovery.rs index 8ca97bd7cd..d217bcfcb4 100644 --- a/crates/tracedecay-daemon-service/src/invocation/work_blocked_interval_recovery.rs +++ b/crates/tracedecay-daemon-service/src/invocation/work_blocked_interval_recovery.rs @@ -357,9 +357,6 @@ mod tests { let profile = tempfile::tempdir().expect("profile root"); let project = tempfile::tempdir().expect("project root"); let tokio_misses = native_runtime::tokio_runtime_missing_count(site); - let _probe = crate::invocation::NATIVE_WRITER_WAITS_PROBE - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner); let writer_waits = tracedecay_rusqlite_runtime::exact_sql::native_thread_writer_waits(); native_runtime::block_on(async { assert!(tokio::runtime::Handle::try_current().is_err()); From 8d434af9fdc59310e47395ce5212a27f50081acd Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 17 Oct 2026 00:41:14 +0000 Subject: [PATCH 07/10] test(daemon): give the terminal-owner drain test a real budget A 5 s shutdown deadline is shorter than the reserves behind the client drain, so that phase was floored at now and ended at the first timer tick. Under load the tick fired before the test released its admitted commit, and the terminal owner ran early. Use the daemon deadline. --- .../tracedecay-daemon-service/src/shutdown/orchestration.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay-daemon-service/src/shutdown/orchestration.rs b/crates/tracedecay-daemon-service/src/shutdown/orchestration.rs index 47ef1272d8..886d300407 100644 --- a/crates/tracedecay-daemon-service/src/shutdown/orchestration.rs +++ b/crates/tracedecay-daemon-service/src/shutdown/orchestration.rs @@ -896,7 +896,9 @@ mod tests { lifecycle.begin_draining(); let terminal_cancelled = Arc::new(AtomicBool::new(false)); let task_cancelled = Arc::clone(&terminal_cancelled); - let deadline = Instant::now() + Duration::from_secs(5); + // A deadline shorter than the reserves behind the client drain floors + // that phase at `now`, so the drain would end at the first timer tick. + let deadline = Instant::now() + DAEMON_SHUTDOWN_DEADLINE; let shutdown_lifecycle = lifecycle.clone(); let shutdown = tokio::spawn(async move { coordinate_daemon_shutdown(&shutdown_lifecycle, deadline, async move { From effff31b600f1cb103c42222ccbe738dccf7a9fe Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 17 Oct 2026 00:41:16 +0000 Subject: [PATCH 08/10] test(daemon): assert the zero-budget branch when spawn eats the wall The wall budget is armed after the provider spawns. Under CPU starvation the spawn took up to 2.6 s of the 2 s budget, production armed nothing, and the payload contract refused the zero-budget terminal. The double now records when the attempt ran, and the test asserts the branch production took instead of assuming the spawn is fast. --- .../src/invocation/work_attempt_exec/tests.rs | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/crates/tracedecay-daemon-service/src/invocation/work_attempt_exec/tests.rs b/crates/tracedecay-daemon-service/src/invocation/work_attempt_exec/tests.rs index d167bfbec5..44cd0876c9 100644 --- a/crates/tracedecay-daemon-service/src/invocation/work_attempt_exec/tests.rs +++ b/crates/tracedecay-daemon-service/src/invocation/work_attempt_exec/tests.rs @@ -127,6 +127,9 @@ struct AttemptRows { /// from the terminal row alone. observed_states: Vec, effects: BTreeMap, + /// When the provider was durably marked running, which is after its + /// spawn and before its wall budget is armed. + running_at: Option, } /// In-memory attempt rows with the same fenced compare-and-swap semantics as @@ -141,6 +144,10 @@ impl AttemptStore { self.inner.lock().unwrap().observed_states.clone() } + fn running_at(&self) -> Option { + self.inner.lock().unwrap().running_at + } + fn sealed_evidence( &self, authority: &WorkAuthority, @@ -313,6 +320,9 @@ impl WorkAttemptStoragePort for AttemptStore { } inner.rows.insert(key, payload); inner.observed_states.push(next.state()); + if next.state() == WorkAttemptStateV1::Running { + inner.running_at = Some(now_micros()); + } Ok(()) } @@ -1553,6 +1563,18 @@ async fn a_wall_exhausted_provider_seals_timed_out_and_emits_the_no_progress_ter }) .await .expect("no-progress page"); + // The budget is armed only once the provider runs, so a spawn that + // outlasts it (seen past 2 s under CPU starvation) arms nothing, and the + // payload contract refuses a zero budget. + let running_at = fixture.rows.running_at().expect("the provider ran"); + if running_at.0 >= deadline.0 { + assert!( + page.events.is_empty(), + "a zero armed budget emits no terminal: {:?}", + page.events + ); + return; + } assert_eq!(page.events.len(), 1, "exactly one no-progress terminal"); let envelope = &page.events[0]; assert_eq!( From 9bee81f46b2df2cdfd9857a2c71f494aab6be806 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 17 Oct 2026 00:47:19 +0000 Subject: [PATCH 09/10] test(daemon): wait for bootstrap completion, not a wall bound Bootstrap tests bounded the worker's completion with 1-10 s timers. Under load the 60 ms retry-budget test missed its 5 s bound. Tests that expect completion now wait on the completion signal; only the deadline test still passes a timeout. --- .../src/profile_host_admission_replay.rs | 75 +++++++------------ crates/tracedecay/src/daemon/branch_admin.rs | 6 +- 2 files changed, 29 insertions(+), 52 deletions(-) diff --git a/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs b/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs index 9a0bc60bbd..f0ba24236e 100644 --- a/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs +++ b/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs @@ -487,7 +487,7 @@ impl ProfileHostAdmissionReplayRegistry { pub async fn wait_bootstrap_completed( &self, profile_root: &Path, - timeout: Duration, + timeout: Option, ) -> BootstrapCompletion { let worker = { let workers = self.bootstrap_workers.lock().await; @@ -498,27 +498,24 @@ impl ProfileHostAdmissionReplayRegistry { let Some(worker) = worker else { return BootstrapCompletion::Missing; }; - let settled = |worker: &ProfileHostAdmissionBootstrapWorker| { - if worker.state.load(Ordering::Acquire) == BOOTSTRAP_RUNNING { - BootstrapCompletion::TimedOut - } else { - BootstrapCompletion::Completed + let completed = async { + loop { + let completed = worker.completed.notified(); + if worker.state.load(Ordering::Acquire) != BOOTSTRAP_RUNNING { + return; + } + completed.await; } }; - let deadline = Instant::now() + timeout; - loop { - let completed = worker.completed.notified(); - if worker.state.load(Ordering::Acquire) != BOOTSTRAP_RUNNING { - return BootstrapCompletion::Completed; - } - if Instant::now() >= deadline { - return BootstrapCompletion::TimedOut; - } - tokio::select! { - () = completed => {} - _ = native_runtime::sleep_until(deadline) => { - return settled(&worker); - } + match timeout { + Some(timeout) => native_runtime::timeout(timeout, completed) + .await + .map_or(BootstrapCompletion::TimedOut, |()| { + BootstrapCompletion::Completed + }), + None => { + completed.await; + BootstrapCompletion::Completed } } } @@ -959,9 +956,7 @@ mod tests { } while tasks.join_next().await.is_some() {} assert_eq!( - registry - .wait_bootstrap_completed(&profile_root, Duration::from_secs(2)) - .await, + registry.wait_bootstrap_completed(&profile_root, None).await, BootstrapCompletion::Completed, "coalesced bootstrap must complete" ); @@ -999,16 +994,12 @@ mod tests { .ensure_bootstrap(&profile_root, Arc::clone(&operation)) .await; assert_eq!( - registry - .wait_bootstrap_completed(&profile_root, Duration::from_secs(1)) - .await, + registry.wait_bootstrap_completed(&profile_root, None).await, BootstrapCompletion::Completed, ); registry.ensure_bootstrap(&profile_root, operation).await; assert_eq!( - registry - .wait_bootstrap_completed(&profile_root, Duration::from_secs(1)) - .await, + registry.wait_bootstrap_completed(&profile_root, None).await, BootstrapCompletion::Completed, ); assert_eq!(attempts.load(Ordering::Acquire), 2); @@ -1043,9 +1034,7 @@ mod tests { .ensure_bootstrap(&profile_root, Arc::clone(&operation)) .await; assert_eq!( - registry - .wait_bootstrap_completed(&profile_root, Duration::from_secs(1)) - .await, + registry.wait_bootstrap_completed(&profile_root, None).await, BootstrapCompletion::Completed, ); registry @@ -1056,9 +1045,7 @@ mod tests { registry.bootstrap_terminal_cache_for = Duration::ZERO; registry.ensure_bootstrap(&profile_root, operation).await; assert_eq!( - registry - .wait_bootstrap_completed(&profile_root, Duration::from_secs(1)) - .await, + registry.wait_bootstrap_completed(&profile_root, None).await, BootstrapCompletion::Completed, ); assert_eq!(attempts.load(Ordering::Acquire), 2); @@ -1086,9 +1073,7 @@ mod tests { registry.ensure_bootstrap(&profile_root, operation).await; assert_eq!( - registry - .wait_bootstrap_completed(&profile_root, Duration::from_secs(1)) - .await, + registry.wait_bootstrap_completed(&profile_root, None).await, BootstrapCompletion::Completed, ); @@ -1141,7 +1126,7 @@ mod tests { .await; assert_eq!( registry - .wait_bootstrap_completed(&retryable_profile, Duration::from_secs(1)) + .wait_bootstrap_completed(&retryable_profile, None) .await, BootstrapCompletion::Completed, ); @@ -1169,7 +1154,7 @@ mod tests { .await; assert_eq!( registry - .wait_bootstrap_completed(&terminal_profile, Duration::from_secs(1)) + .wait_bootstrap_completed(&terminal_profile, None) .await, BootstrapCompletion::Completed, ); @@ -1216,9 +1201,7 @@ mod tests { registry.ensure_bootstrap(&profile_root, operation).await; assert_eq!( - registry - .wait_bootstrap_completed(&profile_root, Duration::from_secs(5)) - .await, + registry.wait_bootstrap_completed(&profile_root, None).await, BootstrapCompletion::Completed, "a permanently retryable bootstrap must still terminate" ); @@ -1892,9 +1875,7 @@ mod tests { assert!(tokio::runtime::Handle::try_current().is_err()); registry.ensure_bootstrap(&profile_root, operation).await; assert_eq!( - registry - .wait_bootstrap_completed(&profile_root, Duration::from_secs(10)) - .await, + registry.wait_bootstrap_completed(&profile_root, None).await, BootstrapCompletion::Completed ); assert!(matches!( @@ -2042,7 +2023,7 @@ mod tests { ); assert_eq!( registry - .wait_bootstrap_completed(&profile_root, grace) + .wait_bootstrap_completed(&profile_root, Some(grace)) .await, BootstrapCompletion::TimedOut ); diff --git a/crates/tracedecay/src/daemon/branch_admin.rs b/crates/tracedecay/src/daemon/branch_admin.rs index 60a02f7a0f..cc9db3e926 100644 --- a/crates/tracedecay/src/daemon/branch_admin.rs +++ b/crates/tracedecay/src/daemon/branch_admin.rs @@ -2324,14 +2324,10 @@ mod tests { .ensure_profile_host_admission_bootstrap(&profile_root) .await .unwrap(); - // The worker performs a real session-runtime-registry open before it - // can surface the spool error, so give it a generous (still bounded) - // wait; completion notifies immediately, so a healthy run never - // sleeps this long. assert_eq!( administration .profile_host_admission_replay - .wait_bootstrap_completed(&profile_root, Duration::from_mins(1)) + .wait_bootstrap_completed(&profile_root, None) .await, BootstrapCompletion::Completed, "production bootstrap worker must publish its terminal state" From adbf0910c99891bd599891f3b4af39679d709e07 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Sat, 17 Oct 2026 01:51:29 +0000 Subject: [PATCH 10/10] test(daemon): let the idle signal, not 250 ms, end the probe wait The test bounded wait_idle with a 250 ms timer, which a loaded runner missed. wait_idle already returns false at its grace, so the result alone proves the notification arrived; the grace is now long enough that a lost notification hangs to the harness timeout. --- .../src/profile_host_admission_replay.rs | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs b/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs index f0ba24236e..a2b73c46f0 100644 --- a/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs +++ b/crates/tracedecay-daemon-service/src/profile_host_admission_replay.rs @@ -1672,9 +1672,11 @@ mod tests { let wait_registry = Arc::clone(®istry); let wait_db_path = db_path.clone(); + // The grace never ends this wait: a lost idle notification leaves it + // parked until the harness timeout. let wait = tokio::spawn(async move { wait_registry - .wait_idle(&wait_db_path, Duration::from_secs(1)) + .wait_idle(&wait_db_path, Duration::from_hours(1)) .await }); probe_started.notified().await; @@ -1682,10 +1684,7 @@ mod tests { release_probe.notify_one(); assert!( - tokio::time::timeout(Duration::from_millis(250), wait) - .await - .expect("idle notification must avoid the full grace") - .expect("wait task"), + wait.await.expect("wait task"), "a fresh idle probe must observe the completed replay" ); registry.shutdown().await;