diff --git a/crates/tracedecay-cli/tests/core_cli_suite/cli_non_interactive_test.rs b/crates/tracedecay-cli/tests/core_cli_suite/cli_non_interactive_test.rs index 3391fe3c5d6..9c309d9b570 100644 --- a/crates/tracedecay-cli/tests/core_cli_suite/cli_non_interactive_test.rs +++ b/crates/tracedecay-cli/tests/core_cli_suite/cli_non_interactive_test.rs @@ -42,18 +42,20 @@ fn ephemeral_safe_fixture_base(test_path: &str) -> Option { assert!(!tracedecay_global_db::is_ephemeral_path(&root)); return Some(root); } - // Worker scratch paths and macOS's per-user temporary directory can leave - // too little room for the nested child home and its Unix daemon socket. - // Use a short, private root on Unix. The private child TMPDIR - // still defines the ephemeral boundary, with durable fixtures beside it. - let scratch = if cfg!(unix) { - TempDir::new_in("/tmp") + // Linux fleet scratch can exceed the Unix socket path limit. Darwin's + // sandbox grants writes to the user's temporary directory, not /tmp. + let parent = if cfg!(target_os = "linux") { + PathBuf::from("/tmp") } else { - TempDir::new() - } - .expect("isolated temp-policy fixture"); - let temporary = scratch.path().join("temporary"); - let durable = scratch.path().join("durable"); + std::env::temp_dir() + }; + let scratch = tempfile::Builder::new() + .prefix("f") + .tempdir_in(parent) + .expect("isolated temp-policy fixture"); + // These components also prefix the child's 104-byte Darwin socket address. + let temporary = scratch.path().join("t"); + let durable = scratch.path().join("d"); std::fs::create_dir(&temporary).unwrap(); std::fs::create_dir(&durable).unwrap(); crate::common::rerun_test_in_child( @@ -1259,15 +1261,15 @@ fn init_wait_holds_until_first_generation_is_ready() { ); } -/// Unix daemon endpoints must fit even when the runner's temporary directory -/// has a long sandbox prefix. The profile remains private to the fixture. -fn daemon_contract_home() -> TempDir { - if cfg!(unix) { - TempDir::new_in("/tmp") - } else { - TempDir::new() - } - .expect("private daemon fixture home") +/// Allocate a short socket path within the caller's private fixture root. +fn daemon_contract_home(root: &Path) -> TempDir { + // The outer scratch already isolates this home. Keep its component one byte + // so the default socket fits even in the native worker scratch directory. + tempfile::Builder::new() + .prefix("h") + .rand_bytes(0) + .tempdir_in(root) + .expect("private daemon fixture home") } fn assert_project_storage_absent(home: &Path, project: &Path) { @@ -1309,7 +1311,7 @@ fn init_without_a_running_daemon_refuses_before_any_index_write() { ) else { return; }; - let home = daemon_contract_home(); + let home = daemon_contract_home(&durable); let project = tempfile::Builder::new() .prefix("no-daemon-init-project-") .tempdir_in(durable) @@ -1351,7 +1353,7 @@ fn status_and_query_reads_refuse_typed_daemon_unavailable_without_one() { ) else { return; }; - let home = daemon_contract_home(); + let home = daemon_contract_home(&durable); let project = tempfile::Builder::new() .prefix("no-daemon-read-project-") .tempdir_in(durable) @@ -1519,7 +1521,7 @@ fn pricing_and_usage_reads_make_no_http_request() { ) else { return; }; - let home = daemon_contract_home(); + let home = daemon_contract_home(&durable); let project = tempfile::Builder::new() .prefix("local-usage-project-") .tempdir_in(durable) @@ -2836,7 +2838,7 @@ fn list_all_reports_orphan_manifest_reconstructable_store() { ) else { return; }; - let home = TempDir::new().unwrap(); + let home = daemon_contract_home(&std::env::temp_dir()); let project = tempfile::Builder::new() .prefix("list-orphan-project-") .tempdir_in(durable) diff --git a/crates/tracedecay-cli/tests/core_cli_suite/tool_daemon_test.rs b/crates/tracedecay-cli/tests/core_cli_suite/tool_daemon_test.rs index 0c65098e48c..429644ba3c3 100644 --- a/crates/tracedecay-cli/tests/core_cli_suite/tool_daemon_test.rs +++ b/crates/tracedecay-cli/tests/core_cli_suite/tool_daemon_test.rs @@ -543,21 +543,6 @@ fn configuration_tool_success( } } -fn wait_for_daemon_socket(socket_path: &Path) { - let deadline = Instant::now() + Duration::from_secs(10); - common::poll_until( - deadline, - Duration::from_millis(25), - || UnixStream::connect(socket_path).is_ok().then_some(()), - || { - format!( - "timed out waiting for daemon socket at {}", - socket_path.display() - ) - }, - ); -} - /// Enroll a project for the native hook capture contract: an enrollment /// marker binds the project root to a profile shard, and a daemon-issued /// hook configuration binding is published under that shard's data root so @@ -1718,21 +1703,9 @@ fn daemon_socket_is_owner_only() { let home = TempDir::new().unwrap(); let home_path = canonical_existing_path(home.path()); let socket_path = common::daemon_socket_path(&home_path); - let _ = std::fs::remove_file(&socket_path); - let mut daemon = common::DaemonProcess::new( - tracedecay_command_with_home(&home_path) - .arg("daemon") - .arg("run") - .arg("--socket") - .arg(&socket_path) - .stdin(Stdio::null()) - .stdout(Stdio::null()) - .stderr(Stdio::null()) - .spawn() - .expect("tracedecay daemon should start"), - ); - - wait_for_daemon_socket(&socket_path); + let mut daemon = common::spawn_tracedecay_daemon_with(&home_path, |command| { + command.arg("--socket").arg(&socket_path); + }); let mode = std::fs::metadata(&socket_path) .expect("socket metadata") .permissions() diff --git a/crates/tracedecay-store-runtime/src/writer_gate.rs b/crates/tracedecay-store-runtime/src/writer_gate.rs index 668af592408..f3dd1c0ebe7 100644 --- a/crates/tracedecay-store-runtime/src/writer_gate.rs +++ b/crates/tracedecay-store-runtime/src/writer_gate.rs @@ -15,13 +15,18 @@ //! # 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 -//! inversion is possible. +//! Admitted operations hold this hierarchy. A queued store operation first +//! reserves its local resources without holding daemon admission. It only +//! *tries* the daemon read gate with that reservation held; if the daemon is +//! busy, it releases every local resource before awaiting the daemon. Thus +//! no local waiter pins daemon admission, and no task waits for the daemon +//! while holding a resource another daemon reader could need. //! //! # Exclusivity argument //! @@ -66,6 +71,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 +86,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 +102,7 @@ impl WriterScope { /// its weak entry. #[derive(Default)] struct StoreGate { + maintenance: Arc>, owner: Arc>, content: Arc>, } @@ -97,6 +114,28 @@ impl StoreGate { StoreWriterClass::Content => Arc::clone(&self.content), } } + + fn try_reserve( + &self, + class: Option, + ) -> Option<(StoreMaintenanceGuard, Option>)> { + match class { + Some(class) => { + let maintenance = Arc::clone(&self.maintenance).try_read_owned().ok()?; + let class_guard = self.class_mutex(class).try_lock_owned().ok()?; + Some(( + StoreMaintenanceGuard::Shared(maintenance), + Some(class_guard), + )) + } + None => Some(( + StoreMaintenanceGuard::Exclusive( + Arc::clone(&self.maintenance).try_write_owned().ok()?, + ), + None, + )), + } + } } /// Admission held for the duration of one writer operation. @@ -105,6 +144,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 +160,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 +220,7 @@ impl StoreWriterGates { match scope { WriterScope::Daemon => WriterAdmissionGuard { _class: None, + _maintenance: None, _daemon: DaemonGuard::Exclusive( tracing::Instrument::instrument( Arc::clone(&self.daemon).write_owned(), @@ -183,23 +231,78 @@ impl StoreWriterGates { _gate: None, }, WriterScope::Store { data_root, class } => { - 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 class_guard = tracing::Instrument::instrument( - gate.class_mutex(*class).lock_owned(), - tracing::trace_span!("daemon.writer_gate.acquire.class"), - ) - .await; - WriterAdmissionGuard { - _class: Some(class_guard), + self.acquire_store(data_root, Some(*class)).await + } + WriterScope::StoreMaintenance { data_root } => { + self.acquire_store(data_root, None).await + } + } + } + + /// Reserve the store without retaining daemon admission while queued. + /// `None` is exclusive maintenance; owner/content reserve a shared store + /// guard plus their class mutex. + async fn acquire_store( + &self, + data_root: &Path, + class: Option, + ) -> WriterAdmissionGuard { + let gate = self.store_gate(data_root); + loop { + let (maintenance, class_guard) = match class { + Some(class) => { + 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"), + ) + .await; + ( + StoreMaintenanceGuard::Shared(maintenance), + Some(class_guard), + ) + } + None => { + let maintenance = tracing::Instrument::instrument( + Arc::clone(&gate.maintenance).write_owned(), + tracing::trace_span!("daemon.writer_gate.acquire.store_maintenance"), + ) + .await; + (StoreMaintenanceGuard::Exclusive(maintenance), None) + } + }; + if let Ok(daemon) = Arc::clone(&self.daemon).try_read_owned() { + return WriterAdmissionGuard { + _class: class_guard, + _maintenance: Some(maintenance), _daemon: DaemonGuard::Shared(daemon), _gate: Some(gate), - } + }; + } + // Take our fair daemon turn without awaiting any local resource + // under it. An immediate local probe can use this turn even when + // another daemon writer is queued; dropping the turn first could + // starve a store behind continuous daemon-wide work. + drop(class_guard); + drop(maintenance); + let daemon = tracing::Instrument::instrument( + Arc::clone(&self.daemon).read_owned(), + tracing::trace_span!("daemon.writer_gate.acquire.store"), + ) + .await; + if let Some((maintenance, class_guard)) = gate.try_reserve(class) { + return WriterAdmissionGuard { + _class: class_guard, + _maintenance: Some(maintenance), + _daemon: DaemonGuard::Shared(daemon), + _gate: Some(gate), + }; } + drop(daemon); } } @@ -209,25 +312,38 @@ 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 class_guard = gate.class_mutex(*class).try_lock_owned().ok()?; - Some(WriterAdmissionGuard { - _class: Some(class_guard), - _daemon: DaemonGuard::Shared(daemon), - _gate: Some(gate), - }) + self.try_acquire_store(data_root, Some(*class)) } + WriterScope::StoreMaintenance { data_root } => self.try_acquire_store(data_root, None), } } + + fn try_acquire_store( + &self, + data_root: &Path, + class: Option, + ) -> Option { + let daemon = Arc::clone(&self.daemon).try_read_owned().ok()?; + let gate = self.store_gate(data_root); + let (maintenance, class_guard) = gate.try_reserve(class)?; + Some(WriterAdmissionGuard { + _class: class_guard, + _maintenance: Some(maintenance), + _daemon: DaemonGuard::Shared(daemon), + _gate: Some(gate), + }) + } } #[cfg(test)] mod tests { + use std::future::{Future, poll_fn}; + use std::task::Poll; use std::time::Duration; use super::*; @@ -322,6 +438,145 @@ 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_scope = store("/a", StoreWriterClass::Owner); + let following = gates.acquire(&following_scope); + 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 queued_maintenance_yields_to_a_daemon_writer_without_a_lock_cycle() { + let gates = StoreWriterGates::default(); + let held = gates.acquire(&store("/a", StoreWriterClass::Content)).await; + let scope = WriterScope::maintenance("/a"); + let maintenance = gates.acquire(&scope); + tokio::pin!(maintenance); + poll_fn(|cx| { + assert!(maintenance.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + let follower_scope = store("/a", StoreWriterClass::Owner); + let follower = gates.acquire(&follower_scope); + tokio::pin!(follower); + poll_fn(|cx| { + assert!(follower.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + let daemon_scope = WriterScope::Daemon; + let daemon = gates.acquire(&daemon_scope); + tokio::pin!(daemon); + poll_fn(|cx| { + assert!(daemon.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + drop(held); + // Neither queued local future is repolled. Their reservations must + // not retain daemon admission, even after their store becomes free. + let exclusive = tokio::time::timeout(Duration::from_secs(5), &mut daemon) + .await + .expect("daemon writer progresses independently of queued store futures"); + assert!(gates.try_acquire(&scope).is_none()); + poll_fn(|cx| { + assert!(maintenance.as_mut().poll(cx).is_pending()); + assert!(follower.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + drop(exclusive); + let retained = tokio::time::timeout(Duration::from_secs(5), &mut maintenance) + .await + .expect("maintenance resumes after daemon writer"); + drop(retained); + let following = tokio::time::timeout(Duration::from_secs(5), &mut follower) + .await + .expect("store follower resumes after maintenance"); + drop(following); + } + + #[tokio::test] + async fn a_queued_store_uses_its_fair_daemon_turn() { + for scope in [ + WriterScope::maintenance("/a"), + store("/a", StoreWriterClass::Content), + ] { + let gates = StoreWriterGates::default(); + let held = gates.acquire(&WriterScope::Daemon).await; + let store = gates.acquire(&scope); + tokio::pin!(store); + poll_fn(|cx| { + assert!(store.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + let daemon_scope = WriterScope::Daemon; + let following = gates.acquire(&daemon_scope); + tokio::pin!(following); + poll_fn(|cx| { + assert!(following.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + drop(held); + // The later daemon writer stays queued and unpolled. The store + // must consume its earlier shared turn rather than discard it. + let admitted = tokio::time::timeout(Duration::from_secs(5), &mut store) + .await + .expect("store uses its fair daemon turn before the next daemon writer"); + assert!(gates.try_acquire(&WriterScope::Daemon).is_none()); + drop(admitted); + let next = tokio::time::timeout(Duration::from_secs(5), &mut following) + .await + .expect("following daemon writer progresses after the store"); + drop(next); + } + } + #[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 d0cae8b43ec..497ff5db3e3 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,6 +1933,16 @@ impl StoreAdministration { operation().await } + /// Transfers only the canonical gate into the supervised retention job. + /// The acquisition retains no mounted graph or project database client. + pub(super) fn retention_writer_acquisition( + &self, + data_root: PathBuf, + ) -> impl Future + Send + 'static { + let gate = Arc::clone(&self.gate); + async move { gate.acquire(&WriterScope::maintenance(data_root)).await } + } + pub(super) async fn try_with_writer( &self, operation: Operation, @@ -1931,10 +1951,21 @@ 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 ea2070195f9..9cc365460f0 100644 --- a/crates/tracedecay/src/daemon/maintenance.rs +++ b/crates/tracedecay/src/daemon/maintenance.rs @@ -1,8 +1,10 @@ -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}; +use futures_util::FutureExt; use tokio::sync::Mutex; use tracedecay_code_index::parallelism::{ collect_idle_installed_worker_heaps, collect_installed_worker_heaps_when_idle, @@ -25,13 +27,203 @@ 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; +/// One owned fair acquisition and its work. Cancellation only governs the +/// acquisition; admitted journaled work finishes cooperatively. +struct RetentionWriterJob { + cancellation: tracedecay_runtime_core::cancellation::CancellationToken, + task: NativeTask>, +} + +/// The coordinator supervises queued store units independently of its cadence. +/// A job consumes its fair turn and runs/releases admission even when no tick +/// can run. Completed outcomes stay here until the next census observes them, +/// preventing duplicate jobs for the same canonical store and work key. +struct RetentionWriterJobs { + pending: std::sync::Mutex>>, +} + +impl Default for RetentionWriterJobs { + fn default() -> Self { + Self { + pending: std::sync::Mutex::new(BTreeMap::new()), + } + } +} + +impl RetentionWriterJobs { + fn cancel_all(&self) { + for job in self + .pending + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .values() + { + job.cancellation.cancel(); + } + } + + fn retain(&self, active: &BTreeSet<(String, PathBuf)>) { + self.pending + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .retain(|key, job| { + if active.contains(key) { + return true; + } + job.cancellation.cancel(); + // Keep admitted work supervised until it settles. Cancelling + // an acquisition releases its queued permit on the next poll. + job.task.try_finished().is_none_or(|outcome| { + record_retention_job_outcome(outcome); + false + }) + }); + } + + async fn shutdown(&self) { + let jobs = std::mem::take( + &mut *self + .pending + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner), + ); + for job in jobs.values() { + job.cancellation.cancel(); + } + for (_, job) in jobs { + join_cancelled_maintenance_task( + Some(job.task), + "retention_store", + DAEMON_TASK_ABORT_DEADLINE, + ) + .await; + } + } +} + +impl Drop for RetentionWriterJobs { + fn drop(&mut self) { + for job in self + .pending + .get_mut() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .values() + { + job.cancellation.cancel(); + } + } +} + +fn record_retention_job_outcome( + outcome: native_runtime::NativeTaskOutcome>, +) -> Option { + match outcome { + native_runtime::NativeTaskOutcome::Completed(output) => output, + native_runtime::NativeTaskOutcome::Cancelled => { + tracing::error!( + event = "retention_store_task_cancelled", + "retention store owner did not settle normally" + ); + None + } + native_runtime::NativeTaskOutcome::Panicked(error) => { + tracing::error!( + event = "retention_store_task_panicked", + ?error, + "retention store owner failed" + ); + None + } + } +} + +async fn admit_retention_writer( + administration: &StoreAdministration, + cancellation: &tracedecay_runtime_core::cancellation::CancellationToken, + scope: WriterScope, + jobs: &RetentionWriterJobs, + work_key: &str, + operation: Operation, +) -> Option +where + Operation: FnOnce() -> OperationFuture + Send + 'static, + OperationFuture: Future + Send, + Output: Clone + Send + Sync + 'static, +{ + if cancellation.is_cancelled() { + jobs.cancel_all(); + return None; + } + let WriterScope::StoreMaintenance { data_root } = scope else { + // Global maintenance remains a nonqueued probe: it must never put + // unrelated store writers behind an unbounded daemon-wide wait. + return administration.try_with_writer_in(scope, operation).await; + }; + let writer = { + let mut pending = jobs + .pending + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if cancellation.is_cancelled() { + return None; + } + let key = (work_key.to_owned(), data_root.clone()); + if let Some(job) = pending.get(&key) { + let outcome = job.task.try_finished()?; + pending.remove(&key); + return record_retention_job_outcome(outcome); + } + let mut acquisition = administration + .retention_writer_acquisition(data_root) + .boxed(); + if let Some(writer) = acquisition.as_mut().now_or_never() { + writer + } else { + let job_cancellation = tracedecay_runtime_core::cancellation::CancellationToken::new(); + let stopped = job_cancellation.clone(); + let stopping = cancellation.clone(); + // Transfer the exact queued future, together with the operation, + // to an actively polled owner. Never retain an idle lock future. + let task = match NativeTask::spawn(async move { + let writer = tokio::select! { + biased; + () = stopping.cancelled() => return None, + () = stopped.cancelled() => return None, + writer = acquisition => writer, + }; + if stopping.is_cancelled() || stopped.is_cancelled() { + return None; + } + let _writer = writer; + Some(operation().await) + }) { + Ok(task) => task, + Err(error) => { + tracing::error!(event = "retention_store_task_unavailable", %error, "could not start retention store owner"); + return None; + } + }; + pending.insert( + key, + RetentionWriterJob { + cancellation: job_cancellation, + task, + }, + ); + return None; + } + }; + let _writer = writer; + Some(operation().await) +} /// Every registered project whose profile shard exists on disk, keyed for the /// tick's work list. @@ -162,8 +354,8 @@ fn spawn_maintenance_task( /// the blocking closure keeps that client past the terminal store close, which /// then leaves the store's WAL untruncated. The task therefore gets the abort /// deadline to finish cooperatively and is aborted only past it. -async fn join_cancelled_maintenance_task( - task: Option>, +async fn join_cancelled_maintenance_task( + task: Option>, owner: &'static str, abort_deadline: Duration, ) { @@ -171,7 +363,27 @@ async fn join_cancelled_maintenance_task( return; }; let reason = match native_runtime::timeout(abort_deadline, task.wait()).await { - Ok(_) => return, + Ok(native_runtime::NativeTaskOutcome::Completed(_)) => return, + Ok(native_runtime::NativeTaskOutcome::Cancelled) => { + log_daemon_event( + "maintenance_task_failed", + &[ + ("owner", owner.to_owned()), + ("reason", "task_cancelled".to_owned()), + ], + ); + return; + } + Ok(native_runtime::NativeTaskOutcome::Panicked(error)) => { + log_daemon_event( + "maintenance_task_failed", + &[ + ("owner", owner.to_owned()), + ("reason", format!("{error:?}")), + ], + ); + return; + } Err(TimeoutError::Elapsed) => "cooperative_join_exceeded_abort_deadline".to_string(), Err(error @ TimeoutError::TimerUnavailable(_)) => error.to_string(), }; @@ -618,6 +830,8 @@ pub(super) struct MaintenanceCoordinator { /// store is starved when the mounted set exceeds `MAINTENANCE_STORE_PAGE_LIMIT`. store_cursor: Arc>>, store_continuations: Arc>, + /// Supervised canonical store jobs, including admitted work and its outcome. + writer_jobs: 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 +853,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_jobs: Arc::new(RetentionWriterJobs::default()), 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( @@ -655,19 +870,38 @@ impl Default for MaintenanceCoordinator { /// retirement of an idle project refuses while anything leases its stores, so /// a tick that kept every listed store for its whole run refused those /// retirements. A store unmounted since the listing is skipped. +#[derive(Clone)] 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), + /// owns. The job retains profile authority, never a project-store client. + Registered { + project_id: String, + store: RegisteredProjectStoreV1, + profile_database: tracedecay_global_db::RegisteredGlobalDbLeaseV1, + }, } impl MaintenanceStoreWork { fn database_path(&self) -> Option<&Path> { match self { - Self::Session(database_path) | Self::Graph(database_path) => Some(database_path), - Self::Registered(_) => None, + 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), } } } @@ -678,7 +912,6 @@ struct MountedGraphListing { database_path: PathBuf, project_root: PathBuf, data_root: PathBuf, - code_index_store_root: PathBuf, } /// The work-list key of a mounted project graph: its root and serving branch. @@ -824,6 +1057,7 @@ impl MaintenanceCoordinator { /// join below stays correct whether or not it already ran. pub(super) fn cancel(&self) { self.cancellation.cancel(); + self.writer_jobs.cancel_all(); } pub(super) async fn shutdown(&self) { @@ -837,6 +1071,7 @@ impl MaintenanceCoordinator { DAEMON_TASK_ABORT_DEADLINE, ) .await; + self.writer_jobs.shutdown().await; join_cancelled_maintenance_task( self.resident_memory_sampler.lock().await.take(), "resident_memory_sampler", @@ -906,6 +1141,9 @@ impl MaintenanceCoordinator { let profile_database = match administration.registered_profile_database().await { Ok(database) => database, Err(error) => { + // Without profile authority there is no current work + // census to justify retaining store queue positions. + self.writer_jobs.cancel_all(); log_daemon_event( "retention_degraded", &[ @@ -919,7 +1157,7 @@ impl MaintenanceCoordinator { }; self.run_tick( profile_root, - profile_database.as_ref(), + &profile_database, administration, code_index_schedulers, retention, @@ -944,7 +1182,7 @@ impl MaintenanceCoordinator { async fn run_tick( &self, profile_root: &Path, - profile_database: &tracedecay_global_db::RegisteredGlobalDb, + profile_database: &tracedecay_global_db::RegisteredGlobalDbLeaseV1, administration: &StoreAdministration, code_index_schedulers: &tracedecay_code_index_runtime::code_index_scheduler::CodeIndexSchedulerRegistryV1, retention: &tracedecay_configuration::RetentionConfig, @@ -979,10 +1217,6 @@ impl MaintenanceCoordinator { database_path: graph.db().database_path().to_path_buf(), project_root: graph.project_root().to_path_buf(), data_root: graph.store_layout().data_root.clone(), - code_index_store_root: code_index_store_root( - &graph.store_layout().data_root, - &graph.store_layout().project_root, - ), }) .collect::>(); #[cfg(any(test, feature = "test-helpers"))] @@ -991,7 +1225,7 @@ impl MaintenanceCoordinator { } // The listing copies in-memory mounts and is not CPU work, so a tick // parked at the listing holds no background CPU unit. - let Some(_background_cpu) = background_cpu.try_acquire() else { + let Some(tick_cpu) = background_cpu.try_acquire() else { return MaintenanceTickOutcome::Retry; }; let (registered_projects, registry_listed) = @@ -1020,10 +1254,6 @@ impl MaintenanceCoordinator { .collect(); sweep_response_handles(handle_roots, self.cancellation.clone()).await; } - let mounted_store_roots = project_graphs - .iter() - .map(|graph| graph.code_index_store_root.clone()) - .collect::>(); let mut active_telemetry_paths = BTreeSet::from([profile_database.db_path().to_path_buf()]); active_telemetry_paths.extend(session_databases.iter().cloned()); active_telemetry_paths.extend( @@ -1048,18 +1278,34 @@ 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 { work.push(( format!("r:{project_id}"), - MaintenanceStoreWork::Registered(store), + MaintenanceStoreWork::Registered { + project_id, + store, + profile_database: profile_database.clone(), + }, )); } 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_jobs.retain( + &work + .iter() + .filter_map(|(key, unit)| match unit.writer_scope() { + WriterScope::StoreMaintenance { data_root } => Some((key.clone(), data_root)), + WriterScope::Daemon | WriterScope::Store { .. } => None, + }) + .collect(), + ); let after = self.store_cursor.lock().await.clone(); let (window, _) = select_store_window(&keys, after.as_deref(), MAINTENANCE_STORE_PAGE_LIMIT); @@ -1084,8 +1330,8 @@ impl MaintenanceCoordinator { .map(|graph| graph.data_root.clone()) .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::Registered { store, .. } => Some(store.data_root.clone()), + MaintenanceStoreWork::Session(_) | MaintenanceStoreWork::Graph { .. } => None, })); maintenance_observations.retain_text_artifact_scan_projects(&scan_data_roots); } @@ -1116,6 +1362,8 @@ impl MaintenanceCoordinator { // advances past attempted units even on cancellation. A continuation // advances bounded lossless storage work and code-generation retention // over the same bounded store window rather than pinning one project. + // CPU capacity belongs to actual work, never a queued acquisition. + drop(tick_cpu); let mut attempted = 0usize; let mut deferred = 0u64; let mut outcome = MaintenanceTickOutcome::Complete; @@ -1124,12 +1372,34 @@ impl MaintenanceCoordinator { outcome = MaintenanceTickOutcome::Retry; break; } - let admitted = administration - .try_with_writer(|| async { + let scope = work[index].1.writer_scope(); + let job_admin = administration.clone(); + let job_cancellation = self.cancellation.clone(); + let job_cpu = Arc::clone(background_cpu); + let job_schedulers = code_index_schedulers.clone(); + let job_retention = retention.clone(); + let job_observations = maintenance_observations.clone(); + let job_key = keys[index].clone(); + let job_unit = work[index].1.clone(); + let admitted = admit_retention_writer( + administration, + &self.cancellation, + scope, + &self.writer_jobs, + &keys[index], + move || async move { + let administration = &job_admin; + let cancellation = &job_cancellation; + let code_index_schedulers = &job_schedulers; + let retention = &job_retention; + let maintenance_observations = job_observations; + let Some(_background_cpu) = job_cpu.try_acquire() else { + return MaintenanceTickOutcome::Retry; + }; let Ok(_clients) = administration.maintenance_store_clients().try_lock() else { return MaintenanceTickOutcome::Retry; }; - let (key, unit) = &work[index]; + let (key, unit) = (&job_key, &job_unit); match unit { MaintenanceStoreWork::Session(database_path) => { let Some(database) = administration @@ -1146,7 +1416,10 @@ impl MaintenanceCoordinator { run_registered_store_retention(&database, retention).await } } - MaintenanceStoreWork::Graph(_) => { + MaintenanceStoreWork::Graph { + database_path, + data_root, + } => { let Some(graph) = administration .mounted_project_graphs() .await @@ -1155,30 +1428,50 @@ impl MaintenanceCoordinator { else { return MaintenanceTickOutcome::Complete; }; + // A same-key remount can replace its store between + // listing and admission. The old gate cannot + // authorize maintenance against the replacement. + if graph.db().database_path() != database_path + || graph.store_layout().data_root != *data_root + { + return MaintenanceTickOutcome::Retry; + } run_project_generation_maintenance( &project_store_maintenance_lease(&graph), code_index_schedulers, &maintenance_observations, - &self.cancellation, + cancellation, retention.compaction.as_ref(), continuation, ) .await } - MaintenanceStoreWork::Registered(store) => { + MaintenanceStoreWork::Registered { project_id, store, profile_database } => { + match profile_database.code_project_exists(project_id).await { + Ok(true) => {} + Ok(false) => return MaintenanceTickOutcome::Complete, + Err(error) => { + tracing::error!(event = "retention_registry_unavailable", %error); + return MaintenanceTickOutcome::Retry; + } + } + let mounted_store_roots = administration.mounted_project_graphs().await + .iter().map(|graph| code_index_store_root(&graph.store_layout().data_root, graph.project_root())) + .collect(); run_registered_project_generation_maintenance( store, &mounted_store_roots, code_index_schedulers, profile_database, &maintenance_observations, - &self.cancellation, + cancellation, ) .await } } - }) - .await; + }, + ) + .await; attempted = attempted.saturating_add(1); match admitted { Some(unit_outcome) => { @@ -1205,6 +1498,10 @@ impl MaintenanceCoordinator { outcome = MaintenanceTickOutcome::Retry; } + let Some(_background_cpu) = background_cpu.try_acquire() else { + return outcome.combine(MaintenanceTickOutcome::Retry); + }; + // Profile-wide maintenance is intentionally excluded from a bounded // continuation: only the owning phase is eligible for the short // cadence. @@ -1530,16 +1827,595 @@ 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; + + #[test] + fn released_store_admits_ordinary_writer_without_another_retention_tick() { + native_runtime::block_on(async { + let administration = StoreAdministration::default(); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_tx, release_rx) = tokio::sync::oneshot::channel(); + let holding = administration.clone(); + let writer = NativeTask::spawn(async move { + holding + .with_writer_in( + WriterScope::store("/between-ticks", StoreWriterClass::Content), + || async { + entered_tx.send(()).expect("writer entered"); + release_rx.await.expect("writer released"); + }, + ) + .await; + }) + .expect("native writer"); + entered_rx.await.expect("store held"); + let cancellation = CancellationToken::new(); + let reservations = super::RetentionWriterJobs::default(); + let retained = Arc::new(AtomicUsize::new(0)); + let completed = Arc::clone(&retained); + assert_eq!( + super::admit_retention_writer( + &administration, + &cancellation, + WriterScope::maintenance("/between-ticks"), + &reservations, + "retention", + move || async move { + completed.fetch_add(1, Ordering::SeqCst); + }, + ) + .await, + None + ); + // Queue a real later writer before releasing the current writer. + let following = administration.with_writer_in( + WriterScope::store("/between-ticks", StoreWriterClass::Owner), + || async { 43_u8 }, + ); + tokio::pin!(following); + poll_fn(|cx| { + assert!(following.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + release_tx.send(()).expect("release between ticks"); + assert!(matches!( + writer.wait().await, + native_runtime::NativeTaskOutcome::Completed(()) + )); + // No retention helper or tick is polled after release, including + // when the next tick cannot obtain background CPU admission. + assert_eq!( + native_runtime::timeout(Duration::from_secs(2), &mut following) + .await + .expect("ordinary writer must resume without a retention tick"), + 43 + ); + assert_eq!( + retained.load(Ordering::SeqCst), + 1, + "the earlier fair retention turn ran exactly once" + ); + reservations.shutdown().await; + }) + .expect("native runtime"); + } + + #[test] + fn shutdown_joins_queued_retention_that_has_started_between_ticks() { + native_runtime::block_on(async { + let administration = StoreAdministration::default(); + let coordinator = MaintenanceCoordinator::default(); + let (holding_tx, holding_rx) = tokio::sync::oneshot::channel(); + let (release_writer, released_writer) = tokio::sync::oneshot::channel(); + let holding = administration.clone(); + let writer = NativeTask::spawn(async move { + holding + .with_writer_in( + WriterScope::store("/shutdown-between-ticks", StoreWriterClass::Content), + || async { + holding_tx.send(()).expect("writer entered"); + released_writer.await.expect("writer released"); + }, + ) + .await; + }) + .expect("native writer"); + holding_rx.await.expect("store held"); + let (entered_tx, entered_rx) = tokio::sync::oneshot::channel(); + let (release_work, released_work) = tokio::sync::oneshot::channel(); + let finished = Arc::new(AtomicUsize::new(0)); + let completed = Arc::clone(&finished); + assert_eq!( + super::admit_retention_writer( + &administration, + &coordinator.cancellation, + WriterScope::maintenance("/shutdown-between-ticks"), + &coordinator.writer_jobs, + "retention", + move || async move { + entered_tx.send(()).expect("retention entered"); + released_work.await.expect("retention released"); + completed.fetch_add(1, Ordering::SeqCst); + MaintenanceTickOutcome::Complete + }, + ) + .await, + None + ); + release_writer.send(()).expect("release store writer"); + assert!(matches!( + writer.wait().await, + native_runtime::NativeTaskOutcome::Completed(()) + )); + entered_rx.await.expect("queued work starts without a tick"); + let shutdown = coordinator.shutdown(); + tokio::pin!(shutdown); + poll_fn(|cx| { + assert!( + shutdown.as_mut().poll(cx).is_pending(), + "shutdown joins admitted work" + ); + Poll::Ready(()) + }) + .await; + assert!(coordinator.cancellation.is_cancelled()); + assert_eq!(finished.load(Ordering::SeqCst), 0); + assert_eq!( + administration.try_with_writer(|| async {}).await, + None, + "admitted work retains writer admission through shutdown" + ); + release_work.send(()).expect("finish admitted work"); + native_runtime::timeout(Duration::from_secs(2), &mut shutdown) + .await + .expect("cooperative shutdown"); + assert_eq!(finished.load(Ordering::SeqCst), 1); + assert_eq!(administration.try_with_writer(|| async {}).await, Some(())); + }) + .expect("native runtime"); + } + + /// Retry ticks submit only one fair job, which owns no daemon read guard + /// while the target store is busy. + #[tokio::test] + async fn retention_runs_one_fair_job_across_retry_ticks() { + 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("/retained-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 store gate"); + let cancellation = CancellationToken::new(); + let reservations = super::RetentionWriterJobs::default(); + for _ in 0..4 { + assert_eq!( + super::admit_retention_writer( + &administration, + &cancellation, + WriterScope::maintenance("/retained-store"), + &reservations, + "retention", + || async { 42_u8 }, + ) + .await, + None, + "a held store defers only its own maintenance unit" + ); + assert_eq!(reservations.pending.lock().unwrap().len(), 1); + } + + assert_eq!( + super::admit_retention_writer( + &administration, + &cancellation, + WriterScope::maintenance("/retained-store"), + &reservations, + "another-branch", + || async { 45_u8 }, + ) + .await, + None, + ); + assert_eq!( + reservations.pending.lock().unwrap().len(), + 2, + "distinct work units in one store retain distinct fair queue positions" + ); + + // Keep unrelated writer activity running through retention admission. + // Same-store readers must wait behind the retained fair acquisitions. + let competing_admin = administration.clone(); + 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.is_cancelled() { + competing_admin + .with_writer_in( + WriterScope::store("/another-store", StoreWriterClass::Content), + || async { + entries.fetch_add(1, Ordering::SeqCst); + tokio::task::yield_now().await; + }, + ) + .await; + } + }); + tokio::time::timeout(Duration::from_secs(5), async { + while writer_entries.load(Ordering::SeqCst) < 4 { + tokio::task::yield_now().await; + } + }) + .await + .expect("unrelated writers churn before retention queues"); + + let later_writer = administration.with_writer_in( + WriterScope::store("/retained-store", StoreWriterClass::Owner), + || 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; + + // 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"); + holder.await.expect("first writer finishes"); + assert_eq!( + tokio::time::timeout(Duration::from_secs(5), &mut later_writer) + .await + .expect("later writer resumes after retention"), + 43 + ); + + tokio::time::timeout(Duration::from_secs(5), async { + while reservations + .pending + .lock() + .unwrap() + .values() + .any(|job| !job.task.is_finished()) + { + tokio::task::yield_now().await; + } + }) + .await + .expect("both fair jobs settle before their outcomes are observed"); + assert_eq!( + super::admit_retention_writer( + &administration, + &cancellation, + WriterScope::maintenance("/retained-store"), + &reservations, + "retention", + || async { 42_u8 }, + ) + .await, + Some(42) + ); + assert_eq!( + super::admit_retention_writer( + &administration, + &cancellation, + WriterScope::maintenance("/retained-store"), + &reservations, + "another-branch", + || async { 45_u8 }, + ) + .await, + Some(45), + "the first work unit cannot consume another branch's queued acquisition", + ); + // 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 writer_entries.load(Ordering::SeqCst) <= before_release { + tokio::task::yield_now().await; + } + }) + .await + .expect("unrelated writers resume after maintenance admission"); + stop_churn.cancel(); + assert!(reservations.pending.lock().unwrap().is_empty()); + competitors.await.expect("unrelated writer churn finishes"); + assert!(writer_entries.load(Ordering::SeqCst) > before_release); + } + + #[tokio::test] + async fn cancelled_retention_never_constructs_work_before_admission() { + let administration = StoreAdministration::default(); + let cancellation = CancellationToken::new(); + cancellation.cancel(); + let constructed = Arc::new(AtomicUsize::new(0)); + let reservations = super::RetentionWriterJobs::default(); + for scope in [WriterScope::maintenance("/held-store"), WriterScope::Daemon] { + assert_eq!( + super::admit_retention_writer( + &administration, + &cancellation, + scope, + &reservations, + "retention", + { + let constructed = Arc::clone(&constructed); + move || { + constructed.fetch_add(1, Ordering::SeqCst); + std::future::ready(7_u8) + } + }, + ) + .await, + None, + "cancelled retention refuses even an uncontended acquisition" + ); + } + 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(); + 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 coordinator = MaintenanceCoordinator::default(); + let constructed = Arc::new(AtomicUsize::new(0)); + assert_eq!( + super::admit_retention_writer( + &administration, + &coordinator.cancellation, + WriterScope::maintenance("/held-store"), + &coordinator.writer_jobs, + "retention", + { + let constructed = Arc::clone(&constructed); + move || { + constructed.fetch_add(1, Ordering::SeqCst); + std::future::ready(MaintenanceTickOutcome::Complete) + } + }, + ) + .await, + None + ); + let follower = administration.with_writer_in( + WriterScope::store("/held-store", StoreWriterClass::Content), + || async { 8_u8 }, + ); + tokio::pin!(follower); + poll_fn(|cx| { + assert!(follower.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + coordinator.shutdown().await; + assert!(coordinator.writer_jobs.pending.lock().unwrap().is_empty()); + assert_eq!(constructed.load(Ordering::SeqCst), 0); + // 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("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) + ); + } + + #[tokio::test] + async fn retired_store_jobs_release_followers_and_replacement_uses_its_own_gate() { + let temp = tempfile::tempdir().expect("store roots"); + let retired = temp.path().join("retired"); + let replacement = temp.path().join("replacement"); + std::fs::create_dir_all(&retired).expect("retired root"); + std::fs::create_dir_all(&replacement).expect("replacement root"); + let administration = StoreAdministration::default(); + let coordinator = MaintenanceCoordinator::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 owner_scope = + crate::daemon::branch_admin::store_writer_scope(&retired, StoreWriterClass::Owner); + let owner = tokio::spawn(async move { + holding_admin + .with_writer_in(owner_scope, || async { + entered_tx.send(()).expect("owner admitted"); + release_rx.await.expect("owner released"); + }) + .await; + }); + entered_rx.await.expect("held retired store"); + let work = Arc::new(AtomicUsize::new(0)); + // Different spellings of the same canonical store share one pending + // acquisition, rather than competing maintenance jobs. + for root in [&retired, &retired.join(".")] { + assert_eq!( + super::admit_retention_writer( + &administration, + &coordinator.cancellation, + super::store_maintenance_writer_scope(root), + &coordinator.writer_jobs, + "retention", + { + let work = Arc::clone(&work); + move || async move { + work.fetch_add(1, Ordering::SeqCst); + MaintenanceTickOutcome::Complete + } + }, + ) + .await, + None + ); + } + assert_eq!(coordinator.writer_jobs.pending.lock().unwrap().len(), 1); + let following = administration.with_writer_in( + crate::daemon::branch_admin::store_writer_scope(&retired, StoreWriterClass::Content), + || async { 8_u8 }, + ); + tokio::pin!(following); + poll_fn(|cx| { + assert!(following.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + let replacement_scope = super::store_maintenance_writer_scope(&replacement); + let WriterScope::StoreMaintenance { data_root } = &replacement_scope else { + panic!("replacement has canonical identity"); + }; + // The current mounted-work census keeps only the replacement identity. + coordinator + .writer_jobs + .retain(&std::collections::BTreeSet::from([( + "retention".to_owned(), + data_root.clone(), + )])); + assert_eq!( + tokio::time::timeout(Duration::from_secs(2), &mut following) + .await + .expect("retirement drops the old queue position while its owner is still active"), + 8 + ); + assert_eq!( + super::admit_retention_writer( + &administration, + &coordinator.cancellation, + replacement_scope, + &coordinator.writer_jobs, + "retention", + { + let work = Arc::clone(&work); + move || async move { + work.fetch_add(1, Ordering::SeqCst); + MaintenanceTickOutcome::Complete + } + }, + ) + .await, + Some(MaintenanceTickOutcome::Complete) + ); + assert_eq!( + work.load(Ordering::SeqCst), + 1, + "only the replacement executes maintenance" + ); + coordinator.shutdown().await; + assert!(coordinator.writer_jobs.pending.lock().unwrap().is_empty()); + release_tx.send(()).expect("release retired owner"); + owner.await.expect("retired owner settles"); + } + + /// 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() { + 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, + WriterScope::maintenance("/retention-store"), + &super::RetentionWriterJobs::default(), + "retention", + 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(); + 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) + ); + } #[tokio::test] async fn mounted_store_identity_tracks_native_removal_outcomes() { @@ -1717,21 +2593,20 @@ mod tests { .expect("sampler joins after cancellation"); } - /// A registered project nothing has mounted since the daemon started is - /// collected by the first full tick and converges on its continuation, - /// instead of waiting for a client to mount it. A tick that runs while a - /// capacity release holds the maintenance store clients defers it. + /// Busy registered store A cannot park free store B or the profile sweep. + /// After release, the supervised fair job lets A converge between ticks. + /// Capacity release still excludes store clients. #[test] - fn a_full_tick_collects_an_unmounted_registered_project_without_a_tokio_runtime() { + fn full_ticks_preserve_independent_store_progress_without_a_tokio_runtime() { let hazards = crate::test_support::native_runtime_hazards(); tracedecay_runtime_core::native_runtime::block_on( - a_full_tick_collects_an_unmounted_registered_project(), + full_ticks_preserve_independent_store_progress(), ) .expect("native runtime"); assert_eq!(crate::test_support::native_runtime_hazards(), hazards); } - async fn a_full_tick_collects_an_unmounted_registered_project() { + async fn full_ticks_preserve_independent_store_progress() { assert!(tokio::runtime::Handle::try_current().is_err()); let temp = tempfile::TempDir::new().expect("temp root"); let profile_root = temp.path().join("profile"); @@ -1806,6 +2681,165 @@ mod tests { ); drop(release); + // A stays busy across more than two complete retry ticks. New work + // in B and the profile must still finish while A's gate remains held. + 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_data_root = retained_data_root.clone(); + let writer = NativeTask::spawn(async move { + holding_admin + .with_writer_in( + crate::daemon::branch_admin::store_writer_scope( + &writer_data_root, + 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("project content writer owns shared store gate"); + for _ in 0..2 { + let deferred = native_runtime::timeout( + Duration::from_secs(10), + coordinator.run_tick( + &profile_root, + &profile_database, + &administration, + &schedulers, + &retention, + branch_gc, + None, + ), + ) + .await + .expect("busy A cannot park a complete tick"); + assert_eq!(deferred, MaintenanceTickOutcome::Retry); + assert_eq!(generation_files(), 6, "A cannot collect under its writer"); + } + let other_checkout = temp.path().join("other-checkout"); + std::fs::create_dir_all(&other_checkout).expect("other checkout"); + let other = profile_database + .upsert_code_project("proj_z_independent", &other_checkout, None, None, None) + .await + .expect("register independent project B"); + assert!( + project.project_id < other.project_id, + "A precedes B in the work window" + ); + let other_store = + tracedecay_code_index_retention::code_index_generations::code_index_store_root( + &tracedecay_runtime_core::storage::profile_sharded_data_root( + &profile_root, + &other.project_id, + ), + Path::new(&other.canonical_root), + ); + tracedecay_code_index_retention::code_index_generations::fixture::write_generation_store_fixture( + &other_store, 6, + ); + let handles = profile_root.join(super::RESPONSE_HANDLES_DIRECTORY); + tracedecay_session_memory::response_handles::store_response_handle(&handles, "expired", 1) + .expect("expired profile handle"); + tracedecay_session_memory::response_handles::store_response_handle( + &handles, + "fresh", + tracedecay_runtime_core::tracedecay::current_timestamp(), + ) + .expect("fresh profile handle"); + for _ in 0..2 { + let deferred = native_runtime::timeout( + Duration::from_secs(10), + coordinator.run_tick( + &profile_root, + &profile_database, + &administration, + &schedulers, + &retention, + branch_gc, + None, + ), + ) + .await + .expect("B and profile maintenance finish while A stays busy beyond two ticks"); + assert_eq!(deferred, MaintenanceTickOutcome::Retry); + assert_eq!(generation_files(), 6, "A's writer is still held"); + assert_eq!( + std::fs::read_dir(other_store.join("code-generations-v1")) + .expect("B generations") + .count(), + 1, + "B reclaims its five superseded generations before A releases", + ); + assert_eq!( + tracedecay_session_memory::response_handles::inventory_response_handles(&handles) + .expect("profile handle inventory") + .file_count, + 1, + "the profile sweep removes the expired handle before A releases", + ); + let pending = coordinator.writer_jobs.pending.lock().unwrap(); + assert_eq!(pending.len(), 1, "only one acquisition, for busy A"); + let WriterScope::StoreMaintenance { data_root } = + super::store_maintenance_writer_scope(&retained_data_root) + else { + panic!("fixture A has canonical store identity"); + }; + assert!(pending.contains_key(&(format!("r:{}", project.project_id), data_root))); + } + let following = administration.with_writer_in( + super::super::branch_admin::store_writer_scope( + &retained_data_root, + StoreWriterClass::Owner, + ), + || async {}, + ); + tokio::pin!(following); + poll_fn(|cx| { + assert!(following.as_mut().poll(cx).is_pending()); + Poll::Ready(()) + }) + .await; + release_writer.send(()).expect("release competing writer"); + assert!( + matches!( + writer.wait().await, + native_runtime::NativeTaskOutcome::Completed(()) + ), + "competing writer must complete normally" + ); + native_runtime::timeout(Duration::from_secs(10), &mut following) + .await + .expect("ordinary writer resumes after retention without another tick"); + assert_eq!( + generation_files(), + 1, + "A collects before any further tick is polled" + ); + native_runtime::timeout(Duration::from_secs(10), async { + while coordinator + .writer_jobs + .pending + .lock() + .unwrap() + .values() + .any(|job| !job.task.is_finished()) + { + tokio::task::yield_now().await; + } + }) + .await + .expect("A's supervised owner settles"); coordinator .run_tick( &profile_root, @@ -1817,7 +2851,12 @@ mod tests { None, ) .await; - assert_eq!(generation_files(), 1, "the full tick collects the backlog"); + assert!(coordinator.writer_jobs.pending.lock().unwrap().is_empty()); + assert_eq!( + generation_files(), + 1, + "queued retention reclaims five verified superseded generations after writer release" + ); let continuation = coordinator .run_tick( diff --git a/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs b/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs index b6dd4a3e4f5..647e85d13cb 100644 --- a/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs +++ b/crates/tracedecay/tests/runtime_acceptance_suite/advisory_runtime_acceptance.rs @@ -2,7 +2,7 @@ use std::collections::{BTreeSet, VecDeque}; use std::io::SeekFrom; -use std::process::{Command, Stdio}; +use std::process::Command; use std::sync::{Arc, Mutex}; use std::time::Duration; @@ -978,18 +978,13 @@ async fn packaged_host_ingest_delivers_a_registered_advisory_cycle() { String::from_utf8_lossy(&output.stderr) ); } - common::initialize_tracedecay_cli_project(environment.home(), &project); let daemon_log = environment.home().join("advisory-daemon.log"); - let _daemon = common::spawn_tracedecay_daemon_with(environment.home(), |command| { - // `.cargo/config.toml` sets TRACEDECAY_DISABLE_GLOBAL_DB=1 so cargo - // children never touch the operator ledger. This journey depends on - // the registered profile accounting owner for hint-outcome settlement. - command - .env_remove("TRACEDECAY_DISABLE_GLOBAL_DB") - .stderr(Stdio::from( - std::fs::File::create(&daemon_log).expect("create isolated advisory daemon log"), - )); - }); + let _daemon = + common::spawn_tracedecay_daemon_logged(environment.home(), &daemon_log, |command| { + // Hint-outcome settlement requires the isolated profile's accounting owner. + command.env_remove("TRACEDECAY_DISABLE_GLOBAL_DB"); + }); + common::initialize_tracedecay_cli_project(environment.home(), &project); let transcript = project.join("cursor-proximity.jsonl"); std::fs::write( &transcript, 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 b0f714cdffb..c94b3bc22ef 100644 --- a/crates/tracedecay/tests/runtime_acceptance_suite/tool_client_transport.rs +++ b/crates/tracedecay/tests/runtime_acceptance_suite/tool_client_transport.rs @@ -79,6 +79,70 @@ fn run_command_with_timeout(mut command: Command, timeout: Duration) -> ChildRes } } +fn run_tool_after_admission( + mut command: Command, + requests: mpsc::Receiver, + deadline: Instant, +) -> Output { + let child = command + .stdin(Stdio::null()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .expect("spawn CLI for scripted daemon"); + let mut child = common::TestChildProcess::new(child); + let admitted = loop { + let remaining = deadline.saturating_duration_since(Instant::now()); + match requests.recv_timeout(remaining.min(Duration::from_millis(50))) { + Ok(admitted) => break admitted, + Err(RecvTimeoutError::Timeout) => { + assert!( + child + .try_wait() + .expect("poll CLI before admission") + .is_none(), + "CLI exited before the daemon received its request" + ); + assert!(Instant::now() < deadline, "CLI request admission timed out"); + } + Err(RecvTimeoutError::Disconnected) => { + panic!("scripted daemon stopped before receiving the CLI request") + } + } + }; + assert!(admitted <= deadline, "CLI request admission timed out"); + // Admission diagnoses which phase stalled; it does not grant a second + // watchdog. Startup, response delivery and process exit share the same + // absolute deadline, including delayed observation of a queued receipt. + let remaining = deadline + .checked_duration_since(Instant::now()) + .filter(|remaining| !remaining.is_zero()) + .expect("CLI response deadline expired after admission"); + let output = child + .wait_with_output(remaining) + .expect("CLI must finish delivering the admitted response within its deadline"); + assert!( + output.status.success(), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + output +} + +#[test] +#[should_panic(expected = "CLI response deadline expired after admission")] +fn admission_receipt_cannot_restart_an_expired_child_watchdog() { + // Use a real subprocess without recursively running this test. A receipt + // admitted at the deadline can be observed later from the channel, but + // must not grant that child a fresh response-delivery timeout. + let mut command = Command::new(std::env::current_exe().expect("test executable")); + command.arg("--list"); + let (admitted_tx, admitted_rx) = mpsc::channel(); + let deadline = Instant::now(); + admitted_tx.send(deadline).expect("queue admitted request"); + run_tool_after_admission(command, admitted_rx, deadline); +} + fn init_project(home: &Path, project: &Path) { std::fs::create_dir_all(project.join("src")).expect("create project source"); std::fs::write(project.join("src/lib.rs"), "pub fn marker() {}\n") @@ -161,11 +225,11 @@ fn response_bytes(request: &DaemonInvocationRequest, scope: ResolvedScope, text: files_response(request, scope, files_result([text.to_owned()])) } -fn output_payload(result: &ChildResult) -> Value { - let envelope = serde_json::from_slice(&result.output.stdout).unwrap_or_else(|error| { +fn output_payload(output: &Output) -> Value { + let envelope = serde_json::from_slice(&output.stdout).unwrap_or_else(|error| { panic!( "invalid CLI JSON ({error}); stderr: {}", - String::from_utf8_lossy(&result.output.stderr) + String::from_utf8_lossy(&output.stderr) ) }); tracedecay::daemon::tool_json_payload(&envelope, "transport fixture") @@ -190,7 +254,7 @@ fn spawn_scripted_daemon( project: &Path, connections: usize, script: F, -) -> (mpsc::Receiver<()>, JoinHandle<()>) +) -> (mpsc::Receiver, JoinHandle<()>) where F: Fn(UnixStream, DaemonInvocationRequest, ResolvedScope) + Send + Sync + 'static, { @@ -275,7 +339,9 @@ where .. } )); - request_tx.send(()).expect("publish request receipt"); + request_tx + .send(Instant::now()) + .expect("publish request receipt"); let script = Arc::clone(&script); let scope = scope.clone(); workers.push(std::thread::spawn(move || script(stream, request, scope))); @@ -324,10 +390,14 @@ fn generic_tool_accepts_slow_byte_stream() { &project, 1, |mut stream, request, scope| { + // Pace each byte against one clock so scheduler delays do not + // accumulate across the entire response frame. + let mut send_at = Instant::now(); for byte in response_bytes(&request, scope, "slow-ok") { + std::thread::sleep(send_at.saturating_duration_since(Instant::now())); stream.write_all(&[byte]).expect("write slow byte"); stream.flush().expect("flush slow byte"); - std::thread::sleep(Duration::from_millis(2)); + send_at += Duration::from_millis(2); } }, ); @@ -423,7 +493,7 @@ fn generic_tool_reports_unavailable_truncation_storage() { "{}", String::from_utf8_lossy(&result.output.stderr) ); - let payload = output_payload(&result); + let payload = output_payload(&result.output); assert_eq!(payload["truncated"], true, "{payload}"); assert_eq!(payload["handle_available"], false, "{payload}"); assert!( @@ -448,30 +518,26 @@ fn generic_tool_reports_unavailable_truncation_storage() { #[test] fn generic_tool_retrieves_oversized_typed_result() { let (_home, _project, _socket_dir, home, project, socket) = fixture(); - let expected = serde_json::to_value(oversized_files()).expect("expected file listing"); - let (_requests, server) = spawn_scripted_daemon( + let files = oversized_files(); + let expected = serde_json::to_value(&files).expect("expected file listing"); + let (requests, server) = spawn_scripted_daemon( socket.clone(), &home, &project, 1, - |mut stream, request, scope| { + move |mut stream, request, scope| { stream - .write_all(&files_response(&request, scope, oversized_files())) + .write_all(&files_response(&request, scope, files.clone())) .expect("write oversized result"); }, ); - let result = run_command_with_timeout( + let output = run_tool_after_admission( tool_command(&home, &project, &socket, "large"), - CHILD_TIMEOUT, + requests, + Instant::now() + CHILD_TIMEOUT, ); server.join().expect("join scripted daemon"); - assert!(!result.killed_by_harness, "oversized response hung"); - assert!( - result.output.status.success(), - "{}", - String::from_utf8_lossy(&result.output.stderr) - ); - let preview = output_payload(&result); + let preview = output_payload(&output); assert_eq!(preview["truncated"], true, "{preview}"); let handle = preview["handle"].as_str().expect("stored response handle"); assert_eq!(preview["retrieve_tool"], "tracedecay_retrieve"); @@ -501,7 +567,7 @@ fn generic_tool_retrieves_oversized_typed_result() { "{}", String::from_utf8_lossy(&retrieved.output.stderr) ); - let page = output_payload(&retrieved); + let page = output_payload(&retrieved.output); pages += 1; assert_eq!(page["handle"], handle); assert_eq!(page["offset"], offset); @@ -545,34 +611,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 +625,14 @@ 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( + let output = run_tool_after_admission( tool_command(&home, &project, &socket, "large"), - CHILD_TIMEOUT, + requests, + Instant::now() + CHILD_TIMEOUT, ); server.join().expect("join scripted daemon"); - let envelope = assert_complete_json_document(&result); + 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 +640,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() ); } @@ -732,7 +778,7 @@ fn cancelling_generic_tool_reaps_child_and_closes_request() { // load, and fail fast if the CLI dies or the daemon stops first. loop { match requests.recv_timeout(Duration::from_millis(50)) { - Ok(()) => break, + Ok(_) => break, Err(RecvTimeoutError::Timeout) => assert!( child.try_wait().expect("poll cancellable CLI").is_none(), "CLI exited before the fake daemon received its request"