From e6bbe775e2071c012b1ef8b3e38f493626333708 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Wed, 14 Oct 2026 16:48:51 +0000 Subject: [PATCH 1/2] test(sessions): join prefetch builds before counting active builds prepared_generation_uses_bounded_parallelism_and_retained_bytes awaited only its demand tasks, then asserted that no page build was still active. A prefetch that waited on a demand build rebuilds its page when that page has already left the process-global cache, which concurrent tests in the suite binary cause. That rebuild was still running at the assertion (left: 1). The prefetch starter now returns its task handles alongside the abort handles, and the test joins every prefetch before asserting. Refs #3950 --- .../jsonl_observation_admission.rs | 28 +++++++++++++------ .../jsonl_observation_admission/tests.rs | 9 +++++- 2 files changed, 28 insertions(+), 9 deletions(-) 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(), From c97790b7992e232704adfb2ba89ce39d2e3c5007 Mon Sep 17 00:00:00 2001 From: ScriptedAlchemy Date: Wed, 14 Oct 2026 17:13:37 +0000 Subject: [PATCH 2/2] test(observability): let the writer fence wait for the release registered_persistence_failure_releases_only_after_join_and_writer_fence gave shutdown a 250 ms budget. The writer fence spends that budget waiting for the test to commit its held writer, which the test does only after it observes the worker join. When the test task was scheduled late, the fence timed out, shutdown returned observability_shutdown_deadline, and the producer stayed joined but not settled. Production ordering is correct: settled is stored only after the worker join and a completed fence. The test now gives shutdown a 30 s budget so the fence ends on the writer release, and drops the pre-shutdown sleep that only kept the worker reply inside the old budget. Refs #3950 --- .../tests/daemon_suite/invocation_observability.rs | 8 +++----- 1 file changed, 3 insertions(+), 5 deletions(-) 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());