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

harmos_storage/directory/
watch.rs
