use crate::NativeRootIdentity;
use crate::cancellation::{CancellationError, CancellationToken};
use crate::kernel::{LogicalName, NamespacePath, NamespacePathError};
use crate::model::{FilesystemProfile, VolumeLimits};
use crate::performance::{
MeasuredResult, OperationFailure, OperationReceipt, WorkBudget, WorkCounters, WorkError,
};
use notify::event::{ModifyKind, RenameMode};
use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
use std::collections::BTreeSet;
use std::mem::size_of;
use std::path::{Component, Path, PathBuf};
use std::sync::atomic::{AtomicU32, Ordering};
use std::sync::mpsc::{Receiver, SyncSender, TryRecvError, TrySendError, sync_channel};
use std::sync::{Arc, Condvar, Mutex};
use std::time::{Duration, Instant};
use thiserror::Error;
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum NativeWatchBackend {
LinuxInotify,
MacosFsevents,
WindowsReadDirectoryChanges,
Unsupported,
}
impl NativeWatchBackend {
#[must_use]
pub const fn as_str(self) -> &'static str {
match self {
Self::LinuxInotify => "linux-inotify",
Self::MacosFsevents => "macos-fsevents",
Self::WindowsReadDirectoryChanges => "windows-read-directory-changes",
Self::Unsupported => "unsupported",
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct NativeWatchCapabilities {
pub backend: NativeWatchBackend,
pub recursive: bool,
pub persistent_restart: bool,
pub root_identity_fencing: bool,
}
#[must_use]
pub const fn native_watch_capabilities() -> NativeWatchCapabilities {
#[cfg(target_os = "linux")]
let backend = NativeWatchBackend::LinuxInotify;
#[cfg(target_os = "macos")]
let backend = NativeWatchBackend::MacosFsevents;
#[cfg(target_os = "windows")]
let backend = NativeWatchBackend::WindowsReadDirectoryChanges;
#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))]
let backend = NativeWatchBackend::Unsupported;
NativeWatchCapabilities {
backend,
recursive: !matches!(backend, NativeWatchBackend::Unsupported),
persistent_restart: false,
root_identity_fencing: cfg!(any(unix, windows)),
}
}
#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub struct WatchEpoch(u64);
impl WatchEpoch {
#[must_use]
pub const fn from_u64(value: u64) -> Self {
Self(value)
}
#[must_use]
pub const fn get(self) -> u64 {
self.0
}
}
#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
pub struct WatchSequence(u64);
impl WatchSequence {
#[must_use]
pub const fn from_u64(value: u64) -> Self {
Self(value)
}
#[must_use]
pub const fn get(self) -> u64 {
self.0
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum WatchChange {
Created(NamespacePath),
Modified(NamespacePath),
MetadataChanged(NamespacePath),
Removed(NamespacePath),
Renamed {
from: NamespacePath,
to: NamespacePath,
},
}
#[derive(Clone, Copy, Debug, Eq, Error, PartialEq)]
pub enum WatchInvalidationReason {
#[error("an initial authenticated baseline is required")]
InitialSnapshotRequired,
#[error("the native notification queue overflowed")]
QueueOverflow,
#[error("the native backend requires a rescan")]
NativeRescanRequired,
#[error("the native backend reported an asynchronous error")]
BackendError,
#[error("a native path cannot be represented by volume semantics")]
UnrepresentablePath,
#[error("a native rename could not be paired exactly")]
AmbiguousRename,
#[error("the watched root changed identity")]
RootChanged,
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub enum WatchBatch {
Changes {
epoch: WatchEpoch,
first_sequence: WatchSequence,
next_sequence: WatchSequence,
changes: Vec<WatchChange>,
},
RescanRequired {
epoch: WatchEpoch,
reason: WatchInvalidationReason,
},
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct NativeWatchOptions {
pub limits: VolumeLimits,
pub maximum_queued_changes: u32,
pub recursive: bool,
}
#[cfg(unix)]
fn open_native_root(path: &Path) -> std::io::Result<std::fs::File> {
use std::os::unix::fs::OpenOptionsExt;
let mut options = std::fs::OpenOptions::new();
options
.read(true)
.custom_flags(libc::O_DIRECTORY | libc::O_NOFOLLOW);
options.open(path)
}
#[cfg(windows)]
fn open_native_root(path: &Path) -> std::io::Result<std::fs::File> {
use std::os::windows::fs::MetadataExt;
use std::os::windows::fs::OpenOptionsExt;
use windows::Win32::Storage::FileSystem::{
FILE_ATTRIBUTE_REPARSE_POINT, FILE_FLAG_BACKUP_SEMANTICS, FILE_FLAG_OPEN_REPARSE_POINT,
FILE_SHARE_DELETE, FILE_SHARE_READ, FILE_SHARE_WRITE,
};
let mut options = std::fs::OpenOptions::new();
options
.read(true)
.share_mode(FILE_SHARE_READ.0 | FILE_SHARE_WRITE.0 | FILE_SHARE_DELETE.0)
.custom_flags(FILE_FLAG_BACKUP_SEMANTICS.0 | FILE_FLAG_OPEN_REPARSE_POINT.0);
let file = options.open(path)?;
if file.metadata()?.file_attributes() & FILE_ATTRIBUTE_REPARSE_POINT.0 != 0 {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidInput,
"watch root is a reparse point",
));
}
Ok(file)
}
#[cfg(not(any(unix, windows)))]
fn open_native_root(path: &Path) -> std::io::Result<std::fs::File> {
std::fs::File::open(path)
}
const fn native_filesystem_profile() -> FilesystemProfile {
#[cfg(unix)]
{
FilesystemProfile::Posix
}
#[cfg(windows)]
{
FilesystemProfile::Windows
}
#[cfg(not(any(unix, windows)))]
{
FilesystemProfile::Portable
}
}
impl NativeWatchOptions {
#[must_use]
pub const fn new(limits: VolumeLimits) -> Self {
Self {
limits,
maximum_queued_changes: 16_384,
recursive: true,
}
}
fn validate(self) -> Result<Self, NativeWatchError> {
if self.maximum_queued_changes == 0 {
return Err(NativeWatchError::InvalidOptions);
}
Ok(self)
}
}
#[derive(Debug)]
struct SharedState {
invalidation: Option<WatchInvalidationReason>,
#[cfg(target_os = "linux")]
pending_rename: Option<PendingRename>,
fence: Option<PathBuf>,
excluded: Vec<PathBuf>,
}
#[cfg(target_os = "linux")]
#[derive(Debug)]
struct PendingRename {
tracker: usize,
from: PathBuf,
to: Option<PathBuf>,
}
impl SharedState {
const fn new(invalidation: Option<WatchInvalidationReason>) -> Self {
Self {
invalidation,
#[cfg(target_os = "linux")]
pending_rename: None,
fence: None,
excluded: Vec::new(),
}
}
fn seal_invalidation(&mut self) -> Option<WatchInvalidationReason> {
#[cfg(target_os = "linux")]
if self.invalidation.is_none() && self.pending_rename.is_some() {
self.invalidation = Some(WatchInvalidationReason::AmbiguousRename);
}
self.invalidation
}
}
struct NativeEventContext {
root: PathBuf,
root_identity: NativeRootIdentity,
profile: FilesystemProfile,
limits: VolumeLimits,
#[cfg(target_os = "linux")]
maximum_queued_changes: u32,
}
pub struct NativeWatch {
watcher: RecommendedWatcher,
root: PathBuf,
recursive: bool,
watched_directories: BTreeSet<PathBuf>,
receiver: Receiver<WatchChange>,
queued: Arc<AtomicU32>,
shared: Arc<Mutex<SharedState>>,
fenced: Arc<Condvar>,
fences: u64,
excluded: Vec<(NamespacePath, PathBuf)>,
epoch: WatchEpoch,
next_sequence: WatchSequence,
rescan_in_progress: bool,
root_identity: NativeRootIdentity,
}
impl NativeWatch {
#[must_use]
pub const fn capabilities(&self) -> NativeWatchCapabilities {
native_watch_capabilities()
}
pub fn open(
root: impl AsRef<Path>,
options: NativeWatchOptions,
) -> Result<Self, NativeWatchError> {
Self::open_with_profile(root, native_filesystem_profile(), options)
}
pub fn open_with_profile(
root: impl AsRef<Path>,
profile: FilesystemProfile,
options: NativeWatchOptions,
) -> Result<Self, NativeWatchError> {
let options = options.validate()?;
let requested_root = root.as_ref();
let admission = requested_root
.symlink_metadata()
.map_err(|error| NativeWatchError::Io(error.to_string()))?;
if admission.file_type().is_symlink() {
return Err(NativeWatchError::RootIsNotDirectory);
}
let root = requested_root
.canonicalize()
.map_err(|error| NativeWatchError::Io(error.to_string()))?;
let root_file =
open_native_root(&root).map_err(|error| NativeWatchError::Io(error.to_string()))?;
let metadata = root_file
.metadata()
.map_err(|error| NativeWatchError::Io(error.to_string()))?;
if !metadata.is_dir() {
return Err(NativeWatchError::RootIsNotDirectory);
}
let root_identity = NativeRootIdentity::from_file(&root_file)
.map_err(|error| NativeWatchError::Io(error.to_string()))?;
let capacity = usize::try_from(options.maximum_queued_changes)
.map_err(|_| NativeWatchError::InvalidOptions)?;
let (sender, receiver) = sync_channel(capacity);
let queued = Arc::new(AtomicU32::new(0));
let shared = Arc::new(Mutex::new(SharedState::new(Some(
WatchInvalidationReason::InitialSnapshotRequired,
))));
let fenced = Arc::new(Condvar::new());
let callback_shared = Arc::clone(&shared);
let callback_fenced = Arc::clone(&fenced);
let callback_queued = Arc::clone(&queued);
let callback_context = NativeEventContext {
root: root.clone(),
root_identity,
profile,
limits: options.limits,
#[cfg(target_os = "linux")]
maximum_queued_changes: options.maximum_queued_changes,
};
let mut watcher = notify::recommended_watcher(move |event: notify::Result<Event>| {
accept_native_event(
&event,
&callback_context,
&sender,
&callback_shared,
&callback_queued,
);
if let (Ok(event), Ok(mut state)) = (&event, callback_shared.lock())
&& state
.fence
.as_ref()
.is_some_and(|fence| event.paths.contains(fence))
{
state.fence = None;
callback_fenced.notify_all();
}
})
.map_err(|error| NativeWatchError::Backend(error.to_string()))?;
watcher
.watch(
&root,
if options.recursive {
RecursiveMode::Recursive
} else {
RecursiveMode::NonRecursive
},
)
.map_err(|error| NativeWatchError::Backend(error.to_string()))?;
Ok(Self {
watcher,
watched_directories: BTreeSet::from([root.clone()]),
root,
recursive: options.recursive,
receiver,
queued,
shared,
fenced,
fences: 0,
excluded: Vec::new(),
epoch: WatchEpoch(0),
next_sequence: WatchSequence(0),
rescan_in_progress: false,
root_identity,
})
}
pub fn watch_directory(&mut self, directory: &NamespacePath) -> Result<(), NativeWatchError> {
if self.recursive || directory.is_root() {
return Ok(());
}
let relative = crate::native_capture::namespace_to_host_path(directory)
.map_err(|_| NativeWatchError::UnrepresentablePath)?;
let mut candidate = self.root.clone();
for component in relative.components() {
let Component::Normal(component) = component else {
return Err(NativeWatchError::UnrepresentablePath);
};
candidate.push(component);
let metadata = match candidate.symlink_metadata() {
Ok(metadata) => metadata,
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
candidate.pop();
break;
}
Err(error) => return Err(NativeWatchError::Io(error.to_string())),
};
if metadata.file_type().is_symlink() {
return Err(NativeWatchError::SymbolicLinkDirectory);
}
if !metadata.is_dir() {
return Err(NativeWatchError::ObservedPathNotDirectory);
}
}
if self.watched_directories.contains(&candidate) {
return Ok(());
}
self.watcher
.watch(&candidate, RecursiveMode::NonRecursive)
.map_err(|error| NativeWatchError::Backend(error.to_string()))?;
self.watched_directories.insert(candidate);
Ok(())
}
fn forget_removed_directories(&mut self, changes: &[WatchChange]) {
if self.recursive {
return;
}
let mut removed = Vec::new();
for change in changes {
match change {
WatchChange::Removed(path) => removed.push(path),
WatchChange::Renamed { from, to } => {
removed.push(from);
removed.push(to);
}
WatchChange::Created(_)
| WatchChange::Modified(_)
| WatchChange::MetadataChanged(_) => {}
}
}
for path in removed {
let Ok(relative) = crate::native_capture::namespace_to_host_path(path) else {
continue;
};
let removed = self.root.join(relative);
let stale = self
.watched_directories
.iter()
.filter(|watched| **watched != self.root && watched.starts_with(&removed))
.cloned()
.collect::<Vec<_>>();
for watched in stale {
let _ = self.watcher.unwatch(&watched);
self.watched_directories.remove(&watched);
}
}
}
fn reset_demand_watches(&mut self) {
if self.recursive {
return;
}
let stale = self
.watched_directories
.iter()
.filter(|path| **path != self.root)
.cloned()
.collect::<Vec<_>>();
for path in stale {
let _ = self.watcher.unwatch(&path);
self.watched_directories.remove(&path);
}
}
pub fn accept_lazy_baseline(&mut self) -> Result<WatchBatch, NativeWatchError> {
self.begin_rescan()?;
self.finish_rescan()
}
#[must_use]
pub const fn root_identity(&self) -> NativeRootIdentity {
self.root_identity
}
pub fn verify_root_identity(
&self,
observed: NativeRootIdentity,
) -> Result<(), NativeWatchError> {
if observed == self.root_identity {
return Ok(());
}
self.shared
.lock()
.map_err(|_| NativeWatchError::StatePoisoned)?
.invalidation = Some(WatchInvalidationReason::RootChanged);
Err(NativeWatchError::RootChanged)
}
pub fn begin_rescan(&mut self) -> Result<WatchEpoch, NativeWatchError> {
if self.rescan_in_progress {
return Err(NativeWatchError::RescanAlreadyInProgress);
}
let mut shared = self
.shared
.lock()
.map_err(|_| NativeWatchError::StatePoisoned)?;
while self.receiver.try_recv().is_ok() {
self.queued.fetch_sub(1, Ordering::AcqRel);
}
shared.invalidation = None;
#[cfg(target_os = "linux")]
{
shared.pending_rename = None;
}
self.epoch = WatchEpoch(
self.epoch
.0
.checked_add(1)
.ok_or(NativeWatchError::SequenceExhausted)?,
);
self.next_sequence = WatchSequence(0);
drop(shared);
self.reset_demand_watches();
self.rescan_in_progress = true;
Ok(self.epoch)
}
#[cfg(target_os = "windows")]
fn accept_verified_baseline<T, E>(
&mut self,
verify: impl FnOnce() -> Result<Option<T>, E>,
) -> Result<Result<Option<T>, E>, NativeWatchError> {
if self.rescan_in_progress {
return Err(NativeWatchError::RescanAlreadyInProgress);
}
let mut shared = self
.shared
.lock()
.map_err(|_| NativeWatchError::StatePoisoned)?;
if shared.invalidation != Some(WatchInvalidationReason::InitialSnapshotRequired)
|| self.epoch.0 != 0
{
return Err(NativeWatchError::RescanAlreadyInProgress);
}
let verified = match verify() {
Ok(Some(verified)) => verified,
Ok(None) => return Ok(Ok(None)),
Err(error) => return Ok(Err(error)),
};
self.epoch = WatchEpoch(
self.epoch
.0
.checked_add(1)
.ok_or(NativeWatchError::SequenceExhausted)?,
);
self.next_sequence = WatchSequence(0);
shared.invalidation = None;
Ok(Ok(Some(verified)))
}
#[cfg(target_os = "windows")]
pub fn accept_windows_usn_baseline(
&mut self,
checkpoint: crate::WindowsUsnCheckpoint,
) -> Result<Result<crate::WindowsUsnContinuity, crate::WindowsUsnError>, NativeWatchError> {
let root = self.root.clone();
self.accept_verified_baseline(|| {
crate::validate_windows_usn_checkpoint(&root, checkpoint).map(|continuity| {
(continuity == crate::WindowsUsnContinuity::Unchanged).then_some(continuity)
})
})
.map(|result| {
result.map(|admitted| {
admitted.unwrap_or(crate::WindowsUsnContinuity::BaselineRequired(
crate::WindowsUsnDiscontinuity::VolumeAdvanced,
))
})
})
}
pub fn finish_rescan(&mut self) -> Result<WatchBatch, NativeWatchError> {
if !self.rescan_in_progress {
return Err(NativeWatchError::NoRescanInProgress);
}
let observed = open_native_root(&self.root)
.and_then(|root| NativeRootIdentity::from_file(&root))
.ok();
if observed != Some(self.root_identity) {
self.shared
.lock()
.map_err(|_| NativeWatchError::StatePoisoned)?
.invalidation = Some(WatchInvalidationReason::RootChanged);
}
self.rescan_in_progress = false;
let mut shared = self
.shared
.lock()
.map_err(|_| NativeWatchError::StatePoisoned)?;
Ok(shared.seal_invalidation().map_or(
WatchBatch::Changes {
epoch: self.epoch,
first_sequence: self.next_sequence,
next_sequence: self.next_sequence,
changes: Vec::new(),
},
|reason| WatchBatch::RescanRequired {
epoch: self.epoch,
reason,
},
))
}
pub fn abort_rescan(
&mut self,
reason: WatchInvalidationReason,
) -> Result<(), NativeWatchError> {
if !self.rescan_in_progress {
return Err(NativeWatchError::NoRescanInProgress);
}
self.rescan_in_progress = false;
self.shared
.lock()
.map_err(|_| NativeWatchError::StatePoisoned)?
.invalidation = Some(reason);
Ok(())
}
pub fn exclude_subtrees(&mut self, subtrees: &[NamespacePath]) -> Result<(), NativeWatchError> {
let excluded = subtrees
.iter()
.map(|subtree| {
if subtree.is_root() {
return Err(NativeWatchError::UnrepresentablePath);
}
crate::native_capture::namespace_to_host_path(subtree)
.map(|relative| self.root.join(relative))
.map_err(|_| NativeWatchError::UnrepresentablePath)
})
.collect::<Result<Vec<_>, _>>()?;
self.shared
.lock()
.map_err(|_| NativeWatchError::StatePoisoned)?
.excluded
.clone_from(&excluded);
self.excluded = subtrees.iter().cloned().zip(excluded).collect();
Ok(())
}
pub fn fence(&mut self, timeout: Duration) -> Result<(), NativeWatchError> {
let delivered = self.fence_cookie().is_ok_and(|cookie| {
self.await_cookie(&cookie, timeout)
.is_ok_and(|delivered| delivered)
});
let mut state = self
.shared
.lock()
.map_err(|_| NativeWatchError::StatePoisoned)?;
state.fence = None;
if !delivered && state.invalidation.is_none() {
state.invalidation = Some(WatchInvalidationReason::NativeRescanRequired);
}
Ok(())
}
fn fence_cookie(&mut self) -> Result<PathBuf, NativeWatchError> {
let (directory, host_directory) = self.fence_directory()?;
self.watch_directory(&directory)?;
self.fences = self.fences.wrapping_add(1);
Ok(host_directory.join(format!(
".acyclic-fence-{}-{}",
std::process::id(),
self.fences
)))
}
fn await_cookie(&self, cookie: &Path, timeout: Duration) -> Result<bool, NativeWatchError> {
let mut state = self
.shared
.lock()
.map_err(|_| NativeWatchError::StatePoisoned)?;
if state.invalidation.is_some() {
return Ok(false);
}
state.fence = Some(cookie.to_path_buf());
drop(state);
std::fs::File::create_new(cookie)
.and_then(|_| std::fs::remove_file(cookie))
.map_err(|error| NativeWatchError::Io(error.to_string()))?;
let deadline = Instant::now() + timeout;
let mut state = self
.shared
.lock()
.map_err(|_| NativeWatchError::StatePoisoned)?;
while state.fence.is_some() && state.invalidation.is_none() {
let remaining = deadline.saturating_duration_since(Instant::now());
if remaining.is_zero() {
return Ok(false);
}
state = self
.fenced
.wait_timeout(state, remaining)
.map_err(|_| NativeWatchError::StatePoisoned)?
.0;
}
Ok(state.invalidation.is_none())
}
fn fence_directory(&self) -> Result<(NamespacePath, PathBuf), NativeWatchError> {
for (subtree, host) in &self.excluded {
match host.symlink_metadata() {
Ok(metadata) if metadata.is_dir() => return Ok((subtree.clone(), host.clone())),
Ok(_) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(NativeWatchError::Io(error.to_string())),
}
}
let (subtree, host) = self
.excluded
.first()
.cloned()
.ok_or(NativeWatchError::InvalidOptions)?;
std::fs::create_dir(&host).map_err(|error| NativeWatchError::Io(error.to_string()))?;
Ok((subtree, host))
}
pub fn poll(
&mut self,
maximum_changes: u32,
budget: WorkBudget,
cancellation: &CancellationToken,
) -> MeasuredResult<OperationReceipt<WatchBatch>, NativeWatchError> {
cancellation
.check()
.map_err(|error| OperationFailure::before_work(error.into()))?;
if maximum_changes == 0 {
return Err(OperationFailure::before_work(
NativeWatchError::ZeroPollLimit,
));
}
if self.rescan_in_progress {
return Err(OperationFailure::before_work(
NativeWatchError::RescanInProgress,
));
}
if let Some(reason) = self.invalidation()? {
return Ok(OperationReceipt {
value: WatchBatch::RescanRequired {
epoch: self.epoch,
reason,
},
work: WorkCounters::default(),
});
}
let root_work = WorkCounters {
backend_read_operations: 1,
..WorkCounters::default()
};
root_work
.verify(budget)
.map_err(|error| OperationFailure::before_work(error.into()))?;
self.verify_current_root()?;
if let Some(reason) = self.invalidation()? {
return Ok(OperationReceipt {
value: WatchBatch::RescanRequired {
epoch: self.epoch,
reason,
},
work: root_work,
});
}
self.poll_contiguous(maximum_changes, budget, root_work)
}
fn poll_contiguous(
&mut self,
maximum_changes: u32,
budget: WorkBudget,
initial_work: WorkCounters,
) -> MeasuredResult<OperationReceipt<WatchBatch>, NativeWatchError> {
let queued = self.queued.load(Ordering::Acquire).min(maximum_changes);
if queued == 0 {
return Ok(OperationReceipt {
value: WatchBatch::Changes {
epoch: self.epoch,
first_sequence: self.next_sequence,
next_sequence: self.next_sequence,
changes: Vec::new(),
},
work: initial_work,
});
}
let capacity = usize::try_from(queued)
.map_err(|_| OperationFailure::before_work(NativeWatchError::InvalidOptions))?;
let allocation_bytes = capacity
.checked_mul(size_of::<WatchChange>())
.map(crate::foundation::usize_to_u64)
.ok_or_else(|| {
OperationFailure::before_work(NativeWatchError::Work(WorkError::Overflow))
})?;
let mut work = initial_work
.checked_add(WorkCounters {
allocation_operations: 1,
peak_allocation_bytes: allocation_bytes,
..WorkCounters::default()
})
.map_err(|error| OperationFailure::before_work(error.into()))?;
work.verify(budget)
.map_err(|error| OperationFailure::before_work(error.into()))?;
let mut changes = Vec::new();
changes
.try_reserve_exact(capacity)
.map_err(|_| OperationFailure::new(NativeWatchError::AllocationFailed, work))?;
while changes.len() < capacity {
match self.receiver.try_recv() {
Ok(change) => {
self.queued.fetch_sub(1, Ordering::AcqRel);
changes.push(change);
}
Err(TryRecvError::Empty) => break,
Err(TryRecvError::Disconnected) => {
self.invalidate(WatchInvalidationReason::BackendError)?;
break;
}
}
}
if let Some(reason) = self.invalidation()? {
return Ok(OperationReceipt {
value: WatchBatch::RescanRequired {
epoch: self.epoch,
reason,
},
work,
});
}
let count = crate::foundation::usize_to_u64(changes.len());
let next = self
.next_sequence
.0
.checked_add(count)
.ok_or_else(|| OperationFailure::new(NativeWatchError::SequenceExhausted, work))?;
work = work
.checked_add(WorkCounters {
items_examined: count,
items_returned: count,
..WorkCounters::default()
})
.map_err(|error| OperationFailure::new(error.into(), work))?;
work.verify(budget)
.map_err(|error| OperationFailure::new(error.into(), work))?;
let first_sequence = self.next_sequence;
self.next_sequence = WatchSequence(next);
self.forget_removed_directories(&changes);
Ok(OperationReceipt {
value: WatchBatch::Changes {
epoch: self.epoch,
first_sequence,
next_sequence: self.next_sequence,
changes,
},
work,
})
}
fn verify_current_root(&self) -> Result<(), OperationFailure<NativeWatchError>> {
let observed = open_native_root(&self.root)
.and_then(|root| NativeRootIdentity::from_file(&root))
.ok();
if observed == Some(self.root_identity) {
return Ok(());
}
self.shared
.lock()
.map_err(|_| OperationFailure::before_work(NativeWatchError::StatePoisoned))?
.invalidation = Some(WatchInvalidationReason::RootChanged);
Ok(())
}
fn invalidation(
&self,
) -> Result<Option<WatchInvalidationReason>, OperationFailure<NativeWatchError>> {
self.shared
.lock()
.map(|mut state| state.seal_invalidation())
.map_err(|_| OperationFailure::before_work(NativeWatchError::StatePoisoned))
}
fn invalidate(
&self,
reason: WatchInvalidationReason,
) -> Result<(), OperationFailure<NativeWatchError>> {
let mut shared = self
.shared
.lock()
.map_err(|_| OperationFailure::before_work(NativeWatchError::StatePoisoned))?;
shared.invalidation.get_or_insert(reason);
Ok(())
}
}
impl crate::demand::DemandDirectoryObserver for Mutex<NativeWatch> {
fn observe_directory(
&self,
directory: &NamespacePath,
) -> Result<(), crate::demand::DemandError> {
self.lock()
.map_err(|_| NativeWatchError::StatePoisoned)?
.watch_directory(directory)?;
Ok(())
}
}
fn accept_native_event(
event: ¬ify::Result<Event>,
context: &NativeEventContext,
sender: &SyncSender<WatchChange>,
shared: &Arc<Mutex<SharedState>>,
queued: &Arc<AtomicU32>,
) {
let Ok(mut state) = shared.lock() else {
return;
};
if state.invalidation.is_some() {
return;
}
if let Ok(event) = event
&& !event.paths.is_empty()
&& event.paths.iter().all(|path| {
state
.excluded
.iter()
.any(|excluded| path.starts_with(excluded))
})
{
return;
}
let mapped = event
.as_ref()
.map_err(|_| WatchInvalidationReason::BackendError)
.and_then(|event| map_native_event(event, context));
let changes = match mapped {
Ok(MappedNativeEvent::Changes(changes)) => changes,
#[cfg(target_os = "linux")]
Ok(MappedNativeEvent::RenameHalf {
tracker,
mode,
path,
}) => {
match mode {
RenameMode::From if state.pending_rename.is_none() => {
if queued.load(Ordering::Acquire) >= context.maximum_queued_changes {
state.invalidation = Some(WatchInvalidationReason::QueueOverflow);
return;
}
state.pending_rename = Some(PendingRename {
tracker,
from: path,
to: None,
});
}
RenameMode::To => {
let Some(pending) = state.pending_rename.as_mut() else {
state.invalidation = Some(WatchInvalidationReason::AmbiguousRename);
return;
};
if pending.tracker != tracker || pending.to.is_some() {
state.invalidation = Some(WatchInvalidationReason::AmbiguousRename);
} else {
pending.to = Some(path);
}
}
_ => state.invalidation = Some(WatchInvalidationReason::AmbiguousRename),
}
return;
}
#[cfg(target_os = "linux")]
Ok(MappedNativeEvent::RenameBoth {
tracker,
from,
to,
changes,
}) => {
let matched = state.pending_rename.as_ref().is_some_and(|pending| {
pending.tracker == tracker
&& pending.from == from
&& pending.to.as_ref() == Some(&to)
});
if !matched {
state.invalidation = Some(WatchInvalidationReason::AmbiguousRename);
return;
}
state.pending_rename = None;
changes
}
Err(reason) => {
state.invalidation = Some(reason);
return;
}
};
for change in changes {
#[cfg(target_os = "linux")]
if queued
.load(Ordering::Acquire)
.saturating_add(u32::from(state.pending_rename.is_some()))
>= context.maximum_queued_changes
{
state.invalidation = Some(WatchInvalidationReason::QueueOverflow);
break;
}
match sender.try_send(change) {
Ok(()) => {
queued.fetch_add(1, Ordering::Release);
}
Err(TrySendError::Full(_) | TrySendError::Disconnected(_)) => {
state.invalidation = Some(WatchInvalidationReason::QueueOverflow);
break;
}
}
}
}
enum MappedNativeEvent {
Changes(Vec<WatchChange>),
#[cfg(target_os = "linux")]
RenameHalf {
tracker: usize,
mode: RenameMode,
path: PathBuf,
},
#[cfg(target_os = "linux")]
RenameBoth {
tracker: usize,
from: PathBuf,
to: PathBuf,
changes: Vec<WatchChange>,
},
}
fn map_native_event(
event: &Event,
context: &NativeEventContext,
) -> Result<MappedNativeEvent, WatchInvalidationReason> {
validate_replayed_root_creation(event, &context.root, context.root_identity)?;
#[cfg(target_os = "linux")]
if let EventKind::Modify(ModifyKind::Name(mode)) = event.kind {
if event.need_rescan() {
return Err(WatchInvalidationReason::NativeRescanRequired);
}
let tracker = event
.attrs
.tracker()
.ok_or(WatchInvalidationReason::AmbiguousRename)?;
if matches!(mode, RenameMode::From | RenameMode::To) {
let [path] = event.paths.as_slice() else {
return Err(WatchInvalidationReason::AmbiguousRename);
};
let relative =
relative_namespace_path(&context.root, path, context.profile, context.limits)?;
if relative.is_root() {
return Err(WatchInvalidationReason::RootChanged);
}
return Ok(MappedNativeEvent::RenameHalf {
tracker,
mode,
path: path.clone(),
});
}
if mode == RenameMode::Both {
let changes = map_event(event, &context.root, context.profile, context.limits)?;
let [from, to] = event.paths.as_slice() else {
return Err(WatchInvalidationReason::AmbiguousRename);
};
return Ok(MappedNativeEvent::RenameBoth {
tracker,
from: from.clone(),
to: to.clone(),
changes,
});
}
}
map_event(event, &context.root, context.profile, context.limits).map(MappedNativeEvent::Changes)
}
fn validate_replayed_root_creation(
event: &Event,
root: &Path,
admitted_root_identity: NativeRootIdentity,
) -> Result<(), WatchInvalidationReason> {
#[cfg(target_os = "macos")]
if event.paths.iter().any(|path| path == root) && matches!(event.kind, EventKind::Create(_)) {
let observed = open_native_root(root)
.and_then(|file| NativeRootIdentity::from_file(&file))
.ok();
if observed != Some(admitted_root_identity) {
return Err(WatchInvalidationReason::RootChanged);
}
}
#[cfg(not(target_os = "macos"))]
let _ = (event, root, admitted_root_identity);
Ok(())
}
fn map_event(
event: &Event,
root: &Path,
profile: FilesystemProfile,
limits: VolumeLimits,
) -> Result<Vec<WatchChange>, WatchInvalidationReason> {
if event.need_rescan() {
return Err(WatchInvalidationReason::NativeRescanRequired);
}
if matches!(event.kind, EventKind::Access(_)) {
return Ok(Vec::new());
}
if event.paths.is_empty() || event.paths.len() > 2 {
return Err(WatchInvalidationReason::UnrepresentablePath);
}
let mut paths = Vec::new();
paths
.try_reserve_exact(event.paths.len())
.map_err(|_| WatchInvalidationReason::UnrepresentablePath)?;
for path in &event.paths {
#[cfg(target_os = "macos")]
if path == root && matches!(event.kind, EventKind::Create(_)) {
continue;
}
paths.push(relative_namespace_path(root, path, profile, limits)?);
}
if paths.iter().any(NamespacePath::is_root)
&& matches!(
event.kind,
EventKind::Remove(_) | EventKind::Modify(ModifyKind::Name(_))
)
{
return Err(WatchInvalidationReason::RootChanged);
}
#[cfg(target_os = "macos")]
if paths.iter().any(NamespacePath::is_root) {
return Err(WatchInvalidationReason::NativeRescanRequired);
}
if let EventKind::Modify(ModifyKind::Name(mode)) = event.kind {
return match (mode, paths.as_slice()) {
(RenameMode::Both, [from, to]) => Ok(vec![WatchChange::Renamed {
from: from.clone(),
to: to.clone(),
}]),
#[cfg(target_os = "linux")]
_ => Err(WatchInvalidationReason::AmbiguousRename),
#[cfg(not(target_os = "linux"))]
(_, paths) => Ok(paths.iter().cloned().map(WatchChange::Modified).collect()),
};
}
let mut changes = Vec::new();
changes
.try_reserve_exact(paths.len())
.map_err(|_| WatchInvalidationReason::UnrepresentablePath)?;
for path in paths {
changes.push(match event.kind {
#[cfg(target_os = "macos")]
EventKind::Create(_) => WatchChange::Modified(path),
#[cfg(not(target_os = "macos"))]
EventKind::Create(_) => WatchChange::Created(path),
EventKind::Remove(_) => WatchChange::Removed(path),
#[cfg(target_os = "macos")]
EventKind::Modify(ModifyKind::Metadata(_)) => WatchChange::Modified(path),
#[cfg(not(target_os = "macos"))]
EventKind::Modify(ModifyKind::Metadata(_)) => WatchChange::MetadataChanged(path),
EventKind::Modify(_) | EventKind::Any | EventKind::Other => WatchChange::Modified(path),
EventKind::Access(_) => continue,
});
}
Ok(changes)
}
fn relative_namespace_path(
root: &Path,
path: &Path,
profile: FilesystemProfile,
limits: VolumeLimits,
) -> Result<NamespacePath, WatchInvalidationReason> {
let relative = path
.strip_prefix(root)
.map_err(|_| WatchInvalidationReason::UnrepresentablePath)?;
let mut components = Vec::new();
for component in relative.components() {
let Component::Normal(name) = component else {
return Err(WatchInvalidationReason::UnrepresentablePath);
};
let (encoding, bytes) =
crate::native_name::host_name_bytes(name, profile, limits.maximum_component_bytes)
.map_err(|_| WatchInvalidationReason::UnrepresentablePath)?;
let logical = LogicalName::new(encoding, bytes, limits.maximum_component_bytes)
.map_err(|_| WatchInvalidationReason::UnrepresentablePath)?;
components.push(logical);
}
NamespacePath::new(components, limits)
.map_err(|_error: NamespacePathError| WatchInvalidationReason::UnrepresentablePath)
}
#[derive(Debug, Error)]
pub enum NativeWatchError {
#[error("native watcher options are invalid")]
InvalidOptions,
#[error("native watcher root is not a directory")]
RootIsNotDirectory,
#[error("native watcher directory traverses a symbolic link")]
SymbolicLinkDirectory,
#[error("native watcher observed path is not a directory")]
ObservedPathNotDirectory,
#[error("native watcher directory is not representable on this host")]
UnrepresentablePath,
#[error("native watcher root changed identity")]
RootChanged,
#[error("native watcher I/O failed: {0}")]
Io(String),
#[error("native watcher backend failed: {0}")]
Backend(String),
#[error("native watcher state is poisoned")]
StatePoisoned,
#[error("native watcher baseline scan is already active")]
RescanAlreadyInProgress,
#[error("native watcher baseline scan is not active")]
NoRescanInProgress,
#[error("native watcher baseline scan is in progress")]
RescanInProgress,
#[error("native watcher poll limit must be non-zero")]
ZeroPollLimit,
#[error("native watcher sequence is exhausted")]
SequenceExhausted,
#[error("native watcher output allocation failed")]
AllocationFailed,
#[error(transparent)]
Cancelled(#[from] CancellationError),
#[error(transparent)]
Work(#[from] WorkError),
}
#[cfg(test)]
#[path = "tests/watch.rs"]
mod tests;