harmos_storage/directory/
scan.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Bounded Depth-First Enumeration With Retained Entry Ownership
11
12/* === Modules === */
13
14/* === Re-Exports === */
15
16/* === Imports === */
17
18// Standard Library Imports
19use std::fs::{self, ReadDir};
20use std::path::Path;
21use std::path::PathBuf;
22use std::sync::atomic::{AtomicBool, Ordering};
23use std::sync::{Arc, Mutex};
24// Third Party Imports
25use tokio::task::JoinHandle;
26// Local Imports
27use super::{DirectoryCompletion, DirectoryEntry, Selection, item, join_error, relative, selection};
28use crate::manager::Inner;
29use crate::{Error, ReadBatch, Result};
30use harmos_domain::model::storage::DirectorySelection;
31
32/* === Constants === */
33
34/* === Macros === */
35
36/* === Type Aliases === */
37
38type Scanned = DirectoryEntry;
39type Pending = JoinHandle<Result<Option<Scanned>>>;
40
41/* === Structs === */
42
43pub struct DirectoryReader {
44    context: crate::execution::Context,
45    pub(super) inner: Arc<Inner>,
46    state: Arc<Mutex<Scan>>,
47    pending: Option<Pending>,
48    cancelled: Arc<AtomicBool>,
49}
50
51struct Scan {
52    inner: Arc<Inner>,
53
54    selection: Arc<Selection>,
55    stack: Vec<Frame>,
56    visited: usize,
57    entries: u64,
58    complete: bool,
59    cancelled: Arc<AtomicBool>,
60}
61
62struct Frame {
63    directory: ReadDir,
64    _path: PathBuf,
65}
66
67/* === Enums === */
68
69/* === Traits === */
70
71/* === Implementations === */
72
73impl DirectoryReader {
74    pub(super) async fn new(inner: Arc<Inner>, value: DirectorySelection) -> Result<Self> {
75        let selected = selection(value).await?;
76        Self::selected(inner, selected).await
77    }
78
79    pub(super) async fn selected(inner: Arc<Inner>, selection: Arc<Selection>) -> Result<Self> {
80        let cancelled = Arc::new(AtomicBool::new(false));
81        let scan = Scan {
82            inner: Arc::clone(&inner),
83
84            selection,
85            stack: Vec::new(),
86            visited: 0,
87            entries: 0,
88            complete: false,
89            cancelled: Arc::clone(&cancelled),
90        };
91        let scan = {
92            let context = crate::execution::Context::current();
93            tokio::task::spawn_blocking(move || {
94                context.blocking(|| {
95                    let mut scan = scan;
96                    let selection = Arc::clone(&scan.selection);
97                    scan.enter(&selection.value.root)?;
98                    Ok::<_, Error>(scan)
99                })
100            })
101        }
102        .await
103        .map_err(join_error)??;
104        Ok(Self {
105            context: crate::execution::Context::current(),
106            inner,
107            state: Arc::new(Mutex::new(scan)),
108            pending: None,
109            cancelled,
110        })
111    }
112
113    pub async fn next(&mut self) -> Result<Option<ReadBatch<DirectoryEntry>>> {
114        self.next_value().await?.map(|value| item(&self.inner, value)).transpose()
115    }
116
117    pub fn completion(&self) -> Result<DirectoryCompletion> {
118        if self.pending.is_some() {
119            return Err(Error::NotReady { domain: "directory completion" });
120        }
121        let scan = self.state.lock().unwrap();
122        if !scan.complete {
123            return Err(Error::NotReady { domain: "directory completion" });
124        }
125        Ok(DirectoryCompletion { entries: scan.entries })
126    }
127
128    pub(super) async fn next_value(&mut self) -> Result<Option<Scanned>> {
129        let context = self.context.clone();
130        let span = context.sync(|| tracing::debug_span!("storage.directory.page", outcome = "cancelled"));
131        context
132            .result(span, async {
133                if self.pending.is_none() {
134                    let state = Arc::clone(&self.state);
135                    let context = self.context.clone();
136                    self.pending = Some(tokio::task::spawn_blocking(move || context.blocking(|| state.lock().unwrap().next())));
137                }
138                let result = self.pending.as_mut().unwrap().await;
139                self.pending = None;
140                result.map_err(join_error)?
141            })
142            .await
143    }
144}
145
146impl Drop for DirectoryReader {
147    fn drop(&mut self) {
148        self.cancelled.store(true, Ordering::Release);
149    }
150}
151
152impl Scan {
153    fn enter(&mut self, path: &Path) -> Result<()> {
154        let path = crate::local::validate_local_path(path)?;
155        let directory = fs::read_dir(&path).map_err(crate::local::io_error)?;
156        self.stack.push(Frame { directory, _path: path });
157        Ok(())
158    }
159
160    fn next(&mut self) -> Result<Option<Scanned>> {
161        loop {
162            if self.cancelled.load(Ordering::Acquire) {
163                return Err(Error::Closing { domain: "directory scan" });
164            }
165            let Some(frame) = self.stack.last_mut() else {
166                self.complete = true;
167                return Ok(None);
168            };
169            let Some(entry) = frame.directory.next() else {
170                self.stack.pop();
171                continue;
172            };
173            let entry = entry.map_err(crate::local::io_error)?;
174            self.visited = self.visited.checked_add(1).ok_or(Error::ResourceExhausted { resource: "directory entries" })?;
175            if self.visited > self.inner.settings.filesystem.max_entries.get() {
176                return Err(Error::ResourceExhausted { resource: "directory entries" });
177            }
178            let name = entry.file_name();
179
180            let path = frame._path.join(name);
181            let metadata = match fs::symlink_metadata(&path) {
182                Ok(metadata) => metadata,
183                Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
184                Err(error) => return Err(crate::local::io_error(error)),
185            };
186            if crate::local::linked(&metadata) {
187                continue;
188            }
189            if metadata.is_dir() && self.selection.value.recursive {
190                self.enter(&path)?;
191            }
192            let Some(path) = relative(&self.selection.value.root, &path, self.selection.value.recursive)? else { continue };
193            if !super::selection::selected(&path, &self.selection.value.rules) {
194                continue;
195            }
196            self.entries += 1;
197            let entry = DirectoryEntry {
198                path,
199                directory: metadata.is_dir(),
200                length: if metadata.is_file() { metadata.len() } else { 0 },
201            };
202            return Ok(Some(entry));
203        }
204    }
205}
206
207/* === Public Functions === */
208
209/* === Private Functions === */
210
211/* === Tests === */
harmos_storage/directory/scan.rs

harmos_storage/directory/
scan.rs
