// Copyright 2025 International Digital Economy Academy
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
///|
using @event_loop {type FileIdentity}
///|
/// A file system event generated by `@fs.Watcher`.
///
/// - `Modify(path)`: the regular file at `path` has been modified
/// - `Create(path)`: a new file or directory is created at `path`
/// - `Remove(path)`: the file or directory at `path` is removed
/// - `Rename(old~, new~)`, the file previously at `old` has been moved to `new`
///
/// All paths are relative to the base directory of the watcher, using `/` as path separator.
pub(all) enum FsEvent {
Modify(String)
Create(String)
Remove(String)
Rename(old~ : String, new~ : String)
} derive(Eq, Debug)
///|
/// Info about a watched file in the watched tree.
/// There are three stages in each `wait` iteration of the watcher:
///
/// - The collect pass: when the watcher is idle, the backend silently
/// collect file system changes in the background.
/// It may, from time to time, mark modified files and directories as dirty.
/// This marking is implemented through the `modified` and `children_to_scan` fields:
///
/// * `modified`: the file/directory itself is modified.
/// For directory this mean the list of files in the directory has changed
/// * `children_to_scan`: for directory only. Some children in the sub tree is modified
///
/// - The scan pass, implemented through `Watcher::synchronize_tree`.
/// In this pass, we scan those modified directories to collect up-to-date knowledge
/// on the structure of the watched tree.
/// Since many operations in the scan pass are path-based,
/// the scan must be performed from parent to child.
/// The dirty mark in the collect pass is used to avoid unnecessary scanning.
///
/// During the scan pass, new changes may happen concurrently on the file system.
/// So the scan pass is run in parallel with the collect pass.
/// When new events arrive during scanning, it is possible that
/// a previously scanned directory will be marked as dirty again.
/// The scan pass must be capable of handling this case.
/// In addition, the scan pass must be also capable of handling scan failure,
/// in order to handle file system changes that happen concurrently with scanning.
///
/// The scan pass consumes and clears `modified` and `children_to_scan`,
/// and record discovered changes in `changed_entries` and `children_with_events`,
/// which are used by the next report pass:
///
/// * `changed_entries`: for directory only. Record the list of locations in the directory
/// that have changed since the last query. The value store for each location is the
/// ID of the old file on that location, or `None` if the file previously does not exist.
/// * `children_with_events`: for directory only.
/// Record the list of children that contain any event to report.
///
/// The watcher maintain a tree snapshot of the file system through the `children` field,
/// The state of the this snapshot, and the meaning of other related fields
/// such as `children_with_events`, should always be in-sync with
/// the latest progress of the scan pass.
///
/// - The report pass. This pass consumes `changed_entries` and `children_with_events`,
/// and report the list of events for the user, through the return value of `wait()`.
/// Event reporting is not performed during the scan pass,
/// because the scan pass may need to rescan already-scanned directory from time to time,
/// while we want to make the returned event list minimal.
/// The report pass is implemented though `emit_events_for_file`.
/// Since we report events in a path-based manner, the report pass also need to operate
/// from parent to children, that's why we need to maintain `children_with_events`
/// instead of a global list.
priv struct WatchedFile {
/// A platform dependent "watch-descriptor":
/// - for `inotify`, this is the inotify watch descriptor of the file
/// - for `kqueue`, this is just a open handle to the file
/// - for Windows, this field is meaningless
wd : @fd_util.Fd
/// The identity of this file.
///
/// Mount point note: for mount points/symbolic links/reparse points,
/// `identity` is the file identity of the directory entry in the parent,
/// not the actual physical identity of the mounted file system/redirected file.
identity : FileIdentity
paths : Set[(FileIdentity, String)]
is_dir : Bool
/// Whether the file itself has been modified since the last query.
/// For directory this means the list of files have changed.
/// The collect pass set this field while the scan pass reset it after scanning.
/// The status of this list should be based on the latest progress of the scan pass.
mut modified : Bool
// the following fields are for directory only
/// The children of a directory, indexed by their file name.
/// The status of this map should be based on the latest progress of the scan pass.
children : Map[String, FileIdentity]
/// The list of children that still need scanning in the scan pass.
/// The collect pass add members to this list and the scan pass remove members from it.
/// The status of this list should be based on the latest progress of the scan pass.
children_to_scan : Map[String, FileIdentity]
/// The list of entries whose content have changed since the last query,
/// as discovered by the scan pass.
/// For entry name the identity of the old file at this location
/// (or `None` if the entry did not previously exist) is stored.
/// The scan pass set this list, while the report pass consume and clear this list
changed_entries : Map[String, FileIdentity?]
/// The list of children (recursively) that contain event for the report pass.
/// The scan pass set this list, while the report pass consume and clear this list
children_with_events : Map[String, FileIdentity]
}
///|
priv trait WatcherBackend {
fn close(Self) -> Unit
async fn wait(Self) -> Unit
async fn add_file(
Self,
StringView,
is_dir~ : Bool,
identity~ : FileIdentity,
context~ : String,
) -> @fd_util.Fd
fn remove_file(Self, @fd_util.Fd, context~ : String) -> Unit raise
}
///|
/// A file system watcher for watching changes in a directory.
struct Watcher {
mut backend : &WatcherBackend?
root_id : FileIdentity
base_path : String
watched : Map[FileIdentity, WatchedFile]
pending_remove : Set[FileIdentity]
ignored_paths : (String) -> Bool
report_child_event : Bool
mut synchronize_task : @coroutine.Coroutine?
}
///|
pub fn Watcher::close(self : Watcher) -> Unit {
if self.backend is Some(backend) {
self.backend = None
backend.close()
}
if self.synchronize_task is Some(task) {
task.cancel()
}
}
///|
#cfg(platform="windows")
fn Watcher::new_backend(
self : Watcher,
root : @event_loop.IoHandle,
root_id~ : FileIdentity,
debounce_timeout~ : Int,
max_debounce_delay~ : Int,
context~ : String,
) -> &WatcherBackend {
ignore(context)
WindowsWatcher(self, root, root_id~, debounce_timeout~, max_debounce_delay~)
}
///|
#cfg(not(platform="windows"))
fn Watcher::new_backend(
self : Watcher,
root : @event_loop.IoHandle,
root_id~ : FileIdentity,
debounce_timeout~ : Int,
max_debounce_delay~ : Int,
context~ : String,
) -> &WatcherBackend raise {
ignore(root_id)
root.close()
match @event_loop.platform {
Linux =>
InotifyWatcher(self, debounce_timeout~, max_debounce_delay~, context~)
as &WatcherBackend
MacOS =>
KqueueWatcher(self, debounce_timeout~, max_debounce_delay~, context~)
as &WatcherBackend
Windows => panic()
}
}
///|
/// Create a new file system watcher that watches the directory `path` recursively.
/// Currently only recursive directory watching is supported.
///
/// When files are created, removed, modified or renamed inside `path`,
/// the watcher will get notified and report events for the change.
/// New files will be watched automatically, and removed files will be unwatched automatically.
///
/// File system events often come in batch.
/// For example, removing a directory recursively will result in several removal event in a row.
/// By default `Watcher` will perform debouncing internally:
/// when many events are happening in sequence, instead of immediately return on the first event,
/// `Watcher::wait_any` will wait until no event is happening
/// within the last `debounce_timeout` ms (defaults to 20ms),
/// or if `max_debounce_delay` ms (defaults to 200ms) has elapsed since the first event.
///
/// `ignored_paths`, if present, can be used to filter out paths that the user don't want to watch.
/// When a file or directory is going to be watched, its path
/// (relative to root of watched tree, using `/` as path separator) will be supplied to `ignored_paths`.
/// If `ignored_paths` return `true`, the file or directory will be ignored.
/// When a directory is ignored, all files/directories within it are also ignored.
/// So to ignore a single directory,
/// `ignored_paths` only need to handle paths of the files inside that ignored directory.
///
/// When a create/remove event is reported for a directory:
/// - if `report_child_event=true`, respective create/remove events will be emitted
/// for everything inside that directory.
/// This mode is useful if tracking the exact list of files in desirable.
/// - If `report_child_event=false` (the default),
/// only a single event for the directory itself will be emitted,
/// making the watcher less noisy
///
/// If `report_event_on_init=true` (`false` by default),
/// the first `wait` call will return immediately after watcher creation,
/// reporting events describing the initial structure of the watched directory.
/// This is useful for keeping the knowledge of the caller in sync with the watcher.
/// Note that you probably want to set `report_child_event=true` as well in this case.
///
/// Currently, the behavior when `path` itself is renamed or removed is undefined.
pub async fn Watcher::Watcher(
path : String,
debounce_timeout? : Int = 20,
max_debounce_delay? : Int = 200,
report_child_event? : Bool = false,
report_event_on_init? : Bool = false,
ignored_paths? : (String) -> Bool = _ => false,
) -> Watcher {
let context = "@fs.Watcher()"
let path = path.trim_end(chars="/")
let path = if path is "" { "/" } else { path.to_owned() }
let (root_file, root_id) = @event_loop.open(
path,
if @event_loop.platform is Windows {
3
} else {
0
},
create=0,
append=false,
sync=0,
mode=0,
context~,
)
guard root_file.kind() is Directory else {
root_file.close()
raise @os_error.OSError(@os_error.errno_ENOTDIR, context~)
}
let self = {
backend: None,
root_id,
base_path: path,
watched: Map([]),
pending_remove: Set([]),
ignored_paths,
report_child_event,
synchronize_task: None,
}
try {
let backend = self.new_backend(
root_file,
root_id~,
debounce_timeout~,
max_debounce_delay~,
context~,
)
self.backend = Some(backend)
let root_wd = backend.add_file(
path,
is_dir=true,
identity=root_id,
context~,
)
let root = {
wd: root_wd,
identity: root_id,
paths: Set([]),
is_dir: true,
children: Map([]),
modified: true,
children_to_scan: Map([]),
changed_entries: Map([]),
children_with_events: Map([]),
}
self.watched[root_id] = root
self.synchronize_tree(context~)
if !report_event_on_init {
self.emit_events_for_file(root, path="", on_event=(_, _) => (), context~)
}
self
} catch {
err => {
self.close()
raise err
}
}
}
///|
fn rel_path(parent : String, name : String) -> String {
if parent is "" {
name
} else {
"\{parent}/\{name}"
}
}
///|
/// Callback for handling event.
/// The first parameter is a rename hint,
/// if it is `Some(id)`, the second event parameter must be `Create` or `Remove`,
/// and `id` is the identity of the created/removed file.
/// Note that some `Create`/`Remove` event may not be a potential rename candidate,
/// in this case the first parameter would be `None`.
priv struct EventCallback((FileIdentity?, FsEvent) -> Unit)
///|
fn Watcher::emit_events_for_file(
self : Watcher,
file : WatchedFile,
on_event~ : EventCallback,
path~ : String,
context~ : String,
) -> Unit raise {
guard file.is_dir else {
// The only possible event for a regular file is `Modify`,
// so no extra information about event detail is recorded for regular files.
on_event(None, Modify(path))
}
let events : Map[FileIdentity, FsEvent] = Map([])
for name, prev_id in file.changed_entries {
let curr_id = file.children.get(name)
let path = rel_path(path, name)
if prev_id is Some(prev_id) {
match events.get(prev_id) {
_ if curr_id is Some(_) => on_event(None, Remove(path))
Some(Create(new)) => {
events.remove(prev_id)
on_event(None, Rename(old=path, new~))
}
Some(_) => {
self.emit_remove_event_for_children(prev_id, on_event~, path~)
on_event(Some(prev_id), Remove(path))
}
None => events[prev_id] = Remove(path)
}
}
if curr_id is Some(curr_id) {
match events.get(curr_id) {
Some(Remove(old)) => {
events.remove(curr_id)
on_event(None, Rename(old~, new=path))
}
Some(_) => {
on_event(Some(curr_id), Create(path))
self.emit_create_event_for_children(curr_id, on_event~, path~)
}
None => events[curr_id] = Create(path)
}
}
}
file.changed_entries.clear()
for id, event in events {
if event is Remove(path) {
self.emit_remove_event_for_children(id, on_event~, path~)
}
on_event(Some(id), event)
if event is Create(path) {
self.emit_create_event_for_children(id, on_event~, path~)
}
}
for name, id in file.children_with_events {
self.emit_events_for_file(
self.watched[id],
on_event~,
path=rel_path(path, name),
context~,
)
}
file.children_with_events.clear()
}
///|
/// Emit create event for the whole tree under `identity`.
/// If the tree is not new, but considered "created" due to failed rename serialization,
/// the event state of the whole tree will be reset by `emit_create_event_for_children`.
fn Watcher::emit_create_event_for_children(
self : Watcher,
identity : FileIdentity,
on_event~ : EventCallback,
path~ : String,
) -> Unit {
let file = self.watched[identity]
guard file.is_dir else { return }
file.changed_entries.clear()
file.children_with_events.clear()
for name, child_id in file.children {
let path = rel_path(path, name)
if self.report_child_event {
on_event(None, Create(path))
}
self.emit_create_event_for_children(child_id, on_event~, path~)
}
}
///|
/// Emit remove event for the whole tree under `identity`.
/// If the sub tree is still present in the watched tree,
/// but considered "removed" due to failed rename serialization,
/// the event state of the whole tree will be reset by `emit_remove_event_for_children`.
fn Watcher::emit_remove_event_for_children(
self : Watcher,
identity : FileIdentity,
on_event~ : EventCallback,
path~ : String,
) -> Unit {
let file = self.watched[identity]
guard file.is_dir else { return }
file.changed_entries.clear()
file.children_with_events.clear()
for name, child_id in file.children {
let path = rel_path(path, name)
self.emit_remove_event_for_children(child_id, on_event~, path~)
if self.report_child_event {
on_event(None, Remove(path))
}
}
}
///|
async fn Watcher::wait_aux(
self : Watcher,
on_event~ : (FileIdentity?, FsEvent) -> Unit,
) -> Unit {
let context = "@fs.Watcher::wait()"
guard self.backend is Some(backend) else {
raise Failure::Failure("watcher already closed")
}
if self.synchronize_task is Some(task) {
task.wait()
}
let root = self.watched[self.root_id]
while !root.has_events() {
backend.wait()
self.synchronize_tree(context~)
}
self.emit_events_for_file(root, path="", on_event~, context~)
while self.pending_remove.iter().next() is Some(identity) {
self.pending_remove.remove(identity)
let file = self.watched[identity]
guard file.paths.is_empty() else { continue }
if file.is_dir {
for child in file.children.keys() {
self.remove_file_at(file, child)
}
}
self.watched.remove(identity)
backend.remove_file(file.wd, context~)
}
}
///|
/// Wait for any change in the watched tree,
/// and return a list of events describing what changed
/// since the last `wait`/`wait_any` call or watcher creation.
///
/// The returned list describe the net change of the watched tree since the last query,
/// instead of recording a precise list of transactions that happen since the last wait.
/// For example, a create event followed by a removed event on the same path
/// will get fused and result in nothing in the returned event list.
///
/// `wait` performs rename detection based on the physical ID of files. The rule is:
/// - renaming for regular files are detected globally for the whole watched tree
/// - renaming for directory are only detected locally within the same parent directory.
/// Directory renaming across directory are treated as separated remove + create events.
/// - if the destination of a rename already contains something,
/// a remove event will be emitted for the old file on the destination.
/// When the watcher cannot serialize the file system changes into simple rename events,
/// for example two files are atomically swapped,
/// a remove event on the old path + a create event on the new path will be created.
///
/// This method is cancellation safe.
/// When it get cancelled and raises the canellation error,
/// it is guaranteed that no event will be consumed.
pub async fn Watcher::wait(self : Watcher) -> Array[FsEvent] {
let events = []
let pending_rename : Map[FileIdentity, FsEvent] = Map([])
fn on_event(rename_id, event : FsEvent) {
guard rename_id is Some(id) &&
!(self.watched.get(id) is Some({ is_dir: true, .. })) else {
events.push(event)
}
match (pending_rename.get(id), event) {
(None, Create(_) | Remove(_)) => pending_rename[id] = event
(Some(Remove(old)), Create(new)) | (Some(Create(new)), Remove(old)) => {
pending_rename.remove(id)
events.push(Rename(old~, new~))
}
_ => events.push(event)
}
}
self.wait_aux(on_event~)
for _, event in pending_rename {
events.push(event)
}
events
}
///|
/// Wait for any change in the watched tree
/// since the last `wait`/`wait_any` call or watcher creation.
///
/// This method is cancellation safe.
/// When it get cancelled and raises the canellation error,
/// it is guaranteed that no event will be consumed.
pub async fn Watcher::wait_any(self : Watcher) -> Unit {
self.wait_aux(on_event=(_, _) => ())
}
///|
fn WatchedFile::is_clean(file : WatchedFile) -> Bool {
!file.modified && file.children_to_scan.is_empty()
}
///|
fn WatchedFile::has_events(file : WatchedFile) -> Bool {
!(file.is_dir &&
file.changed_entries.is_empty() &&
file.children_with_events.is_empty())
}
///|
async fn Watcher::get_file_at(
self : Watcher,
identity~ : FileIdentity,
is_dir~ : Bool,
path~ : String,
context~ : String,
) -> WatchedFile {
if self.watched.get(identity) is Some(existing) {
return existing
}
let full_path = "\{self.base_path}/\{path}"
guard self.backend is Some(backend) else {
raise Failure::Failure("watcher already closed")
}
let wd = backend.add_file(full_path, is_dir~, identity~, context~)
let new_file = {
wd,
identity,
paths: Set([]),
is_dir,
children: Map([]),
modified: is_dir, // new directory need rescan
children_to_scan: Map([]),
changed_entries: Map([]),
children_with_events: Map([]),
}
self.watched[identity] = new_file
new_file
}
///|
async fn Watcher::add_file_at(
self : Watcher,
dir : WatchedFile,
name : String,
is_dir~ : Bool,
identity~ : FileIdentity,
path~ : String,
context~ : String,
) -> Unit {
if (self.ignored_paths)(path) {
return
}
if dir.children.get(name) is Some(curr_id) {
if curr_id == identity {
return
} else {
self.remove_file_at(dir, name)
}
}
let location = (dir.identity, name)
let file = self.get_file_at(path~, is_dir~, identity~, context~)
dir.children[name] = file.identity
file.paths.add(location)
if !file.is_clean() {
dir.children_to_scan[name] = file.identity
}
match dir.changed_entries.get(name) {
None => dir.changed_entries[name] = None
Some(Some(prev_id)) if prev_id == identity =>
dir.changed_entries.remove(name)
Some(_) => ()
}
}
///|
fn Watcher::remove_file_at(
self : Watcher,
dir : WatchedFile,
name : String,
) -> Unit {
let curr_id = dir.children.get(name)
guard curr_id is Some(curr_id) else { return }
let path = (dir.identity, name)
dir.children.remove(name)
dir.children_to_scan.remove(name)
dir.children_with_events.remove(name)
let curr = self.watched[curr_id]
curr.paths.remove(path)
if curr.paths.is_empty() {
self.pending_remove.add(curr_id)
}
match dir.changed_entries.get(name) {
None => dir.changed_entries[name] = Some(curr_id)
Some(None) => dir.changed_entries.remove(name)
Some(Some(_)) => ()
}
}
///|
async fn Watcher::scan_dir(
self : Watcher,
dir : WatchedFile,
path~ : String,
context~ : String,
) -> Unit {
let removed_children = dir.children.copy()
let (dir_fd, dir_id) = @event_loop.open(
"\{self.base_path}/\{path}",
0,
create=0,
append=false,
sync=0,
mode=0,
context~,
)
guard dir_fd.kind() is Directory else {
dir_fd.close()
raise @os_error.OSError(@os_error.errno_ENOTDIR, context~)
}
let dir_obj = Directory::from_io_handle(dir_fd)
defer dir_obj.close()
while dir_obj.next_aux(include_special=false, include_hidden=true, context~)
is Some(ent) {
let new_file_id = { ..dir_id, file_id: ent.file_id }
self.add_file_at(
dir,
ent.name,
is_dir=ent.is_dir,
identity=new_file_id,
path=rel_path(path, ent.name),
context~,
) catch {
@os_error.OSError(_) as err if err.is_ENOENT() => continue
err => raise err
}
removed_children.remove(ent.name)
}
for name in removed_children.keys() {
self.remove_file_at(dir, name)
}
}
///|
async fn Watcher::synchronize_tree(self : Watcher, context~ : String) -> Unit {
if self.synchronize_task is Some(task) {
task.wait()
return
}
let task = @coroutine.spawn <| () => {
defer {
self.synchronize_task = None
}
let root = self.watched[self.root_id]
while !root.is_clean() {
self.synchronize_file(root, path="", context~)
}
}
self.synchronize_task = Some(task)
task.wait()
}
///|
async fn Watcher::synchronize_file(
self : Watcher,
file : WatchedFile,
path~ : String,
context~ : String,
) -> Unit {
if file.modified {
file.modified = false
if file.is_dir {
self.scan_dir(file, path~, context~)
}
} else if file.children_to_scan.iter().next() is Some((name, child_id)) {
let child = self.watched[child_id]
try
self.synchronize_file(child, path=rel_path(path, name), context~)
catch {
@os_error.OSError(_) as err if err.is_ENOENT() || err.is_ENOTDIR() =>
self.mark_as_modified(file.identity)
err => raise err
} noraise {
_ => {
if child.is_clean() {
file.children_to_scan.remove(name)
}
if child.has_events() {
file.children_with_events[name] = child.identity
} else {
file.children_with_events.remove(name)
}
}
}
}
}
///|
fn Watcher::mark_as_modified(self : Watcher, id : FileIdentity) -> Unit {
let file = self.watched[id]
if file.modified {
return
}
file.modified = true
if !file.children_to_scan.is_empty() || id == self.root_id {
return
}
for path in file.paths {
let (dir, name) = path
self.add_dirty_child(dir, name, id)
}
}
///|
fn Watcher::add_dirty_child(
self : Watcher,
dir_id : FileIdentity,
name : String,
child_id : FileIdentity,
) -> Unit {
let dir = self.watched[dir_id]
let was_clean = dir.is_clean()
dir.children_to_scan[name] = child_id
if was_clean && dir_id != self.root_id {
for path in dir.paths {
let (parent, name) = path
self.add_dirty_child(parent, name, dir_id)
}
}
}
///|
fn Watcher::overflow(self : Watcher) -> Unit {
for file in self.watched.values() {
file.modified = true
if file.is_dir {
for name, id in file.children {
file.children_to_scan[name] = id
}
}
}
}