harmos_storage/directory/
mod.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Charged Directory Discovery And Native Change Subscription
11
12/* === Modules === */
13
14mod scan;
15mod selection;
16mod watch;
17
18/* === Re-Exports === */
19
20pub use scan::DirectoryReader;
21pub use watch::DirectoryWatcher;
22
23/* === Imports === */
24
25// Standard Library Imports
26use std::path::Path;
27use std::sync::Arc;
28// Third Party Imports
29// Local Imports
30use crate::arrow::RetainedMemory;
31use crate::capacity::{Held, Pool, Refusal, Reservation};
32use crate::manager::{ENTRY_OVERHEAD, Inner};
33use crate::{Error, MemorySize, Message, ReadBatch, Result, StorageManager};
34use harmos_domain::model::storage::{DirectoryCompletion, DirectoryEntry, DirectoryEvent, DirectorySelection};
35
36/* === Constants === */
37
38/* === Macros === */
39
40/* === Type Aliases === */
41
42/* === Structs === */
43
44struct Selection {
45    value: DirectorySelection,
46    _memory: RetainedMemory,
47}
48
49/* === Enums === */
50
51/* === Traits === */
52
53/* === Implementations === */
54
55impl<C> StorageManager<C> {
56    pub async fn list_directory(&self, selection: DirectorySelection) -> Result<DirectoryReader> {
57        let context = crate::execution::Context::current();
58        let span = context.sync(|| tracing::debug_span!("storage.directory.list_directory", outcome = "cancelled"));
59        context.result(span, async { DirectoryReader::new(Arc::clone(&self.inner), selection).await }).await
60    }
61
62    pub async fn watch_directory(&self, selection: DirectorySelection, include_existing: bool) -> Result<DirectoryWatcher> {
63        let context = crate::execution::Context::current();
64        let span = context.sync(|| tracing::debug_span!("storage.directory.watch_directory", outcome = "cancelled"));
65        context.result(span, async { DirectoryWatcher::new(Arc::clone(&self.inner), selection, include_existing).await }).await
66    }
67}
68
69/* === Public Functions === */
70
71/* === Private Functions === */
72
73fn item<T: MemorySize + Message>(inner: &Arc<Inner>, held: &Held, value: T) -> Result<ReadBatch<T>> {
74    retained_item(inner, value, Reservation::empty(inner, Pool::Read, held))
75}
76
77fn retained_item<T: MemorySize + Message>(inner: &Arc<Inner>, value: T, mut memory: Reservation) -> Result<ReadBatch<T>> {
78    if value.encoded_len() > inner.settings.response.batch_bytes.get() {
79        return Err(Error::ResourceExhausted { resource: "directory response bytes" });
80    }
81    let bytes = value.memory_size().saturating_add(ENTRY_OVERHEAD);
82    if bytes > memory.bytes() {
83        memory.try_grow(bytes - memory.bytes()).map_err(Refusal::into_error)?;
84    }
85    memory.shrink_to(bytes);
86    Ok(ReadBatch::new(value, memory))
87}
88
89async fn selection(inner: &Arc<Inner>, mut value: DirectorySelection, held: &Held) -> Result<Arc<Selection>> {
90    let path = value.root.clone();
91    let context = crate::execution::Context::current();
92    value.root =
93        tokio::task::spawn_blocking(move || context.blocking(|| crate::local::validate_local_path(&path))).await.map_err(join_error)??;
94    let bytes = value.memory_size().saturating_add(ENTRY_OVERHEAD);
95    let memory = Reservation::try_new(inner, Pool::Read, held, bytes).map(Reservation::retain).map_err(Refusal::into_error)?;
96    Ok(Arc::new(Selection { value, _memory: memory }))
97}
98
99fn relative(root: &Path, path: &Path, recursive: bool) -> Result<Option<String>> {
100    let Ok(relative) = path.strip_prefix(root) else { return Ok(None) };
101    if relative.as_os_str().is_empty() || (!recursive && relative.components().count() != 1) {
102        return Ok(None);
103    }
104    let mut result = String::new();
105    for component in relative.components() {
106        let std::path::Component::Normal(value) = component else { return Ok(None) };
107        let value =
108            value.to_str().ok_or(Error::InvalidInput { field: "directory entry", reason: "path cannot be represented as UTF-8" })?;
109        if !result.is_empty() {
110            result.push('/');
111        }
112        result.push_str(value);
113    }
114    Ok(Some(result))
115}
116
117fn join_error(error: tokio::task::JoinError) -> Error {
118    Error::backend("directory task", crate::BackendFailure::Permanent, error)
119}
120
121/* === Tests === */
harmos_storage/directory/mod.rs

harmos_storage/directory/
mod.rs
