aster_forge_webdav/
traversal.rs

1//! Bounded directory paging and recursive traversal resource accounting.
2
3use std::collections::HashSet;
4use std::hash::{BuildHasher, Hash};
5use std::sync::{
6    Arc,
7    atomic::{AtomicBool, Ordering},
8};
9
10use crate::{
11    DavBackendError, DavDirectoryEntry, DavDirectoryEnumerator, DavDirectoryPage,
12    DavDirectoryPageRequest, DavPath,
13};
14use bytes::Bytes;
15
16/// Cancellation checkpoint controlled by the product or transport adapter.
17pub trait DavCancellation: Send + Sync {
18    fn is_cancelled(&self) -> bool;
19}
20
21/// Cloneable cancellation source shared by a transport response and product backend work.
22///
23/// Dropping a clone does not cancel the operation. The owning transport or request lifecycle must
24/// call [`Self::cancel`] when the response body is abandoned or its execution budget expires.
25#[derive(Debug, Clone, Default)]
26pub struct DavCancellationToken {
27    cancelled: Arc<AtomicBool>,
28}
29
30impl DavCancellationToken {
31    #[must_use]
32    pub fn new() -> Self {
33        Self::default()
34    }
35
36    pub fn cancel(&self) {
37        self.cancelled.store(true, Ordering::Release);
38    }
39
40    #[must_use]
41    pub fn is_cancelled(&self) -> bool {
42        self.cancelled.load(Ordering::Acquire)
43    }
44}
45
46impl DavCancellation for DavCancellationToken {
47    fn is_cancelled(&self) -> bool {
48        self.is_cancelled()
49    }
50}
51
52/// Cancellation checkpoint used when the caller has no cancellation source.
53#[derive(Debug, Clone, Copy, Default)]
54pub struct DavNeverCancelled;
55
56impl DavCancellation for DavNeverCancelled {
57    fn is_cancelled(&self) -> bool {
58        false
59    }
60}
61
62/// Maximum directory entries accepted from one product page.
63#[derive(Debug, Clone, Copy, PartialEq, Eq)]
64pub struct DavDirectoryPageLimits {
65    pub maximum_entries: usize,
66    pub maximum_pages: usize,
67}
68
69impl DavDirectoryPageLimits {
70    /// Creates non-zero directory page limits.
71    ///
72    /// # Errors
73    ///
74    /// Returns [`DavDirectoryReadError`] when either directory-page limit is zero.
75    pub const fn new(
76        maximum_entries: usize,
77        maximum_pages: usize,
78    ) -> Result<Self, DavDirectoryReadError> {
79        if maximum_entries == 0 || maximum_pages == 0 {
80            Err(DavDirectoryReadError::InvalidLimit)
81        } else {
82            Ok(Self {
83                maximum_entries,
84                maximum_pages,
85            })
86        }
87    }
88}
89
90/// Invalid product page classification.
91#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
92pub enum DavDirectoryPageValidationError {
93    #[error("directory page size limit must be non-zero")]
94    InvalidLimit,
95    #[error("directory page exceeds the requested entry limit")]
96    TooManyEntries,
97    #[error("empty directory page cannot carry a continuation cursor")]
98    EmptyContinuation,
99    #[error("directory continuation cursor did not advance")]
100    CursorLoop,
101    #[error("directory page contains an empty stable key")]
102    EmptyStableKey,
103    #[error("directory page contains a duplicate entry key")]
104    DuplicateEntry,
105    #[error("directory page entries are not in stable ascending order")]
106    OutOfOrderEntry,
107}
108
109/// Validates product page bounds, ordering and cursor progress before protocol consumption.
110///
111/// # Errors
112///
113/// Returns an error when limits, ordering, entries, or cursor progress are invalid.
114pub fn validate_directory_page<E, C, S>(
115    current_cursor: Option<&C>,
116    seen_cursors: &HashSet<C, S>,
117    previous_stable_key: Option<&[u8]>,
118    maximum_entries: usize,
119    page: &DavDirectoryPage<E, C>,
120) -> Result<(), DavDirectoryPageValidationError>
121where
122    E: DavDirectoryEntry,
123    C: Eq + Hash,
124    S: BuildHasher,
125{
126    if maximum_entries == 0 {
127        return Err(DavDirectoryPageValidationError::InvalidLimit);
128    }
129    if page.entries.len() > maximum_entries {
130        return Err(DavDirectoryPageValidationError::TooManyEntries);
131    }
132    if page.entries.is_empty() && page.next_cursor.is_some() {
133        return Err(DavDirectoryPageValidationError::EmptyContinuation);
134    }
135    if page
136        .next_cursor
137        .as_ref()
138        .is_some_and(|cursor| current_cursor == Some(cursor) || seen_cursors.contains(cursor))
139    {
140        return Err(DavDirectoryPageValidationError::CursorLoop);
141    }
142    let mut previous: Option<&[u8]> = None;
143    for entry in &page.entries {
144        let key = entry.stable_key();
145        if key.is_empty() {
146            return Err(DavDirectoryPageValidationError::EmptyStableKey);
147        }
148        if previous.is_none()
149            && let Some(previous_page_key) = previous_stable_key
150        {
151            match previous_page_key.cmp(key) {
152                std::cmp::Ordering::Equal => {
153                    return Err(DavDirectoryPageValidationError::DuplicateEntry);
154                }
155                std::cmp::Ordering::Greater => {
156                    return Err(DavDirectoryPageValidationError::OutOfOrderEntry);
157                }
158                std::cmp::Ordering::Less => {}
159            }
160        }
161        if let Some(previous) = previous {
162            match previous.cmp(key) {
163                std::cmp::Ordering::Equal => {
164                    return Err(DavDirectoryPageValidationError::DuplicateEntry);
165                }
166                std::cmp::Ordering::Greater => {
167                    return Err(DavDirectoryPageValidationError::OutOfOrderEntry);
168                }
169                std::cmp::Ordering::Less => {}
170            }
171        }
172        previous = Some(key);
173    }
174    Ok(())
175}
176
177/// Opaque cursor state retained between bounded directory page requests.
178#[derive(Debug, Clone, PartialEq, Eq)]
179pub struct DavDirectoryPageState<C: Eq + Hash> {
180    current_cursor: Option<C>,
181    seen_cursors: HashSet<C>,
182    last_stable_key: Option<Bytes>,
183    poisoned: Option<DavDirectoryPageValidationError>,
184    finished: bool,
185    pages_read: usize,
186}
187
188impl<C: Eq + Hash> DavDirectoryPageState<C> {
189    #[must_use]
190    pub fn new() -> Self {
191        Self {
192            current_cursor: None,
193            seen_cursors: HashSet::new(),
194            last_stable_key: None,
195            poisoned: None,
196            finished: false,
197            pages_read: 0,
198        }
199    }
200
201    #[must_use]
202    pub fn cursor(&self) -> Option<&C> {
203        self.current_cursor.as_ref()
204    }
205
206    #[must_use]
207    pub const fn is_finished(&self) -> bool {
208        self.finished
209    }
210
211    #[must_use]
212    pub const fn pages_read(&self) -> usize {
213        self.pages_read
214    }
215}
216
217impl<C: Eq + Hash> Default for DavDirectoryPageState<C> {
218    fn default() -> Self {
219        Self::new()
220    }
221}
222
223/// Validated entries returned by one bounded page request.
224#[derive(Debug, Clone, PartialEq, Eq)]
225pub struct DavValidatedDirectoryPage<E> {
226    pub entries: Vec<E>,
227    pub has_more: bool,
228}
229
230/// Failure while requesting the next bounded directory page.
231#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
232pub enum DavDirectoryReadError {
233    #[error("directory page size limit must be non-zero")]
234    InvalidLimit,
235    #[error("directory enumeration was cancelled")]
236    Cancelled,
237    #[error("directory page limit exceeded")]
238    PageLimitExceeded,
239    #[error(transparent)]
240    Backend(#[from] DavBackendError),
241    #[error(transparent)]
242    InvalidPage(#[from] DavDirectoryPageValidationError),
243}
244
245/// Requests and validates the next directory page.
246///
247/// Cancellation is checked before every backend call. The product cursor remains opaque and is
248/// moved into the state only after the page passes all validation. An invalid product page poisons
249/// the state so subsequent calls return the same validation error without repeating the backend
250/// request.
251///
252/// # Errors
253///
254/// Returns an error on cancellation, backend failure, exhausted limits, or an invalid page.
255pub async fn read_next_directory_page<E, C>(
256    enumerator: &E,
257    path: &DavPath,
258    state: &mut DavDirectoryPageState<E::Cursor>,
259    requested_entries: usize,
260    limits: DavDirectoryPageLimits,
261    cancellation: &C,
262) -> Result<Option<DavValidatedDirectoryPage<E::Entry>>, DavDirectoryReadError>
263where
264    E: DavDirectoryEnumerator,
265    C: DavCancellation,
266{
267    if let Some(error) = state.poisoned {
268        return Err(DavDirectoryReadError::InvalidPage(error));
269    }
270    if state.finished {
271        return Ok(None);
272    }
273    if requested_entries == 0 || limits.maximum_entries == 0 || limits.maximum_pages == 0 {
274        return Err(DavDirectoryReadError::InvalidLimit);
275    }
276    if cancellation.is_cancelled() {
277        return Err(DavDirectoryReadError::Cancelled);
278    }
279    if state.pages_read >= limits.maximum_pages {
280        return Err(DavDirectoryReadError::PageLimitExceeded);
281    }
282    let maximum_entries = requested_entries.min(limits.maximum_entries);
283    let page = enumerator
284        .read_directory_page(DavDirectoryPageRequest {
285            path,
286            cursor: state.current_cursor.as_ref(),
287            maximum_entries,
288        })
289        .await?;
290    if let Err(error) = validate_directory_page(
291        state.current_cursor.as_ref(),
292        &state.seen_cursors,
293        state.last_stable_key.as_deref(),
294        maximum_entries,
295        &page,
296    ) {
297        state.poisoned = Some(error);
298        return Err(DavDirectoryReadError::InvalidPage(error));
299    }
300
301    let has_more = page.next_cursor.is_some();
302    if let Some(last) = page.entries.last() {
303        state.last_stable_key = Some(Bytes::copy_from_slice(last.stable_key()));
304    }
305    if let Some(cursor) = page.next_cursor {
306        if let Some(previous) = state.current_cursor.replace(cursor) {
307            state.seen_cursors.insert(previous);
308        }
309    } else {
310        state.current_cursor = None;
311    }
312    state.finished = !has_more;
313    state.pages_read += 1;
314    Ok(Some(DavValidatedDirectoryPage {
315        entries: page.entries,
316        has_more,
317    }))
318}
319
320/// Hard limits for recursive resource traversal.
321#[derive(Debug, Clone, Copy, PartialEq, Eq)]
322pub struct DavTraversalLimits {
323    pub maximum_visited_resources: usize,
324    pub maximum_queued_work_items: usize,
325    pub maximum_failures: usize,
326    pub maximum_depth: Option<usize>,
327}
328
329impl DavTraversalLimits {
330    #[must_use]
331    pub const fn new(
332        maximum_visited_resources: usize,
333        maximum_queued_work_items: usize,
334        maximum_failures: usize,
335        maximum_depth: Option<usize>,
336    ) -> Self {
337        Self {
338            maximum_visited_resources,
339            maximum_queued_work_items,
340            maximum_failures,
341            maximum_depth,
342        }
343    }
344}
345
346/// Current recursive traversal accounting.
347#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
348pub struct DavTraversalProgress {
349    pub visited_resources: usize,
350    pub queued_work_items: usize,
351    pub failures: usize,
352    pub completed_mutations: usize,
353}
354
355/// Typed recursive traversal stop reason.
356#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
357pub enum DavTraversalErrorKind {
358    #[error("invalid recursive traversal limits")]
359    InvalidLimits,
360    #[error("recursive traversal was cancelled")]
361    Cancelled,
362    #[error("recursive traversal visited-resource limit exceeded")]
363    VisitedResourceLimitExceeded,
364    #[error("recursive traversal work-queue limit exceeded")]
365    QueuedWorkLimitExceeded,
366    #[error("recursive traversal failure limit exceeded")]
367    FailureLimitExceeded,
368    #[error("recursive traversal depth limit exceeded")]
369    DepthLimitExceeded,
370}
371
372/// Recursive traversal failure with partial-execution accounting.
373#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
374#[error("{kind}")]
375pub struct DavTraversalError {
376    pub kind: DavTraversalErrorKind,
377    pub progress: DavTraversalProgress,
378}
379
380impl DavTraversalError {
381    #[must_use]
382    pub const fn partial_execution(self) -> bool {
383        self.progress.completed_mutations != 0
384    }
385}
386
387/// Allocation-free recursive traversal budget state.
388#[derive(Debug, Clone, Copy, PartialEq, Eq)]
389pub struct DavTraversalBudget {
390    limits: DavTraversalLimits,
391    progress: DavTraversalProgress,
392}
393
394impl DavTraversalBudget {
395    ///
396    /// # Errors
397    ///
398    /// Returns [`DavTraversalError`] when any required traversal limit is zero.
399    pub fn new(limits: DavTraversalLimits) -> Result<Self, DavTraversalError> {
400        if limits.maximum_visited_resources == 0
401            || limits.maximum_queued_work_items == 0
402            || limits.maximum_failures == 0
403        {
404            return Err(DavTraversalError {
405                kind: DavTraversalErrorKind::InvalidLimits,
406                progress: DavTraversalProgress::default(),
407            });
408        }
409        Ok(Self {
410            limits,
411            progress: DavTraversalProgress::default(),
412        })
413    }
414
415    #[must_use]
416    pub const fn progress(self) -> DavTraversalProgress {
417        self.progress
418    }
419
420    ///
421    /// # Errors
422    ///
423    /// Returns [`DavTraversalError`] when cancellation has been requested.
424    pub fn checkpoint(&self, cancellation: &impl DavCancellation) -> Result<(), DavTraversalError> {
425        if cancellation.is_cancelled() {
426            Err(self.error(DavTraversalErrorKind::Cancelled))
427        } else {
428            Ok(())
429        }
430    }
431
432    ///
433    /// # Errors
434    ///
435    /// Returns [`DavTraversalError`] when depth or visited-resource limits are exceeded.
436    pub fn visit(&mut self, depth: usize) -> Result<(), DavTraversalError> {
437        if self
438            .limits
439            .maximum_depth
440            .is_some_and(|maximum| depth > maximum)
441        {
442            return Err(self.error(DavTraversalErrorKind::DepthLimitExceeded));
443        }
444        self.progress.visited_resources = checked_increment(
445            self.progress.visited_resources,
446            self.limits.maximum_visited_resources,
447        )
448        .ok_or_else(|| self.error(DavTraversalErrorKind::VisitedResourceLimitExceeded))?;
449        Ok(())
450    }
451
452    ///
453    /// # Errors
454    ///
455    /// Returns [`DavTraversalError`] when the queued-work counter would exceed its limit.
456    pub fn reserve_work(&mut self, additional: usize) -> Result<(), DavTraversalError> {
457        let Some(queued) = self.progress.queued_work_items.checked_add(additional) else {
458            return Err(self.error(DavTraversalErrorKind::QueuedWorkLimitExceeded));
459        };
460        if queued > self.limits.maximum_queued_work_items {
461            return Err(self.error(DavTraversalErrorKind::QueuedWorkLimitExceeded));
462        }
463        self.progress.queued_work_items = queued;
464        Ok(())
465    }
466
467    pub fn complete_work(&mut self) {
468        self.progress.queued_work_items = self.progress.queued_work_items.saturating_sub(1);
469    }
470
471    ///
472    /// # Errors
473    ///
474    /// Returns [`DavTraversalError`] when the failure counter would exceed its limit.
475    pub fn record_failure(&mut self) -> Result<(), DavTraversalError> {
476        self.progress.failures =
477            checked_increment(self.progress.failures, self.limits.maximum_failures)
478                .ok_or_else(|| self.error(DavTraversalErrorKind::FailureLimitExceeded))?;
479        Ok(())
480    }
481
482    pub fn record_completed_mutation(&mut self) {
483        self.progress.completed_mutations = self.progress.completed_mutations.saturating_add(1);
484    }
485
486    const fn error(&self, kind: DavTraversalErrorKind) -> DavTraversalError {
487        DavTraversalError {
488            kind,
489            progress: self.progress,
490        }
491    }
492}
493
494fn checked_increment(value: usize, maximum: usize) -> Option<usize> {
495    value.checked_add(1).filter(|next| *next <= maximum)
496}
aster_forge_webdav/traversal.rs

aster_forge_webdav/
traversal.rs
