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 4e8c3fea90..5138647b0e 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 { @@ -117,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 @@ -131,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, @@ -303,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(()) } @@ -1654,6 +1674,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!( 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()); 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..a2b73c46f0 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)] @@ -488,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; @@ -499,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 } } } @@ -750,7 +746,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 +757,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 +821,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 +885,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 @@ -961,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" ); @@ -984,8 +977,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); @@ -1001,17 +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, ); - tokio::time::sleep(Duration::from_millis(30)).await; 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); @@ -1024,9 +1012,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); @@ -1046,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,12 +1042,10 @@ 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 - .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); @@ -1089,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, ); @@ -1144,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, ); @@ -1172,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, ); @@ -1219,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" ); @@ -1240,7 +1220,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 +1308,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 +1340,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 +1378,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 +1410,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 +1437,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 +1605,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); } @@ -1674,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; @@ -1684,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; @@ -1774,13 +1771,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) { @@ -1858,9 +1874,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!( @@ -2008,7 +2022,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-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); 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-daemon-service/src/shutdown/orchestration.rs b/crates/tracedecay-daemon-service/src/shutdown/orchestration.rs index b0d35f6ffa..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 { @@ -1277,29 +1279,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" 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)) 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; } } 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"