diff --git a/Cargo.lock b/Cargo.lock index 215d06903..ee12166db 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2414,7 +2414,6 @@ dependencies = [ "figment", "futures", "humantime", - "inotify", "jsonschema", "k8s-openapi", "kube", @@ -2422,6 +2421,7 @@ dependencies = [ "log", "nid", "nix 0.29.0", + "notify", "nuid", "once_cell", "opentelemetry", @@ -2807,6 +2807,18 @@ dependencies = [ "version_check", ] +[[package]] +name = "filetime" +version = "0.2.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc0505cd1b6fa6580283f6bdf70a73fcf4aba1184038c90902b92b3dd0df63ed" +dependencies = [ + "cfg-if 1.0.3", + "libc", + "libredox", + "windows-sys 0.60.2", +] + [[package]] name = "find-msvc-tools" version = "0.1.2" @@ -2930,6 +2942,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 = "fuchsia-zircon" version = "0.3.3" @@ -4053,15 +4074,13 @@ checksum = "c8fae54786f62fb2918dcfae3d568594e50eb9b5c25bf04371af6fe7516452fb" [[package]] name = "inotify" -version = "0.11.0" +version = "0.9.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f37dccff2791ab604f9babef0ba14fbe0be30bd368dc541e2b08d07c8aa908f3" +checksum = "f8069d3ec154eb856955c1c0fbffefbf5f3c40a104ec912d4797314c1801abff" dependencies = [ - "bitflags 2.9.4", - "futures-core", + "bitflags 1.3.2", "inotify-sys", "libc", - "tokio", ] [[package]] @@ -4436,6 +4455,26 @@ dependencies = [ "winapi-build", ] +[[package]] +name = "kqueue" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eac30106d7dce88daf4a3fcb4879ea939476d5074a9b7ddd0fb97fa4bed5596a" +dependencies = [ + "kqueue-sys", + "libc", +] + +[[package]] +name = "kqueue-sys" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed9625ffda8729b85e45cf04090035ac368927b8cebc34898e7c120f52e4838b" +dependencies = [ + "bitflags 1.3.2", + "libc", +] + [[package]] name = "kube" version = "2.0.1" @@ -4658,6 +4697,7 @@ checksum = "416f7e718bdb06000964960ffa43b4335ad4012ae8b99060261aa4a8088d5ccb" dependencies = [ "bitflags 2.9.4", "libc", + "redox_syscall", ] [[package]] @@ -5614,6 +5654,24 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0676bb32a98c1a483ce53e500a81ad9c3d5b3f7c920c28c24e9cb0980d0b5bc8" +[[package]] +name = "notify" +version = "6.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6205bd8bb1e454ad2e27422015fb5e4f2bcc7e08fa8f27058670d208324a4d2d" +dependencies = [ + "bitflags 2.9.4", + "filetime", + "fsevent-sys", + "inotify", + "kqueue", + "libc", + "log", + "mio 0.8.11", + "walkdir", + "windows-sys 0.48.0", +] + [[package]] name = "ntapi" version = "0.4.1" diff --git a/lib/bindings/python/Cargo.lock b/lib/bindings/python/Cargo.lock index ad6f4b54c..34345b2ae 100644 --- a/lib/bindings/python/Cargo.lock +++ b/lib/bindings/python/Cargo.lock @@ -1687,13 +1687,13 @@ dependencies = [ "figment", "futures", "humantime", - "inotify", "k8s-openapi", "kube", "local-ip-address", "log", "nid", "nix", + "notify", "nuid", "once_cell", "opentelemetry", @@ -1991,6 +1991,18 @@ dependencies = [ "version_check", ] +[[package]] +name = "filetime" +version = "0.2.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bc0505cd1b6fa6580283f6bdf70a73fcf4aba1184038c90902b92b3dd0df63ed" +dependencies = [ + "cfg-if 1.0.3", + "libc", + "libredox", + "windows-sys 0.60.2", +] + [[package]] name = "fixedbitset" version = "0.5.7" @@ -2045,6 +2057,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 = "fuchsia-zircon" version = "0.3.3" @@ -2970,15 +2991,13 @@ checksum = "c8fae54786f62fb2918dcfae3d568594e50eb9b5c25bf04371af6fe7516452fb" [[package]] name = "inotify" -version = "0.11.0" +version = "0.9.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f37dccff2791ab604f9babef0ba14fbe0be30bd368dc541e2b08d07c8aa908f3" +checksum = "f8069d3ec154eb856955c1c0fbffefbf5f3c40a104ec912d4797314c1801abff" dependencies = [ - "bitflags 2.9.3", - "futures-core", + "bitflags 1.3.2", "inotify-sys", "libc", - "tokio", ] [[package]] @@ -3261,6 +3280,26 @@ dependencies = [ "winapi-build", ] +[[package]] +name = "kqueue" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eac30106d7dce88daf4a3fcb4879ea939476d5074a9b7ddd0fb97fa4bed5596a" +dependencies = [ + "kqueue-sys", + "libc", +] + +[[package]] +name = "kqueue-sys" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed9625ffda8729b85e45cf04090035ac368927b8cebc34898e7c120f52e4838b" +dependencies = [ + "bitflags 1.3.2", + "libc", +] + [[package]] name = "kube" version = "2.0.1" @@ -3434,6 +3473,7 @@ checksum = "391290121bad3d37fbddad76d8f5d1c1c314cfc646d143d7e07a3086ddff0ce3" dependencies = [ "bitflags 2.9.3", "libc", + "redox_syscall", ] [[package]] @@ -3717,6 +3757,18 @@ dependencies = [ "winapi 0.2.8", ] +[[package]] +name = "mio" +version = "0.8.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4a650543ca06a924e8b371db273b2756685faae30f8487da1b56505a8f78b0c" +dependencies = [ + "libc", + "log", + "wasi 0.11.1+wasi-snapshot-preview1", + "windows-sys 0.48.0", +] + [[package]] name = "mio" version = "1.0.4" @@ -3930,6 +3982,24 @@ version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0676bb32a98c1a483ce53e500a81ad9c3d5b3f7c920c28c24e9cb0980d0b5bc8" +[[package]] +name = "notify" +version = "6.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6205bd8bb1e454ad2e27422015fb5e4f2bcc7e08fa8f27058670d208324a4d2d" +dependencies = [ + "bitflags 2.9.3", + "filetime", + "fsevent-sys", + "inotify", + "kqueue", + "libc", + "log", + "mio 0.8.11", + "walkdir", + "windows-sys 0.48.0", +] + [[package]] name = "nu-ansi-term" version = "0.46.0" @@ -7552,6 +7622,15 @@ dependencies = [ "windows-link 0.1.3", ] +[[package]] +name = "windows-sys" +version = "0.48.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "677d2418bec65e3338edb076e806bc1ec15693c5d0104683f2efe857f61056a9" +dependencies = [ + "windows-targets 0.48.5", +] + [[package]] name = "windows-sys" version = "0.52.0" @@ -7579,6 +7658,21 @@ dependencies = [ "windows-targets 0.53.3", ] +[[package]] +name = "windows-targets" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9a2fa6e2155d7247be68c096456083145c183cbbbc2764150dda45a87197940c" +dependencies = [ + "windows_aarch64_gnullvm 0.48.5", + "windows_aarch64_msvc 0.48.5", + "windows_i686_gnu 0.48.5", + "windows_i686_msvc 0.48.5", + "windows_x86_64_gnu 0.48.5", + "windows_x86_64_gnullvm 0.48.5", + "windows_x86_64_msvc 0.48.5", +] + [[package]] name = "windows-targets" version = "0.52.6" @@ -7612,6 +7706,12 @@ dependencies = [ "windows_x86_64_msvc 0.53.0", ] +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b38e32f0abccf9987a4e3079dfb67dcd799fb61361e53e2882c3cbaf0d905d8" + [[package]] name = "windows_aarch64_gnullvm" version = "0.52.6" @@ -7624,6 +7724,12 @@ version = "0.53.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "86b8d5f90ddd19cb4a147a5fa63ca848db3df085e25fee3cc10b39b6eebae764" +[[package]] +name = "windows_aarch64_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc35310971f3b2dbbf3f0690a219f40e2d9afcf64f9ab7cc1be722937c26b4bc" + [[package]] name = "windows_aarch64_msvc" version = "0.52.6" @@ -7636,6 +7742,12 @@ version = "0.53.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c7651a1f62a11b8cbd5e0d42526e55f2c99886c77e007179efff86c2b137e66c" +[[package]] +name = "windows_i686_gnu" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a75915e7def60c94dcef72200b9a8e58e5091744960da64ec734a6c6e9b3743e" + [[package]] name = "windows_i686_gnu" version = "0.52.6" @@ -7660,6 +7772,12 @@ version = "0.53.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ce6ccbdedbf6d6354471319e781c0dfef054c81fbc7cf83f338a4296c0cae11" +[[package]] +name = "windows_i686_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f55c233f70c4b27f66c523580f78f1004e8b5a8b659e05a4eb49d4166cca406" + [[package]] name = "windows_i686_msvc" version = "0.52.6" @@ -7672,6 +7790,12 @@ version = "0.53.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "581fee95406bb13382d2f65cd4a908ca7b1e4c2f1917f143ba16efe98a589b5d" +[[package]] +name = "windows_x86_64_gnu" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53d40abd2583d23e4718fddf1ebec84dbff8381c07cae67ff7768bbf19c6718e" + [[package]] name = "windows_x86_64_gnu" version = "0.52.6" @@ -7684,6 +7808,12 @@ version = "0.53.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2e55b5ac9ea33f2fc1716d1742db15574fd6fc8dadc51caab1c16a3d3b4190ba" +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b7b52767868a23d5bab768e390dc5f5c55825b6d30b86c844ff2dc7414044cc" + [[package]] name = "windows_x86_64_gnullvm" version = "0.52.6" @@ -7696,6 +7826,12 @@ version = "0.53.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0a6e035dd0599267ce1ee132e51c27dd29437f63325753051e71dd9e42406c57" +[[package]] +name = "windows_x86_64_msvc" +version = "0.48.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed94fce61571a4006852b7389a063ab983c02eb1bb37b47f8272ce92d06d9538" + [[package]] name = "windows_x86_64_msvc" version = "0.52.6" diff --git a/lib/runtime/Cargo.toml b/lib/runtime/Cargo.toml index 42770ef10..b98cd8550 100644 --- a/lib/runtime/Cargo.toml +++ b/lib/runtime/Cargo.toml @@ -64,7 +64,7 @@ bincode = { version = "1" } console-subscriber = { version = "0.4", optional = true } educe = { version = "0.6.0" } figment = { version = "0.10.19", features = ["env", "json", "toml", "test"] } -inotify = { version = "0.11" } +notify = { version = "6.1", default-features = false, features = ["macos_fsevent"] } local-ip-address = { version = "0.6.3" } log = { version = "0.4" } nid = { version = "3.0.0", features = ["serde"] } diff --git a/lib/runtime/src/storage/key_value_store/file.rs b/lib/runtime/src/storage/key_value_store/file.rs index 392d403a2..0b9f20a32 100644 --- a/lib/runtime/src/storage/key_value_store/file.rs +++ b/lib/runtime/src/storage/key_value_store/file.rs @@ -14,7 +14,7 @@ use std::{collections::HashMap, pin::Pin}; use anyhow::Context as _; use async_trait::async_trait; use futures::StreamExt; -use inotify::{Event, EventMask, EventStream, Inotify, WatchMask}; +use notify::{Config, Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher}; use parking_lot::Mutex; use crate::storage::key_value_store::KeyValue; @@ -117,8 +117,10 @@ pub struct Directory { impl Directory { fn new(root: PathBuf, p: PathBuf) -> Self { + // Canonicalize root to handle symlinks (e.g., /var -> /private/var on macOS) + let canonical_root = root.canonicalize().unwrap_or_else(|_| root.clone()); Directory { - root, + root: canonical_root, p, owned_files: Arc::new(Mutex::new(HashSet::new())), } @@ -197,64 +199,88 @@ impl KeyValueBucket for Directory { async fn watch( &self, ) -> Result + Send + 'life0>>, StoreError> { - let inotify = Inotify::init().map_err(to_fs_err)?; - inotify - .watches() - .add( - &self.p, - WatchMask::MODIFY | WatchMask::CREATE | WatchMask::DELETE, - ) + let (tx, mut rx) = tokio::sync::mpsc::channel(128); + + let mut watcher = RecommendedWatcher::new( + move |res: Result| { + if let Err(err) = tx.blocking_send(res) { + tracing::error!(error = %err, "Failed to send file watch event"); + } + }, + Config::default(), + ) + .map_err(to_fs_err)?; + + watcher + .watch(&self.p, RecursiveMode::NonRecursive) .map_err(to_fs_err)?; let dir = self.p.clone(); + let root = self.root.clone(); + Ok(Box::pin(async_stream::stream! { - let mut buffer = [0; 1024]; - let mut events = match inotify.into_event_stream(&mut buffer) { - Ok(events) => events, - Err(err) => { - tracing::error!(error = %err, "Failed getting event stream from inotify"); - return; - } - }; - while let Some(Ok(event)) = events.next().await { - let Some(name) = event.name else { - tracing::warn!("Unexpected event on the directory itself"); - continue; - }; - let item_path = dir.join(name); - let key = match item_path.strip_prefix(&self.root) { - Ok(stripped) => stripped.display().to_string().replace("_", "/"), + // Keep watcher alive for the duration of the stream + let _watcher = watcher; + + while let Some(event_result) = rx.recv().await { + let event = match event_result { + Ok(event) => event, Err(err) => { - // Possibly this should be a panic. - // A key cannot be outside the file store root. - tracing::error!( - error = %err, - item_path = %item_path.display(), - root = %self.root.display(), - "Item in file store is not prefixed with file store root. Should be impossible. Ignoring invalid key."); + tracing::error!(error = %err, "Failed receiving file watch event"); continue; } }; - match event.mask { - EventMask::MODIFY | EventMask::CREATE => { - let data: bytes::Bytes = match fs::read(&item_path) { - Ok(data) => data.into(), - Err(err) => { - tracing::warn!(error = %err, item = %item_path.display(), "Failed reading event item. Skipping."); - continue; - } - }; - let item = KeyValue::new(key, data); - yield WatchEvent::Put(item); - } - EventMask::DELETE => { - yield WatchEvent::Delete(Key::from_raw(key)); - } - event_type => { - tracing::warn!(?event_type, dir = %dir.display(), "Unexpected event type"); + for item_path in event.paths { + // Skip if the event is for the directory itself + if item_path == dir { + tracing::warn!("Unexpected event on the directory itself"); continue; } + + // Canonicalize paths to handle symlinks (e.g., /var -> /private/var on macOS) + let canonical_item_path = match item_path.canonicalize() { + Ok(p) => p, + Err(err) => { + tracing::warn!(error = %err, item = %item_path.display(), "Failed to canonicalize path. Using original path."); + item_path.clone() + } + }; + + let key = match canonical_item_path.strip_prefix(&root) { + Ok(stripped) => stripped.display().to_string().replace("_", "/"), + Err(err) => { + // Possibly this should be a panic. + // A key cannot be outside the file store root. + tracing::error!( + error = %err, + item_path = %canonical_item_path.display(), + root = %root.display(), + "Item in file store is not prefixed with file store root. Should be impossible. Ignoring invalid key."); + continue; + } + }; + + match event.kind { + EventKind::Create(_) | EventKind::Modify(_) => { + let data: bytes::Bytes = match fs::read(&item_path) { + Ok(data) => data.into(), + Err(err) => { + tracing::warn!(error = %err, item = %item_path.display(), "Failed reading event item. Skipping."); + continue; + } + }; + let item = KeyValue::new(key, data); + yield WatchEvent::Put(item); + } + EventKind::Remove(_) => { + yield WatchEvent::Delete(Key::from_raw(key)); + } + event_type => { + tracing::debug!(?event_type, dir = %dir.display(), "Ignoring event type"); + continue; + } + } } } }))