harmos_storage/directory/
watch.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Native Subscription Precedes Enumeration; Overflow Requires An Explicit Rescan
11
12/* === Modules === */
13
14/* === Re-Exports === */
15
16/* === Imports === */
17
18// Standard Library Imports
19use std::collections::VecDeque;
20use std::path::Path;
21use std::sync::{
22    Arc,
23    atomic::{AtomicBool, Ordering},
24};
25// Third Party Imports
26use notify::event::{ModifyKind, RenameMode};
27use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
28use tokio::sync::{Notify, mpsc};
29// Local Imports
30use super::{DirectoryEntry, DirectoryEvent, DirectoryReader, Selection, item, relative, selection};
31use crate::manager::Inner;
32use crate::{Error, ReadBatch, Result};
33use harmos_domain::model::storage::DirectorySelection;
34
35/* === Constants === */
36
37/* === Macros === */
38
39/* === Type Aliases === */
40
41/* === Structs === */
42
43pub struct DirectoryWatcher {
44    context: crate::execution::Context,
45    inner: Arc<Inner>,
46
47    selection: Arc<Selection>,
48    initial: Option<DirectoryReader>,
49    initial_complete: bool,
50    initial_marker: ReadBatch<DirectoryEvent>,
51    rescan: ReadBatch<DirectoryEvent>,
52    incoming: mpsc::Receiver<Queued>,
53    ready: VecDeque<ReadBatch<DirectoryEvent>>,
54    overflow: Arc<AtomicBool>,
55    changed: Arc<Notify>,
56    watcher: Option<RecommendedWatcher>,
57}
58
59struct Queued {
60    event: Event,
61}
62
63/* === Enums === */
64
65/* === Traits === */
66
67/* === Implementations === */
68
69impl DirectoryWatcher {
70    pub(super) async fn new(inner: Arc<Inner>, value: DirectorySelection, include_existing: bool) -> Result<Self> {
71        let selected = selection(value).await?;
72        let initial_marker = item(&inner, DirectoryEvent::InitialScanComplete)?;
73        let rescan = item(&inner, DirectoryEvent::RescanRequired)?;
74        let settings = inner.settings.filesystem;
75
76        let (outgoing, incoming) = mpsc::channel(settings.event_capacity.get());
77        let overflow = Arc::new(AtomicBool::new(false));
78        let changed = Arc::new(Notify::new());
79
80        let callback_overflow = Arc::clone(&overflow);
81        let callback_changed = Arc::clone(&changed);
82
83        let span = tracing::debug_span!("storage.directory.subscription", outcome = "cancelled");
84        let context = crate::execution::Context::with_span(span);
85        let callback_context = context.clone();
86        let handler = move |result: notify::Result<Event>| {
87            callback_context.sync(|| {
88                let Ok(event) = result else {
89                    signal(&callback_overflow, &callback_changed);
90                    return;
91                };
92                if matches!(event.kind, EventKind::Access(_)) {
93                    return;
94                }
95                if event.paths.is_empty() || event.paths.len() > settings.event_capacity.get() {
96                    signal(&callback_overflow, &callback_changed);
97                    return;
98                }
99
100                deliver(&outgoing, Queued { event }, &callback_overflow, &callback_changed);
101            })
102        };
103        let configuration = notify::Config::default().with_poll_interval(settings.poll_interval).with_follow_symlinks(false);
104        let mut watcher = RecommendedWatcher::new(handler, configuration).map_err(watch_error)?;
105        let mode = if selected.value.recursive { RecursiveMode::Recursive } else { RecursiveMode::NonRecursive };
106        watcher.watch(&selected.value.root, mode).map_err(watch_error)?;
107        let initial =
108            if include_existing { Some(DirectoryReader::selected(Arc::clone(&inner), Arc::clone(&selected)).await?) } else { None };
109        Ok(Self {
110            context,
111            inner,
112
113            selection: selected,
114            initial,
115            initial_complete: false,
116            initial_marker,
117            rescan,
118            incoming,
119            ready: VecDeque::new(),
120            overflow,
121            changed,
122            watcher: Some(watcher),
123        })
124    }
125
126    pub async fn next(&mut self) -> Result<Option<ReadBatch<DirectoryEvent>>> {
127        let context = self.context.clone();
128        let span = context.sync(|| tracing::trace_span!("storage.directory.watch.page", outcome = "cancelled"));
129        context
130            .result(span, async {
131                let mut lifecycle = self.inner.changed.subscribe();
132                loop {
133                    let closing = {
134                        let state = self.inner.state.lock().unwrap();
135                        state.status.closing || state.status.closed
136                    };
137                    if closing || self.watcher.is_none() {
138                        self.stop();
139                        return Ok(None);
140                    }
141                    if let Some(reader) = &mut self.initial {
142                        let next = tokio::select! {
143                            next = reader.next_value() => Some(next),
144                            _ = lifecycle.changed() => None,
145                        };
146                        let Some(next) = next else { continue };
147                        if let Some(entry) = next? {
148                            return item(&self.inner, DirectoryEvent::Existing(entry)).map(Some);
149                        }
150                        self.initial = None;
151                    }
152                    if !self.initial_complete {
153                        self.initial_complete = true;
154                        return Ok(Some(self.initial_marker.clone()));
155                    }
156                    if let Some(item) = self.ready.pop_front() {
157                        return Ok(Some(item));
158                    }
159
160                    let changed = Arc::clone(&self.changed);
161                    let notified = changed.notified();
162                    if self.overflow.swap(false, Ordering::AcqRel) {
163                        return Ok(Some(self.rescan()));
164                    }
165                    let queued = tokio::select! {
166                        value = self.incoming.recv() => value,
167                        _ = notified => continue,
168                        _ = lifecycle.changed() => continue,
169                    };
170                    let Some(queued) = queued else { return Ok(None) };
171                    if queued.event.paths.len() > self.inner.settings.filesystem.event_capacity.get() {
172                        return Ok(Some(self.rescan()));
173                    }
174                    let events = events(&self.selection.value, queued.event)?;
175                    if events.iter().any(|event| matches!(event, DirectoryEvent::RescanRequired)) {
176                        return Ok(Some(self.rescan()));
177                    }
178                    for event in events {
179                        // One Native Event Can Carry Several Paths; Preserve Every Projection.
180                        let item = item(&self.inner, event)?;
181                        self.ready.push_back(item);
182                    }
183                    if let Some(item) = self.ready.pop_front() {
184                        return Ok(Some(item));
185                    }
186                }
187            })
188            .await
189    }
190
191    fn stop(&mut self) {
192        self.context.sync(|| self.context.span.record("outcome", "completed"));
193        self.watcher.take();
194        self.incoming.close();
195        while self.incoming.try_recv().is_ok() {}
196        self.ready.clear();
197        self.initial = None;
198    }
199
200    fn rescan(&mut self) -> ReadBatch<DirectoryEvent> {
201        self.ready.clear();
202        for _ in 0..self.inner.settings.filesystem.event_capacity.get() {
203            if self.incoming.try_recv().is_err() {
204                break;
205            }
206        }
207        self.rescan.clone()
208    }
209}
210
211impl Drop for DirectoryWatcher {
212    fn drop(&mut self) {
213        self.context.sync(|| drop(self.watcher.take()));
214    }
215}
216
217/* === Public Functions === */
218
219/* === Private Functions === */
220
221fn deliver<T>(outgoing: &mpsc::Sender<T>, value: T, overflow: &AtomicBool, changed: &Notify) {
222    if outgoing.try_send(value).is_err() {
223        signal(overflow, changed);
224    }
225}
226
227fn signal(overflow: &AtomicBool, changed: &Notify) {
228    overflow.store(true, Ordering::Release);
229    changed.notify_one();
230}
231
232fn watch_error(error: notify::Error) -> Error {
233    Error::backend("directory watch", crate::BackendFailure::Permanent, error)
234}
235
236fn selected_path(selection: &DirectorySelection, path: &Path) -> Result<Option<String>> {
237    let Some(path) = relative(&selection.root, path, selection.recursive)? else { return Ok(None) };
238    Ok(super::selection::selected(&path, &selection.rules).then_some(path))
239}
240
241fn entry(selection: &DirectorySelection, path: &Path) -> Result<Option<DirectoryEntry>> {
242    let Some(relative) = selected_path(selection, path)? else { return Ok(None) };
243    let metadata = match std::fs::symlink_metadata(path) {
244        Ok(metadata) => metadata,
245        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
246        Err(error) => return Err(crate::local::io_error(error)),
247    };
248    if crate::local::linked(&metadata) {
249        return Ok(None);
250    }
251    crate::local::validate_local_path(path)?;
252    Ok(Some(DirectoryEntry {
253        path: relative,
254        directory: metadata.is_dir(),
255        length: if metadata.is_file() { metadata.len() } else { 0 },
256    }))
257}
258
259fn events(selection: &DirectorySelection, event: Event) -> Result<Vec<DirectoryEvent>> {
260    let root_changed = matches!(event.kind, EventKind::Remove(_) | EventKind::Modify(ModifyKind::Name(_)))
261        && event.paths.iter().any(|path| path == &selection.root);
262    if event.need_rescan() || event.paths.is_empty() || root_changed {
263        return Ok(vec![DirectoryEvent::RescanRequired]);
264    }
265    let mut result = Vec::new();
266    match event.kind {
267        EventKind::Create(_) | EventKind::Modify(ModifyKind::Data(_) | ModifyKind::Metadata(_) | ModifyKind::Any) => {
268            for path in event.paths {
269                if let Some(entry) = entry(selection, &path)? {
270                    result.push(if matches!(event.kind, EventKind::Create(_)) {
271                        DirectoryEvent::Created(entry)
272                    } else {
273                        DirectoryEvent::Modified(entry)
274                    });
275                }
276            }
277        }
278        EventKind::Remove(_) => {
279            for path in event.paths {
280                if let Some(path) = selected_path(selection, &path)? {
281                    result.push(DirectoryEvent::Removed { path });
282                }
283            }
284        }
285        EventKind::Modify(ModifyKind::Name(RenameMode::Both)) if event.paths.len() == 2 => {
286            let from = selected_path(selection, &event.paths[0])?;
287            let to = selected_path(selection, &event.paths[1])?;
288            match (from, to) {
289                (Some(from), Some(to)) => result.push(DirectoryEvent::Renamed { from, to }),
290                (Some(path), None) => result.push(DirectoryEvent::Removed { path }),
291                (None, Some(_)) => {
292                    if let Some(entry) = entry(selection, &event.paths[1])? {
293                        result.push(DirectoryEvent::Created(entry));
294                    }
295                }
296                (None, None) => {}
297            }
298        }
299        EventKind::Modify(ModifyKind::Name(RenameMode::From)) => {
300            for path in event.paths {
301                if let Some(path) = selected_path(selection, &path)? {
302                    result.push(DirectoryEvent::Removed { path });
303                }
304            }
305        }
306        EventKind::Modify(ModifyKind::Name(RenameMode::To)) => {
307            for path in event.paths {
308                if let Some(entry) = entry(selection, &path)? {
309                    result.push(DirectoryEvent::Created(entry));
310                }
311            }
312        }
313        EventKind::Access(_) => {}
314        _ => result.push(DirectoryEvent::RescanRequired),
315    }
316    Ok(result)
317}
318
319/* === Tests === */
320
321#[cfg(test)]
322mod tests {
323    use super::*;
324
325    #[tokio::test]
326    async fn overflow_remains_observable() {
327        let (outgoing, mut incoming) = mpsc::channel(1);
328        let overflow = AtomicBool::new(false);
329        let changed = Notify::new();
330        deliver(&outgoing, 1, &overflow, &changed);
331        deliver(&outgoing, 2, &overflow, &changed);
332        assert!(overflow.swap(false, Ordering::AcqRel));
333        changed.notified().await;
334        assert_eq!(incoming.recv().await, Some(1));
335        deliver(&outgoing, 3, &overflow, &changed);
336        assert!(!overflow.load(Ordering::Acquire));
337        assert_eq!(incoming.recv().await, Some(3));
338    }
339
340    #[test]
341    fn rename_rescan_projection() {
342        let root = std::path::PathBuf::from("root");
343        let selection = DirectorySelection {
344            root: root.clone(),
345            recursive: true,
346            rules: vec![harmos_domain::model::storage::DirectoryRule { pattern: "**".into(), include: true }],
347        };
348        let renamed =
349            Event::new(EventKind::Modify(ModifyKind::Name(RenameMode::Both))).add_path(root.join("before")).add_path(root.join("after"));
350        assert_eq!(
351            events(&selection, renamed).unwrap(),
352            vec![DirectoryEvent::Renamed {
353                from: "before".into(),
354                to: "after".into()
355            }]
356        );
357        assert_eq!(events(&selection, Event::new(EventKind::Other)).unwrap(), vec![DirectoryEvent::RescanRequired]);
358    }
359}
harmos_storage/directory/watch.rs

harmos_storage/directory/
watch.rs
