Flake batch 38: Windows base-reclaim and call-graph read-bound tests

benchmarks/aft-search/slice-descriptors/land-flake-batch-38.json

@@ -0,0 +1,8 @@
+{
+  "fixtures": [
+    "harness-goldens"
+  ],
+  "kind": "harness",
+  "slice_class": "non_ranking",
+  "targeted_mechanism": "none"
+}

crates/aft/src/commands/configure.rs

@@ -15380,7 +15380,11 @@ mod tests {
         );
     }
 
-    fn assert_callgraph_views_rematerialization_bounded(command: &str, daemon: bool) {
+    fn assert_callgraph_views_rematerialization_bounded(
+        command: &str,
+        daemon: bool,
+        delay_delivery: bool,
+    ) {
         let _git_env = crate::test_env::hermetic_git_env_guard();
         let _serial = crate::blob_store::v3::serial_for_test();
         let temp = tempfile::tempdir().unwrap();
@@ -15469,12 +15473,32 @@ mod tests {
         let (entered_tx, entered_rx) = crossbeam_channel::unbounded();
         let (release_tx, release_rx) = crossbeam_channel::bounded(1);
         let (published_tx, published_rx) = crossbeam_channel::unbounded();
+        let (phase_tx, phase_rx) = crossbeam_channel::unbounded();
+        let (timed_out_tx, timed_out_rx) = crossbeam_channel::unbounded();
+        let (deliver_tx, deliver_rx) = crossbeam_channel::unbounded();
+        let timing_origin = Instant::now();
         let _hook = crate::storage_retention::observe_test_hook(
             &root,
             Arc::new(move |step| {
+                if step.starts_with("cg-timing:") {
+                    let _ = phase_tx.send((
+                        std::thread::current().id(),
+                        step.to_owned(),
+                        Instant::now(),
+                    ));
+                }
+                if matches!(
+                    step,
+                    "cg-timing:receive-timeout" | "cg-timing:query-lock-timeout"
+                ) {
+                    timed_out_tx.send(()).unwrap();
+                    if delay_delivery {
+                        deliver_rx.recv_timeout(Duration::from_secs(120)).unwrap();
+                    }
+                }
                 if step == "callgraph-materialize" {
                     entered_tx.send(()).unwrap();
-                    release_rx.recv_timeout(Duration::from_secs(30)).unwrap();
+                    release_rx.recv_timeout(Duration::from_secs(120)).unwrap();
                 } else if step == "callgraph-reader-acquired" {
                     let _ = published_tx.send(());
                 }
@@ -15490,23 +15514,45 @@ mod tests {
             callers_request.clone()
         };
         let (answer_tx, answer_rx) = crossbeam_channel::unbounded();
+        let timing_ids = std::sync::atomic::AtomicUsize::new(0);
         let spawn_read = || {
             let worker_ctx = ctx.clone();
             let request = request.clone();
             let answer_tx = answer_tx.clone();
             let executor = _executor.clone();
             let root_id = crate::path_identity::ProjectRootId::from_path(&root).unwrap();
+            let timing_root = root.clone();
+            let timing_id = timing_ids.fetch_add(1, Ordering::Relaxed);
+            let (waits_tx, waits_rx) = crossbeam_channel::bounded(1);
             std::thread::spawn(move || {
                 let started = Instant::now();
                 let id = request.id.clone();
                 let zoom = request.command == "zoom";
                 let run = move |ctx: &AppContext| {
-                    let _pin = ctx.pin_config();
-                    if zoom {
-                        crate::commands::zoom::handle_zoom(&request, ctx)
-                    } else {
-                        crate::commands::callers::handle_callers(&request, ctx)
-                    }
+                    let (response, waits) =
+                        crate::views::callgraph::capture_read_waits_for_test(|| {
+                            crate::storage_retention::test_hook(
+                                &timing_root,
+                                &format!("cg-timing:request-{timing_id}-pin-begin"),
+                            );
+                            let _pin = ctx.pin_config();
+                            crate::storage_retention::test_hook(
+                                &timing_root,
+                                "cg-timing:pin-return",
+                            );
+                            let response = if zoom {
+                                crate::commands::zoom::handle_zoom(&request, ctx)
+                            } else {
+                                crate::commands::callers::handle_callers(&request, ctx)
+                            };
+                            crate::storage_retention::test_hook(
+                                &timing_root,
+                                "cg-timing:handler-return",
+                            );
+                            response
+                        });
+                    waits_tx.send(waits).unwrap();
+                    response
                 };
                 let response = if let Some(executor) = executor {
                     executor
@@ -15527,7 +15573,10 @@ mod tests {
                 } else {
                     run(&worker_ctx)
                 };
-                answer_tx.send((started.elapsed(), response)).unwrap();
+                let waits = waits_rx.recv_timeout(Duration::from_secs(120)).unwrap();
+                answer_tx
+                    .send((started.elapsed(), response, waits))
+                    .unwrap();
             })
         };
         let worker = spawn_read();
@@ -15536,21 +15585,91 @@ mod tests {
         // A reader queued behind the first read gets the same total budget,
         // not another full wait after the first read releases the query lock.
         let concurrent = spawn_read();
-        let deadline = Instant::now() + bound + Duration::from_secs(1);
+        // This watchdog only releases stuck fixture threads. The assertions
+        // below check actual wait arguments and timeout outcomes, not dispatch
+        // or response-rendering wall time on a busy runner.
+        let deadline = Instant::now() + Duration::from_secs(120);
+        let first_timeout =
+            timed_out_rx.recv_timeout(deadline.saturating_duration_since(Instant::now()));
+        let second_timeout =
+            timed_out_rx.recv_timeout(deadline.saturating_duration_since(Instant::now()));
+        let materializer_still_gated = published_rx.is_empty();
+        let delivery_was_gated = !delay_delivery || answer_rx.is_empty();
+        if delay_delivery {
+            crossbeam_channel::after(Duration::from_millis(1250))
+                .recv()
+                .unwrap();
+            deliver_tx.send(()).unwrap();
+            deliver_tx.send(()).unwrap();
+        }
         let first = answer_rx.recv_timeout(deadline.saturating_duration_since(Instant::now()));
         let second = answer_rx.recv_timeout(deadline.saturating_duration_since(Instant::now()));
         release_tx.send(()).unwrap();
         worker.join().unwrap();
         concurrent.join().unwrap();
+        for (thread, phase, at) in phase_rx.try_iter() {
+            eprintln!(
+                "cg_timing command={command} daemon={daemon} thread={thread:?} at_us={} {phase}",
+                at.duration_since(timing_origin).as_micros()
+            );
+        }
         assert!(entered.is_ok(), "request never reached materialization");
+        assert!(
+            first_timeout.is_ok() && second_timeout.is_ok(),
+            "a read did not time out while materialization was gated"
+        );
+        assert!(
+            materializer_still_gated,
+            "materialization finished before the timeout observations"
+        );
+        assert!(
+            delivery_was_gated,
+            "post-timeout delivery gate did not hold the reply"
+        );
         for bounded_answer in [first, second] {
-            let answered_in_bound = bounded_answer.is_ok();
-            let (elapsed, answer) = bounded_answer.unwrap_or_else(|_| answer_rx.recv().unwrap());
-            eprintln!("{command}: changed_files=64 read_ms={} bound_ms={} answered_before_release={answered_in_bound}", elapsed.as_millis(), bound.as_millis());
+            let answered_while_gated = bounded_answer.is_ok();
+            let (elapsed, answer, waits) =
+                bounded_answer.unwrap_or_else(|_| answer_rx.recv().unwrap());
+            eprintln!("{command}: changed_files=64 read_ms={} bound_ms={} answered_before_release={answered_while_gated} delayed_delivery={delay_delivery}", elapsed.as_millis(), bound.as_millis());
             assert!(
-                answered_in_bound,
-                "{command} waited {elapsed:?} for materialization past {bound:?}"
+                answered_while_gated,
+                "read did not deliver after its bounded wait"
             );
+            assert!(
+                !waits.is_empty(),
+                "read did not observe the materialization wait policy"
+            );
+            let wait_deadline = waits[0].deadline;
+            assert_eq!(
+                waits[0].stage,
+                crate::views::callgraph::ReadWaitStage::QueryLock
+            );
+            assert!(
+                waits.iter().any(|wait| wait.timed_out),
+                "read did not observe a bounded timeout"
+            );
+            for wait in waits {
+                eprintln!("cg_wait command={command} daemon={daemon} stage={:?} requested_us={} observed_us={} timed_out={}", wait.stage, wait.requested.as_micros(), wait.finished.duration_since(wait.started).as_micros(), wait.timed_out);
+                assert_eq!(
+                    wait.deadline, wait_deadline,
+                    "a wait stage reset the request deadline"
+                );
+                assert!(
+                    wait.requested <= bound,
+                    "wait requested more than the production bound: {wait:?}"
+                );
+                assert_eq!(
+                    wait.requested,
+                    wait_deadline.saturating_duration_since(wait.started),
+                    "wait did not receive only the remaining request budget"
+                );
+            }
+            if delay_delivery {
+                assert!(
+                    elapsed > bound + Duration::from_secs(1),
+                    "controlled response delay was not exercised"
+                );
+            }
             if command == "zoom" {
                 assert!(answer.success, "{:?}", answer.data);
                 assert!(answer.data.to_string().contains("export function target"));
@@ -15579,22 +15698,27 @@ mod tests {
 
     #[test]
     fn callgraph_views_rematerialization_bounds_callers_then_serves_publication() {
-        assert_callgraph_views_rematerialization_bounded("callers", false);
+        assert_callgraph_views_rematerialization_bounded("callers", false, false);
     }
 
     #[test]
     fn callgraph_views_rematerialization_bounds_zoom_then_serves_publication() {
-        assert_callgraph_views_rematerialization_bounded("zoom", false);
+        assert_callgraph_views_rematerialization_bounded("zoom", false, false);
     }
 
     #[test]
     fn callgraph_views_rematerialization_bounds_daemon_callers_then_serves_publication() {
-        assert_callgraph_views_rematerialization_bounded("callers", true);
+        assert_callgraph_views_rematerialization_bounded("callers", true, false);
     }
 
     #[test]
     fn callgraph_views_rematerialization_bounds_daemon_zoom_then_serves_publication() {
-        assert_callgraph_views_rematerialization_bounded("zoom", true);
+        assert_callgraph_views_rematerialization_bounded("zoom", true, false);
+    }
+
+    #[test]
+    fn callgraph_views_wait_bound_excludes_controlled_post_timeout_reply_delay() {
+        assert_callgraph_views_rematerialization_bounded("zoom", false, true);
     }
 
     fn assert_callgraph_acquisition_lifecycle(

crates/aft/src/commands/zoom.rs

@@ -1846,7 +1846,17 @@ fn zoom_one_symbol_inner(
 /// counts as that first use and can start the build.
 fn unavailable_callgraph_field(ctx: &AppContext) -> Option<serde_json::Value> {
     use crate::feature_status::{observe_callgraph_query, IndexEffective};
+    // Timing hooks for the read-bound test. Unconfigured test contexts have no
+    // cache root, so the hooks are skipped there instead of panicking.
+    #[cfg(test)]
+    if let Some(root) = ctx.canonical_cache_root_opt() {
+        crate::storage_retention::test_hook(&root, "cg-timing:zoom-observe-begin");
+    }
     let observation = observe_callgraph_query(ctx);
+    #[cfg(test)]
+    if let Some(root) = ctx.canonical_cache_root_opt() {
+        crate::storage_retention::test_hook(&root, "cg-timing:zoom-observe-return");
+    }
     let code = match observation.effective {
         IndexEffective::Ready => return None,
         IndexEffective::Off => "callgraph_off",

crates/aft/src/context.rs

@@ -13236,6 +13236,8 @@ impl AppContext {
             .lock()
             .as_ref()
             .is_some_and(|view| view.matches(&root, &storage, &family, policy));
+        #[cfg(test)]
+        crate::storage_retention::test_hook(&root, "cg-timing:context-route-ready");
         if !own_view_bound || self.callgraph_borrow.lock().is_some() {
             if let Some(borrowed) =
                 self.borrowed_callgraph_view_reader(&root, &storage, &family, policy, false)
@@ -13257,6 +13259,8 @@ impl AppContext {
                 return Some(borrowed);
             }
         }
+        #[cfg(test)]
+        crate::storage_retention::test_hook(&root, "cg-timing:context-view-select-begin");
         let view = {
             let mut slot = self.callgraph_view.lock();
             if !slot
@@ -13320,13 +13324,19 @@ impl AppContext {
             }
             slot.as_ref().unwrap().clone()
         };
+        #[cfg(test)]
+        crate::storage_retention::test_hook(&root, "cg-timing:context-view-selected");
         self.record_checkout_pending_watcher_gap();
+        #[cfg(test)]
+        crate::storage_retention::test_hook(&root, "cg-timing:context-gap-recorded");
         let result = match acquire(&view) {
             Err(error) if error.reason.contains(crate::views::trigram::BASE_CAP_BUSY) => self
                 .borrowed_callgraph_view_reader(&root, &storage, &family, policy, true)
                 .unwrap_or(Err(error)),
             result => result,
         };
+        #[cfg(test)]
+        crate::storage_retention::test_hook(&root, "cg-timing:context-view-read-return");
         if pinned_config_for(self.config_pin_key()).is_some() {
             CALLGRAPH_QUERY_READERS.with(|readers| {
                 readers
@@ -13342,6 +13352,8 @@ impl AppContext {
                 self.note_classified_view_load_failure("callgraph", &error.reason, false);
             }
         }
+        #[cfg(test)]
+        crate::storage_retention::test_hook(&root, "cg-timing:context-accounted");
         Some(result)
     }
 

crates/aft/src/storage_retention.rs

@@ -6520,6 +6520,8 @@ fn collect_v3_family_at(
     // The final measurement still settles dead reservations and never records
     // a partial total. No publication is admitted through this handle.
     let mut registry = RegistryWriter::open_for_barrier(storage, family, &lock).map_err(v3)?;
+    #[cfg(test)]
+    test_hook(storage, "v3-registry-opened");
     // Base references get the first turn, before member cleanup or accounting
     // can consume the visit. All three phases share the reservation lock.
     if !cancelled()
@@ -16555,7 +16557,24 @@ mod v3_base_reference_collector_tests {
             .unwrap(),
         );
         drop(composed);
-        for sweep in 0..2 {
+        // A slow registry open can exhaust a visit before the reference sweep
+        // starts, even though later visits can still reclaim the unused base.
+        let delay_first_open = std::sync::atomic::AtomicBool::new(true);
+        let _slow_open = observe_test_hook(
+            storage.path(),
+            Arc::new(move |step| {
+                if step == "v3-registry-opened"
+                    && delay_first_open.swap(false, std::sync::atomic::Ordering::SeqCst)
+                {
+                    std::thread::sleep(Duration::from_millis(500));
+                }
+            }),
+        );
+        // Only admitted reference sweeps advance the persisted unreferenced
+        // count. Keep the visit budget short and wait for reclamation instead
+        // of assuming that two collector visits always admit two sweeps.
+        const MAX_SWEEPS: usize = 16;
+        for sweep in 1..=MAX_SWEEPS {
             v3::set_walk_delay_for_test(Duration::from_millis(500));
             let mut report = SweepReport::default();
             let result = collect_one_v3_family(
@@ -16574,9 +16593,25 @@ mod v3_base_reference_collector_tests {
                 report.skipped_families,
                 [format!("blobs/v3/{FAMILY}: measurement_owed")]
             );
-            assert_eq!(registered(storage.path(), &unused.scope), sweep == 0);
-            assert!(registered(storage.path(), &held.scope));
-            assert!(registered(storage.path(), &referenced.scope));
+            assert!(
+                registered(storage.path(), &held.scope),
+                "visit {sweep}: {report:?}"
+            );
+            assert!(
+                registered(storage.path(), &referenced.scope),
+                "visit {sweep}: {report:?}"
+            );
+            if !registered(storage.path(), &unused.scope) {
+                eprintln!(
+                    "unused base reclaimed after {sweep} collector visits with accounting owed"
+                );
+                break;
+            }
+            assert!(
+                sweep < MAX_SWEEPS,
+                "unused base {} still registered after {MAX_SWEEPS} collector visits with accounting owed; last report: {report:?}",
+                unused.scope
+            );
         }
         assert!(!v3::view_dir(storage.path(), FAMILY, &unused.scope)
             .unwrap()