test: wait on real signals in two load-flaky tests

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<AbortHandle> {
+pub(crate) fn start_shared_jsonl_page_prefetch(paths: &[PathBuf]) -> Vec<SharedJsonlPrefetchTask> {
     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<Result<(), futures_util::future::Aborted>>;
+
 fn start_shared_jsonl_page_prefetch_with_cancellation(
     paths: &[PathBuf],
     cancellation: Option<Arc<std::sync::atomic::AtomicBool>>,
-) -> Vec<AbortHandle> {
+) -> 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

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(),

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());