diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs index eb5b635f2a..cf687342ad 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission.rs @@ -993,10 +993,14 @@ impl SharedJsonlPathPin { self.prefetches .lock() .unwrap_or_else(PoisonError::into_inner) - .extend(start_shared_jsonl_page_prefetch_with_cancellation( - paths, - Some(Arc::clone(&self.cancelled)), - )); + .extend( + start_shared_jsonl_page_prefetch_with_cancellation( + paths, + Some(Arc::clone(&self.cancelled)), + ) + .into_iter() + .map(|(abort, _task)| abort), + ); } } @@ -1696,14 +1700,22 @@ async fn shared_jsonl_page_with_frame_limit_and_cancellation( /// ordered source admission; the canonical admission call waits on the exact /// in-flight page it needs and preserves per-source cursor order. #[cfg(test)] -pub(crate) fn start_shared_jsonl_page_prefetch(paths: &[PathBuf]) -> Vec { +pub(crate) fn start_shared_jsonl_page_prefetch(paths: &[PathBuf]) -> Vec { start_shared_jsonl_page_prefetch_with_cancellation(paths, None) + .into_iter() + .map(|(_abort, task)| task) + .collect() } +/// Dropping the handle detaches the page task; joining it observes the end +/// of any build the task started. +type SharedJsonlPrefetchTask = + asupersync::runtime::TaskHandle>; + fn start_shared_jsonl_page_prefetch_with_cancellation( paths: &[PathBuf], cancellation: Option>, -) -> Vec { +) -> Vec<(AbortHandle, SharedJsonlPrefetchTask)> { let Some(cx) = tracedecay_runtime_core::native_runtime::cx() else { return Vec::new(); }; @@ -1737,8 +1749,8 @@ fn start_shared_jsonl_page_prefetch_with_cancellation( ); } }); - if cx.spawn(move |_cx| prefetch).is_ok() { - prefetches.push(abort); + if let Ok(task) = cx.spawn(move |_cx| prefetch) { + prefetches.push((abort, task)); } } prefetches diff --git a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs index 71fdba9d01..c6fcd03a48 100644 --- a/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs +++ b/crates/tracedecay-sessions/src/runtime/observation/jsonl_observation_admission/tests.rs @@ -686,7 +686,7 @@ async fn prepared_generation_uses_bounded_parallelism_and_retained_bytes() { } } - let _prefetches = super::start_shared_jsonl_page_prefetch(&paths); + let prefetches = super::start_shared_jsonl_page_prefetch(&paths); let builds = paths .into_iter() .map(|path| { @@ -706,6 +706,13 @@ async fn prepared_generation_uses_bounded_parallelism_and_retained_bytes() { for build in builds { build.await.expect("prepared generation task"); } + // A prefetch that waited on a demand build rebuilds its page when that + // page has already left the process-global cache, so the demand tasks + // alone do not cover every build this test started. + let cx = tracedecay_runtime_core::native_runtime::cx().expect("native runtime"); + for mut prefetch in prefetches { + let _ = prefetch.join(&cx).await.expect("prefetch task"); + } assert_eq!( observed_builds.active(), diff --git a/crates/tracedecay/tests/daemon_suite/invocation_observability.rs b/crates/tracedecay/tests/daemon_suite/invocation_observability.rs index 9be7e10e2c..5056c322e0 100644 --- a/crates/tracedecay/tests/daemon_suite/invocation_observability.rs +++ b/crates/tracedecay/tests/daemon_suite/invocation_observability.rs @@ -1330,9 +1330,11 @@ async fn registered_persistence_failure_releases_only_after_join_and_writer_fenc database.clone(), identity.clone(), 1, + // The writer fence waits for the test to release the writer, so the + // shutdown budget must outlast any scheduling delay before it does. ObservabilityProducerDeadlinesV1 { persistence, - shutdown: Duration::from_millis(250), + shutdown: Duration::from_secs(30), }, ) .expect("producer"); @@ -1344,14 +1346,10 @@ async fn registered_persistence_failure_releases_only_after_join_and_writer_fenc .begin_write_transaction() .await .expect("hold registered writer"); - // The held writer exhausts the persistence deadline before shutdown - // begins: a startup rollup pass may wait one deadline before the worker - // takes the event, whose write then waits another. registered .producer() .try_emit(envelope(&project_id, "shutdown:blocked")) .expect("enqueue blocked event"); - tokio::time::sleep(persistence * 6).await; let retained = registered.producer(); assert!(!retained.shutdown_settled());