From bbc7ad9c43c947ef9615ed0a0c8f24750b209099 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 23:18:29 -0800 Subject: [PATCH 01/14] fix(retention): queue maintenance after repeated writer refusals Retention fast probes never join the daemon writer queue. Under sustained store-scoped writers, even a fair RwLock cannot admit the maintenance pass. After two per-store nonqueued refusals, join the fair writer queue with cooperative shutdown cancellation; reset/prune the counters on admission. Exercise bounded deferrals, progress with competing store writers, and cancellation of the queued acquisition in native Rust tests. Refs #4616 --- crates/tracedecay/src/daemon/maintenance.rs | 190 +++++++++++++++++++- 1 file changed, 185 insertions(+), 5 deletions(-) diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index ea2070195f..4db847bc33 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -1,4 +1,5 @@ -use std::collections::BTreeSet; +use std::collections::{BTreeMap, BTreeSet}; +use std::future::Future; use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::{Duration, Instant}; @@ -32,6 +33,32 @@ use tracedecay_runtime_core::logging::log_daemon_event; use tracedecay_runtime_core::native_runtime::{self, NativeTask, TimeoutError}; const MAINTENANCE_STORE_PAGE_LIMIT: usize = 8; const REGISTERED_PROJECT_PAGE_LIMIT: usize = 64; +/// Fast probes preserve low-latency work. After two refusals, join Tokio's +/// fair writer queue so a stream of store-scoped readers cannot starve GC. +const MAINTENANCE_WRITER_FAST_DEFERRALS: u8 = 2; + +async fn admit_retention_writer( + administration: &StoreAdministration, + cancellation: &tracedecay_runtime_core::cancellation::CancellationToken, + prior_deferrals: u8, + operation: Operation, +) -> Option +where + Operation: FnOnce() -> OperationFuture, + OperationFuture: Future, +{ + if prior_deferrals < MAINTENANCE_WRITER_FAST_DEFERRALS { + administration.try_with_writer(operation).await + } else { + // A queued writer blocks later store readers behind it. Cancellation + // drops the acquisition future and removes the queue entry on shutdown. + tokio::select! { + biased; + () = cancellation.cancelled() => None, + outcome = administration.with_writer(operation) => Some(outcome), + } + } +} /// Every registered project whose profile shard exists on disk, keyed for the /// tick's work list. @@ -618,6 +645,8 @@ pub(super) struct MaintenanceCoordinator { /// store is starved when the mounted set exceeds `MAINTENANCE_STORE_PAGE_LIMIT`. store_cursor: Arc>>, store_continuations: Arc>, + /// Per-store writer admission refusals, bounded and reset on a real admission. + writer_deferrals: Arc>>, /// Instant of the last branch-store GC pass that succeeded for every /// mounted project. `None` keeps the daily cadence retry-eligible. last_branch_gc: Arc>>, @@ -639,6 +668,7 @@ impl Default for MaintenanceCoordinator { metrics: Arc::new(Mutex::new(MaintenanceMetricsV1::default())), store_cursor: Arc::new(Mutex::new(None)), store_continuations: Arc::new(Mutex::new(StoreMaintenanceContinuations::default())), + writer_deferrals: Arc::new(Mutex::new(BTreeMap::new())), last_branch_gc: Arc::new(Mutex::new(None)), resident_memory_sampler: Arc::new(Mutex::new(None)), resident_memory_log: Arc::new(std::sync::Mutex::new( @@ -1060,6 +1090,10 @@ impl MaintenanceCoordinator { work.sort_by(|left, right| left.0.cmp(&right.0)); let keys = work.iter().map(|(key, _)| key.clone()).collect::>(); self.store_continuations.lock().await.retain_stores(&keys); + self.writer_deferrals + .lock() + .await + .retain(|key, _| keys.binary_search(key).is_ok()); let after = self.store_cursor.lock().await.clone(); let (window, _) = select_store_window(&keys, after.as_deref(), MAINTENANCE_STORE_PAGE_LIMIT); @@ -1124,8 +1158,18 @@ impl MaintenanceCoordinator { outcome = MaintenanceTickOutcome::Retry; break; } - let admitted = administration - .try_with_writer(|| async { + let prior_deferrals = self + .writer_deferrals + .lock() + .await + .get(&keys[index]) + .copied() + .unwrap_or(0); + let admitted = admit_retention_writer( + administration, + &self.cancellation, + prior_deferrals, + || async { let Ok(_clients) = administration.maintenance_store_clients().try_lock() else { return MaintenanceTickOutcome::Retry; }; @@ -1177,8 +1221,17 @@ impl MaintenanceCoordinator { .await } } - }) - .await; + }, + ) + .await; + { + let mut deferrals = self.writer_deferrals.lock().await; + if admitted.is_some() { + deferrals.remove(&keys[index]); + } else if !self.cancellation.is_cancelled() { + deferrals.insert(keys[index].clone(), prior_deferrals.saturating_add(1)); + } + } attempted = attempted.saturating_add(1); match admitted { Some(unit_outcome) => { @@ -1541,6 +1594,133 @@ mod tests { use tracedecay_maintenance::telemetry::StoreTelemetrySamplingRegistry; use tracedecay_maintenance::tick::{MaintenanceContinuation, MaintenanceTickOutcome}; + /// Hold an unrelated store writer across two failed fast probes. The third + /// retention attempt must queue behind it and complete even as more store + /// writers arrive, rather than repeatedly taking non-queued snapshots. + #[tokio::test] + async fn retention_queues_after_bounded_writer_deferrals() { + use crate::daemon::branch_admin::{StoreAdministration, StoreWriterClass, WriterScope}; + use tracedecay_runtime_core::cancellation::CancellationToken; + + let administration = StoreAdministration::default(); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let holding_admin = administration.clone(); + let holder = tokio::spawn(async move { + holding_admin + .with_writer_in( + WriterScope::store("/unrelated-store", StoreWriterClass::Content), + move || async move { + entered_tx.send(()).expect("store writer entered"); + release_rx.await.expect("store writer released"); + }, + ) + .await; + }); + entered_rx.await.expect("writer acquired its shared daemon gate"); + let cancellation = CancellationToken::new(); + for refused in 0..super::MAINTENANCE_WRITER_FAST_DEFERRALS { + assert_eq!( + super::admit_retention_writer(&administration, &cancellation, refused, || async { + 41_u8 + }) + .await, + None, + "a store writer blocks the non-queued daemon probe" + ); + } + + let queued = super::admit_retention_writer( + &administration, + &cancellation, + super::MAINTENANCE_WRITER_FAST_DEFERRALS, + || async { 42_u8 }, + ); + tokio::pin!(queued); + assert!( + tokio::time::timeout(Duration::from_millis(30), &mut queued) + .await + .is_err(), + "the maintenance operation waits behind an existing writer" + ); + + let competing_admin = administration.clone(); + let completed_writers = Arc::new(AtomicUsize::new(0)); + let completed = Arc::clone(&completed_writers); + let competitors = tokio::spawn(async move { + for _ in 0..64 { + competing_admin + .with_writer_in( + WriterScope::store("/another-store", StoreWriterClass::Content), + || { + let completed = Arc::clone(&completed); + async move { + completed.fetch_add(1, Ordering::SeqCst); + tokio::task::yield_now().await; + } + }, + ) + .await; + } + }); + release_tx.send(()).expect("release the original writer"); + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), &mut queued) + .await + .expect("queued retention is admitted despite competing writers"), + Some(42) + ); + holder.await.expect("first writer finishes"); + competitors.await.expect("subsequent store writers finish"); + assert_eq!(completed_writers.load(Ordering::SeqCst), 64); + } + + #[tokio::test] + async fn cancelled_queued_retention_releases_its_writer_place() { + use crate::daemon::branch_admin::{StoreAdministration, StoreWriterClass, WriterScope}; + use tracedecay_runtime_core::cancellation::CancellationToken; + + let administration = StoreAdministration::default(); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let holding_admin = administration.clone(); + let writer = tokio::spawn(async move { + holding_admin + .with_writer_in( + WriterScope::store("/held-store", StoreWriterClass::Owner), + move || async move { + entered_tx.send(()).expect("entered writer"); + release_rx.await.expect("released writer"); + }, + ) + .await; + }); + entered_rx.await.expect("writer holds shared daemon admission"); + let cancellation = CancellationToken::new(); + let waiting = super::admit_retention_writer( + &administration, + &cancellation, + super::MAINTENANCE_WRITER_FAST_DEFERRALS, + || async { 7_u8 }, + ); + tokio::pin!(waiting); + assert!( + tokio::time::timeout(Duration::from_millis(30), &mut waiting) + .await + .is_err() + ); + cancellation.cancel(); + assert_eq!( + tokio::time::timeout(Duration::from_secs(2), &mut waiting) + .await + .expect("cancelled acquisition completes"), + None + ); + release_tx.send(()).expect("release held writer"); + writer.await.expect("held writer exits"); + assert_eq!(administration.try_with_writer(|| async { 9_u8 }).await, Some(9)); + } + #[tokio::test] async fn mounted_store_identity_tracks_native_removal_outcomes() { let temporary = tempfile::tempdir().expect("store identity fixture root"); From c13a88f0a72f9b2e1db3b3ab7ae102e716ce4ab8 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 23:24:14 -0800 Subject: [PATCH 02/14] test(retention): reclaim six-generation backlog after queued writer Drive the existing registered-store production maintenance tick through two real writer refusals and then a queued admission, with six durable code-index generations and a concurrent unrelated store writer. Verify that the writer release enables the same tick to collect five superseded generations. Refs #4616 --- crates/tracedecay/src/daemon/maintenance.rs | 54 ++++++++++++++++++++- 1 file changed, 53 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index 4db847bc33..70b51efe06 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -1986,6 +1986,49 @@ mod tests { ); drop(release); + // A long-lived, unrelated content writer holds the daemon read gate. + // Two real retention ticks must defer, and the next must queue rather + // than missing the same writer again. The fixture's six on-disk + // generations make forward progress observable through actual GC. + let (writer_started, writer_entered) = tokio::sync::oneshot::channel(); + let (release_writer, writer_released) = tokio::sync::oneshot::channel(); + let holding_admin = administration.clone(); + let writer = NativeTask::spawn(async move { + holding_admin + .with_writer_in( + crate::daemon::branch_admin::WriterScope::store( + "/unrelated-project", + crate::daemon::branch_admin::StoreWriterClass::Content, + ), + move || async move { + writer_started.send(()).expect("writer admitted"); + writer_released.await.expect("writer released"); + }, + ) + .await; + }) + .expect("start competing store writer"); + writer_entered.await.expect("unrelated store writer owns shared gate"); + for _ in 0..super::MAINTENANCE_WRITER_FAST_DEFERRALS { + let deferred = coordinator + .run_tick( + &profile_root, + &profile_database, + &administration, + &schedulers, + &retention, + branch_gc, + None, + ) + .await; + assert_eq!(deferred, MaintenanceTickOutcome::Retry); + assert_eq!(generation_files(), 6, "fast probe must not collect under a writer"); + } + + let release_after_queue = std::thread::spawn(move || { + std::thread::sleep(Duration::from_millis(100)); + release_writer.send(()).expect("release competing writer"); + }); coordinator .run_tick( &profile_root, @@ -1997,7 +2040,16 @@ mod tests { None, ) .await; - assert_eq!(generation_files(), 1, "the full tick collects the backlog"); + release_after_queue.join().expect("writer release thread"); + assert!( + matches!(writer.wait().await, native_runtime::NativeTaskOutcome::Completed(())), + "competing writer must complete normally" + ); + assert_eq!( + generation_files(), + 1, + "queued retention reclaims five verified superseded generations after writer release" + ); let continuation = coordinator .run_tick( From 61fd4684c48fb553485a7c29b4d1a20d005ed25b Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 23:26:29 -0800 Subject: [PATCH 03/14] fix(retention): cancel only queued writer admission A shutdown cancellation during an already-admitted retention operation must not drop journaled work that can still own blocking store clients. Separate the cancellable writer wait from the critical section, preserving cooperative operation completion and the existing writer lease after admission. Add a native Rust test that cancels after admission and requires the critical section to complete and release the gate. Refs #4616 --- crates/tracedecay/src/daemon/branch_admin.rs | 24 ++++++++++ crates/tracedecay/src/daemon/maintenance.rs | 49 +++++++++++++++++--- 2 files changed, 66 insertions(+), 7 deletions(-) diff --git a/crates/tracedecay/src/daemon/branch_admin.rs b/crates/tracedecay/src/daemon/branch_admin.rs index d0cae8b43e..e521ed4b89 100644 --- a/crates/tracedecay/src/daemon/branch_admin.rs +++ b/crates/tracedecay/src/daemon/branch_admin.rs @@ -1923,6 +1923,30 @@ impl StoreAdministration { operation().await } + /// Cancellation applies only while queued for daemon-wide writer admission. + /// Once the guard is acquired, the operation must finish cooperatively: + /// dropping an in-flight journal/retention future would abandon physical + /// blocking work whose client survives the daemon's shutdown join. + pub(super) async fn with_writer_cancellable_acquisition( + &self, + cancellation: &tracedecay_runtime_core::cancellation::CancellationToken, + operation: Operation, + ) -> Option + where + Operation: FnOnce() -> OperationFuture, + OperationFuture: Future, + { + let scope = WriterScope::Daemon; + let _writer = tokio::select! { + biased; + () = cancellation.cancelled() => return None, + writer = crate::daemon::core_admission::park_admission( + self.gate.acquire(&scope) + ) => writer, + }; + Some(operation().await) + } + pub(super) async fn try_with_writer( &self, operation: Operation, diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index 70b51efe06..284564fef3 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -50,13 +50,11 @@ where if prior_deferrals < MAINTENANCE_WRITER_FAST_DEFERRALS { administration.try_with_writer(operation).await } else { - // A queued writer blocks later store readers behind it. Cancellation - // drops the acquisition future and removes the queue entry on shutdown. - tokio::select! { - biased; - () = cancellation.cancelled() => None, - outcome = administration.with_writer(operation) => Some(outcome), - } + // Only cancel the wait: once admitted, journal/retention work must + // finish cooperatively rather than be dropped at the writer boundary. + administration + .with_writer_cancellable_acquisition(cancellation, operation) + .await } } @@ -1721,6 +1719,43 @@ mod tests { assert_eq!(administration.try_with_writer(|| async { 9_u8 }).await, Some(9)); } + /// Cancellation after writer admission must not drop a running retention + /// operation: its blocking/journaled work is responsible for cooperating. + #[tokio::test] + async fn cancellation_after_writer_admission_preserves_running_work() { + use crate::daemon::branch_admin::StoreAdministration; + use tracedecay_runtime_core::cancellation::CancellationToken; + + let administration = StoreAdministration::default(); + let cancellation = CancellationToken::new(); + let working = administration.clone(); + let stopping = cancellation.clone(); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let work = tokio::spawn(async move { + super::admit_retention_writer( + &working, + &stopping, + super::MAINTENANCE_WRITER_FAST_DEFERRALS, + move || async move { + entered_tx.send(()).expect("retention entered its critical section"); + release_rx.await.expect("retention work released"); + 43_u8 + }, + ) + .await + }); + entered_rx.await.expect("writer admission succeeded"); + cancellation.cancel(); + release_tx.send(()).expect("finish admitted work"); + assert_eq!( + work.await.expect("retention task joins"), + Some(43), + "shutdown cancellation cannot drop journaled work already admitted" + ); + assert_eq!(administration.try_with_writer(|| async { 44_u8 }).await, Some(44)); + } + #[tokio::test] async fn mounted_store_identity_tracks_native_removal_outcomes() { let temporary = tempfile::tempdir().expect("store identity fixture root"); From c41076a0fdf3aa290d40ec32e7b7532e7be32a1c Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sat, 7 Nov 2026 23:59:46 -0800 Subject: [PATCH 04/14] test(retention): prove fair admission during ongoing writer churn The old finite competitor burst could finish before maintenance acquired the daemon gate. Keep unrelated store writers active through the queued retention admission, asserting churn both before and afterward and only stopping writers after the admission proof. This remains a focused gate regression; sealed-artifact N>3 acceptance requires its own native test. Refs #4616 --- crates/tracedecay/src/daemon/maintenance.rs | 29 ++++++++++++++++++--- 1 file changed, 26 insertions(+), 3 deletions(-) diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index 284564fef3..be2f963577 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -1642,11 +1642,16 @@ mod tests { "the maintenance operation waits behind an existing writer" ); + // Keep unrelated content writers churning until *after* the queued + // retention operation completes. A finite burst that happens to stop + // before retention runs would not prove progress under sustained load. let competing_admin = administration.clone(); let completed_writers = Arc::new(AtomicUsize::new(0)); let completed = Arc::clone(&completed_writers); + let stop_churn = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let stop = Arc::clone(&stop_churn); let competitors = tokio::spawn(async move { - for _ in 0..64 { + while !stop.load(Ordering::SeqCst) { competing_admin .with_writer_in( WriterScope::store("/another-store", StoreWriterClass::Content), @@ -1661,6 +1666,14 @@ mod tests { .await; } }); + tokio::time::timeout(Duration::from_secs(5), async { + while completed_writers.load(Ordering::SeqCst) < 4 { + tokio::task::yield_now().await; + } + }) + .await + .expect("unrelated writers churn before original writer releases"); + let before_release = completed_writers.load(Ordering::SeqCst); release_tx.send(()).expect("release the original writer"); assert_eq!( tokio::time::timeout(Duration::from_secs(5), &mut queued) @@ -1668,9 +1681,19 @@ mod tests { .expect("queued retention is admitted despite competing writers"), Some(42) ); + // A succeeding write proves the churn remained live *through* the + // retention admission; it is stopped only after that observation. + tokio::time::timeout(Duration::from_secs(5), async { + while completed_writers.load(Ordering::SeqCst) <= before_release { + tokio::task::yield_now().await; + } + }) + .await + .expect("unrelated writers resume after maintenance admission"); + stop_churn.store(true, Ordering::SeqCst); holder.await.expect("first writer finishes"); - competitors.await.expect("subsequent store writers finish"); - assert_eq!(completed_writers.load(Ordering::SeqCst), 64); + competitors.await.expect("unrelated writer churn finishes"); + assert!(completed_writers.load(Ordering::SeqCst) > before_release); } #[tokio::test] From 58a76567f9bcbd3b7b6f9a28feb6fb91e8e2e5f5 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 00:27:01 -0800 Subject: [PATCH 05/14] fix(retention): honor shutdown and repair admission regressions Refuse already-cancelled retention before either fast or queued admission, without interrupting operations that already own the writer gate. Start ongoing writer churn before queueing exclusive retention. Explicitly poll queue admission and a later store writer instead of waiting for later readers to make progress behind the exclusive waiter. Check that cancelling the waiter admits its follower while the original shared holder is alive. Add cancellation-before-admission coverage and restore the native runtime import required by the real maintenance-tick regression. Refs #4616 Bazel/native execution remains required; no Cargo or CI changes. --- crates/tracedecay/src/daemon/maintenance.rs | 213 ++++++++++++++------ 1 file changed, 155 insertions(+), 58 deletions(-) diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index be2f963577..fa2a8c4965 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -47,6 +47,11 @@ where Operation: FnOnce() -> OperationFuture, OperationFuture: Future, { + // Shutdown observed before admission must not start another retention + // unit, even when a fast probe could acquire the writer gate immediately. + if cancellation.is_cancelled() { + return None; + } if prior_deferrals < MAINTENANCE_WRITER_FAST_DEFERRALS { administration.try_with_writer(operation).await } else { @@ -1581,25 +1586,27 @@ pub(super) fn retention_maintenance_enabled( #[cfg(test)] mod tests { + use std::future::{Future, poll_fn}; use std::path::Path; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::{Arc, Condvar, Mutex as StdMutex}; + use std::task::Poll; use std::time::Duration; use tokio::sync::Notify; use super::{MaintenanceCoordinator, NativeTask, run_resident_memory_sampler_loop}; + use crate::daemon::branch_admin::{StoreAdministration, StoreWriterClass, WriterScope}; use tracedecay_maintenance::telemetry::StoreTelemetrySamplingRegistry; use tracedecay_maintenance::tick::{MaintenanceContinuation, MaintenanceTickOutcome}; + use tracedecay_runtime_core::cancellation::CancellationToken; + use tracedecay_runtime_core::native_runtime; /// Hold an unrelated store writer across two failed fast probes. The third /// retention attempt must queue behind it and complete even as more store /// writers arrive, rather than repeatedly taking non-queued snapshots. #[tokio::test] async fn retention_queues_after_bounded_writer_deferrals() { - use crate::daemon::branch_admin::{StoreAdministration, StoreWriterClass, WriterScope}; - use tracedecay_runtime_core::cancellation::CancellationToken; - let administration = StoreAdministration::default(); let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); let (release_tx, release_rx) = tokio::sync::oneshot::channel(); @@ -1615,7 +1622,9 @@ mod tests { ) .await; }); - entered_rx.await.expect("writer acquired its shared daemon gate"); + entered_rx + .await + .expect("writer acquired its shared daemon gate"); let cancellation = CancellationToken::new(); for refused in 0..super::MAINTENANCE_WRITER_FAST_DEFERRALS { assert_eq!( @@ -1628,52 +1637,65 @@ mod tests { ); } - let queued = super::admit_retention_writer( - &administration, - &cancellation, - super::MAINTENANCE_WRITER_FAST_DEFERRALS, - || async { 42_u8 }, - ); - tokio::pin!(queued); - assert!( - tokio::time::timeout(Duration::from_millis(30), &mut queued) - .await - .is_err(), - "the maintenance operation waits behind an existing writer" - ); - - // Keep unrelated content writers churning until *after* the queued - // retention operation completes. A finite burst that happens to stop - // before retention runs would not prove progress under sustained load. + // Establish ongoing writer activity before queueing the exclusive + // maintenance request. A fair queued writer blocks later readers; + // waiting for their progress before releasing the holder deadlocks. let competing_admin = administration.clone(); - let completed_writers = Arc::new(AtomicUsize::new(0)); - let completed = Arc::clone(&completed_writers); - let stop_churn = Arc::new(std::sync::atomic::AtomicBool::new(false)); - let stop = Arc::clone(&stop_churn); + let writer_entries = Arc::new(AtomicUsize::new(0)); + let entries = Arc::clone(&writer_entries); + let stop_churn = CancellationToken::new(); + let stop = stop_churn.clone(); let competitors = tokio::spawn(async move { - while !stop.load(Ordering::SeqCst) { + while !stop.is_cancelled() { competing_admin .with_writer_in( WriterScope::store("/another-store", StoreWriterClass::Content), - || { - let completed = Arc::clone(&completed); - async move { - completed.fetch_add(1, Ordering::SeqCst); - tokio::task::yield_now().await; - } + || async { + entries.fetch_add(1, Ordering::SeqCst); + tokio::task::yield_now().await; }, ) .await; } }); tokio::time::timeout(Duration::from_secs(5), async { - while completed_writers.load(Ordering::SeqCst) < 4 { + while writer_entries.load(Ordering::SeqCst) < 4 { tokio::task::yield_now().await; } }) .await - .expect("unrelated writers churn before original writer releases"); - let before_release = completed_writers.load(Ordering::SeqCst); + .expect("unrelated writers churn before retention queues"); + + let queued = super::admit_retention_writer( + &administration, + &cancellation, + super::MAINTENANCE_WRITER_FAST_DEFERRALS, + || async { 42_u8 }, + ); + tokio::pin!(queued); + poll_fn(|cx| { + assert!( + queued.as_mut().poll(cx).is_pending(), + "the retention request must queue behind the held writer" + ); + Poll::Ready(()) + }) + .await; + let later_writer = administration.with_writer_in( + WriterScope::store("/later-store", StoreWriterClass::Content), + || async { 43_u8 }, + ); + tokio::pin!(later_writer); + poll_fn(|cx| { + assert!( + later_writer.as_mut().poll(cx).is_pending(), + "a later store writer must not overtake queued retention" + ); + Poll::Ready(()) + }) + .await; + + let before_release = writer_entries.load(Ordering::SeqCst); release_tx.send(()).expect("release the original writer"); assert_eq!( tokio::time::timeout(Duration::from_secs(5), &mut queued) @@ -1681,26 +1703,59 @@ mod tests { .expect("queued retention is admitted despite competing writers"), Some(42) ); - // A succeeding write proves the churn remained live *through* the - // retention admission; it is stopped only after that observation. + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), &mut later_writer) + .await + .expect("later writer resumes after retention"), + 43 + ); + // Keep the churn alive through retention and observe a subsequent + // acquisition before stopping it: a finite burst is not this proof. tokio::time::timeout(Duration::from_secs(5), async { - while completed_writers.load(Ordering::SeqCst) <= before_release { + while writer_entries.load(Ordering::SeqCst) <= before_release { tokio::task::yield_now().await; } }) .await .expect("unrelated writers resume after maintenance admission"); - stop_churn.store(true, Ordering::SeqCst); + stop_churn.cancel(); holder.await.expect("first writer finishes"); competitors.await.expect("unrelated writer churn finishes"); - assert!(completed_writers.load(Ordering::SeqCst) > before_release); + assert!(writer_entries.load(Ordering::SeqCst) > before_release); } #[tokio::test] - async fn cancelled_queued_retention_releases_its_writer_place() { - use crate::daemon::branch_admin::{StoreAdministration, StoreWriterClass, WriterScope}; - use tracedecay_runtime_core::cancellation::CancellationToken; + async fn cancelled_retention_never_constructs_work_before_admission() { + let administration = StoreAdministration::default(); + let cancellation = CancellationToken::new(); + cancellation.cancel(); + let constructed = AtomicUsize::new(0); + for prior_deferrals in 0..=super::MAINTENANCE_WRITER_FAST_DEFERRALS { + assert_eq!( + super::admit_retention_writer( + &administration, + &cancellation, + prior_deferrals, + || { + constructed.fetch_add(1, Ordering::SeqCst); + std::future::ready(7_u8) + }, + ) + .await, + None, + "cancelled retention must refuse even an uncontended fast probe" + ); + } + assert_eq!(constructed.load(Ordering::SeqCst), 0); + assert_eq!( + administration.try_with_writer(|| async { 9_u8 }).await, + Some(9), + "refused work leaves the writer gate available" + ); + } + #[tokio::test] + async fn cancelled_queued_retention_releases_its_writer_place() { let administration = StoreAdministration::default(); let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); let (release_tx, release_rx) = tokio::sync::oneshot::channel(); @@ -1716,20 +1771,36 @@ mod tests { ) .await; }); - entered_rx.await.expect("writer holds shared daemon admission"); + entered_rx + .await + .expect("writer holds shared daemon admission"); let cancellation = CancellationToken::new(); + let constructed = AtomicUsize::new(0); let waiting = super::admit_retention_writer( &administration, &cancellation, super::MAINTENANCE_WRITER_FAST_DEFERRALS, - || async { 7_u8 }, + || { + constructed.fetch_add(1, Ordering::SeqCst); + std::future::ready(7_u8) + }, ); tokio::pin!(waiting); - assert!( - tokio::time::timeout(Duration::from_millis(30), &mut waiting) - .await - .is_err() + poll_fn(|cx| { + assert!(waiting.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + let follower = administration.with_writer_in( + WriterScope::store("/follower-store", StoreWriterClass::Content), + || async { 8_u8 }, ); + tokio::pin!(follower); + poll_fn(|cx| { + assert!(follower.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; cancellation.cancel(); assert_eq!( tokio::time::timeout(Duration::from_secs(2), &mut waiting) @@ -1737,18 +1808,25 @@ mod tests { .expect("cancelled acquisition completes"), None ); + assert_eq!(constructed.load(Ordering::SeqCst), 0); + assert_eq!( + tokio::time::timeout(Duration::from_secs(2), &mut follower) + .await + .expect("cancellation removes the queued writer before the holder exits"), + 8 + ); release_tx.send(()).expect("release held writer"); writer.await.expect("held writer exits"); - assert_eq!(administration.try_with_writer(|| async { 9_u8 }).await, Some(9)); + assert_eq!( + administration.try_with_writer(|| async { 9_u8 }).await, + Some(9) + ); } /// Cancellation after writer admission must not drop a running retention /// operation: its blocking/journaled work is responsible for cooperating. #[tokio::test] async fn cancellation_after_writer_admission_preserves_running_work() { - use crate::daemon::branch_admin::StoreAdministration; - use tracedecay_runtime_core::cancellation::CancellationToken; - let administration = StoreAdministration::default(); let cancellation = CancellationToken::new(); let working = administration.clone(); @@ -1761,7 +1839,9 @@ mod tests { &stopping, super::MAINTENANCE_WRITER_FAST_DEFERRALS, move || async move { - entered_tx.send(()).expect("retention entered its critical section"); + entered_tx + .send(()) + .expect("retention entered its critical section"); release_rx.await.expect("retention work released"); 43_u8 }, @@ -1770,13 +1850,21 @@ mod tests { }); entered_rx.await.expect("writer admission succeeded"); cancellation.cancel(); + assert_eq!( + administration.try_with_writer(|| async { 44_u8 }).await, + None, + "admitted work retains exclusion until it finishes" + ); release_tx.send(()).expect("finish admitted work"); assert_eq!( work.await.expect("retention task joins"), Some(43), "shutdown cancellation cannot drop journaled work already admitted" ); - assert_eq!(administration.try_with_writer(|| async { 44_u8 }).await, Some(44)); + assert_eq!( + administration.try_with_writer(|| async { 44_u8 }).await, + Some(44) + ); } #[tokio::test] @@ -2066,7 +2154,9 @@ mod tests { .await; }) .expect("start competing store writer"); - writer_entered.await.expect("unrelated store writer owns shared gate"); + writer_entered + .await + .expect("unrelated store writer owns shared gate"); for _ in 0..super::MAINTENANCE_WRITER_FAST_DEFERRALS { let deferred = coordinator .run_tick( @@ -2080,7 +2170,11 @@ mod tests { ) .await; assert_eq!(deferred, MaintenanceTickOutcome::Retry); - assert_eq!(generation_files(), 6, "fast probe must not collect under a writer"); + assert_eq!( + generation_files(), + 6, + "fast probe must not collect under a writer" + ); } let release_after_queue = std::thread::spawn(move || { @@ -2100,7 +2194,10 @@ mod tests { .await; release_after_queue.join().expect("writer release thread"); assert!( - matches!(writer.wait().await, native_runtime::NativeTaskOutcome::Completed(())), + matches!( + writer.wait().await, + native_runtime::NativeTaskOutcome::Completed(()) + ), "competing writer must complete normally" ); assert_eq!( From 41362a6527bdfea1e8609f370c25f56340fbe743 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 02:59:01 -0800 Subject: [PATCH 06/14] fix(retention): queue maintenance per project without global stalls Use a fair store-local maintenance lane shared by owner/content writers, while retaining daemon-wide exclusivity for truly profile-wide work. Queue verified project retention only against that project's writers and preserve independent project opens during a long running sync. Retain nonqueued daemon-wide probes for session work that lacks proven per-store identity. Exercise same-store fairness, independent writer progress, cancellation and a real registered-project GC tick. Refs #4616 --- .../src/writer_gate.rs | 117 ++++++++++++++++- crates/tracedecay/src/daemon/branch_admin.rs | 32 ++++- crates/tracedecay/src/daemon/maintenance.rs | 118 +++++++++++++----- 3 files changed, 224 insertions(+), 43 deletions(-) diff --git a/crates/tracedecay-store-runtime/src/writer_gate.rs b/crates/tracedecay-store-runtime/src/writer_gate.rs index 668af59240..a03ffa5cdf 100644 --- a/crates/tracedecay-store-runtime/src/writer_gate.rs +++ b/crates/tracedecay-store-runtime/src/writer_gate.rs @@ -15,9 +15,10 @@ //! # The lock hierarchy //! //! ```text -//! daemon: RwLock Daemon scope = write, Store scope = read -//! ├── owner: Mutex -//! └── content: Mutex +//! daemon: RwLock Daemon scope = write, store scopes = read +//! └── store: RwLock Maintenance = write, owner/content = read +//! ├── owner: Mutex +//! └── content: Mutex //! ``` //! //! Every acquisition takes these in exactly that order, so no lock-order @@ -66,6 +67,8 @@ pub enum WriterScope { data_root: PathBuf, class: StoreWriterClass, }, + /// Fair exclusive maintenance for one store, without blocking unrelated stores. + StoreMaintenance { data_root: PathBuf }, } impl WriterScope { @@ -79,6 +82,15 @@ impl WriterScope { class, } } + + /// Same canonical store identity as [`Self::store`], with exclusive admission + /// only against this store's owner and content operations. + #[must_use] + pub fn maintenance(data_root: impl Into) -> Self { + Self::StoreMaintenance { + data_root: data_root.into(), + } + } } /// The logical lanes of one store family. Held by `Arc` from the registry and by @@ -86,6 +98,7 @@ impl WriterScope { /// its weak entry. #[derive(Default)] struct StoreGate { + maintenance: Arc>, owner: Arc>, content: Arc>, } @@ -105,6 +118,7 @@ impl StoreGate { /// exist to pin the guards. pub struct WriterAdmissionGuard { _class: Option>, + _maintenance: Option, _daemon: DaemonGuard, /// Keeps the store's gate alive for as long as it is held, so a registry /// prune between acquisition and release cannot hand a second acquirer a @@ -120,6 +134,13 @@ enum DaemonGuard { Exclusive(OwnedRwLockWriteGuard<()>), } +/// Store-local readers and the queued exclusive retention guard. +#[allow(dead_code)] +enum StoreMaintenanceGuard { + Shared(OwnedRwLockReadGuard<()>), + Exclusive(OwnedRwLockWriteGuard<()>), +} + /// Writer-gate registry: one process-wide lane plus one lane set per store family. pub struct StoreWriterGates { daemon: Arc>, @@ -173,6 +194,7 @@ impl StoreWriterGates { match scope { WriterScope::Daemon => WriterAdmissionGuard { _class: None, + _maintenance: None, _daemon: DaemonGuard::Exclusive( tracing::Instrument::instrument( Arc::clone(&self.daemon).write_owned(), @@ -189,6 +211,11 @@ impl StoreWriterGates { ) .await; let gate = self.store_gate(data_root); + let maintenance = tracing::Instrument::instrument( + Arc::clone(&gate.maintenance).read_owned(), + tracing::trace_span!("daemon.writer_gate.acquire.store_read"), + ) + .await; let class_guard = tracing::Instrument::instrument( gate.class_mutex(*class).lock_owned(), tracing::trace_span!("daemon.writer_gate.acquire.class"), @@ -196,6 +223,26 @@ impl StoreWriterGates { .await; WriterAdmissionGuard { _class: Some(class_guard), + _maintenance: Some(StoreMaintenanceGuard::Shared(maintenance)), + _daemon: DaemonGuard::Shared(daemon), + _gate: Some(gate), + } + } + WriterScope::StoreMaintenance { data_root } => { + let daemon = tracing::Instrument::instrument( + Arc::clone(&self.daemon).read_owned(), + tracing::trace_span!("daemon.writer_gate.acquire.store"), + ) + .await; + let gate = self.store_gate(data_root); + let maintenance = tracing::Instrument::instrument( + Arc::clone(&gate.maintenance).write_owned(), + tracing::trace_span!("daemon.writer_gate.acquire.store_maintenance"), + ) + .await; + WriterAdmissionGuard { + _class: None, + _maintenance: Some(StoreMaintenanceGuard::Exclusive(maintenance)), _daemon: DaemonGuard::Shared(daemon), _gate: Some(gate), } @@ -209,15 +256,29 @@ impl StoreWriterGates { match scope { WriterScope::Daemon => Some(WriterAdmissionGuard { _class: None, + _maintenance: None, _daemon: DaemonGuard::Exclusive(Arc::clone(&self.daemon).try_write_owned().ok()?), _gate: None, }), WriterScope::Store { data_root, class } => { let daemon = Arc::clone(&self.daemon).try_read_owned().ok()?; let gate = self.store_gate(data_root); + let maintenance = Arc::clone(&gate.maintenance).try_read_owned().ok()?; let class_guard = gate.class_mutex(*class).try_lock_owned().ok()?; Some(WriterAdmissionGuard { _class: Some(class_guard), + _maintenance: Some(StoreMaintenanceGuard::Shared(maintenance)), + _daemon: DaemonGuard::Shared(daemon), + _gate: Some(gate), + }) + } + WriterScope::StoreMaintenance { data_root } => { + let daemon = Arc::clone(&self.daemon).try_read_owned().ok()?; + let gate = self.store_gate(data_root); + let maintenance = Arc::clone(&gate.maintenance).try_write_owned().ok()?; + Some(WriterAdmissionGuard { + _class: None, + _maintenance: Some(StoreMaintenanceGuard::Exclusive(maintenance)), _daemon: DaemonGuard::Shared(daemon), _gate: Some(gate), }) @@ -228,6 +289,8 @@ impl StoreWriterGates { #[cfg(test)] mod tests { + use std::future::{Future, poll_fn}; + use std::task::Poll; use std::time::Duration; use super::*; @@ -322,6 +385,54 @@ mod tests { ); } + #[tokio::test] + async fn queued_store_maintenance_preserves_other_project_admission() { + let gates = StoreWriterGates::default(); + let held = gates.acquire(&store("/a", StoreWriterClass::Content)).await; + let maintenance_scope = WriterScope::maintenance("/a"); + let queued = gates.acquire(&maintenance_scope); + tokio::pin!(queued); + poll_fn(|cx| { + assert!(queued.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + + let following = gates.acquire(&store("/a", StoreWriterClass::Owner)); + tokio::pin!(following); + poll_fn(|cx| { + assert!( + following.as_mut().poll(cx).is_pending(), + "same-store operations wait behind queued maintenance" + ); + Poll::Ready(()) + }) + .await; + + assert!( + gates + .try_acquire(&store("/b", StoreWriterClass::Owner)) + .is_some(), + "an unrelated project can open while maintenance queues behind a long sync" + ); + drop(held); + let admitted = tokio::time::timeout(Duration::from_secs(5), &mut queued) + .await + .expect("queued maintenance acquires when its own store is free"); + assert!( + gates + .try_acquire(&store("/a", StoreWriterClass::Content)) + .is_none(), + "admitted maintenance excludes writes to its own store" + ); + drop(admitted); + let follower = tokio::time::timeout(Duration::from_secs(5), &mut following) + .await + .expect("same-store owner resumes after maintenance"); + drop(follower); + assert!(gates.try_acquire(&maintenance_scope).is_some()); + } + #[tokio::test] async fn retired_store_gates_are_reclaimed() { let gates = StoreWriterGates::default(); diff --git a/crates/tracedecay/src/daemon/branch_admin.rs b/crates/tracedecay/src/daemon/branch_admin.rs index e521ed4b89..43ae742c7c 100644 --- a/crates/tracedecay/src/daemon/branch_admin.rs +++ b/crates/tracedecay/src/daemon/branch_admin.rs @@ -125,6 +125,16 @@ pub(super) fn store_writer_scope(data_root: &Path, class: StoreWriterClass) -> W } } +/// Resolves maintenance to the same canonical store gate as logical writers. +/// A missing or unresolvable store refuses isolation by falling back to daemon +/// administration rather than silently splitting the writer identity. +pub(super) fn store_maintenance_writer_scope(data_root: &Path) -> WriterScope { + match authority::canonical_identity_path(data_root) { + Ok(canonical) => WriterScope::maintenance(canonical), + Err(_) => WriterScope::Daemon, + } +} + /// Owner-lane scope for one registered database owner. /// /// [`StoreOwnerKey::store_root`](tracedecay_session_runtime::StoreOwnerKey) is already the @@ -1923,20 +1933,20 @@ impl StoreAdministration { operation().await } - /// Cancellation applies only while queued for daemon-wide writer admission. + /// Cancellation applies only while queued for writer admission. /// Once the guard is acquired, the operation must finish cooperatively: /// dropping an in-flight journal/retention future would abandon physical /// blocking work whose client survives the daemon's shutdown join. pub(super) async fn with_writer_cancellable_acquisition( &self, cancellation: &tracedecay_runtime_core::cancellation::CancellationToken, + scope: WriterScope, operation: Operation, ) -> Option where Operation: FnOnce() -> OperationFuture, OperationFuture: Future, { - let scope = WriterScope::Daemon; let _writer = tokio::select! { biased; () = cancellation.cancelled() => return None, @@ -1955,10 +1965,20 @@ impl StoreAdministration { Operation: FnOnce() -> OperationFuture, OperationFuture: Future, { - let writer = self.gate.try_acquire(&WriterScope::Daemon)?; - let output = operation().await; - drop(writer); - Some(output) + self.try_with_writer_in(WriterScope::Daemon, operation).await + } + + pub(super) async fn try_with_writer_in( + &self, + scope: WriterScope, + operation: Operation, + ) -> Option + where + Operation: FnOnce() -> OperationFuture, + OperationFuture: Future, + { + let _writer = self.gate.try_acquire(&scope)?; + Some(operation().await) } /// Resolves the authenticated client's project layout and runs destructive diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index fa2a8c4965..015842f186 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -26,20 +26,22 @@ use tracedecay_maintenance::tick::{ }; use tracedecay_runtime_core::storage::{RESPONSE_HANDLES_DIRECTORY, profile_sharded_data_root}; -use super::branch_admin::StoreAdministration; +use super::branch_admin::{StoreAdministration, store_maintenance_writer_scope}; use tracedecay_daemon_service::shutdown::DAEMON_TASK_ABORT_DEADLINE; use tracedecay_domain::process_heap::request_idle_thread_collection_v1; use tracedecay_runtime_core::logging::log_daemon_event; use tracedecay_runtime_core::native_runtime::{self, NativeTask, TimeoutError}; +use tracedecay_store_runtime::WriterScope; const MAINTENANCE_STORE_PAGE_LIMIT: usize = 8; const REGISTERED_PROJECT_PAGE_LIMIT: usize = 64; -/// Fast probes preserve low-latency work. After two refusals, join Tokio's -/// fair writer queue so a stream of store-scoped readers cannot starve GC. +/// Fast probes preserve low-latency work. After two store-local refusals, +/// queue behind that store's writers without stalling independent projects. const MAINTENANCE_WRITER_FAST_DEFERRALS: u8 = 2; async fn admit_retention_writer( administration: &StoreAdministration, cancellation: &tracedecay_runtime_core::cancellation::CancellationToken, + scope: WriterScope, prior_deferrals: u8, operation: Operation, ) -> Option @@ -53,12 +55,12 @@ where return None; } if prior_deferrals < MAINTENANCE_WRITER_FAST_DEFERRALS { - administration.try_with_writer(operation).await + administration.try_with_writer_in(scope, operation).await } else { // Only cancel the wait: once admitted, journal/retention work must // finish cooperatively rather than be dropped at the writer boundary. administration - .with_writer_cancellable_acquisition(cancellation, operation) + .with_writer_cancellable_acquisition(cancellation, scope, operation) .await } } @@ -690,7 +692,10 @@ impl Default for MaintenanceCoordinator { /// retirements. A store unmounted since the listing is skipped. enum MaintenanceStoreWork { Session(PathBuf), - Graph(PathBuf), + Graph { + database_path: PathBuf, + data_root: PathBuf, + }, /// The code-index scopes of a registered project that no mounted graph /// owns; its database is never opened by this unit. Registered(RegisteredProjectStoreV1), @@ -699,10 +704,21 @@ enum MaintenanceStoreWork { impl MaintenanceStoreWork { fn database_path(&self) -> Option<&Path> { match self { - Self::Session(database_path) | Self::Graph(database_path) => Some(database_path), + Self::Session(database_path) | Self::Graph { database_path, .. } => Some(database_path), Self::Registered(_) => None, } } + + fn writer_scope(&self) -> WriterScope { + match self { + // Session-wide administration has no independently proven canonical + // store gate. Keep its global probe nonqueued rather than blocking + // unrelated project opens behind a potentially long session pass. + Self::Session(_) => WriterScope::Daemon, + Self::Graph { data_root, .. } => store_maintenance_writer_scope(data_root), + Self::Registered(store) => store_maintenance_writer_scope(&store.data_root), + } + } } /// What a tick reads from one mounted project graph, copied out at listing. @@ -1081,7 +1097,10 @@ impl MaintenanceCoordinator { for graph in &project_graphs { work.push(( graph.key.clone(), - MaintenanceStoreWork::Graph(graph.database_path.clone()), + MaintenanceStoreWork::Graph { + database_path: graph.database_path.clone(), + data_root: graph.data_root.clone(), + }, )); } for (project_id, store) in registered_projects { @@ -1122,7 +1141,7 @@ impl MaintenanceCoordinator { .collect::>(); scan_data_roots.extend(work.iter().filter_map(|(_, unit)| match unit { MaintenanceStoreWork::Registered(store) => Some(store.data_root.clone()), - MaintenanceStoreWork::Session(_) | MaintenanceStoreWork::Graph(_) => None, + MaintenanceStoreWork::Session(_) | MaintenanceStoreWork::Graph { .. } => None, })); maintenance_observations.retain_text_artifact_scan_projects(&scan_data_roots); } @@ -1161,16 +1180,24 @@ impl MaintenanceCoordinator { outcome = MaintenanceTickOutcome::Retry; break; } - let prior_deferrals = self - .writer_deferrals - .lock() - .await - .get(&keys[index]) - .copied() - .unwrap_or(0); + let scope = work[index].1.writer_scope(); + // Only store-local scopes may queue. A queued global writer would + // block every unrelated project while one long sync holds a read gate. + let local = !matches!(&scope, WriterScope::Daemon); + let prior_deferrals = if local { + self.writer_deferrals + .lock() + .await + .get(&keys[index]) + .copied() + .unwrap_or(0) + } else { + 0 + }; let admitted = admit_retention_writer( administration, &self.cancellation, + scope, prior_deferrals, || async { let Ok(_clients) = administration.maintenance_store_clients().try_lock() else { @@ -1193,7 +1220,7 @@ impl MaintenanceCoordinator { run_registered_store_retention(&database, retention).await } } - MaintenanceStoreWork::Graph(_) => { + MaintenanceStoreWork::Graph { .. } => { let Some(graph) = administration .mounted_project_graphs() .await @@ -1231,7 +1258,7 @@ impl MaintenanceCoordinator { let mut deferrals = self.writer_deferrals.lock().await; if admitted.is_some() { deferrals.remove(&keys[index]); - } else if !self.cancellation.is_cancelled() { + } else if local && !self.cancellation.is_cancelled() { deferrals.insert(keys[index].clone(), prior_deferrals.saturating_add(1)); } } @@ -1602,9 +1629,9 @@ mod tests { use tracedecay_runtime_core::cancellation::CancellationToken; use tracedecay_runtime_core::native_runtime; - /// Hold an unrelated store writer across two failed fast probes. The third - /// retention attempt must queue behind it and complete even as more store - /// writers arrive, rather than repeatedly taking non-queued snapshots. + /// Hold a writer for the target store across two failed fast probes. + /// The third retention attempt joins its local queue; independent project + /// opens must remain admitted while that writer is still active. #[tokio::test] async fn retention_queues_after_bounded_writer_deferrals() { let administration = StoreAdministration::default(); @@ -1614,7 +1641,7 @@ mod tests { let holder = tokio::spawn(async move { holding_admin .with_writer_in( - WriterScope::store("/unrelated-store", StoreWriterClass::Content), + WriterScope::store("/retained-store", StoreWriterClass::Content), move || async move { entered_tx.send(()).expect("store writer entered"); release_rx.await.expect("store writer released"); @@ -1624,21 +1651,25 @@ mod tests { }); entered_rx .await - .expect("writer acquired its shared daemon gate"); + .expect("writer acquired its store gate"); let cancellation = CancellationToken::new(); for refused in 0..super::MAINTENANCE_WRITER_FAST_DEFERRALS { assert_eq!( - super::admit_retention_writer(&administration, &cancellation, refused, || async { - 41_u8 - }) + super::admit_retention_writer( + &administration, + &cancellation, + WriterScope::maintenance("/retained-store"), + refused, + || async { 41_u8 }, + ) .await, None, - "a store writer blocks the non-queued daemon probe" + "a store writer blocks the nonqueued store-local probe" ); } // Establish ongoing writer activity before queueing the exclusive - // maintenance request. A fair queued writer blocks later readers; + // maintenance request. A fair queued maintenance writer blocks later readers of that store; // waiting for their progress before releasing the holder deadlocks. let competing_admin = administration.clone(); let writer_entries = Arc::new(AtomicUsize::new(0)); @@ -1669,6 +1700,7 @@ mod tests { let queued = super::admit_retention_writer( &administration, &cancellation, + WriterScope::maintenance("/retained-store"), super::MAINTENANCE_WRITER_FAST_DEFERRALS, || async { 42_u8 }, ); @@ -1682,7 +1714,7 @@ mod tests { }) .await; let later_writer = administration.with_writer_in( - WriterScope::store("/later-store", StoreWriterClass::Content), + WriterScope::store("/retained-store", StoreWriterClass::Owner), || async { 43_u8 }, ); tokio::pin!(later_writer); @@ -1695,6 +1727,17 @@ mod tests { }) .await; + // An independent store must still open while maintenance is queued. + let independent = administration.with_writer_in( + WriterScope::store("/independent-store", StoreWriterClass::Owner), + || async { 44_u8 }, + ); + assert_eq!( + tokio::time::timeout(Duration::from_secs(2), independent) + .await + .expect("unrelated project open must not stall behind queued maintenance"), + 44 + ); let before_release = writer_entries.load(Ordering::SeqCst); release_tx.send(()).expect("release the original writer"); assert_eq!( @@ -1735,6 +1778,7 @@ mod tests { super::admit_retention_writer( &administration, &cancellation, + WriterScope::maintenance("/held-store"), prior_deferrals, || { constructed.fetch_add(1, Ordering::SeqCst); @@ -1779,6 +1823,7 @@ mod tests { let waiting = super::admit_retention_writer( &administration, &cancellation, + WriterScope::maintenance("/held-store"), super::MAINTENANCE_WRITER_FAST_DEFERRALS, || { constructed.fetch_add(1, Ordering::SeqCst); @@ -1837,6 +1882,7 @@ mod tests { super::admit_retention_writer( &working, &stopping, + WriterScope::maintenance("/retention-store"), super::MAINTENANCE_WRITER_FAST_DEFERRALS, move || async move { entered_tx @@ -2132,19 +2178,23 @@ mod tests { ); drop(release); - // A long-lived, unrelated content writer holds the daemon read gate. + // A long-lived writer for this project holds the store-local read gate. // Two real retention ticks must defer, and the next must queue rather // than missing the same writer again. The fixture's six on-disk // generations make forward progress observable through actual GC. let (writer_started, writer_entered) = tokio::sync::oneshot::channel(); let (release_writer, writer_released) = tokio::sync::oneshot::channel(); let holding_admin = administration.clone(); + let retained_data_root = tracedecay_runtime_core::storage::profile_sharded_data_root( + &profile_root, + &project.project_id, + ); let writer = NativeTask::spawn(async move { holding_admin .with_writer_in( - crate::daemon::branch_admin::WriterScope::store( - "/unrelated-project", - crate::daemon::branch_admin::StoreWriterClass::Content, + crate::daemon::branch_admin::store_writer_scope( + &retained_data_root, + StoreWriterClass::Content, ), move || async move { writer_started.send(()).expect("writer admitted"); @@ -2156,7 +2206,7 @@ mod tests { .expect("start competing store writer"); writer_entered .await - .expect("unrelated store writer owns shared gate"); + .expect("project content writer owns shared store gate"); for _ in 0..super::MAINTENANCE_WRITER_FAST_DEFERRALS { let deferred = coordinator .run_tick( From 2bd73a4fac5be20797837e421ba976478ffb7912 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 05:24:10 -0800 Subject: [PATCH 07/14] style(retention): match Bazel rustfmt on admission paths --- crates/tracedecay/src/daemon/branch_admin.rs | 3 ++- crates/tracedecay/src/daemon/maintenance.rs | 4 +--- 2 files changed, 3 insertions(+), 4 deletions(-) diff --git a/crates/tracedecay/src/daemon/branch_admin.rs b/crates/tracedecay/src/daemon/branch_admin.rs index 43ae742c7c..5c89131616 100644 --- a/crates/tracedecay/src/daemon/branch_admin.rs +++ b/crates/tracedecay/src/daemon/branch_admin.rs @@ -1965,7 +1965,8 @@ impl StoreAdministration { Operation: FnOnce() -> OperationFuture, OperationFuture: Future, { - self.try_with_writer_in(WriterScope::Daemon, operation).await + self.try_with_writer_in(WriterScope::Daemon, operation) + .await } pub(super) async fn try_with_writer_in( diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index 015842f186..8e7e541def 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -1649,9 +1649,7 @@ mod tests { ) .await; }); - entered_rx - .await - .expect("writer acquired its store gate"); + entered_rx.await.expect("writer acquired its store gate"); let cancellation = CancellationToken::new(); for refused in 0..super::MAINTENANCE_WRITER_FAST_DEFERRALS { assert_eq!( From 5c37cdedf9b8bea37e61961257c3a65502740853 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 07:24:51 -0800 Subject: [PATCH 08/14] fix(retention): keep queued test scope alive for Bazel The store-local writer admission regression borrowed a temporary WriterScope across a pinned asynchronous acquisition. Native Bazel Rust compilation fails with E0716 at writer_gate.rs:401; retain the exact scope in a binding until that future completes. Existing behavioral regressions continue proving fairness and cross-project isolation. Refs #4616 --- crates/tracedecay-store-runtime/src/writer_gate.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay-store-runtime/src/writer_gate.rs b/crates/tracedecay-store-runtime/src/writer_gate.rs index a03ffa5cdf..e2d572d8e2 100644 --- a/crates/tracedecay-store-runtime/src/writer_gate.rs +++ b/crates/tracedecay-store-runtime/src/writer_gate.rs @@ -398,7 +398,8 @@ mod tests { }) .await; - let following = gates.acquire(&store("/a", StoreWriterClass::Owner)); + let following_scope = store("/a", StoreWriterClass::Owner); + let following = gates.acquire(&following_scope); tokio::pin!(following); poll_fn(|cx| { assert!( From 0b2a0621ca2616a9f7fe0cba95508048cae1d169 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 13:31:34 -0800 Subject: [PATCH 09/14] test(retention): queue the cancellation follower on the held store --- crates/tracedecay/src/daemon/maintenance.rs | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index 8e7e541def..68eb3a5026 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -1835,7 +1835,7 @@ mod tests { }) .await; let follower = administration.with_writer_in( - WriterScope::store("/follower-store", StoreWriterClass::Content), + WriterScope::store("/held-store", StoreWriterClass::Content), || async { 8_u8 }, ); tokio::pin!(follower); @@ -1852,14 +1852,19 @@ mod tests { None ); assert_eq!(constructed.load(Ordering::SeqCst), 0); + poll_fn(|cx| { + assert!(follower.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + release_tx.send(()).expect("release held writer"); + writer.await.expect("held writer exits"); assert_eq!( tokio::time::timeout(Duration::from_secs(2), &mut follower) .await - .expect("cancellation removes the queued writer before the holder exits"), + .expect("next store writer proceeds after the holder exits"), 8 ); - release_tx.send(()).expect("release held writer"); - writer.await.expect("held writer exits"); assert_eq!( administration.try_with_writer(|| async { 9_u8 }).await, Some(9) From 0dbda7ea574b8576b5a8c5ec4822001b45bd09b2 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 01:52:34 -0800 Subject: [PATCH 10/14] test(automation): bind scheduler fixtures to backend identities Repair three hosted automation_runner_test failures by supplying the configured executable identity required by the production prefilter. Use the same identity for the prefilter and retained runner, preserving manual-schedule, exact-ledger, live-lock and lock-race assertions. Add a production-prefilter regression for unconfigured refusal, deduplication and same-occurrence recovery after binding, with zero backend calls and no effectful terminal. Do not relax backend admission, change timeouts, or alter the graph publication changes. Refs #4532 --- .../tests/automation_runner_test/jobs.rs | 124 ++++++++++++++++-- 1 file changed, 110 insertions(+), 14 deletions(-) diff --git a/crates/tracedecay/tests/automation_runner_test/jobs.rs b/crates/tracedecay/tests/automation_runner_test/jobs.rs index 740283d56e..7342786381 100644 --- a/crates/tracedecay/tests/automation_runner_test/jobs.rs +++ b/crates/tracedecay/tests/automation_runner_test/jobs.rs @@ -86,6 +86,7 @@ fn enabled_job_config() -> AutomationSettingsV1 { struct ContentBackend { calls: AtomicUsize, content: &'static str, + executable: Option, } impl ContentBackend { @@ -93,9 +94,15 @@ impl ContentBackend { Self { calls: AtomicUsize::new(0), content, + executable: None, } } + fn with_executable(mut self, executable: &Path) -> Self { + self.executable = Some(executable.to_path_buf()); + self + } + fn calls(&self) -> usize { self.calls.load(Ordering::SeqCst) } @@ -126,7 +133,7 @@ impl AgentTaskBackend for ContentBackend { } fn executable(&self) -> Option<&std::path::Path> { - None + self.executable.as_deref() } } @@ -988,6 +995,10 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let mut job = sample_job("sched-job"); job.schedule = Some("manual".to_string()); let config = enabled_job_config(); + // Reach the schedule gate with a bound identity, not the earlier + // backend-unconfigured refusal. This gate never invokes the backend. + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = ContentBackend::new("unused").with_executable(executable.path()); // Manual schedule: every scheduler tick of the same occurrence reaches // the prefilter, which records the skip once. @@ -996,7 +1007,7 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let run = evaluate_and_record_scheduler_skip( &dashboard_root, &config, - None, + backend.executable(), &job, "sched-occurrence", None, @@ -1019,7 +1030,7 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { let repeated = evaluate_and_record_scheduler_skip( &dashboard_root, &config, - None, + backend.executable(), &job, "sched-occurrence", None, @@ -1031,6 +1042,76 @@ async fn scheduler_trigger_skips_repeat_and_respects_lock_discipline() { repeated.ledger_record, records[0], "an immediate repeat skip must return the exact durable prior row" ); + assert_eq!(backend.calls(), 0); +} + +#[tokio::test] +async fn scheduler_prefilter_rejects_unconfigured_backend_and_recovers_without_effect() { + let temp = tempdir().unwrap(); + let dashboard_root = temp.path().join("dashboard"); + let job = sample_job("backend-recovery"); + let config = enabled_job_config(); + let occurrence = "backend-recovery-occurrence"; + let backend = ContentBackend::new("unused"); + + let first = evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .expect("a due job still requires a configured backend"); + assert_eq!(first.report["reason"], json!("backend_unconfigured")); + assert_eq!(first.ledger_record.status, AutomationRunStatus::Skipped); + assert_ne!(first.run_id, occurrence); + let repeated = evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .unwrap(); + assert_eq!(repeated.ledger_record, first.ledger_record); + + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = backend.with_executable(executable.path()); + assert!( + evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .is_none(), + "binding the backend must admit the same due occurrence without cooldown" + ); + assert_eq!(backend.calls(), 0); + assert_eq!( + load_run_records(&dashboard_root, 10).await.unwrap(), + vec![first.ledger_record], + "advisory admission must not append an effectful terminal" + ); + assert!( + tracedecay_automation_runtime::automation::run_ledger::find_run_record_exact_bounded( + &dashboard_root, + occurrence, + ) + .await + .unwrap() + .is_none() + ); } #[tokio::test] @@ -1206,15 +1287,22 @@ async fn scheduler_prefilter_live_lock_wins_over_not_due_summary() { let config = enabled_job_config(); let occurrence = "locked-prefilter-occurrence"; - let prefilter = - evaluate_and_record_scheduler_skip(&dashboard_root, &config, None, &job, occurrence, None) - .await - .unwrap() - .expect("the live job lock must produce a diagnostic"); + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = ContentBackend::new("unused").with_executable(executable.path()); + let prefilter = evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .expect("the live job lock must produce a diagnostic"); assert_eq!(prefilter.report["reason"], json!("scheduler_lock_active")); assert_ne!(prefilter.run_id, occurrence); - let backend = ContentBackend::new("unused"); let admitted_job = tracedecay_automation_runtime::automation::jobs::admit_job(&job).unwrap(); let retained = run_user_job_with_backend_for_retained_settlement( &dashboard_root, @@ -1262,12 +1350,21 @@ async fn retained_scheduler_runner_reacquires_after_due_prefilter() { job.schedule = Some("every 1h".to_owned()); let config = enabled_job_config(); let occurrence = "prefilter-gap-occurrence"; + let executable = tempfile::NamedTempFile::new().unwrap(); + let backend = ContentBackend::new("unused").with_executable(executable.path()); assert!( - evaluate_and_record_scheduler_skip(&dashboard_root, &config, None, &job, occurrence, None) - .await - .unwrap() - .is_none(), + evaluate_and_record_scheduler_skip( + &dashboard_root, + &config, + backend.executable(), + &job, + occurrence, + None, + ) + .await + .unwrap() + .is_none(), "a new interval job is advisory-due before effect admission" ); let _raced_in = @@ -1276,7 +1373,6 @@ async fn retained_scheduler_runner_reacquires_after_due_prefilter() { .unwrap() .expect("the advisory prefilter must release its reservation lock"); - let backend = ContentBackend::new("unused"); let admitted_job = tracedecay_automation_runtime::automation::jobs::admit_job(&job).unwrap(); let retained = run_user_job_with_backend_for_retained_settlement( &dashboard_root, From 0b520f3c26ba551c61b8913cd9702eff87dd5300 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 13:51:16 -0800 Subject: [PATCH 11/14] test(retention): verify content resumes before owner release --- crates/tracedecay/src/daemon/maintenance.rs | 13 +++++-------- 1 file changed, 5 insertions(+), 8 deletions(-) diff --git a/crates/tracedecay/src/daemon/maintenance.rs b/crates/tracedecay/src/daemon/maintenance.rs index 68eb3a5026..a62aa22d87 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -1852,19 +1852,16 @@ mod tests { None ); assert_eq!(constructed.load(Ordering::SeqCst), 0); - poll_fn(|cx| { - assert!(follower.as_mut().poll(cx).is_pending()); - Poll::Ready(()) - }) - .await; - release_tx.send(()).expect("release held writer"); - writer.await.expect("held writer exits"); + // Owner and content otherwise run concurrently. Only queued exclusive + // maintenance blocked this content writer on the same store. assert_eq!( tokio::time::timeout(Duration::from_secs(2), &mut follower) .await - .expect("next store writer proceeds after the holder exits"), + .expect("content writer resumes while the owner is still active"), 8 ); + release_tx.send(()).expect("release held writer"); + writer.await.expect("held writer exits"); assert_eq!( administration.try_with_writer(|| async { 9_u8 }).await, Some(9) From b299f0047691424fee71962344c0a2cb337cc8c8 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 15:06:56 -0800 Subject: [PATCH 12/14] test(cli): time held responses after request admission --- .../tool_client_transport.rs | 55 +++++++++---------- 1 file changed, 25 insertions(+), 30 deletions(-) diff --git a/crates/tracedecay/tests/runtime_acceptance_suite/tool_client_transport.rs b/crates/tracedecay/tests/runtime_acceptance_suite/tool_client_transport.rs index b0f714cdff..e45871856b 100644 --- a/crates/tracedecay/tests/runtime_acceptance_suite/tool_client_transport.rs +++ b/crates/tracedecay/tests/runtime_acceptance_suite/tool_client_transport.rs @@ -545,34 +545,12 @@ fn hold_stream_after_files( let _ = BufReader::new(stream).read_line(&mut extra); } -fn assert_complete_json_document(result: &ChildResult) -> Value { - assert!( - !result.killed_by_harness, - "CLI --json hung after writing files: elapsed={:?} stdout_len={} stderr={}", - result.elapsed, - result.output.stdout.len(), - String::from_utf8_lossy(&result.output.stderr) - ); - assert!( - result.output.status.success(), - "{}", - String::from_utf8_lossy(&result.output.stderr) - ); - serde_json::from_slice(&result.output.stdout).unwrap_or_else(|error| { - panic!( - "--json must be one complete parseable document ({error}); bytes={} head={}", - result.output.stdout.len(), - String::from_utf8_lossy(&result.output.stdout[..result.output.stdout.len().min(200)]) - ) - }) -} - #[test] fn generic_tool_json_preserves_large_result_while_daemon_waits() { let (_home, _project, _socket_dir, home, project, socket) = fixture(); let files = oversized_files(); let expected = files.count; - let (_requests, server) = spawn_scripted_daemon( + let (requests, server) = spawn_scripted_daemon( socket.clone(), &home, &project, @@ -581,12 +559,29 @@ fn generic_tool_json_preserves_large_result_while_daemon_waits() { hold_stream_after_files(stream, request, scope, files.clone()); }, ); - let result = run_command_with_timeout( - tool_command(&home, &project, &socket, "large"), - CHILD_TIMEOUT, - ); + let child = tool_command(&home, &project, &socket, "large") + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn CLI for held response"); + let mut child = common::TestChildProcess::new(child); + // Measure completion while the admitted request's connection remains open. + // The daemon has its own bounded acceptance window for cold CLI startup. + requests + .recv_timeout(ACCEPT_HANG_GUARD) + .expect("daemon received the large-result request"); + let output = child + .wait_with_output(CHILD_TIMEOUT) + .expect("CLI must finish without waiting for daemon EOF"); server.join().expect("join scripted daemon"); - let envelope = assert_complete_json_document(&result); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let envelope: Value = + serde_json::from_slice(&output.stdout).expect("--json must contain one complete document"); assert_eq!(envelope["isError"], false, "{envelope}"); assert_eq!(envelope["structuredContent"]["count"], expected); let files = envelope["structuredContent"]["files"] @@ -594,9 +589,9 @@ fn generic_tool_json_preserves_large_result_while_daemon_waits() { .unwrap_or_else(|| panic!("held-stream --json must keep the full listing: {envelope}")); assert_eq!(files.len(), expected); assert!( - result.output.stdout.len() > 65_536, + output.stdout.len() > 65_536, "held-stream --json must keep the full listing above 64KiB: {}", - result.output.stdout.len() + output.stdout.len() ); } From fedc68eb803da5f2d88e0af151450ee09e7980dd Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 15:45:00 -0800 Subject: [PATCH 13/14] test(runtime): bound CLI startup before scripted daemon request Refs #4616 --- .../tool_client_transport.rs | 29 +++++++++++++++---- 1 file changed, 24 insertions(+), 5 deletions(-) diff --git a/crates/tracedecay/tests/runtime_acceptance_suite/tool_client_transport.rs b/crates/tracedecay/tests/runtime_acceptance_suite/tool_client_transport.rs index e45871856b..020a5d17f4 100644 --- a/crates/tracedecay/tests/runtime_acceptance_suite/tool_client_transport.rs +++ b/crates/tracedecay/tests/runtime_acceptance_suite/tool_client_transport.rs @@ -566,11 +566,30 @@ fn generic_tool_json_preserves_large_result_while_daemon_waits() { .spawn() .expect("spawn CLI for held response"); let mut child = common::TestChildProcess::new(child); - // Measure completion while the admitted request's connection remains open. - // The daemon has its own bounded acceptance window for cold CLI startup. - requests - .recv_timeout(ACCEPT_HANG_GUARD) - .expect("daemon received the large-result request"); + // Request admission and CLI startup share the original child deadline. + // Poll the child as well as the receipt: waiting only for the daemon's + // 120-second accept leak guard would mask an early CLI exit or failure. + let request_deadline = Instant::now() + CHILD_TIMEOUT; + loop { + match requests.recv_timeout(Duration::from_millis(50)) { + Ok(()) => break, + Err(RecvTimeoutError::Timeout) => { + assert!( + child.try_wait().expect("poll CLI before daemon request").is_none(), + "CLI exited before the fake daemon received its request" + ); + assert!( + Instant::now() < request_deadline, + "CLI did not send a daemon request within the child startup deadline" + ); + } + Err(RecvTimeoutError::Disconnected) => { + panic!("fake daemon stopped before the CLI sent its request") + } + } + } + // Once admitted, prove the full response is written while the daemon + // deliberately holds the connection open, not after an EOF or stall. let output = child .wait_with_output(CHILD_TIMEOUT) .expect("CLI must finish without waiting for daemon EOF"); From d70ad107cabc2ce357f11a9c547cb1d1dd659330 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Sun, 8 Nov 2026 16:29:42 -0800 Subject: [PATCH 14/14] style(cli): format the pre-admission process assertion --- .../tests/runtime_acceptance_suite/tool_client_transport.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/tracedecay/tests/runtime_acceptance_suite/tool_client_transport.rs b/crates/tracedecay/tests/runtime_acceptance_suite/tool_client_transport.rs index 020a5d17f4..49501498fa 100644 --- a/crates/tracedecay/tests/runtime_acceptance_suite/tool_client_transport.rs +++ b/crates/tracedecay/tests/runtime_acceptance_suite/tool_client_transport.rs @@ -575,7 +575,10 @@ fn generic_tool_json_preserves_large_result_while_daemon_waits() { Ok(()) => break, Err(RecvTimeoutError::Timeout) => { assert!( - child.try_wait().expect("poll CLI before daemon request").is_none(), + child + .try_wait() + .expect("poll CLI before daemon request") + .is_none(), "CLI exited before the fake daemon received its request" ); assert!(