diff --git a/Cargo.lock b/Cargo.lock index d159765..df25bfc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -159,7 +159,7 @@ version = "1.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" dependencies = [ - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -170,7 +170,7 @@ checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" dependencies = [ "anstyle", "once_cell_polyfill", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -818,6 +818,7 @@ dependencies = [ "codespace-pty", "hex", "libc", + "notify", "serde", "serde_json", "sha2", @@ -1941,7 +1942,7 @@ dependencies = [ "libc", "option-ext", "redox_users 0.5.3", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -2175,7 +2176,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -2411,6 +2412,15 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" +[[package]] +name = "fsevent-sys" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76ee7a02da4d231650c7cea31349b889be2f45ddb3ef3032d2ec8185f6313fd2" +dependencies = [ + "libc", +] + [[package]] name = "futures" version = "0.3.34" @@ -4111,6 +4121,26 @@ dependencies = [ "serde_core", ] +[[package]] +name = "inotify" +version = "0.11.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cc00ea907cab49550b7da656f80ebb97be1b997d931fbcd28d39734e17ce592" +dependencies = [ + "bitflags 2.13.2", + "inotify-sys", + "libc", +] + +[[package]] +name = "inotify-sys" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c033f80b2c113cdf91ab7a33faa9cbc014726dcad99880c8609af2a370edf37d" +dependencies = [ + "libc", +] + [[package]] name = "inout" version = "0.1.4" @@ -4298,6 +4328,26 @@ version = "0.2.19" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a4933f3f57a8e9d9da04db23fb153356ecaf00cbd14aee46279c33dc80925c37" +[[package]] +name = "kqueue" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8d763e5b24120b4ddf50de6c92308156765aabfbbccebf401da7cff2d70a41ea" +dependencies = [ + "kqueue-sys", + "libc", +] + +[[package]] +name = "kqueue-sys" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07293a4e297ac234359b510362495713f75ea345d5307140414f20c69ffeb087" +dependencies = [ + "bitflags 2.13.2", + "libc", +] + [[package]] name = "kstring" version = "2.0.4" @@ -4626,6 +4676,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4b18443e9c262bfe8fa82f51666e2642c53393f7e5c27b3e1aeab922cff5b9d8" dependencies = [ "libc", + "log", "wasi", "windows-sys 0.61.2", ] @@ -4800,13 +4851,40 @@ version = "0.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9737e026353e5cd0736f98eddae28665118eb6f6600902a7f50db585621fecb6" +[[package]] +name = "notify" +version = "8.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d3d07927151ff8575b7087f245456e549fea62edf0ec4e565a5ee50c8402bc3" +dependencies = [ + "bitflags 2.13.2", + "fsevent-sys", + "inotify", + "kqueue", + "libc", + "log", + "mio", + "notify-types", + "walkdir", + "windows-sys 0.60.2", +] + +[[package]] +name = "notify-types" +version = "2.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42b8cfee0e339a0337359f3c88165702ac6e600dc01c0cc9579a92d62b08477a" +dependencies = [ + "bitflags 2.13.2", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -5639,7 +5717,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -6396,7 +6474,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -6988,7 +7066,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" dependencies = [ "libc", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -7325,7 +7403,7 @@ dependencies = [ "getrandom 0.4.3", "once_cell", "rustix", - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] @@ -8329,7 +8407,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.52.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/crates/codex-runtime/Cargo.lock b/crates/codex-runtime/Cargo.lock index 7a4f500..ce26833 100644 --- a/crates/codex-runtime/Cargo.lock +++ b/crates/codex-runtime/Cargo.lock @@ -829,6 +829,7 @@ dependencies = [ "codespace-pty", "hex", "libc", + "notify", "serde", "serde_json", "sha2", @@ -2240,7 +2241,7 @@ checksum = "0ce92ff622d6dadf7349484f42c93271a0d49b7cc4d466a936405bacbe10aa78" dependencies = [ "cfg-if", "rustix", - "windows-sys 0.52.0", + "windows-sys 0.59.0", ] [[package]] @@ -2386,6 +2387,15 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" +[[package]] +name = "fsevent-sys" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "76ee7a02da4d231650c7cea31349b889be2f45ddb3ef3032d2ec8185f6313fd2" +dependencies = [ + "libc", +] + [[package]] name = "futures" version = "0.3.34" @@ -4090,6 +4100,26 @@ dependencies = [ "serde_core", ] +[[package]] +name = "inotify" +version = "0.11.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cc00ea907cab49550b7da656f80ebb97be1b997d931fbcd28d39734e17ce592" +dependencies = [ + "bitflags 2.13.2", + "inotify-sys", + "libc", +] + +[[package]] +name = "inotify-sys" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c033f80b2c113cdf91ab7a33faa9cbc014726dcad99880c8609af2a370edf37d" +dependencies = [ + "libc", +] + [[package]] name = "inout" version = "0.1.4" @@ -4277,6 +4307,26 @@ version = "0.2.19" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a4933f3f57a8e9d9da04db23fb153356ecaf00cbd14aee46279c33dc80925c37" +[[package]] +name = "kqueue" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8d763e5b24120b4ddf50de6c92308156765aabfbbccebf401da7cff2d70a41ea" +dependencies = [ + "kqueue-sys", + "libc", +] + +[[package]] +name = "kqueue-sys" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07293a4e297ac234359b510362495713f75ea345d5307140414f20c69ffeb087" +dependencies = [ + "bitflags 2.13.2", + "libc", +] + [[package]] name = "kstring" version = "2.0.4" @@ -4594,6 +4644,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4b18443e9c262bfe8fa82f51666e2642c53393f7e5c27b3e1aeab922cff5b9d8" dependencies = [ "libc", + "log", "wasi", "windows-sys 0.61.2", ] @@ -4768,6 +4819,33 @@ version = "0.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9737e026353e5cd0736f98eddae28665118eb6f6600902a7f50db585621fecb6" +[[package]] +name = "notify" +version = "8.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d3d07927151ff8575b7087f245456e549fea62edf0ec4e565a5ee50c8402bc3" +dependencies = [ + "bitflags 2.13.2", + "fsevent-sys", + "inotify", + "kqueue", + "libc", + "log", + "mio", + "notify-types", + "walkdir", + "windows-sys 0.60.2", +] + +[[package]] +name = "notify-types" +version = "2.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "42b8cfee0e339a0337359f3c88165702ac6e600dc01c0cc9579a92d62b08477a" +dependencies = [ + "bitflags 2.13.2", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" diff --git a/crates/runner/Cargo.toml b/crates/runner/Cargo.toml index d37bde4..4d1028a 100644 --- a/crates/runner/Cargo.toml +++ b/crates/runner/Cargo.toml @@ -19,6 +19,7 @@ sha2 = "0.10" hex = "0.4" libc = "0.2" tokio = { version = "1", features = ["macros", "rt-multi-thread", "process", "time", "io-util", "net", "sync"] } +notify = "8" [dev-dependencies] tempfile = "3" diff --git a/crates/runner/src/apply.rs b/crates/runner/src/apply.rs index 12409e5..d08263f 100644 --- a/crates/runner/src/apply.rs +++ b/crates/runner/src/apply.rs @@ -10,6 +10,7 @@ use crate::PathSandbox; impl InProcessRunner { pub async fn read_file(&self, ws: &Workspace, path: &str) -> Result { ws.require_file_read()?; + self.touch_watch(ws); PathSandbox::new(ws.clone()).read_file(path).await } @@ -19,11 +20,13 @@ impl InProcessRunner { glob: Option<&str>, ) -> Result { ws.require_file_read()?; + self.touch_watch(ws); PathSandbox::new(ws.clone()).find(glob).await } pub async fn file_version(&self, ws: &Workspace, path: &str) -> Result { ws.require_file_read()?; + self.touch_watch(ws); PathSandbox::new(ws.clone()).version(path).await } @@ -33,6 +36,7 @@ impl InProcessRunner { req: RunnerApplyPatchRequest, ) -> Result { ws.require_file_write()?; + self.touch_watch(ws); let sandbox = PathSandbox::new(ws.clone()); for (path, expected) in &req.expected_versions { let actual = sandbox.version(path).await?; diff --git a/crates/runner/src/lib.rs b/crates/runner/src/lib.rs index d2a46ff..03ee4e9 100644 --- a/crates/runner/src/lib.rs +++ b/crates/runner/src/lib.rs @@ -109,6 +109,7 @@ mod process; mod rollback; mod socket; mod uds; +mod watch; mod wire; pub use api::{ @@ -135,6 +136,7 @@ pub use socket::{ runner_socket_path, RUNNER_SOCKET_NAME, }; pub use uds::{DisconnectHook, UdsRunner, RUNNER_CALL_DEADLINE}; +pub use watch::{FsWatchEvent, FsWatchKind, WatchClosed, WatchSubscription}; pub use wire::{ host_worker, read_frame, serve_runner_connection, write_frame, RunnerEvent, RunnerOp, RunnerOpResult, WireEnvelope, WireKind, WIRE_PROTOCOL, diff --git a/crates/runner/src/process.rs b/crates/runner/src/process.rs index a295851..2378a6a 100644 --- a/crates/runner/src/process.rs +++ b/crates/runner/src/process.rs @@ -62,6 +62,7 @@ pub struct InProcessRunner { inner: Arc>>, on_release: ShellRelease, retention: RetentionPolicy, + pub(crate) watches: crate::watch::WatchSet, } struct Slot { @@ -110,6 +111,7 @@ impl InProcessRunner { inner: Arc::new(Mutex::new(HashMap::new())), on_release, retention, + watches: crate::watch::WatchSet::default(), } } diff --git a/crates/runner/src/watch.rs b/crates/runner/src/watch.rs new file mode 100644 index 0000000..e999bd8 --- /dev/null +++ b/crates/runner/src/watch.rs @@ -0,0 +1,462 @@ +//! Internal workspace filesystem observation. Events are invalidation +//! hints: missing, coalesced, or restarted watches must never weaken +//! `expected_versions` / `VERSION_CONFLICT`. + +use std::collections::HashMap; +use std::path::{Component, Path, PathBuf}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; + +use codespace_domain::{ErrorBody, ErrorCode}; +use codespace_policy::Workspace; +use notify::event::{ModifyKind, RenameMode}; +use notify::{Config, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher}; +use tokio::sync::broadcast; + +use crate::process::InProcessRunner; + +const WATCH_LAG: usize = 512; + +/// One observation delivered by a live watcher. `seq` is that watcher's +/// delivery order, not a filesystem causal revision. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct FsWatchEvent { + pub epoch: u64, + pub seq: u64, + pub kind: FsWatchKind, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum FsWatchKind { + Create { path: String }, + Modify { path: String }, + Remove { path: String }, + Rename { from: String, to: String }, + ResyncRequired, +} + +impl FsWatchKind { + pub fn paths(&self) -> Vec<&str> { + match self { + Self::Create { path } | Self::Modify { path } | Self::Remove { path } => { + vec![path.as_str()] + } + Self::Rename { from, to } => vec![from.as_str(), to.as_str()], + Self::ResyncRequired => Vec::new(), + } + } + + pub fn touches(&self, path: &str) -> bool { + self.paths().contains(&path) + } +} + +/// Normalized watch consumer. Lag is a local `ResyncRequired`, not a skip. +pub struct WatchSubscription { + rx: broadcast::Receiver, + epoch: Arc, +} + +/// The watcher is gone; the subscriber cannot recover from this stream. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct WatchClosed; + +impl WatchSubscription { + pub async fn recv(&mut self) -> Result { + match self.rx.recv().await { + Ok(event) => Ok(event), + Err(broadcast::error::RecvError::Lagged(_)) => Ok(FsWatchEvent { + epoch: self.epoch.load(Ordering::SeqCst), + seq: 0, + kind: FsWatchKind::ResyncRequired, + }), + Err(broadcast::error::RecvError::Closed) => Err(WatchClosed), + } + } +} + +struct WatchShared { + root: PathBuf, + roots: Vec, + epoch: Arc, + seq: AtomicU64, + tx: broadcast::Sender, + fail_next_spawn: AtomicBool, +} + +/// Recursive watcher for one workspace root. +pub struct WorkspaceWatch { + shared: Arc, + watcher: Mutex>, +} + +#[derive(Clone, Default)] +pub(crate) struct WatchSet { + inner: Arc>>>, +} + +impl WatchSet { + fn watch_for(&self, root: &Path) -> Result, ErrorBody> { + let key = std::fs::canonicalize(root).unwrap_or_else(|_| root.to_path_buf()); + let mut map = self.inner.lock().expect("watch set"); + if let Some(existing) = map.get(&key) { + return Ok(existing.clone()); + } + let watch = WorkspaceWatch::start(root.to_path_buf())?; + map.insert(key, watch.clone()); + Ok(watch) + } + + fn touch(&self, root: &Path) { + let _ = self.watch_for(root); + } +} + +impl WorkspaceWatch { + pub fn start(root: PathBuf) -> Result, ErrorBody> { + let roots = watch_root_aliases(&root); + let watched = roots[0].clone(); + let (tx, _) = broadcast::channel(WATCH_LAG); + let shared = Arc::new(WatchShared { + root: watched, + roots, + epoch: Arc::new(AtomicU64::new(1)), + seq: AtomicU64::new(0), + tx, + fail_next_spawn: AtomicBool::new(false), + }); + let watcher = spawn_watcher(shared.clone())?; + Ok(Arc::new(Self { + shared, + watcher: Mutex::new(Some(watcher)), + })) + } + + pub fn subscribe(&self) -> WatchSubscription { + WatchSubscription { + rx: self.shared.tx.subscribe(), + epoch: Arc::clone(&self.shared.epoch), + } + } + + pub fn epoch(&self) -> u64 { + self.shared.epoch.load(Ordering::SeqCst) + } + + pub fn force_resync(&self) { + emit_resync(&self.shared); + } + + pub fn restart(&self) -> Result { + match spawn_watcher(self.shared.clone()) { + Ok(new) => { + let mut slot = self.watcher.lock().expect("watcher"); + *slot = Some(new); + self.shared.epoch.fetch_add(1, Ordering::SeqCst); + self.shared.seq.store(0, Ordering::SeqCst); + drop(slot); + emit_resync(&self.shared); + Ok(self.epoch()) + } + Err(err) => { + emit_resync(&self.shared); + Err(err) + } + } + } +} + +impl Drop for WorkspaceWatch { + fn drop(&mut self) { + if let Ok(mut slot) = self.watcher.lock() { + *slot = None; + } + } +} + +impl InProcessRunner { + /// Best-effort start. Observation must never fail a file operation. + pub(crate) fn touch_watch(&self, ws: &Workspace) { + self.watches.touch(&ws.root); + } + + pub fn subscribe_watch(&self, ws: &Workspace) -> Result { + Ok(self.watches.watch_for(&ws.root)?.subscribe()) + } + + pub fn restart_watch(&self, ws: &Workspace) -> Result { + self.watches.watch_for(&ws.root)?.restart() + } + + pub fn force_watch_resync(&self, ws: &Workspace) -> Result<(), ErrorBody> { + self.watches.watch_for(&ws.root)?.force_resync(); + Ok(()) + } + + #[doc(hidden)] + pub fn fail_next_watch_spawn(&self, ws: &Workspace) -> Result<(), ErrorBody> { + self.watches + .watch_for(&ws.root)? + .shared + .fail_next_spawn + .store(true, Ordering::SeqCst); + Ok(()) + } + + #[doc(hidden)] + pub fn overflow_watch_for_tests(&self, ws: &Workspace) -> Result<(), ErrorBody> { + let watch = self.watches.watch_for(&ws.root)?; + for i in 0..(WATCH_LAG + 8) { + emit( + &watch.shared, + FsWatchKind::Modify { + path: format!("overflow-{i}"), + }, + ); + } + Ok(()) + } +} + +fn spawn_watcher(shared: Arc) -> Result { + if shared.fail_next_spawn.swap(false, Ordering::SeqCst) { + return Err(ErrorBody::new( + ErrorCode::FileOperationFailed, + "workspace watch: injected spawn failure", + )); + } + let callback_shared = shared.clone(); + let mut watcher = RecommendedWatcher::new( + move |result: Result| match result { + Ok(event) => handle_event(&callback_shared, event), + Err(_) => emit_resync(&callback_shared), + }, + Config::default().with_follow_symlinks(false), + ) + .map_err(watch_err)?; + watcher + .watch(&shared.root, RecursiveMode::Recursive) + .map_err(watch_err)?; + Ok(watcher) +} + +fn handle_event(shared: &WatchShared, event: Event) { + if event.need_rescan() { + emit_resync(shared); + return; + } + match event.kind { + EventKind::Access(_) => {} + EventKind::Other => emit_resync(shared), + EventKind::Any => { + let rels = relative_paths(shared, &event.paths); + if rels.is_empty() { + if event.paths.is_empty() { + emit_resync(shared); + } + return; + } + for path in rels { + emit(shared, FsWatchKind::Modify { path }); + } + } + EventKind::Create(_) => { + for path in relative_paths(shared, &event.paths) { + emit(shared, FsWatchKind::Create { path }); + } + } + EventKind::Remove(_) => { + for path in relative_paths(shared, &event.paths) { + emit(shared, FsWatchKind::Remove { path }); + } + } + EventKind::Modify(ModifyKind::Name(mode)) => handle_rename(shared, mode, &event.paths), + EventKind::Modify(_) => { + for path in relative_paths(shared, &event.paths) { + emit(shared, FsWatchKind::Modify { path }); + } + } + } +} + +fn handle_rename(shared: &WatchShared, mode: RenameMode, paths: &[PathBuf]) { + match mode { + RenameMode::Both => match (paths.first(), paths.get(1)) { + (Some(from), Some(to)) => emit_mapped_rename(shared, from, to), + (Some(only), None) => emit_inside(shared, only, |path| FsWatchKind::Modify { path }), + _ => {} + }, + RenameMode::From => { + for path in paths { + emit_inside(shared, path, |p| FsWatchKind::Remove { path: p }); + } + } + RenameMode::To => { + for path in paths { + emit_inside(shared, path, |p| FsWatchKind::Create { path: p }); + } + } + _ if paths.len() >= 2 => emit_mapped_rename(shared, &paths[0], &paths[1]), + _ => { + for path in paths { + emit_inside(shared, path, |p| FsWatchKind::Modify { path: p }); + } + } + } +} + +fn emit_mapped_rename(shared: &WatchShared, from: &Path, to: &Path) { + if let Some(kind) = map_rename_sides( + workspace_relative_event_path(&shared.roots, from), + workspace_relative_event_path(&shared.roots, to), + ) { + emit(shared, kind); + } +} + +fn emit_inside(shared: &WatchShared, path: &Path, kind: impl FnOnce(String) -> FsWatchKind) { + if let Some(rel) = workspace_relative_event_path(&shared.roots, path) { + emit(shared, kind(rel)); + } +} + +fn map_rename_sides(from: Option, to: Option) -> Option { + match (from, to) { + (Some(from), Some(to)) => Some(FsWatchKind::Rename { from, to }), + (Some(from), None) => Some(FsWatchKind::Remove { path: from }), + (None, Some(to)) => Some(FsWatchKind::Create { path: to }), + (None, None) => None, + } +} + +fn relative_paths(shared: &WatchShared, paths: &[PathBuf]) -> Vec { + paths + .iter() + .filter_map(|path| workspace_relative_event_path(&shared.roots, path)) + .collect() +} + +/// Lexical workspace names only. File type and access policy are not consulted. +fn workspace_relative_event_path(roots: &[PathBuf], event_path: &Path) -> Option { + let primary = roots.first()?; + let abs = if event_path.is_absolute() { + event_path.to_path_buf() + } else { + primary.join(event_path) + }; + for root in roots { + if let Some(rel) = strip_under(root, &abs) { + return Some(rel); + } + } + None +} + +fn watch_root_aliases(root: &Path) -> Vec { + let canon = std::fs::canonicalize(root).unwrap_or_else(|_| root.to_path_buf()); + let mut roots = vec![canon.clone()]; + if root != canon.as_path() { + roots.push(root.to_path_buf()); + } + roots +} + +fn strip_under(root: &Path, abs: &Path) -> Option { + let rel = abs.strip_prefix(root).ok()?; + if rel.as_os_str().is_empty() { + return None; + } + if rel.components().any(|c| { + matches!( + c, + Component::ParentDir | Component::Prefix(_) | Component::RootDir + ) + }) { + return None; + } + let text = rel.to_str()?.replace('\\', "/"); + if text.starts_with('/') || text.split('/').any(|part| part == "..") { + return None; + } + Some(text) +} + +fn emit(shared: &WatchShared, kind: FsWatchKind) { + let epoch = shared.epoch.load(Ordering::SeqCst); + let seq = shared.seq.fetch_add(1, Ordering::SeqCst) + 1; + let _ = shared.tx.send(FsWatchEvent { epoch, seq, kind }); +} + +fn emit_resync(shared: &WatchShared) { + emit(shared, FsWatchKind::ResyncRequired); +} + +fn watch_err(err: notify::Error) -> ErrorBody { + ErrorBody::new( + ErrorCode::FileOperationFailed, + format!("workspace watch: {err}"), + ) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::tempdir; + + fn roots_for(dir: &Path) -> Vec { + watch_root_aliases(dir) + } + + #[test] + fn namespace_keeps_workspace_names_including_symlinks() { + let dir = tempdir().unwrap(); + std::fs::write(dir.path().join("keep.txt"), "keep\n").unwrap(); + std::os::unix::fs::symlink("/etc/passwd", dir.path().join("link")).unwrap(); + let roots = roots_for(dir.path()); + assert_eq!( + workspace_relative_event_path(&roots, &dir.path().join("keep.txt")).as_deref(), + Some("keep.txt") + ); + assert_eq!( + workspace_relative_event_path(&roots, &dir.path().join("link")).as_deref(), + Some("link") + ); + } + + #[test] + fn namespace_drops_outside_and_escape() { + let dir = tempdir().unwrap(); + let outside = dir.path().parent().unwrap().join("outside.txt"); + std::fs::write(&outside, "nope\n").unwrap(); + let roots = roots_for(dir.path()); + assert_eq!( + workspace_relative_event_path(&roots, &dir.path().join("..").join("outside.txt")), + None + ); + assert_eq!(workspace_relative_event_path(&roots, &outside), None); + } + + #[test] + fn rename_sides_are_classified_independently() { + assert_eq!( + map_rename_sides(Some("a.txt".into()), Some("b.txt".into())), + Some(FsWatchKind::Rename { + from: "a.txt".into(), + to: "b.txt".into(), + }) + ); + assert_eq!( + map_rename_sides(Some("a.txt".into()), None), + Some(FsWatchKind::Remove { + path: "a.txt".into(), + }) + ); + assert_eq!( + map_rename_sides(None, Some("b.txt".into())), + Some(FsWatchKind::Create { + path: "b.txt".into(), + }) + ); + assert_eq!(map_rename_sides(None, None), None); + } +} diff --git a/crates/runner/tests/fs_watch.rs b/crates/runner/tests/fs_watch.rs new file mode 100644 index 0000000..3932712 --- /dev/null +++ b/crates/runner/tests/fs_watch.rs @@ -0,0 +1,409 @@ +//! Internal filesystem observation: events are hints, not apply guards. + +use std::collections::BTreeMap; +use std::path::Path; +use std::sync::Arc; +use std::time::Duration; + +use codespace_domain::{ErrorCode, Profile, WorkspaceId}; +use codespace_policy::Workspace; +use codespace_runner::{ + FsWatchEvent, FsWatchKind, InProcessRunner, Runner, RunnerApplyPatchRequest, WatchSubscription, +}; +use tempfile::tempdir; + +fn workspace(root: &Path) -> Workspace { + Workspace::new( + WorkspaceId("demo".into()), + root.to_path_buf(), + Profile::WorkspaceWrite, + ) +} + +fn runner() -> InProcessRunner { + InProcessRunner::new(Arc::new(|_| {})) +} + +async fn recv_until( + rx: &mut WatchSubscription, + timeout: Duration, + mut pred: impl FnMut(&FsWatchEvent) -> bool, +) -> Option { + let deadline = tokio::time::Instant::now() + timeout; + loop { + let left = deadline.saturating_duration_since(tokio::time::Instant::now()); + if left.is_zero() { + return None; + } + match tokio::time::timeout(left, rx.recv()).await { + Ok(Ok(event)) => { + if pred(&event) { + return Some(event); + } + } + Ok(Err(_)) => return None, + Err(_) => return None, + } + } +} + +async fn drain_for(rx: &mut WatchSubscription, timeout: Duration) -> Vec { + let mut out = Vec::new(); + let deadline = tokio::time::Instant::now() + timeout; + loop { + let left = deadline.saturating_duration_since(tokio::time::Instant::now()); + if left.is_zero() { + break; + } + match tokio::time::timeout(left, rx.recv()).await { + Ok(Ok(event)) => out.push(event), + _ => break, + } + } + out +} + +async fn prime_watch(runner: &InProcessRunner, ws: &Workspace) -> WatchSubscription { + let mut rx = runner.subscribe_watch(ws).expect("subscribe"); + let probe = ws.root.join(".watch-prime"); + std::fs::write(&probe, "prime\n").unwrap(); + let seen = recv_until(&mut rx, Duration::from_secs(5), |event| { + event.kind.touches(".watch-prime") + }) + .await; + assert!(seen.is_some(), "watcher did not observe the prime file"); + let _ = std::fs::remove_file(&probe); + let _ = recv_until(&mut rx, Duration::from_millis(500), |event| { + event.kind.touches(".watch-prime") + }) + .await; + rx +} + +#[tokio::test] +async fn external_modify_is_observed() { + let dir = tempdir().unwrap(); + std::fs::write(dir.path().join("keep.txt"), "keep\n").unwrap(); + let ws = workspace(dir.path()); + let runner = runner(); + let mut rx = prime_watch(&runner, &ws).await; + + std::fs::write(dir.path().join("keep.txt"), "edited\n").unwrap(); + let event = recv_until(&mut rx, Duration::from_secs(5), |event| { + event.kind.touches("keep.txt") + }) + .await + .expect("modify event"); + assert_eq!(event.epoch, 1); +} + +#[tokio::test] +async fn create_remove_and_rename_dirty_source_and_dest() { + let dir = tempdir().unwrap(); + std::fs::write(dir.path().join("keep.txt"), "keep\n").unwrap(); + let ws = workspace(dir.path()); + let runner = runner(); + let mut rx = prime_watch(&runner, &ws).await; + + std::fs::write(dir.path().join("created.txt"), "new\n").unwrap(); + recv_until(&mut rx, Duration::from_secs(5), |event| { + event.kind.touches("created.txt") + }) + .await + .expect("create"); + + std::fs::rename( + dir.path().join("created.txt"), + dir.path().join("renamed.txt"), + ) + .unwrap(); + let mut saw_from = false; + let mut saw_to = false; + recv_until(&mut rx, Duration::from_secs(5), |event| { + for path in event.kind.paths() { + if path == "created.txt" { + saw_from = true; + } + if path == "renamed.txt" { + saw_to = true; + } + } + if let FsWatchKind::Rename { from, to } = &event.kind { + if from == "created.txt" && to == "renamed.txt" { + saw_from = true; + saw_to = true; + } + } + saw_from && saw_to + }) + .await + .expect("rename dirty source and dest"); + + std::fs::remove_file(dir.path().join("renamed.txt")).unwrap(); + recv_until(&mut rx, Duration::from_secs(5), |event| { + event.kind.touches("renamed.txt") + }) + .await + .expect("remove"); +} + +#[tokio::test] +async fn coalesced_writes_still_read_current_bytes() { + let dir = tempdir().unwrap(); + std::fs::write(dir.path().join("keep.txt"), "0").unwrap(); + let ws = workspace(dir.path()); + let runner = runner(); + let mut rx = prime_watch(&runner, &ws).await; + + for i in 1..=20 { + std::fs::write(dir.path().join("keep.txt"), i.to_string()).unwrap(); + } + let _ = drain_for(&mut rx, Duration::from_millis(400)).await; + let read = runner.read(&ws, "keep.txt").await.expect("read"); + assert_eq!(read.content, "20"); +} + +#[tokio::test] +async fn forced_resync_stays_on_the_same_epoch() { + let dir = tempdir().unwrap(); + std::fs::write(dir.path().join("keep.txt"), "keep\n").unwrap(); + let ws = workspace(dir.path()); + let runner = runner(); + let mut rx = prime_watch(&runner, &ws).await; + runner.force_watch_resync(&ws).unwrap(); + let event = recv_until(&mut rx, Duration::from_secs(2), |event| { + event.kind == FsWatchKind::ResyncRequired + }) + .await + .expect("resync"); + assert_eq!(event.epoch, 1); +} + +#[tokio::test] +async fn subscriber_lag_becomes_resync_required() { + let dir = tempdir().unwrap(); + std::fs::write(dir.path().join("keep.txt"), "keep\n").unwrap(); + let ws = workspace(dir.path()); + let runner = runner(); + let mut rx = prime_watch(&runner, &ws).await; + runner.overflow_watch_for_tests(&ws).unwrap(); + let event = recv_until(&mut rx, Duration::from_secs(2), |event| { + event.kind == FsWatchKind::ResyncRequired && event.epoch == 1 + }) + .await + .expect("lagged resync"); + assert_eq!(event.epoch, 1); +} + +#[tokio::test] +async fn watcher_restart_bumps_epoch_and_invalidates() { + let dir = tempdir().unwrap(); + std::fs::write(dir.path().join("keep.txt"), "keep\n").unwrap(); + let ws = workspace(dir.path()); + let runner = runner(); + let mut rx = prime_watch(&runner, &ws).await; + let epoch = runner.restart_watch(&ws).unwrap(); + assert!(epoch > 1, "epoch={epoch}"); + let event = recv_until(&mut rx, Duration::from_secs(2), |event| { + event.kind == FsWatchKind::ResyncRequired && event.epoch == epoch + }) + .await + .expect("restart resync"); + assert_eq!(event.epoch, epoch); +} + +#[tokio::test] +async fn restart_failure_keeps_the_live_watcher() { + let dir = tempdir().unwrap(); + std::fs::write(dir.path().join("keep.txt"), "keep\n").unwrap(); + let ws = workspace(dir.path()); + let runner = runner(); + let mut rx = prime_watch(&runner, &ws).await; + + runner.fail_next_watch_spawn(&ws).unwrap(); + let err = runner.restart_watch(&ws).expect_err("spawn failure"); + assert_eq!(err.code, ErrorCode::FileOperationFailed); + recv_until(&mut rx, Duration::from_secs(2), |event| { + event.kind == FsWatchKind::ResyncRequired && event.epoch == 1 + }) + .await + .expect("failed restart still resyncs"); + + std::fs::write(dir.path().join("keep.txt"), "still-watched\n").unwrap(); + recv_until(&mut rx, Duration::from_secs(5), |event| { + event.kind.touches("keep.txt") && event.epoch == 1 + }) + .await + .expect("old watcher still live"); + + let epoch = runner.restart_watch(&ws).unwrap(); + assert!(epoch > 1, "epoch={epoch}"); + recv_until(&mut rx, Duration::from_secs(2), |event| { + event.kind == FsWatchKind::ResyncRequired && event.epoch == epoch + }) + .await + .expect("successful restart resync"); +} + +#[tokio::test] +async fn outside_paths_are_not_emitted() { + let dir = tempdir().unwrap(); + std::fs::write(dir.path().join("keep.txt"), "keep\n").unwrap(); + let outside_dir = tempdir().unwrap(); + let secret = outside_dir.path().join("secret.txt"); + std::fs::write(&secret, "secret\n").unwrap(); + + let ws = workspace(dir.path()); + let runner = runner(); + let mut rx = prime_watch(&runner, &ws).await; + + std::fs::write(&secret, "changed\n").unwrap(); + std::fs::write(outside_dir.path().join("sibling.txt"), "nope\n").unwrap(); + + let events = drain_for(&mut rx, Duration::from_millis(600)).await; + assert!( + !events.iter().any(|event| { + event + .kind + .paths() + .iter() + .any(|path| path.contains("secret") || path.contains("sibling")) + }), + "outside paths must not be emitted: {events:?}" + ); +} + +#[tokio::test] +async fn symlink_replacement_invalidates_the_name() { + let dir = tempdir().unwrap(); + std::fs::write(dir.path().join("keep.txt"), "keep\n").unwrap(); + let outside_dir = tempdir().unwrap(); + let secret = outside_dir.path().join("secret.txt"); + std::fs::write(&secret, "secret\n").unwrap(); + + let ws = workspace(dir.path()); + let runner = runner(); + let mut rx = prime_watch(&runner, &ws).await; + + std::os::unix::fs::symlink(&secret, dir.path().join("tmp-link")).unwrap(); + recv_until(&mut rx, Duration::from_secs(5), |event| { + event.kind.touches("tmp-link") + }) + .await + .expect("symlink create invalidates the name"); + + std::fs::rename(dir.path().join("tmp-link"), dir.path().join("keep.txt")).unwrap(); + recv_until(&mut rx, Duration::from_secs(5), |event| { + event.kind.touches("keep.txt") + }) + .await + .expect("destination name invalidated"); + + let err = runner.read(&ws, "keep.txt").await.expect_err("symlink"); + match err { + codespace_runner::RunnerError::Execution(body) => { + assert_eq!(body.code, ErrorCode::SymlinkRejected); + } + other => panic!("expected SYMLINK_REJECTED, got {other:?}"), + } +} + +#[tokio::test] +async fn rename_across_workspace_boundary_maps_each_side() { + let dir = tempdir().unwrap(); + std::fs::write(dir.path().join("leave.txt"), "go\n").unwrap(); + let outside_dir = tempdir().unwrap(); + std::fs::write(outside_dir.path().join("enter.txt"), "in\n").unwrap(); + + let ws = workspace(dir.path()); + let runner = runner(); + let mut rx = prime_watch(&runner, &ws).await; + + std::fs::rename( + dir.path().join("leave.txt"), + outside_dir.path().join("left.txt"), + ) + .unwrap(); + recv_until(&mut rx, Duration::from_secs(5), |event| { + event.kind.touches("leave.txt") + }) + .await + .expect("inside to outside is Remove/touch of source"); + + std::fs::rename( + outside_dir.path().join("enter.txt"), + dir.path().join("entered.txt"), + ) + .unwrap(); + recv_until(&mut rx, Duration::from_secs(5), |event| { + event.kind.touches("entered.txt") + }) + .await + .expect("outside to inside is Create/touch of dest"); +} + +#[tokio::test] +async fn special_file_name_is_still_invalidated() { + let dir = tempdir().unwrap(); + std::fs::write(dir.path().join("keep.txt"), "keep\n").unwrap(); + let ws = workspace(dir.path()); + let runner = runner(); + let mut rx = prime_watch(&runner, &ws).await; + + let fifo = dir.path().join("pipe.fifo"); + let status = std::process::Command::new("mkfifo") + .arg(&fifo) + .status() + .expect("mkfifo"); + assert!(status.success(), "mkfifo should exist"); + recv_until(&mut rx, Duration::from_secs(5), |event| { + event.kind.touches("pipe.fifo") + }) + .await + .expect("special-file name invalidated"); + + let err = runner.read(&ws, "pipe.fifo").await.expect_err("special"); + match err { + codespace_runner::RunnerError::Execution(body) => { + assert_eq!(body.code, ErrorCode::SpecialFileRejected); + } + other => panic!("expected SPECIAL_FILE_REJECTED, got {other:?}"), + } +} + +#[tokio::test] +async fn discarded_watch_events_do_not_bypass_version_conflict() { + let dir = tempdir().unwrap(); + std::fs::write(dir.path().join("keep.txt"), "keep\n").unwrap(); + let ws = workspace(dir.path()); + let runner = runner(); + let mut rx = prime_watch(&runner, &ws).await; + + let version = runner.version(&ws, "keep.txt").await.unwrap(); + std::fs::write(dir.path().join("keep.txt"), "externally-edited\n").unwrap(); + let _ = drain_for(&mut rx, Duration::from_millis(400)).await; + + let err = runner + .apply_patch( + &ws, + RunnerApplyPatchRequest { + patch: + "*** Begin Patch\n*** Update File: keep.txt\n@@\n-keep\n+new\n*** End Patch\n" + .into(), + expected_versions: BTreeMap::from([("keep.txt".into(), version)]), + check_only: false, + }, + ) + .await + .expect_err("version conflict"); + match err { + codespace_runner::RunnerError::Execution(body) => { + assert_eq!(body.code, ErrorCode::VersionConflict); + } + other => panic!("expected VERSION_CONFLICT, got {other:?}"), + } + assert_eq!( + std::fs::read_to_string(dir.path().join("keep.txt")).unwrap(), + "externally-edited\n" + ); +} diff --git a/docs/execution-substrate.md b/docs/execution-substrate.md index a76ccb0..5698b30 100644 --- a/docs/execution-substrate.md +++ b/docs/execution-substrate.md @@ -47,6 +47,15 @@ UDS transport and Linux sandbox preparation have distinct protocols and failure + +## Filesystem observation + +The Runner may start a recursive filesystem watcher for a workspace on the first `read`, `find`, `version`, or `apply_patch` call. Events are workspace-relative invalidation hints (`Create`, `Modify`, `Remove`, `Rename`, `ResyncRequired`) with an `epoch` and a delivery `seq`. They are not an MCP tool, not a permission decision, and not a mutation precondition. Watch paths are workspace-relative invalidation names. Emitting a path does not imply that the path is readable, writable, regular, or non-symlink; normal file operations continue to enforce `PathSandbox`. + +`expected_versions` and `VERSION_CONFLICT` remain the authoritative apply guard. Missing, coalesced, or restarted watch events must never make `apply_patch` succeed when the on-disk hash no longer matches. Overflow, receive failure (including a lagged subscriber), or an unclassifiable event yields `ResyncRequired` on the same epoch (treat any consumer cache as fully untrusted). Restarting the watcher increments `epoch` and also emits `ResyncRequired` only after a replacement watcher is running. This substrate does not classify self-generated versus external writes, keep a lossless event ledger, or send watch events over UDS. + +`find` stays a bounded glob walk. It is not a watch API. + @@ -58,7 +67,7 @@ When the operator sets workspace `approvals` to `confirm`, a policy-allowed `app ## What remains unimplemented -Process exit codes and explicit output-loss metadata are not exposed to MCP. PTY resize, file range/pagination arguments, durable process recovery, container/remote dispatch, and a resource queue scheduler remain absent. Richer internal types and negotiated protocol flags do not imply those features are callable. +Process exit codes and explicit output-loss metadata are not exposed to MCP. PTY resize, file range/pagination arguments, durable process recovery, container/remote dispatch, and a resource queue scheduler remain absent. MCP `fs/watch`, UDS watch events, and write-cause classification are not provided. Richer internal types and negotiated protocol flags do not imply those features are callable. ## Maintaining the boundary diff --git a/docs/ko/execution-substrate.md b/docs/ko/execution-substrate.md index 076a7d2..8a45ea4 100644 --- a/docs/ko/execution-substrate.md +++ b/docs/ko/execution-substrate.md @@ -54,6 +54,15 @@ UDS 전송과 Linux 샌드박스 준비는 서로 다른 프로토콜과 실패 + +## 파일 시스템 관측 + +Runner는 해당 작업 공간에서 처음 `read`·`find`·`version`·`apply_patch`가 호출될 때 재귀 파일 감시자를 시작할 수 있습니다. 이벤트는 작업 공간 상대 경로의 무효화 힌트(`Create`, `Modify`, `Remove`, `Rename`, `ResyncRequired`)이며 `epoch`와 전달 순서 `seq`를 가집니다. MCP 도구가 아니고, 권한 결정도 아니며, 변경 전제도 아닙니다. watch 경로는 작업 공간 상대 무효화 이름입니다. 경로가 방출되어도 그 경로가 읽기·쓰기 가능하거나 일반 파일·비심링크라는 뜻은 아닙니다. 실제 접근은 계속 `PathSandbox`가 강제합니다. + +변경 전제는 여전히 `expected_versions`와 `VERSION_CONFLICT`입니다. 놓친·합쳐진·재시작된 watch 이벤트 때문에 디스크 해시가 달라진 `apply_patch`가 성공해서는 안 됩니다. 오버플로, 수신 실패(뒤처진 구독자 포함), 분류할 수 없는 이벤트는 같은 epoch에서 `ResyncRequired`를 내며 소비자 캐시 전체를 신뢰하지 않아야 합니다. 감시자를 다시 시작할 때는 교체 감시자가 살아 있는 뒤에만 `epoch`가 증가하고 `ResyncRequired`를 냅니다. 이 기반은 자체 apply와 외부 편집을 구분하지 않고, 손실 없는 이벤트 원장을 두지 않으며, UDS로 watch 이벤트를 보내지 않습니다. + +`find`는 제한된 glob 탐색입니다. watch API가 아닙니다. + @@ -67,7 +76,7 @@ UDS 전송과 Linux 샌드박스 준비는 서로 다른 프로토콜과 실패 ## 아직 제공하지 않는 기능 -MCP에는 프로세스 종료 코드와 명시적 출력 유실 정보가 없습니다. PTY 크기 변경, 파일 범위·페이지 인자, 영속적인 프로세스 복구, 컨테이너·원격 실행, 자원 큐 스케줄러도 제공하지 않습니다. 내부 타입이나 협상된 프로토콜 플래그가 존재한다고 해당 기능을 호출할 수 있는 것은 아닙니다. +MCP에는 프로세스 종료 코드와 명시적 출력 유실 정보가 없습니다. PTY 크기 변경, 파일 범위·페이지 인자, 영속적인 프로세스 복구, 컨테이너·원격 실행, 자원 큐 스케줄러도 제공하지 않습니다. MCP `fs/watch`, UDS watch 이벤트, 쓰기 원인 분류도 제공하지 않습니다. 내부 타입이나 협상된 프로토콜 플래그가 존재한다고 해당 기능을 호출할 수 있는 것은 아닙니다. ## 구현 경계 유지 diff --git a/docs/translations.json b/docs/translations.json index aaf5dde..a48e944 100644 --- a/docs/translations.json +++ b/docs/translations.json @@ -261,6 +261,7 @@ "execution-and-observation", "execution-contracts", "execution-substrate", + "filesystem-observation", "four-axes-target-domain", "fs-watch-and-search", "fs-watch와-검색", @@ -298,11 +299,12 @@ "실행과-결과-관측", "아직-제공하지-않는-기능", "정책과-실행-구현", + "파일-시스템-관측", "확인-홀드", "훅과-스킬" ], - "source_sha256": "0cd55e60c0ea0e71c3d44fedaa0bd2e07c6aea3cf150bb9c716a990137414510", - "translation_sha256": "4373d96a0b266706e98fb3728140169838d22edcd0938c018f93c7b792d6566a" + "source_sha256": "07174574aa201e3d86e301916a09348a0fd6a2ff47d70f45c7473cb3cd956bf4", + "translation_sha256": "24e6a8d39012ddb4e863bb243099f83c5a303335180d46262e08166d460d0f1d" }, { "id": "protocol-compatibility", diff --git a/tests/security/adversarial.rs b/tests/security/adversarial.rs index 15cd837..a161a47 100644 --- a/tests/security/adversarial.rs +++ b/tests/security/adversarial.rs @@ -254,6 +254,46 @@ async fn version_conflict_and_context_mismatch_are_not_applied() { client.cancel().await.expect("cancel"); } +#[tokio::test] +async fn discarded_watch_hints_do_not_replace_version_conflict() { + let (root, cfg, ws) = write_ws("workspace-write"); + let db = root.path().join("ops.sqlite"); + let client = spawn_client(&cfg, Some(&db)).await; + + let read = client + .call_tool( + CallToolRequestParams::new(TOOL_READ) + .with_arguments(object!({ "workspace_id": "demo", "path": "keep.txt" })), + ) + .await + .expect("read"); + let version = payload(&read)["version"].as_str().unwrap().to_string(); + std::fs::write(ws.join("keep.txt"), "externally-edited\n").unwrap(); + tokio::time::sleep(Duration::from_millis(200)).await; + + let conflict = client + .call_tool( + CallToolRequestParams::new(TOOL_APPLY_PATCH).with_arguments(object!({ + "workspace_id": "demo", + "patch": "*** Begin Patch\n*** Update File: keep.txt\n@@\n-keep\n+new\n*** End Patch\n", + "expected_versions": { "keep.txt": version }, + "operation_key": "watch-ignored-1" + })), + ) + .await; + let text = err_text(&conflict); + assert!( + text.contains("VERSION_CONFLICT") || text.contains("version"), + "{text}" + ); + assert_eq!( + std::fs::read_to_string(ws.join("keep.txt")).unwrap(), + "externally-edited\n" + ); + + client.cancel().await.expect("cancel"); +} + async fn spawn_http(config: HttpConfig) -> SocketAddr { let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind"); let addr = listener.local_addr().expect("addr");