diff --git a/.alfonso/release-notes/next/honest-walk-partials.md b/.alfonso/release-notes/next/honest-walk-partials.md new file mode 100644 index 000000000..3e6613525 --- /dev/null +++ b/.alfonso/release-notes/next/honest-walk-partials.md @@ -0,0 +1 @@ +- `glob`, `grep` and `outline` on a large directory outside the project (such as `~/.cargo/registry/src`) now return the files found so far, marked as a partial walk, instead of a `read_blocked` error that told agents to change macOS Privacy & Security settings. `read_blocked` is now reserved for one filesystem call that stops responding; it names that path, and a walk that stalls keeps what it had already found. A walk the request stops waiting for now exits and frees its helper slot. diff --git a/benchmarks/aft-search/slice-descriptors/land-honest-walk-partials.json b/benchmarks/aft-search/slice-descriptors/land-honest-walk-partials.json new file mode 100644 index 000000000..00880d5a6 --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-honest-walk-partials.json @@ -0,0 +1,12 @@ +{ + "slice_class": "engine_unwired", + "targeted_mechanism": "none", + "kind": "harness", + "fixtures": [ + "crates/aft/tests/integration/tool_call_parity_test.rs" + ], + "fenced_files": [ + "crates/aft/src/search_index.rs" + ], + "note": "Fallback directory walks run under the request budget and snapshot each match's mtime and the resolved filter root before returning, so partial-result ordering needs no filesystem access after the walk. Complete walks keep mtime-descending order with the same display-path tie-break; indexed search ranking and routing are unchanged. Real-query rows must remain byte-identical." +} diff --git a/crates/aft/src/bounded_io.rs b/crates/aft/src/bounded_io.rs index 224f2ee44..f7de657fa 100644 --- a/crates/aft/src/bounded_io.rs +++ b/crates/aft/src/bounded_io.rs @@ -7,9 +7,7 @@ use std::cell::RefCell; use std::collections::{HashMap, HashSet}; use std::io::{self, Read}; use std::path::{Path, PathBuf}; -#[cfg(test)] -use std::sync::atomic::AtomicUsize; -use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; use std::sync::{Arc, Condvar, LazyLock, Mutex}; use std::time::{Duration, Instant}; @@ -31,6 +29,10 @@ struct Measurements { struct Failure { code: &'static str, message: String, + /// Set when a walk stalled on this path after it had already found + /// results. The reply keeps those results and names the path instead of + /// becoming an error; see `finish_response`. + partial: Option, } impl std::fmt::Display for Failure { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { @@ -307,16 +309,51 @@ pub(crate) fn finish_response(response: Response) -> Response { .clone() }); match error { + Some(Failure { + code, + message, + partial: Some(path), + }) if response.success => keep_partial_reply(response, code, &path, &message), Some(error) => Response::error(&response.id, error.code, error.message), None => response, } } + +/// A walk that stalled after finding results keeps them: the reply is marked +/// incomplete, names the stalled path as a gap, and ends with the same +/// `read_blocked` sentence an error reply would carry. `finish_response` can run +/// more than once on one reply, so this never adds the gap or the line twice. +fn keep_partial_reply(mut response: Response, code: &str, path: &Path, message: &str) -> Response { + if let Some(object) = response.data.as_object_mut() { + object.insert("complete".into(), serde_json::Value::Bool(false)); + let gap = serde_json::json!({"kind": code, "path": path, "reason": message}); + let gaps = object + .entry("gaps") + .or_insert_with(|| serde_json::Value::Array(Vec::new())); + if let Some(gaps) = gaps.as_array_mut() { + if !gaps.contains(&gap) { + gaps.push(gap); + } + } + if let Some(serde_json::Value::String(text)) = object.get_mut("text") { + if !text.ends_with(message) { + if !text.is_empty() { + text.push_str(if text.ends_with('\n') { "\n" } else { "\n\n" }); + } + text.push_str(message); + } + } + } + response +} thread_local! { static CURRENT: RefCell> = const { RefCell::new(None) }; // Rayon shares the request's single-flight probes, but does not contend on // that map per file. Its workers remember successful lookups locally. static DIRECTORIES: RefCell<(u64, HashMap)> = RefCell::new((0, HashMap::new())); static HELPER_DEADLINE: std::cell::Cell> = const { std::cell::Cell::new(None) }; + // Set only on a walk's helper thread; see `run_walk`. + static WALK: RefCell>> = const { RefCell::new(None) }; } pub(crate) struct Scope(Option); impl Drop for Scope { @@ -338,6 +375,15 @@ pub(crate) fn with_budget(budget: Option, work: impl FnOnce() -> T) - pub(crate) fn enter(budget: Option) -> Scope { Scope(CURRENT.with(|current| current.replace(budget))) } +/// The longest one filesystem call may take, and the longest a walk may go +/// without progress before it counts as stalled. +fn single_access_budget() -> Duration { + #[cfg(test)] + if let Some(budget) = regression_tests::single_access_budget_override() { + return budget; + } + IO_BUDGET +} fn deadline(at: Option) -> Instant { request_deadline(at) .into_iter() @@ -348,20 +394,47 @@ fn request_deadline(at: Option) -> Option { .chain(CURRENT.with(|budget| budget.borrow().as_ref().map(|b| b.deadline))) .min() } +fn failure_message(code: &str, path: &Path, reason: &str) -> String { + format!("{code}: {}: {reason}", path.display()) +} +/// The reason given when one OS call did not return in time. +fn blocked_reason(waited: Duration) -> String { + let hint = if cfg!(target_os = "macos") { + "; on macOS, check ck-aft access in Privacy & Security, then retry" + } else { + "" + }; + format!( + "OS access did not finish within {:.3} s{hint}", + waited.as_secs_f64() + ) +} fn failure(path: &Path, code: &'static str, reason: &str) -> io::Error { - let message = format!("{code}: {}: {reason}", path.display()); + let message = failure_message(code, path, reason); if let Some(budget) = current() { - budget - .failure - .lock() - .unwrap_or_else(|e| e.into_inner()) - .get_or_insert_with(|| Failure { + let mut recorded = budget.failure.lock().unwrap_or_else(|e| e.into_inner()); + // The first failure wins, except over a stall whose results were kept: + // a later failure makes the whole reply an error. + if recorded + .as_ref() + .is_none_or(|failure| failure.partial.is_some()) + { + *recorded = Some(Failure { code, message: message.clone(), + partial: None, }); + } budget.failed.store(true, Ordering::Release); } - io::Error::new(io::ErrorKind::TimedOut, Failure { code, message }) + io::Error::new( + io::ErrorKind::TimedOut, + Failure { + code, + message, + partial: None, + }, + ) } fn prior_failure() -> Option { CURRENT.with(|current| { @@ -473,8 +546,8 @@ impl Drop for Permit { self.0.available.notify_one(); } } -/// Spawn only for boundary probes, whole walks, or reads on risky filesystems. -/// A stuck call retains its limiter permit, never executor/root ownership. +/// Spawn only for boundary probes or reads on risky filesystems; whole walks use +/// `run_walk`. A stuck call retains its limiter permit, never executor/root ownership. pub(crate) fn run( path: &Path, at: Option, @@ -516,6 +589,35 @@ fn run_with_timeout( if let Some(error) = prior_failure() { return Err(error); } + // A walk's helper makes its per-directory calls inline instead of taking + // a helper slot per directory. The request thread watches the walk's + // progress, so one of these calls that stops returning is still reported, + // under its own path. + if let Some(walk) = WALK.with(|walk| walk.borrow().clone()) { + if walk.expired() { + // Not a request failure: the walk stops here and reports itself + // truncated. + walk.cut.store(true, Ordering::Relaxed); + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "the walk reached its deadline before this call", + )); + } + if let Some(probe) = &probe { + // This probe runs on the walk's existing helper. Publish its access + // phase just as a newly scheduled helper would, so shared waiters + // use the running call's cap rather than a stale queue deadline. + let started = Instant::now(); + let deadline = request_deadline(at).map_or(started + access_budget, |at| { + at.min(started + access_budget) + }); + probe.set_deadline(deadline, Some(started)); + } + walk.reached(path); + let result = work(path.to_path_buf()); + walk.step(); + return result; + } let queued = Instant::now(); let request_deadline = request_deadline(at); // Without a request deadline, queueing and access each get their own cap. @@ -632,6 +734,384 @@ fn run_with_timeout( } } } + +/// Reply time a walk leaves inside the request budget for the work after it +/// (sorting, presence checks, formatting). +const WALK_REPLY_MARGIN: Duration = Duration::from_millis(500); + +/// State shared by a walk's helper thread and the request thread waiting on it. +/// +/// The request thread does not time the walk as a whole. The walker stops +/// itself at `deadline` and reports a truncated walk. The waiting thread +/// reports `read_blocked` only when the walk makes no progress for the stall +/// window, which means one filesystem call inside it has not returned. +pub(crate) struct WalkControl { + started: Instant, + deadline: Instant, + /// Set by the request thread when it stops waiting, so an abandoned walk + /// exits at its next entry and frees its helper slot. + stop: AtomicBool, + /// Set when a call inside the walk was refused because the walk's + /// deadline had passed; the walk did not finish. + cut: AtomicBool, + entries: AtomicUsize, + /// Time of the last progress, in nanoseconds after `started`. + progress: AtomicU64, + /// The path the walk last started work on, named if the walk stalls. + current: Mutex, +} +impl WalkControl { + fn new(deadline: Instant, path: &Path) -> Self { + Self { + started: Instant::now(), + deadline, + stop: AtomicBool::new(false), + cut: AtomicBool::new(false), + entries: AtomicUsize::new(0), + progress: AtomicU64::new(0), + current: Mutex::new(path.to_path_buf()), + } + } + /// Records one more walked entry. Returns true when the walk must stop now: + /// its deadline passed or the request stopped waiting for it. The walk + /// then drops the entry unprocessed, so it is not counted. + pub(crate) fn advance(&self) -> bool { + if self.expired() { + return true; + } + self.entries.fetch_add(1, Ordering::Relaxed); + self.step(); + false + } + pub(crate) fn expired(&self) -> bool { + self.stop.load(Ordering::Relaxed) || Instant::now() >= self.deadline + } + /// Names the path the walk is about to work on, such as a directory it is + /// about to open. + pub(crate) fn reached(&self, path: &Path) { + let mut current = self.current.lock().unwrap_or_else(|e| e.into_inner()); + current.clear(); + current.push(path); + drop(current); + self.step(); + #[cfg(test)] + regression_tests::before_walk_access(path); + } + /// Records one more entry of a directory the walk is reading or checking, + /// and names `path` as the call about to run. A large directory makes + /// many calls between `reached` updates (one per entry read, one per type + /// check); each returns promptly, but together they can outlast the stall + /// window. Counting each one as progress means only a single call that + /// does not return is reported as a stall. Returns true when the walk must + /// stop now, like `advance`. + pub(crate) fn entry(&self, path: &Path) -> bool { + let mut current = self.current.lock().unwrap_or_else(|e| e.into_inner()); + current.clear(); + current.push(path); + drop(current); + self.step(); + #[cfg(test)] + regression_tests::before_walk_entry(path); + self.expired() + } + /// Stat a discovered match as entry progress, not as another directory + /// access. The request can stop waiting if this one metadata call stalls. + pub(crate) fn metadata(&self, path: &Path) -> io::Result { + if self.entry(path) { + self.cut.store(true, Ordering::Relaxed); + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "the walk reached its deadline before metadata", + )); + } + let result = stat(path.to_path_buf()); + self.step(); + result + } + /// True when a call inside the walk was refused at the walk's deadline. + pub(crate) fn cut_short(&self) -> bool { + self.cut.load(Ordering::Relaxed) + } + pub(crate) fn entries(&self) -> usize { + self.entries.load(Ordering::Relaxed) + } + #[cfg(test)] + pub(crate) fn deadline(&self) -> Instant { + self.deadline + } + /// A control for calling a walker directly in a test, outside `run_walk`. + #[cfg(test)] + pub(crate) fn unbounded_for_test(path: &Path) -> Self { + Self::new(Instant::now() + Duration::from_secs(3600), path) + } + fn step(&self) { + let elapsed = u64::try_from(self.started.elapsed().as_nanos()).unwrap_or(u64::MAX); + self.progress.fetch_max(elapsed, Ordering::Relaxed); + } + fn last_progress(&self) -> Instant { + self.started + Duration::from_nanos(self.progress.load(Ordering::Relaxed)) + } +} + +/// How a walk ended, as seen by the request thread. +pub(crate) enum WalkEnd { + /// The walk returned, complete or stopped at its own bounds. + Finished(T), + /// The request stopped waiting and told the walk to stop. The caller keeps + /// whatever the walk had already shared with it, and reports the walk + /// truncated. `stall` is set when one call made no progress for the stall + /// window. + Stopped { + stall: Option, + entries_visited: usize, + }, +} + +enum WalkMessage { + Started, + Finished(T), +} + +/// One filesystem call inside a walk that stopped making progress. +pub(crate) struct WalkStall { + path: PathBuf, + waited: Duration, +} +impl WalkStall { + #[cfg(test)] + pub(crate) fn path(&self) -> &Path { + &self.path + } + /// Records the stall as the request's `read_blocked` failure. With + /// `kept_partial`, the caller keeps results the walk found before it + /// stalled, and the reply stays a partial success that names the stalled + /// path; otherwise the reply becomes the error. + pub(crate) fn record(&self, kept_partial: bool) { + let reason = blocked_reason(self.waited); + if !kept_partial { + let _ = failure(&self.path, "read_blocked", &reason); + return; + } + if let Some(budget) = current() { + // `failed` stays clear: later calls in this request still run, + // and any of them that fails turns the reply into its error. + budget + .failure + .lock() + .unwrap_or_else(|e| e.into_inner()) + .get_or_insert_with(|| Failure { + code: "read_blocked", + message: failure_message("read_blocked", &self.path, &reason), + partial: Some(self.path.clone()), + }); + } + } +} + +/// Inside a walk's helper, names the path the walk is about to work on, so a +/// stall there is reported under that path. Does nothing outside a walk. +pub(crate) fn walk_reached(path: &Path) { + WALK.with(|walk| { + if let Some(walk) = walk.borrow().as_ref() { + walk.reached(path); + } + }); +} + +/// Inside a walk's helper, records that the walk examined one more entry, +/// such as in a walker's entry filter, which also sees the entries the walk's +/// own loop never receives. Without this, a directory of many filtered-out +/// entries would look like one call making no progress. It does not rename +/// the walk's current path: the filter runs between the walker's own calls. +/// +/// Returns true once the walk must stop: its deadline passed or the request +/// stopped waiting for it. The walk is then marked cut short, so it reports +/// itself truncated. A filter must then keep the entry, not drop it. Dropping +/// an entry does not stop the walker: it goes on reading every remaining entry +/// of each directory it has open, and opening each remaining sibling +/// directory, only to drop them too, while the walk holds its helper slot. +/// The kept entry instead reaches the walk's own loop, whose +/// `WalkControl::advance` sees the same expiry and ends the walk before +/// processing that entry. Every walker whose filter calls this must check +/// `advance` first for each entry it receives. +/// +/// Outside a walk it does nothing and returns false. +pub(crate) fn walk_entry_must_stop(_path: &Path) -> bool { + WALK.with(|walk| { + let walk = walk.borrow(); + let Some(walk) = walk.as_ref() else { + return false; + }; + walk.step(); + #[cfg(test)] + regression_tests::before_walk_entry(_path); + if walk.expired() { + walk.cut.store(true, Ordering::Relaxed); + return true; + } + false + }) +} + +/// Runs a whole directory walk on a helper thread. +/// +/// The walk is bounded at its iterator: `work` must call +/// `WalkControl::advance` for each entry and stop when it returns true. Its +/// deadline is `deadline`, moved earlier only to leave reply time inside the +/// request budget, never cut to the single-call limit. The request thread +/// waits until the walk returns, until the request budget runs out, or until +/// the walk makes no progress for the stall window. A walk that is slow but +/// still moving is never reported as blocked. +/// +/// `Err` means the walk did not run (no helper slot or the deadline already +/// passed); that failure is already recorded on the request. +pub(crate) fn run_walk( + path: &Path, + deadline: Instant, + work: impl FnOnce(PathBuf, &WalkControl) -> T + Send + 'static, +) -> io::Result> { + run_walk_with_limiter(path, deadline, Arc::clone(&HELPERS), work) +} + +/// The walk's own deadline and the time the request thread stops waiting. +fn walk_deadlines(requested: Instant) -> (Instant, Instant) { + let now = Instant::now(); + let window = single_access_budget(); + let request = CURRENT.with(|budget| budget.borrow().as_ref().map(|b| b.deadline)); + let walk = match request { + Some(request) => { + let margin = WALK_REPLY_MARGIN.min(request.saturating_duration_since(now) / 2); + requested.min(request.checked_sub(margin).unwrap_or(request)) + } + None => requested, + }; + let wait = request.map_or(walk + window, |request| request.min(walk + window)); + (walk, wait) +} + +fn run_walk_with_limiter( + path: &Path, + deadline: Instant, + limiter: Arc, + work: impl FnOnce(PathBuf, &WalkControl) -> T + Send + 'static, +) -> io::Result> { + if let Some(error) = prior_failure() { + return Err(error); + } + // A walk started from inside another walk shares the outer walk's bounds. + if let Some(outer) = WALK.with(|walk| walk.borrow().clone()) { + return Ok(WalkEnd::Finished(work(path.to_path_buf(), &outer))); + } + let (walk_deadline, wait_deadline) = walk_deadlines(deadline); + if Instant::now() >= walk_deadline { + return Err(failure( + path, + "read_deadline_exceeded", + "filesystem deadline elapsed before this call started", + )); + } + let Some(permit) = limiter.acquire(walk_deadline) else { + return Err(failure(path, "read_io_queue_timeout", "no filesystem helper slot became available before the deadline; this call did not start")); + }; + let control = Arc::new(WalkControl::new(walk_deadline, path)); + let owned = path.to_path_buf(); + let budget = current(); + let confinement = crate::context::confinement::capture(); + #[cfg(test)] + if let Some(counts) = budget.as_ref().map(|b| Arc::clone(&b.measurements)) { + counts.round_trips.fetch_add(1, Ordering::Relaxed); + counts.spawns.fetch_add(1, Ordering::Relaxed); + } + let (tx, rx) = std::sync::mpsc::sync_channel(2); + let running = Arc::new(Mutex::new(false)); + let helper_running = Arc::clone(&running); + let helper_control = Arc::clone(&control); + std::thread::Builder::new() + .name("aft-filesystem-walk".into()) + .spawn(move || { + #[cfg(test)] + regression_tests::before_helper(&owned); + // Publish helper start and its first progress together. A waiter + // must never charge scheduling delay against the stall window. + let mut running = helper_running.lock().unwrap_or_else(|e| e.into_inner()); + if helper_control.expired() { + return; + } + helper_control.step(); + *running = true; + drop(running); + let _ = tx.send(WalkMessage::Started); + WALK.with(|walk| *walk.borrow_mut() = Some(Arc::clone(&helper_control))); + let result = crate::context::confinement::inherit(confinement, || { + with_budget(budget, || work(owned, &helper_control)) + }); + WALK.with(|walk| *walk.borrow_mut() = None); + // The walk is done with the filesystem: free its slot before the + // reply, so the next request can take it as soon as this one ends. + drop(permit); + let _ = tx.send(WalkMessage::Finished(result)); + })?; + let window = single_access_budget(); + loop { + let wake = if *running.lock().unwrap_or_else(|e| e.into_inner()) { + wait_deadline.min(control.last_progress() + window) + } else { + walk_deadline + }; + match rx.recv_timeout(wake.saturating_duration_since(Instant::now())) { + Ok(WalkMessage::Started) => continue, + Ok(WalkMessage::Finished(result)) => return Ok(WalkEnd::Finished(result)), + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { + control.stop.store(true, Ordering::Relaxed); + if !*running.lock().unwrap_or_else(|e| e.into_inner()) { + return Err(failure( + path, + "read_io_queue_timeout", + "filesystem helper did not start before the deadline; this call did not start", + )); + } + return Err(io::Error::other("filesystem helper stopped")); + } + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { + let now = Instant::now(); + let running = running.lock().unwrap_or_else(|e| e.into_inner()); + if !*running { + // Not yet scheduled: a scheduling delay is not a stall. + if now >= walk_deadline { + control.stop.store(true, Ordering::Relaxed); + return Err(failure( + path, + "read_io_queue_timeout", + "filesystem helper did not start before the deadline; this call did not start", + )); + } + continue; + } + drop(running); + let last = control.last_progress(); + let waited = now.saturating_duration_since(last); + // Request expiry alone does not mean an OS call stalled for + // the single-access window; return an ordinary walk partial. + let stalled = waited >= window; + if stalled || now >= wait_deadline { + control.stop.store(true, Ordering::Relaxed); + let stall = stalled.then(|| WalkStall { + path: control + .current + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone(), + waited, + }); + return Ok(WalkEnd::Stopped { + stall, + entries_visited: control.entries(), + }); + } + } + } + } +} fn directory(path: &Path, at: Option) -> io::Result { if let Some(error) = prior_failure() { return Err(error); @@ -810,12 +1290,12 @@ fn probe_target(path: &Path) -> io::Result<()> { { let metadata = std::fs::symlink_metadata(path)?; if metadata.file_type().is_symlink() { - run(path, None, std::fs::metadata)? + run(path, None, stat)? } else { metadata } } else { - match run(path, None, std::fs::metadata) { + match run(path, None, stat) { Ok(metadata) => metadata, Err(error) => { if error.raw_os_error() == Some(1) { @@ -879,6 +1359,12 @@ fn local_filesystem(_: &Path) -> io::Result { Ok(false) } +/// One `stat` that follows symlinks, as a single bounded access. +fn stat(path: PathBuf) -> io::Result { + #[cfg(test)] + regression_tests::before_stat(&path); + std::fs::metadata(path) +} pub(crate) fn read(path: &Path, at: Option) -> io::Result> { if Instant::now() >= deadline(at) { return Err(failure( @@ -978,12 +1464,17 @@ pub(crate) fn metadata(path: &Path) -> io::Result { return Ok(metadata); } } - run_traversal(path, std::fs::metadata) + run_traversal(path, stat) } #[cfg(test)] #[path = "bounded_io_tests.rs"] mod regression_tests; +#[cfg(test)] +pub(crate) use regression_tests::{ + block_stat, expire_walk_after_entries, many_entry_tree, run_walk_in_own_slot, slow_walk, + slow_walk_entries, stall_walk_at, stall_walk_entry, with_single_access_budget, +}; #[cfg(all(test, unix))] pub(crate) use regression_tests::{stall_helper, stall_running_helper}; #[cfg(all(test, unix))] diff --git a/crates/aft/src/bounded_io_tests.rs b/crates/aft/src/bounded_io_tests.rs index a2b456bd9..095ab78d6 100644 --- a/crates/aft/src/bounded_io_tests.rs +++ b/crates/aft/src/bounded_io_tests.rs @@ -12,6 +12,299 @@ static HELPER_GATES: LazyLock>> = static RUNNING_GATES: LazyLock>> = LazyLock::new(|| Mutex::new(HashMap::new())); +thread_local! { + static SINGLE_ACCESS_BUDGET: std::cell::Cell> = + const { std::cell::Cell::new(None) }; +} + +pub(super) fn single_access_budget_override() -> Option { + SINGLE_ACCESS_BUDGET.with(std::cell::Cell::get) +} + +/// Runs `work` with a shorter single-call limit, which is also the walk stall +/// window, on this thread only. Both are read on the requesting thread. +pub(crate) fn with_single_access_budget(budget: Duration, work: impl FnOnce() -> T) -> T { + struct Restore(Option); + impl Drop for Restore { + fn drop(&mut self) { + SINGLE_ACCESS_BUDGET.with(|budget| budget.set(self.0)); + } + } + let _restore = Restore(SINGLE_ACCESS_BUDGET.with(|current| current.replace(Some(budget)))); + work() +} + +/// A test hook on the per-directory calls a walk makes under `roots`, or with +/// `entries`, on the per-entry calls inside one directory. +struct WalkHook { + roots: Vec, + entries: bool, + delay: Duration, + block_at: Option, + /// For an entry hook: block the call for exactly this path (one of `roots`). + block_exact: bool, + /// Sets the walk's stop flag right after this many hooked calls, as the + /// request does when it stops waiting, so the walk expires at a known + /// entry instead of at a wall-clock deadline a loaded machine can reach + /// at any point. + expire_after: Option, + seen: AtomicUsize, + /// Paths of the calls made while the walk could still go on, in order. + before_expiry: Mutex>, + /// Calls made after the walk's deadline passed or its request stopped + /// waiting: work the walk should no longer be doing. + after_expiry: AtomicUsize, + blocked: Mutex>>, + release: Mutex>>, +} + +static WALK_HOOKS: LazyLock>>> = LazyLock::new(|| Mutex::new(Vec::new())); + +/// Removes its hook when dropped, and releases a blocked call. +pub(crate) struct WalkHookGuard { + hook: Arc, + release: Option>, + /// Receives the path of the call the hook blocked. + pub(crate) blocked: Receiver, +} + +impl WalkHookGuard { + pub(crate) fn release(&mut self) { + self.release.take(); + } + /// Paths of the hooked calls made before the walk had to stop, in order. + pub(crate) fn calls_before_expiry(&self) -> Vec { + self.hook.before_expiry.lock().unwrap().clone() + } + /// How many hooked calls the walk made after it had to stop. + pub(crate) fn calls_after_expiry(&self) -> usize { + self.hook.after_expiry.load(Ordering::Relaxed) + } + /// Checks the partial result of a walk that expired: it holds at least + /// one file, and only files the walk checked before it had to stop. + /// Returns how many files it checked before then are missing from `found`. + pub(crate) fn files_missed_before_expiry(&self, found: &[PathBuf]) -> usize { + let checked = self.calls_before_expiry(); + let checked_files = checked + .iter() + .filter(|path| path.is_file()) + .collect::>(); + assert!(!found.is_empty(), "the walk kept no file it found"); + for path in found { + assert!( + checked_files.contains(path), + "{} was kept but not checked before the walk had to stop", + path.display() + ); + } + checked_files.len() - found.len() + } +} + +/// A canonical tree of 1,250 entries: 20 files at the root, and 30 +/// directories of 40 files each, so a walk expired after a few hundred entry +/// checks still has most of the tree left to read. +pub(crate) fn many_entry_tree() -> (tempfile::TempDir, PathBuf) { + let dir = tempfile::tempdir().unwrap(); + let root = std::fs::canonicalize(dir.path()).unwrap(); + for file in 0..20 { + std::fs::write(root.join(format!("root_{file:02}.rs")), "pub fn f() {}\n").unwrap(); + } + for directory in 0..30 { + let parent = root.join(format!("directory_{directory:02}")); + std::fs::create_dir(&parent).unwrap(); + for file in 0..40 { + std::fs::write(parent.join(format!("file_{file:02}.rs")), "pub fn f() {}\n").unwrap(); + } + } + (dir, root) +} + +/// Runs a walk on a helper limiter of its own with a single slot, and reports +/// whether that slot was already free when the walk's reply reached the caller. +pub(crate) fn run_walk_in_own_slot( + path: &Path, + deadline: Instant, + work: impl FnOnce(PathBuf, &WalkControl) -> T + Send + 'static, +) -> (io::Result>, bool) { + let limiter = Arc::new(Limiter::new(1)); + let end = run_walk_with_limiter(path, deadline, Arc::clone(&limiter), work); + let free = *limiter.live.lock().unwrap() == 0; + (end, free) +} + +impl Drop for WalkHookGuard { + fn drop(&mut self) { + WALK_HOOKS + .lock() + .unwrap() + .retain(|hook| !Arc::ptr_eq(hook, &self.hook)); + self.release.take(); + } +} + +fn walk_hook( + root: &Path, + entries: bool, + delay: Duration, + block_at: Option, + block_exact: bool, + expire_after: Option, +) -> WalkHookGuard { + let mut roots = vec![root.to_path_buf()]; + if let Ok(canonical) = std::fs::canonicalize(root) { + roots.push(canonical); + } + let (blocked_tx, blocked) = channel(); + let (release, release_rx) = channel(); + let hook = Arc::new(WalkHook { + roots, + entries, + delay, + block_at, + block_exact, + expire_after, + seen: AtomicUsize::new(0), + before_expiry: Mutex::new(Vec::new()), + after_expiry: AtomicUsize::new(0), + blocked: Mutex::new(Some(blocked_tx)), + release: Mutex::new(Some(release_rx)), + }); + WALK_HOOKS.lock().unwrap().push(Arc::clone(&hook)); + WalkHookGuard { + hook, + release: Some(release), + blocked, + } +} + +/// Slows every per-directory call a walk makes under `root` by `delay`. The +/// walk keeps making progress, so it is slow, not stalled. +pub(crate) fn slow_walk(root: &Path, delay: Duration) -> WalkHookGuard { + walk_hook(root, false, delay, None, false, None) +} + +/// Slows every per-entry call a walk makes under `root` by `delay`: each entry +/// read, type check, or filter inside a directory. Each call returns, so the +/// walk is slow, not stalled, however many entries one directory holds. +pub(crate) fn slow_walk_entries(root: &Path, delay: Duration) -> WalkHookGuard { + walk_hook(root, true, delay, None, false, None) +} + +/// Counts every per-entry call a walk makes under `root`, and expires the +/// walk right after the `checks`th one by setting its stop flag: the flag the +/// request sets when it stops waiting, which `WalkControl::expired` reads +/// alongside the deadline. The `checks`th call itself counts as made before +/// expiry; the walk's own check that follows it already sees the expiry. +pub(crate) fn expire_walk_after_entries(root: &Path, checks: usize) -> WalkHookGuard { + walk_hook(root, true, Duration::ZERO, None, false, Some(checks)) +} + +/// Blocks the per-entry call a walk makes for exactly `path` until the guard +/// is released or dropped: one entry whose check never returns. +pub(crate) fn stall_walk_entry(path: &Path) -> WalkHookGuard { + walk_hook(path, true, Duration::ZERO, None, true, None) +} + +/// Blocks the `access`th per-directory call a walk makes under `root` until +/// the guard is released or dropped: one call that stops making progress. +pub(crate) fn stall_walk_at(root: &Path, access: usize) -> WalkHookGuard { + walk_hook(root, false, Duration::ZERO, Some(access), false, None) +} + +pub(super) fn before_walk_access(path: &Path) { + run_walk_hook(path, false); +} + +pub(super) fn before_walk_entry(path: &Path) { + run_walk_hook(path, true); +} + +fn run_walk_hook(path: &Path, entries: bool) { + let hook = WALK_HOOKS + .lock() + .unwrap() + .iter() + .find(|hook| { + hook.entries == entries && hook.roots.iter().any(|root| path.starts_with(root)) + }) + .cloned(); + let Some(hook) = hook else { + return; + }; + let seen = hook.seen.fetch_add(1, Ordering::Relaxed) + 1; + // Hooked calls run on the walk's helper thread, where its control is set. + let expired = WALK.with(|walk| walk.borrow().as_ref().is_some_and(|walk| walk.expired())); + if expired { + hook.after_expiry.fetch_add(1, Ordering::Relaxed); + } else { + hook.before_expiry.lock().unwrap().push(path.to_path_buf()); + } + if hook.expire_after == Some(seen) { + WALK.with(|walk| { + if let Some(walk) = walk.borrow().as_ref() { + walk.stop.store(true, Ordering::Relaxed); + } + }); + } + let block = if hook.block_exact { + hook.roots.iter().any(|root| path == root) + } else { + hook.block_at == Some(seen) + }; + if block { + if let Some(blocked) = hook.blocked.lock().unwrap().take() { + let _ = blocked.send(path.to_path_buf()); + } + let release = hook.release.lock().unwrap().take(); + if let Some(release) = release { + let _ = release.recv_timeout(Duration::from_secs(30)); + } + } + if !hook.delay.is_zero() { + std::thread::sleep(hook.delay); + } +} + +static BLOCKED_STATS: LazyLock, Receiver<()>)>>> = + LazyLock::new(|| Mutex::new(HashMap::new())); + +/// Blocks the next single `stat` of `path` until dropped. +pub(crate) struct StatBlock { + path: PathBuf, + pub(crate) reached: Receiver<()>, + release: Option>, +} + +impl Drop for StatBlock { + fn drop(&mut self) { + BLOCKED_STATS.lock().unwrap().remove(&self.path); + self.release.take(); + } +} + +pub(crate) fn block_stat(path: &Path) -> StatBlock { + let (reached_tx, reached) = channel(); + let (release, release_rx) = channel(); + BLOCKED_STATS + .lock() + .unwrap() + .insert(path.to_path_buf(), (reached_tx, release_rx)); + StatBlock { + path: path.to_path_buf(), + reached, + release: Some(release), + } +} + +pub(super) fn before_stat(path: &Path) { + let blocked = BLOCKED_STATS.lock().unwrap().remove(path); + if let Some((reached, release)) = blocked { + let _ = reached.send(()); + let _ = release.recv_timeout(Duration::from_secs(30)); + } +} + #[cfg(unix)] pub(crate) struct HelperStall { path: PathBuf, @@ -713,3 +1006,306 @@ fn running_helper_timeout_has_only_native_platform_hint() { }; assert_eq!(hint, expected_hint); } + +#[test] +fn walk_probe_shared_waiter_uses_running_access_deadline() { + for shorter_waiter in [true, false] { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().to_path_buf(); + let mut gate = stall_walk_at(&path, 1); + let mut budget = Budget::after(Duration::from_secs(5)); + budget.access_budget = Duration::from_millis(200); + let owner_budget = budget.clone(); + let owner_path = path.clone(); + let owner = std::thread::spawn(move || { + with_budget(Some(owner_budget), || { + run_walk(&owner_path, Instant::now() + IO_BUDGET, |path, _| { + probe_directory(&path, None) + }) + }) + }); + gate.blocked.recv_timeout(IO_BUDGET).unwrap(); + // The inline probe has started. Waiters follow its access cap unless + // their own request deadline expires first. + let wait = if shorter_waiter { + Duration::from_millis(50) + } else { + Duration::from_secs(1) + }; + let error = with_budget(Some(budget), || { + probe_directory(&path, Some(Instant::now() + wait)) + }) + .unwrap_err(); + gate.release(); + owner.join().unwrap().unwrap(); + let expected = if shorter_waiter { + "read_io_queue_timeout" + } else { + "read_blocked" + }; + assert_eq!(timeout_code(&error), expected, "{error}"); + } +} + +#[test] +fn walk_request_expiry_before_stall_window_is_not_read_blocked() { + let (started_tx, started_rx) = channel(); + let (release_tx, release_rx) = channel(); + let caller = std::thread::spawn(move || { + with_budget(Some(Budget::after(Duration::from_millis(300))), || { + run_walk( + Path::new("short-walk"), + Instant::now() + IO_BUDGET, + move |_, _| { + started_tx.send(()).unwrap(); + let _ = release_rx.recv_timeout(IO_BUDGET); + }, + ) + }) + }); + started_rx.recv_timeout(IO_BUDGET).unwrap(); + let end = caller.join().unwrap().unwrap(); + release_tx.send(()).unwrap(); + assert!( + matches!(end, WalkEnd::Stopped { stall: None, .. }), + "request expiry alone must not claim a stalled OS call" + ); +} + +#[cfg(unix)] +#[test] +fn queued_walk_gets_full_stall_window_after_helper_starts() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("queued-walk"); + let gate = stall_helper(&path); + let (started_tx, started_rx) = channel(); + let (release_tx, release_rx) = channel(); + let (result_tx, result_rx) = channel(); + let caller = std::thread::spawn(move || { + with_single_access_budget(Duration::from_millis(400), || { + let end = run_walk_with_limiter( + &path, + Instant::now() + Duration::from_secs(5), + Arc::new(Limiter::new(1)), + move |_, _| { + started_tx.send(()).unwrap(); + let _ = release_rx.recv_timeout(IO_BUDGET); + 42 + }, + ); + result_tx.send(end).unwrap(); + }); + }); + gate.reached.recv_timeout(IO_BUDGET).unwrap(); + // Queue longer than the stall window, then let the access use less than + // its own window. Scheduling time must not make the running walk stall. + assert!(result_rx.recv_timeout(Duration::from_millis(600)).is_err()); + drop(gate); + started_rx.recv_timeout(IO_BUDGET).unwrap(); + let early = result_rx.recv_timeout(Duration::from_millis(250)); + let _ = release_tx.send(()); + caller.join().unwrap(); + assert!( + early.is_err(), + "queue time was charged against the access window" + ); + let end = result_rx.recv_timeout(IO_BUDGET).unwrap().unwrap(); + assert!(matches!(end, WalkEnd::Finished(42))); +} + +#[cfg(unix)] +#[test] +fn unscheduled_walk_expires_at_queue_deadline_and_never_runs_late() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("unscheduled-walk"); + let gate = stall_helper(&path); + let fired = Arc::new(AtomicBool::new(false)); + let work_fired = Arc::clone(&fired); + let limiter = Arc::new(Limiter::new(1)); + let walker_limiter = Arc::clone(&limiter); + let (result_tx, result_rx) = channel(); + let caller = std::thread::spawn(move || { + with_budget(Some(Budget::after(Duration::from_secs(5))), || { + let end = run_walk_with_limiter( + &path, + Instant::now() + Duration::from_millis(200), + walker_limiter, + move |_, _| { + work_fired.store(true, Ordering::Relaxed); + }, + ); + result_tx.send(end).unwrap(); + }); + }); + gate.reached.recv_timeout(IO_BUDGET).unwrap(); + let early = result_rx.recv_timeout(Duration::from_secs(1)); + drop(gate); + caller.join().unwrap(); + let end = early.expect("an unstarted walk must stop waiting at its queue deadline"); + assert_eq!(timeout_code(&end.err().unwrap()), "read_io_queue_timeout"); + assert!(!fired.load(Ordering::Relaxed)); +} + +#[test] +fn walk_deadline_is_not_cut_to_the_single_access_limit() { + let limiter = Arc::new(Limiter::new(1)); + let budget = Budget::after(Duration::from_secs(30)); + let requested = Instant::now() + Duration::from_secs(20); + // The walk keeps moving for three times the single-call limit. A healthy + // walk that outlasts one call's limit must still finish. + let end = with_single_access_budget(Duration::from_millis(100), || { + with_budget(Some(budget.clone()), || { + run_walk_with_limiter(Path::new("slow-walk"), requested, limiter, |_, control| { + let started = Instant::now(); + while started.elapsed() < Duration::from_millis(300) { + if control.advance() { + return None; + } + std::thread::sleep(Duration::from_millis(10)); + } + Some(control.deadline()) + }) + }) + }); + let Ok(WalkEnd::Finished(Some(deadline))) = end else { + panic!("a walk that kept making progress was stopped or reported as blocked"); + }; + assert_eq!( + deadline, requested, + "the walk keeps the deadline it asked for" + ); + assert!(budget.failure.lock().unwrap().is_none()); +} + +#[test] +fn request_budget_still_bounds_a_walks_explicit_deadline() { + let limiter = Arc::new(Limiter::new(1)); + let budget = Budget::after(Duration::from_millis(400)); + let started = Instant::now(); + let end = with_budget(Some(budget.clone()), || { + run_walk_with_limiter( + Path::new("long-walk"), + Instant::now() + Duration::from_secs(30), + limiter, + |_, control| { + let mut entries = 0usize; + while !control.advance() { + entries += 1; + std::thread::sleep(Duration::from_millis(5)); + } + (entries, control.deadline()) + }, + ) + }); + let Ok(WalkEnd::Finished((entries, deadline))) = end else { + panic!("the walk must stop itself at its deadline and return"); + }; + assert!(entries > 0); + assert!( + deadline < budget.deadline, + "the walk's deadline leaves reply time inside the request budget" + ); + assert!(started.elapsed() < Duration::from_secs(2)); + assert!(budget.failure.lock().unwrap().is_none()); +} + +#[test] +fn stalled_walk_names_the_stalled_path_and_its_stop_flag_frees_the_slot() { + let limiter = Arc::new(Limiter::new(4)); + let (blocked_tx, blocked_rx) = channel(); + let (release_tx, release_rx) = channel::<()>(); + let budget = Budget::after(Duration::from_secs(30)); + let walker_limiter = Arc::clone(&limiter); + let end = with_single_access_budget(Duration::from_millis(150), || { + with_budget(Some(budget.clone()), || { + run_walk_with_limiter( + Path::new("walk-root"), + Instant::now() + Duration::from_secs(30), + walker_limiter, + move |_, control| { + control.advance(); + control.reached(Path::new("walk-root/stuck")); + blocked_tx.send(()).unwrap(); + let _ = release_rx.recv_timeout(Duration::from_secs(10)); + // Once the stuck call returns, the walk would go on to its + // own 30 s deadline unless the request told it to stop. + while !control.advance() { + std::thread::sleep(Duration::from_millis(5)); + } + }, + ) + }) + }); + blocked_rx.recv_timeout(Duration::from_secs(2)).unwrap(); + let Ok(WalkEnd::Stopped { + stall: Some(stall), .. + }) = end + else { + panic!("a walk stuck in one call must be reported as stalled"); + }; + assert_eq!(stall.path(), Path::new("walk-root/stuck")); + assert!( + budget.failure.lock().unwrap().is_none(), + "the caller decides how to record the stall" + ); + assert_eq!( + *limiter.live.lock().unwrap(), + 1, + "the stuck walk holds its slot" + ); + release_tx.send(()).unwrap(); + let released = Instant::now(); + while *limiter.live.lock().unwrap() != 0 { + assert!( + released.elapsed() < Duration::from_secs(2), + "the abandoned walk kept its helper slot after the request ended" + ); + std::thread::sleep(Duration::from_millis(10)); + } +} + +#[test] +fn stall_record_keeps_partial_results_or_fails_the_reply() { + let stall = WalkStall { + path: PathBuf::from("root/stuck"), + waited: Duration::from_secs(5), + }; + let budget = Budget::after(Duration::from_secs(5)); + let response = with_budget(Some(budget.clone()), || { + stall.record(true); + assert!( + prior_failure().is_none(), + "later calls in the request still run" + ); + // The router finishes the reply twice; the stall is named once. + let once = finish_response(Response::success( + "walk", + serde_json::json!({"text": "1 file matching *.rs", "complete": true}), + )); + finish_response(once) + }); + let expected = format!( + "read_blocked: {}: {}", + Path::new("root/stuck").display(), + blocked_reason(Duration::from_secs(5)) + ); + assert!(response.success, "{response:?}"); + assert_eq!(response.data["complete"], false); + assert_eq!( + response.data["text"], + format!("1 file matching *.rs\n\n{expected}") + ); + assert_eq!( + response.data["gaps"], + serde_json::json!([{"kind": "read_blocked", "path": "root/stuck", "reason": expected}]) + ); + + let budget = Budget::after(Duration::from_secs(5)); + let response = with_budget(Some(budget), || { + stall.record(false); + finish_response(Response::success("walk", serde_json::json!({"text": ""}))) + }); + assert!(!response.success); + assert_eq!(response.data["code"], "read_blocked"); + assert_eq!(response.data["message"], expected); +} diff --git a/crates/aft/src/commands/glob.rs b/crates/aft/src/commands/glob.rs index edfa3d3a8..c9593a8d1 100644 --- a/crates/aft/src/commands/glob.rs +++ b/crates/aft/src/commands/glob.rs @@ -18,6 +18,7 @@ use super::multi_path::{canonical_key, resolve_path_or_multi, SearchPathResoluti struct GlobDiscovery { files: Vec, indexed_mtimes: HashMap, + walked_mtimes: HashMap>, walk_truncated: bool, source: &'static str, entries_visited: usize, @@ -104,6 +105,7 @@ fn handle_glob_inner(req: &RawRequest, ctx: &AppContext) -> Response { let total_started = Instant::now(); let mut parent_gaps = Vec::new(); let mut indexed_mtimes = HashMap::new(); + let mut walked_mtimes = HashMap::new(); let ignored_targets = search_roots .iter() .map(|root| { @@ -160,6 +162,7 @@ fn handle_glob_inner(req: &RawRequest, ctx: &AppContext) -> Response { .map(|_| req.session()), ); indexed_mtimes = discovery.indexed_mtimes; + walked_mtimes = discovery.walked_mtimes; ( discovery.files, discovery.walk_truncated, @@ -201,6 +204,12 @@ fn handle_glob_inner(req: &RawRequest, ctx: &AppContext) -> Response { "mixed/fallback" }; for discovery in &discoveries { + walked_mtimes.extend( + discovery + .walked_mtimes + .iter() + .map(|(path, time)| (path.clone(), *time)), + ); indexed_mtimes.extend( discovery .indexed_mtimes @@ -208,7 +217,10 @@ fn handle_glob_inner(req: &RawRequest, ctx: &AppContext) -> Response { .map(|(path, time)| (path.clone(), *time)), ); } - let files = merge_glob_files(discoveries.into_iter().flat_map(|d| d.files).collect()); + let files = merge_glob_files( + discoveries.into_iter().flat_map(|d| d.files).collect(), + &walked_mtimes, + ); ( files, walk_truncated, @@ -230,16 +242,18 @@ fn handle_glob_inner(req: &RawRequest, ctx: &AppContext) -> Response { ); // Glob's public contract is newest-first. Sort before truncating so the cap // keeps the most recently modified matches instead of the lexically first. - sort_paths_by_cached_mtime_desc(&mut files, &project_root, &indexed_mtimes); + sort_paths_by_cached_mtime_desc(&mut files, &project_root, &indexed_mtimes, &walked_mtimes); // Backstop against a stale index (for example a borrowed snapshot that // still lists another checkout's files): only paths present on disk in this // checkout are returned. Only the page is checked, and the page is refilled - // from the following matches, so present files keep their order. - let presence = crate::search_index::take_present_on_disk( + // from the following matches, so present files keep their order. Walked + // paths were just observed; checking them again could stall after discovery. + let presence = crate::search_index::take_present_on_disk_with_presence( files, DEFAULT_MAX_RESULTS, |path| path.as_path(), "glob", + |path| !walked_mtimes.contains_key(path) && crate::search_index::path_missing_on_disk(path), ); let missing_on_disk = presence.dropped; let presence_budget_exhausted = presence.budget_exhausted; @@ -386,6 +400,7 @@ fn glob_root( scope_has_files: !files.is_empty(), files, indexed_mtimes: HashMap::new(), + walked_mtimes: HashMap::new(), walk_truncated: false, source: "task-output", entries_visited, @@ -426,6 +441,7 @@ fn glob_root( entries_visited, files, indexed_mtimes, + walked_mtimes: HashMap::new(), walk_truncated: false, source: "index", walk_time: Duration::ZERO, @@ -464,6 +480,7 @@ fn glob_root( return GlobDiscovery { files: outcome.files, indexed_mtimes: HashMap::new(), + walked_mtimes: outcome.walked_mtimes, walk_truncated: outcome.walk_truncated, source: "fallback", entries_visited: outcome.entries_visited, @@ -503,11 +520,19 @@ pub(crate) fn glob_root_files_for_test( (discovery.files, discovery.source, discovery.entries_visited) } -fn merge_glob_files(files: Vec) -> Vec { +fn merge_glob_files( + files: Vec, + walked_mtimes: &HashMap>, +) -> Vec { let mut seen = HashSet::new(); let mut deduped = Vec::new(); for file in files { - if seen.insert(canonical_key(&file)) { + let key = if walked_mtimes.contains_key(&file) { + crate::inspect::job::normalize_path(&file) + } else { + crate::inspect::job::normalize_path(&canonical_key(&file)) + }; + if seen.insert(key) { deduped.push(file); } } @@ -538,6 +563,7 @@ fn fallback_glob( GlobDiscovery { files: outcome.files, indexed_mtimes: HashMap::new(), + walked_mtimes: outcome.walked_mtimes, walk_truncated: outcome.walk_truncated, source: "fallback", entries_visited: outcome.entries_visited, @@ -857,4 +883,230 @@ mod tests { "1 file matching **/*.rs\n\nsrc/a.rs\n\n(Results are truncated: showing first 100 results. Consider using a more specific path or pattern.)" ); } + + /// A project and, outside it, a tree laid out like a crate registry, so a + /// glob of the tree takes the filesystem walk rather than the index. + fn registry_fixture(crates: usize) -> (tempfile::TempDir, AppContext, PathBuf) { + let dir = tempfile::tempdir().unwrap(); + let base = std::fs::canonicalize(dir.path()).unwrap(); + let project = base.join("project"); + let registry = base.join("registry"); + std::fs::create_dir_all(&project).unwrap(); + for index in 0..crates { + let source = registry.join(format!("crate-{index:03}/src")); + std::fs::create_dir_all(&source).unwrap(); + std::fs::write(source.join("lib.rs"), "pub fn lib() {}\n").unwrap(); + std::fs::write(source.parent().unwrap().join("Cargo.toml"), "[package]\n").unwrap(); + } + let ctx = AppContext::new( + crate::context::default_language_provider_factory(), + crate::config::Config { + project_root: Some(project), + ..Default::default() + }, + ); + (dir, ctx, registry) + } + + fn glob_request(path: &Path, pattern: &str) -> RawRequest { + serde_json::from_value(serde_json::json!({ + "id": "registry-glob", "command": "glob", "path": path, "pattern": pattern + })) + .unwrap() + } + + #[test] + fn slow_large_walk_returns_an_honest_partial_not_read_blocked() { + let (_dir, ctx, registry) = registry_fixture(120); + // Each directory the walk enters takes 15 ms, so the whole tree needs + // about 3.6 s: longer than this request's budget allows. + let _slow = crate::bounded_io::slow_walk(®istry, Duration::from_millis(15)); + let request = glob_request(®istry, "**/lib.rs"); + let budget = crate::bounded_io::Budget::after(Duration::from_millis(1500)); + let started = Instant::now(); + let response = crate::bounded_io::with_budget(Some(budget), || handle_glob(&request, &ctx)); + let elapsed = started.elapsed(); + let body = &response.data; + assert!(response.success, "{body}"); + assert!(elapsed < Duration::from_secs(3), "{elapsed:?}"); + assert_eq!(body["walk_truncated"], true, "{body}"); + assert_eq!(body["complete"], false, "{body}"); + let found = body["files"].as_array().unwrap().len(); + assert!(found > 0 && found < 120, "a partial list: {body}"); + let envelope = &body["files_list_envelope"]; + assert!( + envelope.to_string().contains("\"walk\""), + "the list says the walk stopped it: {envelope}" + ); + let text = body.to_string(); + assert!(!text.contains("read_blocked"), "{text}"); + assert!(!text.contains("Privacy"), "{text}"); + } + + #[test] + fn walk_stall_after_matches_keeps_them_and_names_the_stalled_directory() { + let (_dir, ctx, registry) = registry_fixture(6); + // Each crate is two directories (the crate and its src). The fifth + // directory the walk enters, the third crate, never answers; the two + // crates walked before it hold one match each. + let mut stall = crate::bounded_io::stall_walk_at(®istry, 5); + let request = glob_request(®istry, "**/lib.rs"); + let response = + crate::bounded_io::with_single_access_budget(Duration::from_millis(200), || { + crate::bounded_io::with_budget( + Some(crate::bounded_io::Budget::after(Duration::from_secs(20))), + || handle_glob(&request, &ctx), + ) + }); + let stalled = stall.blocked.recv_timeout(Duration::from_secs(2)).unwrap(); + stall.release(); + let body = &response.data; + assert!(response.success, "{body}"); + assert_eq!(body["complete"], false, "{body}"); + assert_eq!(body["walk_truncated"], true, "{body}"); + assert_eq!(body["files"].as_array().unwrap().len(), 2, "{body}"); + assert!( + body["files_list_envelope"].to_string().contains("\"walk\""), + "{body}" + ); + let gaps = body["gaps"].as_array().unwrap(); + assert_eq!(gaps.len(), 1, "{body}"); + assert_eq!(gaps[0]["kind"], "read_blocked"); + assert_eq!(gaps[0]["path"], serde_json::json!(stalled)); + let line = gaps[0]["reason"].as_str().unwrap(); + assert!( + line.starts_with(&format!( + "read_blocked: {}: OS access did not finish within ", + stalled.display() + )), + "{line}" + ); + assert!( + body["text"].as_str().unwrap().ends_with(line), + "the stall is the text's last line: {body}" + ); + assert_eq!( + line.contains("Privacy & Security"), + cfg!(target_os = "macos"), + "{line}" + ); + } + + #[test] + fn fallback_glob_does_not_reopen_discovered_paths_after_a_stall() { + let (_dir, ctx, registry) = registry_fixture(1); + let file = registry.join("crate-000/src/lib.rs"); + let mut stall = crate::bounded_io::stall_walk_entry(&file); + let request = glob_request(&file, "*.rs"); + crate::search_index::audit_work_reset(); + let started = Instant::now(); + let response = + crate::bounded_io::with_single_access_budget(Duration::from_millis(100), || { + crate::bounded_io::with_budget( + Some(crate::bounded_io::Budget::after(Duration::from_secs(2))), + || handle_glob(&request, &ctx), + ) + }); + let elapsed = started.elapsed(); + stall.blocked.recv_timeout(Duration::from_secs(1)).unwrap(); + stall.release(); + let body = &response.data; + assert!(elapsed < Duration::from_secs(1), "{elapsed:?}"); + assert!(response.success, "{body}"); + assert_eq!(body["files"], serde_json::json!([file]), "{body}"); + assert_eq!(body["walk_truncated"], true, "{body}"); + assert_eq!(body["complete"], false, "{body}"); + let work = crate::search_index::audit_work(); + assert_eq!(work.sort_stats, 0, "{work:?}"); + assert_eq!(work.sort_canonicalizes, 0, "{work:?}"); + assert_eq!(work.presence_checks, 0, "{work:?}"); + } + + #[test] + fn walk_stall_before_any_match_is_the_read_blocked_error() { + let (_dir, ctx, registry) = registry_fixture(3); + let mut stall = crate::bounded_io::stall_walk_at(®istry, 1); + let request = glob_request(®istry, "**/lib.rs"); + let response = + crate::bounded_io::with_single_access_budget(Duration::from_millis(200), || { + crate::bounded_io::with_budget( + Some(crate::bounded_io::Budget::after(Duration::from_secs(20))), + || handle_glob(&request, &ctx), + ) + }); + let stalled = stall.blocked.recv_timeout(Duration::from_secs(2)).unwrap(); + stall.release(); + assert!(!response.success, "{:?}", response.data); + assert_eq!(response.data["code"], "read_blocked"); + let message = response.data["message"].as_str().unwrap(); + assert!( + message.starts_with(&format!( + "read_blocked: {}: OS access did not finish within ", + stalled.display() + )), + "{message}" + ); + } + + #[test] + fn many_filtered_entries_in_one_directory_are_progress_not_a_stall() { + let (_dir, ctx, registry) = registry_fixture(1); + let junk = registry.join("junk"); + std::fs::create_dir(&junk).unwrap(); + // macOS resource-fork companions: the walker's entry filter drops + // every one, so none of them reaches the walk's own loop. + for index in 0..40 { + std::fs::write(junk.join(format!("._resource_{index:02}")), "").unwrap(); + } + // 40 entries at 15 ms each is 0.6 s inside one directory, three times + // the stall window, while every check returns. + let _slow = crate::bounded_io::slow_walk_entries(&junk, Duration::from_millis(15)); + let request = glob_request(®istry, "**/lib.rs"); + let response = + crate::bounded_io::with_single_access_budget(Duration::from_millis(200), || { + crate::bounded_io::with_budget( + Some(crate::bounded_io::Budget::after(Duration::from_secs(20))), + || handle_glob(&request, &ctx), + ) + }); + let body = &response.data; + assert!(response.success, "{body}"); + assert_ne!(body["complete"], false, "{body}"); + assert_eq!(body["files"].as_array().map(Vec::len), Some(1), "{body}"); + assert!(!body.to_string().contains("read_blocked"), "{body}"); + } + + #[test] + fn stalled_single_stat_is_read_blocked_with_hint_naming_its_path() { + let (_dir, ctx, registry) = registry_fixture(1); + let target = registry.join("crate-000"); + let block = crate::bounded_io::block_stat(&target); + let request = glob_request(&target, "**/lib.rs"); + let response = + crate::bounded_io::with_single_access_budget(Duration::from_millis(200), || { + crate::bounded_io::with_budget( + Some(crate::bounded_io::Budget::after(Duration::from_secs(20))), + || handle_glob(&request, &ctx), + ) + }); + block + .reached + .recv_timeout(Duration::from_secs(2)) + .expect("the stat must have started"); + drop(block); + assert!(!response.success, "{:?}", response.data); + assert_eq!(response.data["code"], "read_blocked"); + let message = response.data["message"].as_str().unwrap(); + let hint = if cfg!(target_os = "macos") { + "; on macOS, check ck-aft access in Privacy & Security, then retry" + } else { + "" + }; + let prefix = format!( + "read_blocked: {}: OS access did not finish within ", + target.display() + ); + assert!(message.starts_with(&prefix), "{message}"); + assert!(message.ends_with(&format!(" s{hint}")), "{message}"); + } } diff --git a/crates/aft/src/commands/outline.rs b/crates/aft/src/commands/outline.rs index 38551e7be..7d7da1550 100644 --- a/crates/aft/src/commands/outline.rs +++ b/crates/aft/src/commands/outline.rs @@ -2354,6 +2354,30 @@ fn discover_outline_files(directory: &Path) -> OutlineFileDiscovery { discovery } +/// How long outline's directory discovery may walk; parsing the files it finds +/// still has to fit in the request after it. +const OUTLINE_WALK_BUDGET: std::time::Duration = std::time::Duration::from_secs(5); + +/// Paths a discovery walk has found so far. The walk adds to it as it goes, so +/// the request thread still has them if it stops waiting for the walk. +#[derive(Default)] +struct FoundPaths(std::sync::Mutex>); + +impl FoundPaths { + fn len(&self) -> usize { + self.0.lock().unwrap_or_else(|e| e.into_inner()).len() + } + fn is_empty(&self) -> bool { + self.len() == 0 + } + fn push(&self, path: String) { + self.0.lock().unwrap_or_else(|e| e.into_inner()).push(path); + } + fn take(&self) -> Vec { + std::mem::take(&mut *self.0.lock().unwrap_or_else(|e| e.into_inner())) + } +} + fn discover_outline_files_with_options( directory: &Path, breadth_first: bool, @@ -2372,18 +2396,57 @@ fn discover_outline_files_with_options( ignored_entries: 0, }; } - crate::bounded_io::run(directory, None, move |directory| { - Ok(discover_outline_files_unbounded(&directory, breadth_first)) - }) - .unwrap_or_else(|_| OutlineFileDiscovery { - entries_examined: 0, - files: Vec::new(), - directories: Vec::new(), - walk_truncated: true, - collection_truncated: true, - skipped_foreign_mounts: 0, - ignored_entries: 0, - }) + let files = Arc::new(FoundPaths::default()); + let directories = Arc::new(FoundPaths::default()); + let walk_files = Arc::clone(&files); + let walk_directories = Arc::clone(&directories); + let walked = crate::bounded_io::run_walk( + directory, + std::time::Instant::now() + OUTLINE_WALK_BUDGET, + move |directory, control| { + discover_outline_files_unbounded( + &directory, + breadth_first, + control, + &walk_files, + &walk_directories, + ) + }, + ); + let mut discovery = match walked { + Ok(crate::bounded_io::WalkEnd::Finished(discovery)) => discovery, + Ok(crate::bounded_io::WalkEnd::Stopped { + stall, + entries_visited, + }) => { + if let Some(stall) = stall { + stall.record(!files.is_empty()); + } + OutlineFileDiscovery { + entries_examined: entries_visited, + files: Vec::new(), + directories: Vec::new(), + walk_truncated: true, + collection_truncated: false, + skipped_foreign_mounts: 0, + ignored_entries: 0, + } + } + Err(_) => OutlineFileDiscovery { + entries_examined: 0, + files: Vec::new(), + directories: Vec::new(), + walk_truncated: true, + collection_truncated: true, + skipped_foreign_mounts: 0, + ignored_entries: 0, + }, + }; + discovery.files = files.take(); + discovery.directories = directories.take(); + discovery.files.sort(); + discovery.directories.sort(); + discovery } /// The files and directories one outline walk discovers below `directory` @@ -2397,9 +2460,14 @@ pub(crate) fn discover_outline_paths_for_test(directory: &Path, files_mode: bool paths } -fn discover_outline_files_unbounded(directory: &Path, breadth_first: bool) -> OutlineFileDiscovery { - let mut files = Vec::new(); - let mut directories = Vec::new(); +/// Fills `files` and `directories`; the caller sorts them. +fn discover_outline_files_unbounded( + directory: &Path, + breadth_first: bool, + control: &crate::bounded_io::WalkControl, + files: &FoundPaths, + directories: &FoundPaths, +) -> OutlineFileDiscovery { let mut entries_examined = 0; let mut walk_truncated = false; let mut collection_truncated = false; @@ -2413,37 +2481,37 @@ fn discover_outline_files_unbounded(directory: &Path, breadth_first: bool) -> Ou let ignore_stack = OutlineIgnoreStack::for_target_root(directory); entries_examined = collect_outline_files_breadth_first_with_device_lookup( directory, - &mut files, - &mut directories, + files, + directories, &mut walk_truncated, &mut collection_truncated, &mut skipped_foreign_mounts, &boundary, &ignore_stack, &mut ignored_entries, + control, ); } else { collect_outline_files_with_device_lookup( directory, - &mut files, - &mut directories, + files, + directories, &mut walk_truncated, &mut collection_truncated, &mut skipped_foreign_mounts, &boundary, crate::walk_boundary::filesystem_device_id, + control, ); } } else { collection_truncated = true; } - files.sort(); - directories.sort(); OutlineFileDiscovery { entries_examined, - files, - directories, + files: Vec::new(), + directories: Vec::new(), walk_truncated, collection_truncated, skipped_foreign_mounts, @@ -2468,15 +2536,17 @@ fn outline_target_walk_builder(directory: &Path) -> ignore::WalkBuilder { builder } +#[allow(clippy::too_many_arguments)] fn collect_outline_files_with_device_lookup( directory: &Path, - files: &mut Vec, - directories: &mut Vec, + files: &FoundPaths, + directories: &FoundPaths, walk_truncated: &mut bool, collection_truncated: &mut bool, skipped_foreign_mounts: &mut usize, boundary: &crate::walk_boundary::DeviceBoundary, device_lookup: F, + control: &crate::bounded_io::WalkControl, ) where F: Fn(&Path) -> std::io::Result> + Copy + Send + Sync + 'static, { @@ -2497,6 +2567,14 @@ fn collect_outline_files_with_device_lookup( let boundary_failed = Arc::clone(&boundary_failed); let boundary = *boundary; builder.filter_entry(move |entry| { + // Entries this filter drops never reach the walk's loop below, so + // count them as progress here. Past the walk's deadline, keep the + // entry: the loop's `advance` then ends the walk without + // processing it, where dropping it would only make the walker read + // on to the next one, opening each remaining sibling directory. + if crate::bounded_io::walk_entry_must_stop(entry.path()) { + return true; + } if !crate::bash_background::registry::safe_task_entry(entry.path()) { return false; } @@ -2524,6 +2602,7 @@ fn collect_outline_files_with_device_lookup( return false; } } + crate::bounded_io::walk_reached(entry.path()); match boundary.should_descend_with(entry.path(), device_lookup) { Ok(true) => true, Ok(false) => { @@ -2539,7 +2618,9 @@ fn collect_outline_files_with_device_lookup( } for entry in builder.build() { - if files.len() >= OUTLINE_FILE_COLLECTION_CAP { + // Checked before any work on the entry: past the deadline, the filter + // above lets one entry through only so this stops the walk. + if files.len() >= OUTLINE_FILE_COLLECTION_CAP || control.advance() { *walk_truncated = true; break; } @@ -2563,39 +2644,57 @@ fn collect_outline_files_with_device_lookup( } } *skipped_foreign_mounts += skipped_mounts.load(Ordering::Relaxed); + // The filter above marks the walk cut short once its deadline passes. + if control.cut_short() { + *walk_truncated = true; + } if boundary_failed.load(Ordering::Relaxed) { *collection_truncated = true; } } +#[allow(clippy::too_many_arguments)] fn collect_outline_files_breadth_first_with_device_lookup( directory: &Path, - files: &mut Vec, - directories: &mut Vec, + files: &FoundPaths, + directories: &FoundPaths, walk_truncated: &mut bool, collection_truncated: &mut bool, skipped_foreign_mounts: &mut usize, boundary: &crate::walk_boundary::DeviceBoundary, root_ignore_stack: &OutlineIgnoreStack, ignored_entries: &mut usize, + control: &crate::bounded_io::WalkControl, ) -> usize { let root_stack = root_ignore_stack.clone(); let mut pending = VecDeque::from([(directory.to_path_buf(), root_stack)]); let mut entries_examined = 0usize; while let Some((current, ignore_stack)) = pending.pop_front() { - if files.len() >= OUTLINE_FILE_COLLECTION_CAP { + if files.len() >= OUTLINE_FILE_COLLECTION_CAP || control.advance() { *walk_truncated = true; return entries_examined; } - let Ok(entries) = std::fs::read_dir(¤t) else { + control.reached(¤t); + let Ok(reader) = std::fs::read_dir(¤t) else { *collection_truncated = true; continue; }; let remaining = ENTRY_BUDGET.saturating_sub(entries_examined); - let mut entries = entries - .take(remaining.saturating_add(1)) - .collect::>(); + let mut reader = reader.take(remaining.saturating_add(1)); + let mut entries = Vec::new(); + loop { + // Every entry read is its own call; record each one so a large + // directory that keeps answering is slow, not stalled. + if control.entry(¤t) { + *walk_truncated = true; + return entries_examined; + } + let Some(entry) = reader.next() else { + break; + }; + entries.push(entry); + } entries_examined += entries.len(); if entries.len() > remaining { entries.truncate(remaining); @@ -2615,7 +2714,14 @@ fn collect_outline_files_breadth_first_with_device_lookup( let mut child_directories = Vec::new(); let mut child_files = Vec::new(); for entry in entries { - if !crate::bash_background::registry::safe_task_entry(&entry.path()) { + let path = entry.path(); + // The task-bundle check and the type check can each touch the + // filesystem; a stall here names this entry. + if control.entry(&path) { + *walk_truncated = true; + return entries_examined; + } + if !crate::bash_background::registry::safe_task_entry(&path) { continue; } let Ok(file_type) = entry.file_type() else { @@ -2626,15 +2732,19 @@ fn collect_outline_files_breadth_first_with_device_lookup( continue; } if file_type.is_dir() { - child_directories.push(entry.path()); + child_directories.push(path); } else if file_type.is_file() { - child_files.push(entry.path()); + child_files.push(path); } } // Queue directories before counting sibling files so the traversal keeps // its breadth-first coverage when the 10,000-file limit is reached. for path in child_directories { + if control.advance() { + *walk_truncated = true; + return entries_examined; + } if should_skip_directory(&path) || ignore_stack.is_ignored(&path, true) { *ignored_entries += 1; continue; @@ -2642,6 +2752,7 @@ fn collect_outline_files_breadth_first_with_device_lookup( if !ignore_stack.admits(&path) { continue; } + control.reached(&path); match boundary.should_descend(&path) { Ok(true) => { directories.push(path.to_string_lossy().to_string()); @@ -4877,6 +4988,121 @@ impl Boxed for GenericBox { pub fn boxed(&self) {} } ); } + #[test] + fn slow_outline_discovery_returns_a_partial_not_read_blocked() { + let temp = tempfile::tempdir().expect("tempdir"); + let root = std::fs::canonicalize(temp.path()).expect("canonical root"); + for index in 0..150 { + let directory = root.join(format!("pkg_{index:03}")); + std::fs::create_dir(&directory).expect("create package"); + std::fs::write(directory.join("lib.rs"), "pub fn package() {}\n") + .expect("write package"); + } + // About 2.3 s of directory work against a 1.5 s request budget. + let _slow = crate::bounded_io::slow_walk(&root, std::time::Duration::from_millis(15)); + let ctx = AppContext::new( + Box::new(crate::parser::TreeSitterProvider::new()), + crate::config::Config { + project_root: Some(root.clone()), + ..crate::config::Config::default() + }, + ); + let request: RawRequest = serde_json::from_value(serde_json::json!({ + "id": "slow-outline", "command": "outline", "directory": root + })) + .expect("request"); + let budget = crate::bounded_io::Budget::after(std::time::Duration::from_millis(1500)); + let response = + crate::bounded_io::with_budget(Some(budget), || handle_outline(&request, &ctx)); + let body = &response.data; + assert!(response.success, "{body}"); + assert_eq!(body["walk_truncated"], true, "{body}"); + assert_eq!(body["complete"], false, "{body}"); + let text = body.to_string(); + assert!(text.contains("pkg_"), "the walked part is outlined: {text}"); + assert!(!text.contains("read_blocked"), "{text}"); + } + + fn files_mode_request(root: &Path) -> (AppContext, RawRequest) { + let ctx = AppContext::new( + Box::new(crate::parser::TreeSitterProvider::new()), + crate::config::Config { + project_root: Some(root.to_path_buf()), + ..crate::config::Config::default() + }, + ); + let request: RawRequest = serde_json::from_value(serde_json::json!({ + "id": "files-mode-walk", "command": "outline", "directory": root, "files": true + })) + .expect("request"); + (ctx, request) + } + + /// Runs `request` with a 200 ms stall window inside a 20 s budget, so only + /// a stall, never the budget, can end the walk early. + fn outline_with_short_stall_window(ctx: &AppContext, request: &RawRequest) -> Response { + crate::bounded_io::with_single_access_budget(std::time::Duration::from_millis(200), || { + crate::bounded_io::with_budget( + Some(crate::bounded_io::Budget::after( + std::time::Duration::from_secs(20), + )), + || handle_outline(request, ctx), + ) + }) + } + + #[test] + fn slow_entries_of_one_large_directory_are_progress_not_a_stall() { + let temp = tempfile::tempdir().expect("tempdir"); + let root = std::fs::canonicalize(temp.path()).expect("canonical root"); + for index in 0..40 { + std::fs::write(root.join(format!("file_{index:02}.rs")), "pub fn f() {}\n") + .expect("write file"); + } + // All in one directory: 40 entry reads and 40 type checks at 10 ms + // each is about 0.8 s without reaching a new directory, four times + // the stall window. Each call still returns, so this is not a stall. + let _slow = + crate::bounded_io::slow_walk_entries(&root, std::time::Duration::from_millis(10)); + let (ctx, request) = files_mode_request(&root); + let response = outline_with_short_stall_window(&ctx, &request); + let body = &response.data; + assert!(response.success, "{body}"); + assert_eq!(body["complete"], true, "{body}"); + assert_eq!(body["files"].as_array().map(Vec::len), Some(40), "{body}"); + assert!(!body.to_string().contains("read_blocked"), "{body}"); + } + + #[test] + fn hung_entry_check_is_read_blocked_naming_that_entry() { + let temp = tempfile::tempdir().expect("tempdir"); + let root = std::fs::canonicalize(temp.path()).expect("canonical root"); + for name in ["a.rs", "b.rs", "c.rs", "d.rs"] { + std::fs::write(root.join(name), "pub fn f() {}\n").expect("write file"); + } + let hung = root.join("c.rs"); + let mut stall = crate::bounded_io::stall_walk_entry(&hung); + let (ctx, request) = files_mode_request(&root); + let response = outline_with_short_stall_window(&ctx, &request); + let blocked = stall + .blocked + .recv_timeout(std::time::Duration::from_secs(2)) + .expect("the walk reached the hung entry"); + stall.release(); + assert_eq!(blocked, hung); + // No file was collected before the stall, so the reply is the error. + assert!(!response.success, "{:?}", response.data); + assert_eq!(response.data["code"], "read_blocked", "{:?}", response.data); + let message = response.data["message"].as_str().unwrap_or_default(); + assert!( + message.starts_with(&format!( + "read_blocked: {}: OS access did not finish within ", + hung.display() + )), + "{message}" + ); + } + #[test] fn outline_walk_skips_and_reports_injected_foreign_mount() { let temp = tempfile::tempdir().expect("tempdir"); @@ -4889,8 +5115,8 @@ impl Boxed for GenericBox { pub fn boxed(&self) {} } std::fs::write(foreign.join("skip.rs"), "pub fn skip() {}\n").expect("write foreign file"); let boundary = crate::walk_boundary::DeviceBoundary::from_device_for_test(41); - let mut files = Vec::new(); - let mut directories = Vec::new(); + let found_files = FoundPaths::default(); + let found_directories = FoundPaths::default(); let mut walk_truncated = false; let mut collection_truncated = false; let mut skipped_foreign_mounts = 0usize; @@ -4906,14 +5132,16 @@ impl Boxed for GenericBox { pub fn boxed(&self) {} } collect_outline_files_with_device_lookup( &root, - &mut files, - &mut directories, + &found_files, + &found_directories, &mut walk_truncated, &mut collection_truncated, &mut skipped_foreign_mounts, &boundary, lookup, + &crate::bounded_io::WalkControl::unbounded_for_test(&root), ); + let files = found_files.take(); assert_eq!(skipped_foreign_mounts, 1, "foreign mount is disclosed"); assert!( @@ -4936,6 +5164,79 @@ impl Boxed for GenericBox { pub fn boxed(&self) {} } ); } + /// Walks the 1,250-entry test tree and expires the walk after its first + /// 200 entry checks, and checks that the walk stops there: no more + /// entries read, its helper slot free when the reply arrives, and the + /// files it found before then kept. Returns how many files it checked in + /// time but did not keep. + fn assert_expired_outline_walk_stops(breadth_first: bool) -> usize { + use std::time::{Duration, Instant}; + let (_dir, root) = crate::bounded_io::many_entry_tree(); + // Breadth-first makes 101 entry checks in the root (51 reads, 50 type + // checks) before it keeps the root's files, so 200 always leaves it + // files to keep; either order is still far from the whole tree. + let hook = crate::bounded_io::expire_walk_after_entries(&root, 200); + let files = Arc::new(FoundPaths::default()); + let directories = Arc::new(FoundPaths::default()); + let walk_files = Arc::clone(&files); + let walk_directories = Arc::clone(&directories); + // The hook expires the walk; the 60 s deadline and 10 s stall window + // only guard against a hang, so a loaded runner cannot move the + // point where the walk expires or turn a slow but correct stop into + // an abandoned walk. A walk that went on reading after it expired + // still fails: the count below catches every further entry check. + let (end, slot_free) = + crate::bounded_io::with_single_access_budget(Duration::from_secs(10), || { + crate::bounded_io::run_walk_in_own_slot( + &root, + Instant::now() + Duration::from_secs(60), + move |root, control| { + discover_outline_files_unbounded( + &root, + breadth_first, + control, + &walk_files, + &walk_directories, + ) + }, + ) + }); + let after_expiry = hook.calls_after_expiry(); + assert!( + slot_free, + "the expired walk still held its helper slot after the reply" + ); + // At most the one entry check that saw the expiry; nothing after it. + assert!( + after_expiry <= 1, + "the walk checked {after_expiry} entries after it expired" + ); + let Ok(crate::bounded_io::WalkEnd::Finished(discovery)) = end else { + panic!("the expired walk must end on its own, not be abandoned"); + }; + assert!(discovery.walk_truncated, "an expired walk is truncated"); + let found = files + .take() + .into_iter() + .map(PathBuf::from) + .collect::>(); + hook.files_missed_before_expiry(&found) + } + + #[test] + fn expired_depth_first_outline_walk_stops_reading_entries_and_frees_its_slot() { + let missed = assert_expired_outline_walk_stops(false); + assert!(missed <= 1, "{missed} files checked in time were dropped"); + } + + #[test] + fn expired_breadth_first_outline_walk_stops_reading_entries_and_frees_its_slot() { + // Breadth-first keeps a directory's files only once all its entries + // are checked, so files of the directory the expiry interrupted + // may be missing; the kept files are still only ones checked in time. + assert_expired_outline_walk_stops(true); + } + fn make_symbol( name: &str, kind: SymbolKind, diff --git a/crates/aft/src/grep_executor.rs b/crates/aft/src/grep_executor.rs index 55d9a0eb8..08b72e23b 100644 --- a/crates/aft/src/grep_executor.rs +++ b/crates/aft/src/grep_executor.rs @@ -1,13 +1,14 @@ -use std::collections::HashSet; +use std::collections::{HashMap, HashSet}; use std::env; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; -use std::sync::Arc; -use std::time::{Duration, Instant}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant, SystemTime}; use ignore::WalkBuilder; use rayon::prelude::*; +use crate::bounded_io::{WalkControl, WalkEnd}; use crate::commands::multi_path::{ canonical_key, dedupe_nested_paths, resolve_path_or_multi, SearchPathResolution, }; @@ -57,12 +58,20 @@ pub(crate) const FALLBACK_WALK_BUDGET: Duration = Duration::from_secs(10); #[derive(Clone, Debug)] pub struct FallbackWalkOutcome { pub files: Vec, + pub(crate) walked_mtimes: HashMap>, pub walk_truncated: bool, /// Foreign filesystem mounts skipped before a recursive fallback could open them. pub skipped_foreign_mounts: usize, pub entries_visited: usize, } +#[derive(Default)] +struct FallbackWalkSnapshot { + files: Vec, + mtimes: HashMap>, + resolved_filter_root: PathBuf, +} + /// Counts are updated before include/exclude filtering. Empty-scope probes count /// rejected directories once, without enumerating their contents. #[derive(Clone, Copy, Debug, Default)] @@ -398,16 +407,29 @@ fn bounded_scope_has_files( ) -> Option { let filters = filters.clone(); let filter_root = filter_root.to_path_buf(); - crate::bounded_io::run(search_root, None, move |search_root| { - Ok(scope_has_files_unbounded( - &filter_root, - &search_root, - &filters, - ignored_target, - )) - }) - .ok() - .flatten() + let walked = crate::bounded_io::run_walk( + search_root, + Instant::now() + FALLBACK_WALK_BUDGET, + move |search_root, control| { + scope_has_files_unbounded( + &filter_root, + &search_root, + &filters, + ignored_target, + control, + ) + }, + ); + match walked { + Ok(WalkEnd::Finished(answer)) => answer, + Ok(WalkEnd::Stopped { stall, .. }) => { + if let Some(stall) = stall { + stall.record(false); + } + None + } + Err(_) => None, + } } fn scope_has_files_unbounded( @@ -415,8 +437,8 @@ fn scope_has_files_unbounded( search_root: &Path, filters: &PathFilters, ignored_target: bool, + control: &WalkControl, ) -> Option { - let started = Instant::now(); let skipped = Arc::new(AtomicUsize::new(0)); let Some(builder) = fallback_target_walk_builder(search_root, Arc::clone(&skipped), ignored_target) @@ -426,7 +448,7 @@ fn scope_has_files_unbounded( let walker = builder.build(); for (visited, entry) in walker.enumerate() { if visited >= MAX_FALLBACK_WALK_FILES.saturating_mul(8) - || started.elapsed() >= FALLBACK_WALK_BUDGET + || control.advance() || crate::executor::current_job_cancelled() { return None; @@ -440,7 +462,7 @@ fn scope_has_files_unbounded( return Some(true); } } - (skipped.load(Ordering::Relaxed) == 0).then_some(false) + (skipped.load(Ordering::Relaxed) == 0 && !control.cut_short()).then_some(false) } pub fn execute( @@ -958,6 +980,14 @@ fn target_walk_builder( crate::context::apply_project_ignore_rules(&mut builder, search_root) .parents(!crate::context::confinement::active()) .filter_entry(move |entry| { + // Entries this filter drops never reach the walk's own loop, so + // count them as progress here. Past the walk's deadline, keep the + // entry: the loop's `advance` then ends the walk without + // processing it, where dropping it would only make the walker read + // on to the next one. + if crate::bounded_io::walk_entry_must_stop(entry.path()) { + return true; + } let file_type = entry.file_type(); if crate::project_content::is_excluded_with_file_type(entry.path(), file_type.as_ref()) { @@ -1101,16 +1131,29 @@ pub(crate) fn diagnose_scope_counts( ) -> Option { let filters = filters.clone(); let filter_root = filter_root.to_path_buf(); - crate::bounded_io::run(search_root, None, move |search_root| { - Ok(diagnose_scope_counts_unbounded( - &filter_root, - &search_root, - &filters, - ignored_target, - )) - }) - .ok() - .flatten() + let walked = crate::bounded_io::run_walk( + search_root, + Instant::now() + FALLBACK_WALK_BUDGET, + move |search_root, control| { + diagnose_scope_counts_unbounded( + &filter_root, + &search_root, + &filters, + ignored_target, + control, + ) + }, + ); + match walked { + Ok(WalkEnd::Finished(counts)) => counts, + Ok(WalkEnd::Stopped { stall, .. }) => { + if let Some(stall) = stall { + stall.record(false); + } + None + } + Err(_) => None, + } } fn diagnose_scope_counts_unbounded( @@ -1118,13 +1161,13 @@ fn diagnose_scope_counts_unbounded( search_root: &Path, filters: &PathFilters, ignored_target: bool, + control: &WalkControl, ) -> Option { - let started = Instant::now(); let skipped = Arc::new(AtomicUsize::new(0)); let mut counts = ScopeFileCounts::default(); let within_budget = |visited: usize| { visited < MAX_FALLBACK_WALK_FILES.saturating_mul(8) - && started.elapsed() < FALLBACK_WALK_BUDGET + && !control.advance() && !crate::executor::current_job_cancelled() }; // A starting path a confined walk may not use has nothing in scope; it is @@ -1164,9 +1207,10 @@ fn diagnose_scope_counts_unbounded( accepted.insert(entry.file_name().to_os_string()); } } - if skipped.load(Ordering::Relaxed) > 0 { + if skipped.load(Ordering::Relaxed) > 0 || control.cut_short() { return None; } + control.reached(&directory); for entry in std::fs::read_dir(&directory).ok()? { if !within_budget(counts.entries_visited) { return None; @@ -1255,6 +1299,7 @@ pub(crate) fn source_first_fallback_walk_files( } FallbackWalkOutcome { files, + walked_mtimes: HashMap::new(), walk_truncated, skipped_foreign_mounts: skipped_foreign_mounts.load(Ordering::Relaxed), entries_visited, @@ -1296,66 +1341,114 @@ fn bounded_fallback_walk_files_with_limits_target( budget: Duration, ignored_target: bool, ) -> FallbackWalkOutcome { - let filter_root = filter_root.to_path_buf(); + let owned_filter_root = filter_root.to_path_buf(); let filters = filters.clone(); - crate::bounded_io::run( + // The walker shares matches as it finds them, so a walk the request stops + // waiting for still returns what it found. + let found = Arc::new(Mutex::new(FallbackWalkSnapshot::default())); + let skipped_foreign_mounts = Arc::new(AtomicUsize::new(0)); + let walk_found = Arc::clone(&found); + let walk_skipped = Arc::clone(&skipped_foreign_mounts); + let walked = crate::bounded_io::run_walk( search_root, - Some(Instant::now() + budget), - move |search_root| { - Ok(walk_files_unbounded( - &filter_root, + Instant::now() + budget, + move |search_root, control| { + walk_files_unbounded( + &owned_filter_root, &search_root, &filters, max_files, - budget, ignored_target, - )) + control, + &walk_found, + walk_skipped, + ) }, - ) - .unwrap_or_else(|_| FallbackWalkOutcome { - files: Vec::new(), - walk_truncated: true, - skipped_foreign_mounts: 0, - entries_visited: 0, - }) + ); + let (walk_truncated, entries_visited) = match walked { + Ok(WalkEnd::Finished(walked)) => walked, + Ok(WalkEnd::Stopped { + stall, + entries_visited, + }) => { + if let Some(stall) = stall { + let kept = !found + .lock() + .unwrap_or_else(|e| e.into_inner()) + .files + .is_empty(); + stall.record(kept); + } + (true, entries_visited) + } + Err(_) => (true, 0), + }; + let FallbackWalkSnapshot { + mut files, + mtimes, + resolved_filter_root, + } = std::mem::take(&mut *found.lock().unwrap_or_else(|e| e.into_inner())); + // Only in-memory keys are used here: the filesystem may have stopped + // responding while the walker was collecting the last match's mtime. + sort_walked_paths_by_mtime_desc(&mut files, &resolved_filter_root, &mtimes); + FallbackWalkOutcome { + files, + walked_mtimes: mtimes, + walk_truncated, + skipped_foreign_mounts: skipped_foreign_mounts.load(Ordering::Relaxed), + entries_visited, + } +} + +/// Patterns for an explicitly named file are matched relative to its parent. +fn walk_filter_root<'a>(filter_root: &'a Path, search_root: &'a Path) -> &'a Path { + if filter_root == search_root && search_root.is_file() { + search_root.parent().unwrap_or(search_root) + } else { + filter_root + } } +/// Returns whether the walk stopped early, and how many entries it visited. +#[allow(clippy::too_many_arguments)] fn walk_files_unbounded( filter_root: &Path, search_root: &Path, filters: &PathFilters, max_files: usize, - budget: Duration, ignored_target: bool, -) -> FallbackWalkOutcome { - let skipped_foreign_mounts = Arc::new(AtomicUsize::new(0)); + control: &WalkControl, + found: &Mutex, + skipped_foreign_mounts: Arc, +) -> (bool, usize) { // Built before the file test below, which would stat the starting path. let Some(builder) = fallback_target_walk_builder( search_root, Arc::clone(&skipped_foreign_mounts), ignored_target, ) else { - return FallbackWalkOutcome { - files: Vec::new(), - walk_truncated: false, - skipped_foreign_mounts: 0, - entries_visited: 0, - }; + return (false, 0); }; - let filter_root = if filter_root == search_root && search_root.is_file() { - search_root.parent().unwrap_or(search_root) - } else { - filter_root - }; - let started = Instant::now(); - let mut files = Vec::new(); + let filter_root = walk_filter_root(filter_root, search_root); + let resolved_filter_root = crate::inspect::job::canonicalize_normalized(filter_root); + found + .lock() + .unwrap_or_else(|e| e.into_inner()) + .resolved_filter_root = resolved_filter_root; + let mut matched = 0usize; let mut walk_truncated = false; let mut entries_visited = 0usize; let walker = builder.build(); for entry in walker.filter_map(|entry| entry.ok()) { + // First, before counting or examining the entry: past the deadline, + // the walker's filter lets one entry through only so this stops. + if control.advance() { + walk_truncated = true; + break; + } entries_visited += 1; - if started.elapsed() >= budget || entries_visited > max_files.saturating_mul(8).max(1024) { + if entries_visited > max_files.saturating_mul(8).max(1024) { walk_truncated = true; break; } @@ -1367,24 +1460,33 @@ fn walk_files_unbounded( } let path = entry.into_path(); if filters.matches(filter_root, &path) { - files.push(path); - if files.len() > max_files { + if matched == max_files { walk_truncated = true; - files.truncate(max_files); break; } + matched += 1; + { + let mut found = found.lock().unwrap_or_else(|e| e.into_inner()); + found.files.push(path.clone()); + found.mtimes.insert(path.clone(), None); + } + // Publish the match before metadata, and never hold the snapshot + // lock across I/O: a stalled stat must not hide a discovered file. + let mtime = control + .metadata(&path) + .and_then(|metadata| metadata.modified()) + .ok(); + found + .lock() + .unwrap_or_else(|e| e.into_inner()) + .mtimes + .insert(path, mtime); } } - - sort_walked_paths_by_mtime_desc(&mut files, filter_root); - FallbackWalkOutcome { - files, - walk_truncated, - skipped_foreign_mounts: skipped_foreign_mounts.load(Ordering::Relaxed), - entries_visited, - } + (walk_truncated || control.cut_short(), entries_visited) } +#[allow(clippy::too_many_arguments)] fn for_each_bounded_fallback_walk_file_with_limits( filter_root: &Path, search_root: &Path, @@ -1403,33 +1505,64 @@ where let project_root = project_root.to_path_buf(); let filters = filters.clone(); let cancellation = crate::executor::current_job_cancellation(); - let walked = crate::bounded_io::run( + // Publish matches in a shared vector as they are found, so the request can + // return those partial results even if it stops waiting for the walk. + let found = Arc::new(Mutex::new(Vec::new())); + let skipped_foreign_mounts = Arc::new(AtomicUsize::new(0)); + let walk_found = Arc::clone(&found); + let walk_skipped = Arc::clone(&skipped_foreign_mounts); + let walked = crate::bounded_io::run_walk( search_root, - Some(Instant::now() + budget), - move |search_root| { - let mut files = Vec::new(); - let progress = walk_each_file_unbounded( + Instant::now() + budget, + move |search_root, control| { + walk_each_file_unbounded( &filter_root, &search_root, &filters, &project_root, path_exclusion, max_files, - budget, - &mut |path| files.push(path.clone()), + &mut |path| { + walk_found + .lock() + .unwrap_or_else(|e| e.into_inner()) + .push(path.clone()) + }, ignored_target, cancellation.as_ref(), - ); - Ok((files, progress)) + control, + walk_skipped, + ) }, ); + let files = || std::mem::take(&mut *found.lock().unwrap_or_else(|e| e.into_inner())); match walked { - Ok((files, progress)) => { - for file in files { + Ok(WalkEnd::Finished(progress)) => { + for file in files() { on_file(&file); } progress } + Ok(WalkEnd::Stopped { + stall, + entries_visited, + }) => { + let files = files(); + if let Some(stall) = stall { + stall.record(!files.is_empty()); + } + for file in &files { + on_file(file); + } + FallbackWalkProgress { + bound: Some(WalkBound::TimeBudget), + skipped_foreign_mounts: skipped_foreign_mounts.load(Ordering::Relaxed), + counts: ScopeFileCounts { + entries_visited, + ..ScopeFileCounts::default() + }, + } + } Err(_) => FallbackWalkProgress { bound: Some(WalkBound::TimeBudget), ..FallbackWalkProgress::default() @@ -1437,6 +1570,7 @@ where } } +#[allow(clippy::too_many_arguments)] fn walk_each_file_unbounded( filter_root: &Path, search_root: &Path, @@ -1444,15 +1578,14 @@ fn walk_each_file_unbounded( project_root: &Path, path_exclusion: Option, max_files: usize, - budget: Duration, on_file: &mut F, ignored_target: bool, cancellation: Option<&crate::executor::JobCancellation>, + control: &WalkControl, + skipped_foreign_mounts: Arc, ) -> FallbackWalkProgress { - let started = Instant::now(); let mut files_seen = 0usize; let mut counts = ScopeFileCounts::default(); - let skipped_foreign_mounts = Arc::new(AtomicUsize::new(0)); let Some(builder) = fallback_target_walk_builder( search_root, Arc::clone(&skipped_foreign_mounts), @@ -1463,17 +1596,19 @@ fn walk_each_file_unbounded( let walker = builder.build(); for entry in walker.filter_map(|entry| entry.ok()) { - counts.entries_visited += 1; - if cancellation.is_some_and(|token| token.cancel_requested_before_commit()) { + // First, before counting or examining the entry: past the deadline, + // the walker's filter lets one entry through only so this stops. + if control.advance() { return FallbackWalkProgress { - bound: Some(WalkBound::Cancelled), + bound: Some(WalkBound::TimeBudget), skipped_foreign_mounts: skipped_foreign_mounts.load(Ordering::Relaxed), counts, }; } - if started.elapsed() >= budget { + counts.entries_visited += 1; + if cancellation.is_some_and(|token| token.cancel_requested_before_commit()) { return FallbackWalkProgress { - bound: Some(WalkBound::TimeBudget), + bound: Some(WalkBound::Cancelled), skipped_foreign_mounts: skipped_foreign_mounts.load(Ordering::Relaxed), counts, }; @@ -1506,7 +1641,8 @@ fn walk_each_file_unbounded( } } FallbackWalkProgress { - bound: None, + // A directory refused at the walk's deadline was never searched. + bound: control.cut_short().then_some(WalkBound::TimeBudget), skipped_foreign_mounts: skipped_foreign_mounts.load(Ordering::Relaxed), counts, } @@ -2493,6 +2629,51 @@ mod tests { assert!(text.contains("ran out of its time budget"), "{text}"); } + #[test] + fn slow_fallback_grep_walk_returns_a_partial_not_read_blocked() { + let dir = tempfile::tempdir().unwrap(); + let root = std::fs::canonicalize(dir.path()).unwrap(); + for index in 0..120 { + let directory = root.join(format!("directory_{index:03}")); + std::fs::create_dir(&directory).unwrap(); + std::fs::write(directory.join("file.rs"), "pub fn needle() {}\n").unwrap(); + } + // About 1.8 s of directory work against a 1.5 s request budget. + let _slow = crate::bounded_io::slow_walk(&root, Duration::from_millis(15)); + let budget = crate::bounded_io::Budget::after(Duration::from_millis(1500)); + let (result, finished) = crate::bounded_io::with_budget(Some(budget), || { + let result = fallback_grep_with_limits( + &root, + &root, + &root, + &literal("needle"), + &PathFilters::default(), + 1000, + IndexStatus::Fallback, + None, + MAX_FALLBACK_WALK_FILES, + FALLBACK_WALK_BUDGET, + ); + let finished = crate::bounded_io::finish_response(Response::success( + "grep", + serde_json::json!({"text": ""}), + )); + (result, finished) + }); + assert_eq!(result.walk_bound, Some(WalkBound::TimeBudget)); + assert!(result.walk_truncated); + assert!( + result.files_searched > 0 && result.files_searched < 120, + "the files walked before the deadline are still searched: {}", + result.files_searched + ); + assert!( + finished.success, + "a slow walk is not a request failure: {:?}", + finished.data + ); + } + #[test] fn single_root_uses_requested_max() { let scope = GrepScope { @@ -2565,6 +2746,68 @@ mod tests { assert_eq!(outcome.files.len(), 2); } + #[test] + fn fallback_walk_keeps_discovered_file_when_its_mtime_stalls() { + let dir = tempfile::tempdir().unwrap(); + let file = std::fs::canonicalize(dir.path()).unwrap().join("found.rs"); + std::fs::write(&file, "pub fn found() {}\n").unwrap(); + let blocked = crate::bounded_io::block_stat(&file); + crate::search_index::audit_work_reset(); + let started = Instant::now(); + let (outcome, response) = + crate::bounded_io::with_single_access_budget(Duration::from_millis(100), || { + crate::bounded_io::with_budget( + Some(crate::bounded_io::Budget::after(Duration::from_secs(2))), + || { + let outcome = + bounded_fallback_walk_files(&file, &file, &PathFilters::default()); + let response = crate::bounded_io::finish_response(Response::success( + "walk", + serde_json::json!({"files": outcome.files}), + )); + (outcome, response) + }, + ) + }); + let elapsed = started.elapsed(); + blocked + .reached + .recv_timeout(Duration::from_secs(1)) + .unwrap(); + drop(blocked); + assert!(elapsed < Duration::from_secs(1), "{elapsed:?}"); + assert_eq!(outcome.files, vec![file]); + assert!(outcome.walk_truncated, "the mtime lookup did not finish"); + assert!(response.success, "{:?}", response.data); + assert_eq!(crate::search_index::audit_work().sort_stats, 0); + } + + #[test] + fn complete_fallback_walk_orders_by_mtime_then_display_path_without_restatting() { + let dir = tempfile::tempdir().unwrap(); + let root = std::fs::canonicalize(dir.path()).unwrap(); + for (name, seconds) in [ + ("a-old.rs", 1_700_000_000), + ("z-new.rs", 1_700_000_002), + ("z-tie.rs", 1_700_000_001), + ("nested/a-tie.rs", 1_700_000_001), + ] { + let path = root.join(name); + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + std::fs::write(&path, "pub fn f() {}\n").unwrap(); + filetime::set_file_mtime(&path, filetime::FileTime::from_unix_time(seconds, 0)) + .unwrap(); + } + crate::search_index::audit_work_reset(); + let outcome = bounded_fallback_walk_files(&root, &root, &PathFilters::default()); + assert!(!outcome.walk_truncated); + assert_eq!( + outcome.files, + ["z-new.rs", "nested/a-tie.rs", "z-tie.rs", "a-old.rs"].map(|name| root.join(name)) + ); + assert_eq!(crate::search_index::audit_work().sort_stats, 0); + } + #[test] fn fallback_project_walk_prunes_large_ignored_trees() { // build is also a fixed directory skip; vendor makes sure Git's own @@ -3232,4 +3475,113 @@ mod tests { "median: per-root={per_root_median}ns shared={shared_query_median}ns speedup={speedup:.2}x" ); } + + /// How many entry checks the expiry tests below let a walk make before + /// the hook expires it: enough to keep files, far short of the whole tree. + const EXPIRE_AFTER_ENTRY_CHECKS: usize = 200; + + /// The walk deadline for the expiry tests below. The hook expires the + /// walk long before it; this only bounds a walk that never stops. + const HANG_GUARD_DEADLINE: Duration = Duration::from_secs(60); + + #[test] + fn expired_fallback_walk_stops_reading_entries_and_frees_its_slot() { + let (_dir, root) = crate::bounded_io::many_entry_tree(); + let hook = crate::bounded_io::expire_walk_after_entries(&root, EXPIRE_AFTER_ENTRY_CHECKS); + let found = Arc::new(Mutex::new(FallbackWalkSnapshot::default())); + let walk_found = Arc::clone(&found); + // The hook expires the walk; the deadline and the 10 s stall window + // only guard against a hang, so a loaded runner cannot move the point + // where the walk expires or turn a slow but correct stop into an + // abandoned walk. A walk that went on reading after it expired still + // fails: the count below catches every further entry check. + let (end, slot_free) = + crate::bounded_io::with_single_access_budget(Duration::from_secs(10), || { + crate::bounded_io::run_walk_in_own_slot( + &root, + Instant::now() + HANG_GUARD_DEADLINE, + move |root, control| { + walk_files_unbounded( + &root, + &root, + &PathFilters::default(), + 100_000, + false, + control, + &walk_found, + Arc::new(AtomicUsize::new(0)), + ) + }, + ) + }); + let after_expiry = hook.calls_after_expiry(); + assert!( + slot_free, + "the expired walk still held its helper slot after the reply" + ); + // At most the one entry the filter lets through so the walk's loop + // sees the expiry; nothing after it is read. + assert!( + after_expiry <= 1, + "the walk checked {after_expiry} entries after it expired" + ); + let Ok(WalkEnd::Finished((truncated, entries_visited))) = end else { + panic!("the expired walk must end on its own, not be abandoned"); + }; + assert!(truncated, "an expired walk is truncated"); + let found = found.lock().unwrap().files.clone(); + let missed = hook.files_missed_before_expiry(&found); + assert!(missed <= 1, "{missed} files checked in time were dropped"); + // The count covers the root (never filtered, so never hooked) plus + // each entry the walk took in before it expired. Exactly one + // filter-checked entry is not counted: the one the walk refused when + // it expired, checked before expiry (so in `calls_before_expiry`) + // or after it (the single call in `after_expiry`). Counting that + // refused entry would exceed this bound by one. + let checked_in_time = hook.calls_before_expiry().len(); + assert!( + entries_visited <= checked_in_time + after_expiry, + "{entries_visited} entries counted, {checked_in_time} checked in time" + ); + } + + #[test] + fn expired_scope_probe_stops_reading_entries_and_frees_its_slot() { + let dir = tempfile::tempdir().unwrap(); + let root = std::fs::canonicalize(dir.path()).unwrap(); + // Only empty directories: the probe finds no file and walks them all. + for directory in 0..1_000 { + std::fs::create_dir(root.join(format!("empty_{directory:04}"))).unwrap(); + } + let hook = crate::bounded_io::expire_walk_after_entries(&root, EXPIRE_AFTER_ENTRY_CHECKS); + let (end, slot_free) = + crate::bounded_io::with_single_access_budget(Duration::from_secs(10), || { + crate::bounded_io::run_walk_in_own_slot( + &root, + Instant::now() + HANG_GUARD_DEADLINE, + |root, control| { + scope_has_files_unbounded( + &root, + &root, + &PathFilters::default(), + false, + control, + ) + }, + ) + }); + let after_expiry = hook.calls_after_expiry(); + assert!( + slot_free, + "the expired probe still held its helper slot after the reply" + ); + assert!( + after_expiry <= 1, + "the probe checked {after_expiry} entries after it expired" + ); + let Ok(WalkEnd::Finished(answer)) = end else { + panic!("the expired probe must end on its own, not be abandoned"); + }; + assert_eq!(answer, None, "an expired probe cannot prove an empty scope"); + } } diff --git a/crates/aft/src/list_surfaces.rs b/crates/aft/src/list_surfaces.rs index 630ebdf0b..2408a3815 100644 --- a/crates/aft/src/list_surfaces.rs +++ b/crates/aft/src/list_surfaces.rs @@ -169,7 +169,7 @@ pub static LIST_SURFACES: &[SurfaceEntry] = &[ ReasonEntry { reason: Reason::Cap, kind: ReasonKind::Selecting, - predicate_name: "SearchTrailer::shared_envelope_projection, StopState::S1MoreAtDepth, StopState::S2Exhausted, more_available, handle_external_semantic_or_hybrid_search, handle_external_grep_search, handle_semantic_or_hybrid_search, handle_grep_search, from_matches, run_engine_ranking, blast_radius_annotation_for_result, enrich_snippets_from_source_reference, enrich_snippets_from_source_with_context, truncate_chars, split_semantic_results, handle_split_search, rank_hits", + predicate_name: "SearchTrailer::shared_envelope_projection, StopState::S1MoreAtDepth, StopState::S2Exhausted, more_available, handle_external_semantic_or_hybrid_search, handle_external_grep_search, handle_semantic_or_hybrid_search, handle_grep_search, from_matches, run_engine_ranking, blast_radius_annotation_for_result, enrich_snippets_from_source_reference, enrich_snippets_from_source_with_context, truncate_chars, split_semantic_results, handle_split_search, rank_hits, take_present_on_disk_with_presence", }, ], }, @@ -202,12 +202,12 @@ pub static LIST_SURFACES: &[SurfaceEntry] = &[ ReasonEntry { reason: Reason::Walk, kind: ReasonKind::Bounding, - predicate_name: "GlobDiscovery, handle_glob, handle_glob_inner, fallback_glob, glob_root, walk_truncated, skipped_foreign_mounts", + predicate_name: "GlobDiscovery, handle_glob, handle_glob_inner, fallback_glob, glob_root, fallback_glob_does_not_reopen_discovered_paths_after_a_stall, walk_truncated, skipped_foreign_mounts", }, ReasonEntry { reason: Reason::Cap, kind: ReasonKind::Selecting, - predicate_name: "DEFAULT_MAX_RESULTS, MAX_DISPLAY_DIRECTORIES, MAX_DISPLAY_FILES_PER_DIRECTORY, handle_glob, handle_glob_inner, format_glob_text, rendered_glob_file_count", + predicate_name: "DEFAULT_MAX_RESULTS, MAX_DISPLAY_DIRECTORIES, MAX_DISPLAY_FILES_PER_DIRECTORY, handle_glob, handle_glob_inner, format_glob_text, rendered_glob_file_count, take_present_on_disk_with_presence", }, ], }, @@ -226,7 +226,7 @@ pub static LIST_SURFACES: &[SurfaceEntry] = &[ ReasonEntry { reason: Reason::Walk, kind: ReasonKind::Bounding, - predicate_name: "OutlineFileDiscovery, discover_outline_files_with_options, discover_outline_files_unbounded, collect_outline_files_with_device_lookup, collect_outline_files_breadth_first_with_device_lookup, outline_walk_skips_and_reports_injected_foreign_mount, ITERATIONS, collection_truncated, walk_truncated, skipped_foreign_mounts", + predicate_name: "OutlineFileDiscovery, discover_outline_files_with_options, discover_outline_files_unbounded, collect_outline_files_with_device_lookup, collect_outline_files_breadth_first_with_device_lookup, outline_walk_skips_and_reports_injected_foreign_mount, slow_outline_discovery_returns_a_partial_not_read_blocked, assert_expired_outline_walk_stops, ITERATIONS, collection_truncated, walk_truncated, skipped_foreign_mounts", }, ], }, @@ -591,9 +591,9 @@ pub static EXCLUSIONS: &[ExclusionEntry] = &[ }, ExclusionEntry { file: "commands/glob.rs", - enclosing_item: "exact_file_glob_with_ready_index_matches_basename_without_widening_scope, exact_file_glob_fallback_matches_basename_without_widening_scope", + enclosing_item: "exact_file_glob_with_ready_index_matches_basename_without_widening_scope, exact_file_glob_fallback_matches_basename_without_widening_scope, slow_large_walk_returns_an_honest_partial_not_read_blocked, walk_stall_after_matches_keeps_them_and_names_the_stalled_directory", location_or_primitive: "test walk_truncated assertions", - reason: "test-only assertions on walk_truncated for indexed and fallback exact-file discovery; they do not cut a list or produce a tool response", + reason: "test-only assertions on walk_truncated for indexed and fallback exact-file discovery and for slow or stalled fallback walks; they do not cut a list or produce a tool response", }, // Checkpoint and restore results name their first files and then say how // many more there are; the checkpoint itself always covers every file. diff --git a/crates/aft/src/search_index.rs b/crates/aft/src/search_index.rs index 7813c21a2..9f0351b20 100644 --- a/crates/aft/src/search_index.rs +++ b/crates/aft/src/search_index.rs @@ -6047,6 +6047,17 @@ pub(crate) fn take_present_on_disk( limit: usize, path_of: impl Fn(&T) -> &Path, surface: &str, +) -> PresentOnDisk { + take_present_on_disk_with_presence(items, limit, path_of, surface, path_missing_on_disk) +} + +/// Reuse presence observed by a walk, while still checking stale indexed paths. +pub(crate) fn take_present_on_disk_with_presence( + items: impl IntoIterator, + limit: usize, + path_of: impl Fn(&T) -> &Path, + surface: &str, + missing: impl Fn(&Path) -> bool, ) -> PresentOnDisk { let stat_budget = limit.saturating_add(MISSING_ON_DISK_REFILL_BUDGET); let mut present = Vec::new(); @@ -6066,7 +6077,7 @@ pub(crate) fn take_present_on_disk( continue; } stats += 1; - if path_missing_on_disk(path_of(&item)) { + if missing(path_of(&item)) { dropped += 1; if dropped_examples.len() < 3 { dropped_examples.push(path_of(&item).display().to_string()); @@ -6244,27 +6255,52 @@ pub(crate) fn validate_cached_relative_path(path: &Path) -> Option { /// Metadata and display keys are snapshotted before sorting: the comparator must /// remain a total order even if files change or disappear during the sort. pub(crate) fn sort_paths_by_mtime_desc(paths: &mut [PathBuf], stable_root: &Path) { - sort_paths_by_mtime_desc_with_key_normalization(paths, stable_root, true, None); + sort_paths_by_mtime_desc_with_key_normalization(paths, stable_root, true, None, None); } /// Sort paths emitted by one filesystem walk without resolving every path again. /// -/// Walk entries already share one resolved root and do not traverse symlinks, so -/// lexical normalization produces the same display tie-break key. Calling -/// `canonicalize` for every match only repeats filesystem lookups the walk just -/// performed. -pub(crate) fn sort_walked_paths_by_mtime_desc(paths: &mut [PathBuf], stable_root: &Path) { - sort_paths_by_mtime_desc_with_key_normalization(paths, stable_root, false, None); +/// The walker supplies the resolved root and mtimes. Missing mtimes sort last; +/// neither they nor the root may be looked up again after the walk has stopped. +pub(crate) fn sort_walked_paths_by_mtime_desc( + paths: &mut [PathBuf], + resolved_root: &Path, + mtimes: &HashMap>, +) { + let display_paths = paths + .iter() + .map(|path| (path.clone(), walked_display_sort_key(resolved_root, path))) + .collect(); + sort_paths_by_snapshot(paths, mtimes, &display_paths); } -/// Indexed paths already have canonical keys and mtimes. Unindexed roots still -/// use the disk-backed ordering, so mixed-root glob retains the same contract. +fn walked_display_sort_key(resolved_root: &Path, path: &Path) -> String { + let resolved = if path.is_absolute() { + path.to_path_buf() + } else { + resolved_root.join(path) + }; + normalized_display_sort_key( + Some(&crate::inspect::job::normalize_path(resolved_root)), + &crate::inspect::job::normalize_path(&resolved), + ) +} + +/// Indexed paths have canonical keys; walked paths have snapshotted mtimes, +/// including failed lookups. The caller has already resolved the stable root. pub(crate) fn sort_paths_by_cached_mtime_desc( paths: &mut [PathBuf], stable_root: &Path, mtimes: &HashMap, + walked_mtimes: &HashMap>, ) { - sort_paths_by_mtime_desc_with_key_normalization(paths, stable_root, true, Some(mtimes)); + sort_paths_by_mtime_desc_with_key_normalization( + paths, + stable_root, + true, + Some(mtimes), + Some(walked_mtimes), + ); } fn sort_paths_by_mtime_desc_with_key_normalization( @@ -6272,17 +6308,26 @@ fn sort_paths_by_mtime_desc_with_key_normalization( stable_root: &Path, canonicalize_paths: bool, cached_mtimes: Option<&HashMap>, + walked_mtimes: Option<&HashMap>>, ) { use std::collections::HashMap; - let stable_root = crate::inspect::job::canonicalize_normalized(stable_root); + let stable_root = if walked_mtimes.is_some() { + crate::inspect::job::normalize_path(stable_root) + } else { + crate::inspect::job::canonicalize_normalized(stable_root) + }; let mut mtimes: HashMap> = HashMap::with_capacity(paths.len()); let mut display_paths: HashMap = HashMap::with_capacity(paths.len()); for path in paths.iter() { mtimes.entry(path.clone()).or_insert_with(|| { - cached_mtimes - .and_then(|times| times.get(path)) - .copied() - .or_else(|| path_modified_time(path)) + if let Some(time) = walked_mtimes.and_then(|times| times.get(path)) { + *time + } else { + cached_mtimes + .and_then(|times| times.get(path)) + .copied() + .or_else(|| path_modified_time(path)) + } }); display_paths.entry(path.clone()).or_insert_with(|| { let resolved = if path.is_absolute() { @@ -6292,6 +6337,7 @@ fn sort_paths_by_mtime_desc_with_key_normalization( }; let comparison_path = if canonicalize_paths && !cached_mtimes.is_some_and(|times| times.contains_key(path)) + && !walked_mtimes.is_some_and(|times| times.contains_key(path)) { #[cfg(test)] audit_record(|work| work.sort_canonicalizes += 1); @@ -6302,6 +6348,14 @@ fn sort_paths_by_mtime_desc_with_key_normalization( normalized_display_sort_key(Some(&stable_root), &comparison_path) }); } + sort_paths_by_snapshot(paths, &mtimes, &display_paths); +} + +fn sort_paths_by_snapshot( + paths: &mut [PathBuf], + mtimes: &HashMap>, + display_paths: &HashMap, +) { paths.sort_by(|left, right| { let left_mtime = mtimes.get(left).and_then(|v| *v); let right_mtime = mtimes.get(right).and_then(|v| *v); @@ -12758,9 +12812,42 @@ mod tests { let mut canonicalized = paths.clone(); sort_paths_by_mtime_desc(&mut canonicalized, dir.path()); - sort_walked_paths_by_mtime_desc(&mut paths, dir.path()); + let mtimes = paths + .iter() + .map(|path| (path.clone(), path_modified_time(path))) + .collect(); + let root = crate::inspect::job::canonicalize_normalized(dir.path()); + audit_work_reset(); + sort_walked_paths_by_mtime_desc(&mut paths, &root, &mtimes); assert_eq!(paths, canonicalized); + assert_eq!(audit_work().sort_stats, 0); + assert_eq!(audit_work().sort_canonicalizes, 0); + } + + #[test] + fn walked_path_sort_keeps_unknown_mtimes_last_without_filesystem_access() { + let dir = tempfile::tempdir().unwrap(); + let root = crate::inspect::job::canonicalize_normalized(dir.path()); + let mut paths = + ["z-unknown.rs", "old.rs", "a-unknown.rs", "new.rs"].map(|name| root.join(name)); + let mtimes = HashMap::from([ + (root.join("old.rs"), Some(SystemTime::UNIX_EPOCH)), + ( + root.join("new.rs"), + Some(SystemTime::UNIX_EPOCH + Duration::from_secs(1)), + ), + (root.join("z-unknown.rs"), None), + ]); + drop(dir); + audit_work_reset(); + sort_walked_paths_by_mtime_desc(&mut paths, &root, &mtimes); + assert_eq!( + paths, + ["new.rs", "old.rs", "a-unknown.rs", "z-unknown.rs"].map(|name| root.join(name)) + ); + assert_eq!(audit_work().sort_stats, 0); + assert_eq!(audit_work().sort_canonicalizes, 0); } /// Manual release-mode probe for the sort paid by one fallback glob walk. @@ -12786,12 +12873,17 @@ mod tests { } } + let root = crate::inspect::job::canonicalize_normalized(temp.path()); + let mtimes = paths + .iter() + .map(|path| (path.clone(), path_modified_time(path))) + .collect(); let mut micros_per_operation = Vec::with_capacity(SAMPLES); for _ in 0..SAMPLES { let started = Instant::now(); for _ in 0..ITERATIONS { let mut sample = paths.clone(); - sort_walked_paths_by_mtime_desc(&mut sample, temp.path()); + sort_walked_paths_by_mtime_desc(&mut sample, &root, &mtimes); std::hint::black_box(sample); } micros_per_operation.push(started.elapsed().as_micros() / ITERATIONS as u128);