codex_file_system/
find_up.rs

1use crate::ExecutorFileSystem;
2use crate::FileSystemResult;
3use crate::FileSystemSandboxContext;
4use codex_utils_absolute_path::AbsolutePathBuf;
5use codex_utils_path_uri::PathUri;
6use futures::StreamExt;
7use std::io;
8
9// Keep enough ordinary metadata calls in flight to cover typical ancestor chains in one remote
10// round trip, while leaving room for independent startup discovery to run at the same time.
11const MAX_CONCURRENT_PROBES: usize = 256;
12
13/// Controls how an upward marker search handles metadata errors other than `NotFound`.
14#[derive(Clone, Copy, Debug, Eq, PartialEq)]
15pub enum FindUpErrorPolicy {
16    /// Return the first error in lexical search order.
17    Propagate,
18    /// Treat errors as missing markers and continue searching.
19    Ignore,
20}
21
22/// Finds the nearest ancestor containing one of the provided marker names.
23///
24/// Marker paths are probed in lexical order from `start` toward the filesystem root. A bounded
25/// number of ordinary metadata calls are kept in flight so remote filesystems can pipeline them
26/// without requiring a batch protocol operation.
27pub async fn find_nearest_ancestor_with_markers(
28    file_system: &dyn ExecutorFileSystem,
29    start: &PathUri,
30    markers: Vec<String>,
31    error_policy: FindUpErrorPolicy,
32    sandbox: Option<&FileSystemSandboxContext>,
33) -> FileSystemResult<Option<PathUri>> {
34    find_nearest_ancestor(
35        file_system,
36        start.clone(),
37        markers,
38        PathUri::parent,
39        |ancestor, marker| {
40            ancestor
41                .join(marker)
42                .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))
43        },
44        error_policy,
45        sandbox,
46    )
47    .await
48}
49
50/// Finds the nearest native ancestor containing one of the provided marker names.
51///
52/// Ancestors and marker paths remain native until each complete probe is converted to a URI. This
53/// preserves paths that require an opaque [`PathUri`] fallback.
54pub async fn find_nearest_native_ancestor_with_markers(
55    file_system: &dyn ExecutorFileSystem,
56    start: &AbsolutePathBuf,
57    markers: Vec<String>,
58    error_policy: FindUpErrorPolicy,
59    sandbox: Option<&FileSystemSandboxContext>,
60) -> FileSystemResult<Option<AbsolutePathBuf>> {
61    find_nearest_ancestor(
62        file_system,
63        start.clone(),
64        markers,
65        AbsolutePathBuf::parent,
66        |ancestor, marker| Ok(PathUri::from_abs_path(&ancestor.join(marker))),
67        error_policy,
68        sandbox,
69    )
70    .await
71}
72
73async fn find_nearest_ancestor<P, Parent, MarkerPath>(
74    file_system: &dyn ExecutorFileSystem,
75    start: P,
76    markers: Vec<String>,
77    parent: Parent,
78    mut marker_path: MarkerPath,
79    error_policy: FindUpErrorPolicy,
80    sandbox: Option<&FileSystemSandboxContext>,
81) -> FileSystemResult<Option<P>>
82where
83    P: Clone + Send,
84    Parent: FnMut(&P) -> Option<P> + Send,
85    MarkerPath: FnMut(&P, &str) -> FileSystemResult<PathUri> + Send,
86{
87    let mut ancestors = std::iter::successors(Some(start), parent);
88    let mut ancestor = ancestors.next();
89    let mut marker_index = 0;
90    let probes = std::iter::from_fn(move || {
91        let current_ancestor = ancestor.clone()?;
92        let marker = markers.get(marker_index)?;
93        let marker_path = marker_path(&current_ancestor, marker);
94
95        marker_index += 1;
96        if marker_index == markers.len() {
97            marker_index = 0;
98            ancestor = ancestors.next();
99        }
100
101        Some((current_ancestor, marker_path))
102    });
103    let mut results = futures::stream::iter(probes)
104        .map(|(ancestor, marker_path)| async move {
105            let marker_path = marker_path?;
106            match file_system.get_metadata(&marker_path, sandbox).await {
107                Ok(_) => Ok(Some(ancestor)),
108                Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(None),
109                Err(err) => match error_policy {
110                    FindUpErrorPolicy::Propagate => Err(err),
111                    FindUpErrorPolicy::Ignore => Ok(None),
112                },
113            }
114        })
115        .buffered(MAX_CONCURRENT_PROBES);
116
117    while let Some(result) = results.next().await {
118        if let Some(ancestor) = result? {
119            return Ok(Some(ancestor));
120        }
121    }
122    Ok(None)
123}
codex_file_system/find_up.rs

codex_file_system/
find_up.rs
