From 7ca1ecfdb9506e8cfe50e9f248a456406f00734f Mon Sep 17 00:00:00 2001 From: Adam Spofford Date: Wed, 12 Aug 2026 06:00:25 -0700 Subject: [PATCH 1/3] Maybe prevent fs event stream leaking --- crates/icp/src/network/managed/launcher.rs | 96 ++++++++++++++++++++-- 1 file changed, 87 insertions(+), 9 deletions(-) diff --git a/crates/icp/src/network/managed/launcher.rs b/crates/icp/src/network/managed/launcher.rs index 38d27d65..9a653ae1 100644 --- a/crates/icp/src/network/managed/launcher.rs +++ b/crates/icp/src/network/managed/launcher.rs @@ -6,7 +6,12 @@ use serde::Deserialize; use snafu::prelude::*; use std::{io::ErrorKind, process::Stdio, time::Duration}; use sysinfo::{Pid, ProcessesToUpdate, Signal, System}; -use tokio::{process::Child, select, sync::mpsc::Sender, time::Instant}; +use tokio::{ + process::Child, + select, + sync::mpsc::{Receiver, Sender}, + time::Instant, +}; use tracing::{info, warn}; use crate::{ @@ -343,11 +348,11 @@ pub fn wait_for_single_line_file( ) -> Result> + use<>, WaitForFileError> { let dir = path.parent().unwrap(); // notify will get here faster - let (rec_tx, mut rec_rx) = tokio::sync::mpsc::channel(10); + let (rec_tx, rec_rx) = tokio::sync::mpsc::channel(10); let mut rec_watcher = notify::recommended_watcher(WatchRecv(rec_tx)).context(WatchSnafu { path: &dir })?; // poll is more reliable when dealing with vfs like 9p, notably in WSL2 - let (poll_tx, mut poll_rx) = tokio::sync::mpsc::channel(10); + let (poll_tx, poll_rx) = tokio::sync::mpsc::channel(10); let mut poll_watcher = notify::PollWatcher::new( WatchRecv(poll_tx), notify::Config::default() @@ -364,14 +369,15 @@ pub fn wait_for_single_line_file( _ = poll_watcher.poll(); let path = path.to_path_buf(); let dir = dir.to_path_buf(); + let mut session = WatchSession { + rec_rx, + poll_rx, + _rec_watcher: rec_watcher, + _poll_watcher: poll_watcher, + }; Ok(async move { - let _rec_watcher = rec_watcher; - let _poll_watcher = poll_watcher; loop { - let evt = select! { - rec = rec_rx.recv() => rec, - poll = poll_rx.recv() => poll, - }; + let evt = session.next_event().await; let Some(res) = evt else { unreachable!("watcher dropped while waiting for file"); }; @@ -436,6 +442,32 @@ pub struct LauncherStatus { pub const CUSTOM_DOMAINS_FEATURE: &str = "custom-domains"; +/// Keeps each watcher together with its receiver, relying on struct fields being dropped in +/// declaration order so the receivers always go first. +/// +/// Both watchers hand events over with [`Sender::blocking_send`], and both stop by waiting for +/// their callback thread to go idle - the FSEvents backend does so by busy-waiting. A callback +/// parked on a full channel would therefore make that wait spin forever; closing the channels +/// first releases it. Grouping the four into one value keeps that order whether the future is +/// dropped part-way through or before it is ever polled. +struct WatchSession { + rec_rx: Receiver>, + poll_rx: Receiver>, + _rec_watcher: notify::RecommendedWatcher, + _poll_watcher: notify::PollWatcher, +} + +impl WatchSession { + /// Takes `&mut self` rather than the receivers, so that awaiting this keeps the whole session + /// - watchers included - alive. + async fn next_event(&mut self) -> Option> { + select! { + rec = self.rec_rx.recv() => rec, + poll = self.poll_rx.recv() => poll, + } + } +} + struct WatchRecv(Sender>); impl EventHandler for WatchRecv { @@ -448,6 +480,52 @@ impl EventHandler for WatchRecv { mod tests { use super::*; + #[tokio::test] + async fn resolves_once_the_file_has_a_full_line() { + let dir = camino_tempfile::Utf8TempDir::new().unwrap(); + let file = dir.path().join("status.json"); + let fut = wait_for_single_line_file(&file).unwrap(); + + std::thread::spawn({ + let file = file.clone(); + move || { + std::thread::sleep(Duration::from_millis(200)); + std::fs::write(&file, b"partial").unwrap(); + std::thread::sleep(Duration::from_millis(200)); + std::fs::write(&file, b"complete\n").unwrap(); + } + }); + + let content = tokio::time::timeout(Duration::from_secs(20), fut) + .await + .expect("timed out waiting for the file") + .unwrap(); + assert_eq!(content, "complete\n"); + } + + /// Saturates the channels without ever polling the future, then drops it. If the receivers + /// were not dropped ahead of the watchers, the callback thread would still be parked in + /// `blocking_send` and the watchers' shutdown would never finish. + #[test] + fn dropping_an_unpolled_watcher_does_not_wedge() { + let dir = camino_tempfile::Utf8TempDir::new().unwrap(); + let dir = dir.path().to_path_buf(); + let fut = wait_for_single_line_file(&dir.join("status.json")).unwrap(); + + for i in 0..500 { + std::fs::write(dir.join(format!("churn{i}")), b"x").unwrap(); + } + std::thread::sleep(Duration::from_millis(1500)); + + let (tx, rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + drop(fut); + let _ = tx.send(()); + }); + rx.recv_timeout(Duration::from_secs(20)) + .expect("dropping the future wedged"); + } + #[test] fn output_tail_keeps_short_content_verbatim() { assert_eq!(output_tail("line one\nline two"), "line one\nline two"); From 3a3ddda3f266d7b2e94d7bf837417ad8b763fa3a Mon Sep 17 00:00:00 2001 From: Adam Spofford Date: Wed, 12 Aug 2026 09:51:01 -0700 Subject: [PATCH 2/3] copilot --- crates/icp/src/network/managed/launcher.rs | 28 ++++++++++++---------- 1 file changed, 16 insertions(+), 12 deletions(-) diff --git a/crates/icp/src/network/managed/launcher.rs b/crates/icp/src/network/managed/launcher.rs index 9a653ae1..c8df6c24 100644 --- a/crates/icp/src/network/managed/launcher.rs +++ b/crates/icp/src/network/managed/launcher.rs @@ -353,28 +353,32 @@ pub fn wait_for_single_line_file( notify::recommended_watcher(WatchRecv(rec_tx)).context(WatchSnafu { path: &dir })?; // poll is more reliable when dealing with vfs like 9p, notably in WSL2 let (poll_tx, poll_rx) = tokio::sync::mpsc::channel(10); - let mut poll_watcher = notify::PollWatcher::new( + let poll_watcher = notify::PollWatcher::new( WatchRecv(poll_tx), notify::Config::default() .with_poll_interval(Duration::from_millis(100)) .with_compare_contents(true), ) .context(WatchSnafu { path: &dir })?; - rec_watcher + // Assembled before either watcher is registered, so that every exit from here on unwinds + // through the session's field order rather than these locals'. + let mut session = WatchSession { + rec_rx, + poll_rx, + rec_watcher, + poll_watcher, + }; + session + .rec_watcher .watch(dir.as_std_path(), notify::RecursiveMode::NonRecursive) .context(WatchSnafu { path: &dir })?; - poll_watcher + session + .poll_watcher .watch(dir.as_std_path(), notify::RecursiveMode::NonRecursive) .context(WatchSnafu { path: &dir })?; - _ = poll_watcher.poll(); + _ = session.poll_watcher.poll(); let path = path.to_path_buf(); let dir = dir.to_path_buf(); - let mut session = WatchSession { - rec_rx, - poll_rx, - _rec_watcher: rec_watcher, - _poll_watcher: poll_watcher, - }; Ok(async move { loop { let evt = session.next_event().await; @@ -453,8 +457,8 @@ pub const CUSTOM_DOMAINS_FEATURE: &str = "custom-domains"; struct WatchSession { rec_rx: Receiver>, poll_rx: Receiver>, - _rec_watcher: notify::RecommendedWatcher, - _poll_watcher: notify::PollWatcher, + rec_watcher: notify::RecommendedWatcher, + poll_watcher: notify::PollWatcher, } impl WatchSession { From 67187017d50852bab815c45dcf87a376b6c05488 Mon Sep 17 00:00:00 2001 From: Adam Spofford Date: Thu, 13 Aug 2026 04:24:01 -0700 Subject: [PATCH 3/3] fix --- crates/icp/src/network/managed/launcher.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/icp/src/network/managed/launcher.rs b/crates/icp/src/network/managed/launcher.rs index c8df6c24..dd03c742 100644 --- a/crates/icp/src/network/managed/launcher.rs +++ b/crates/icp/src/network/managed/launcher.rs @@ -349,7 +349,7 @@ pub fn wait_for_single_line_file( let dir = path.parent().unwrap(); // notify will get here faster let (rec_tx, rec_rx) = tokio::sync::mpsc::channel(10); - let mut rec_watcher = + let rec_watcher = notify::recommended_watcher(WatchRecv(rec_tx)).context(WatchSnafu { path: &dir })?; // poll is more reliable when dealing with vfs like 9p, notably in WSL2 let (poll_tx, poll_rx) = tokio::sync::mpsc::channel(10);