mooncake-track
/
dynamo
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: use notify instead of inotify (#4250) 

Signed-off-by: PeaBrane <yanrpei@gmail.com>
This commit is contained in:
Yan Ru Pei 2025-11-12 11:53:39 -05:00 committed by GitHub
parent 3ea22fcf39
commit 38242c8d7b
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 282 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

70
Cargo.lock generated
View File

@ -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"

148
lib/bindings/python/Cargo.lock generated
View File

@ -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"

2
lib/runtime/Cargo.toml
View File

@ -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"] }

124
lib/runtime/src/storage/key_value_store/file.rs
View File

@ -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<Pin<Box<dyn futures::Stream<Item = WatchEvent> + 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<Event, notify::Error>| {
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;
}
}
}
}
}))