ailloli_ui_fs_runtime/lib.rs
1//! Worker-owned filesystem sources and bounded UI delivery.
2//!
3//! A [`FileTreeRuntime`] moves provider I/O off the UI thread, accepts work
4//! through a bounded request queue, and returns bounded batches through a
5//! wake-aware runtime inbox. [`FileTreeReconcileScheduler`] supplies targeted
6//! polling only when a provider lacks native watch support.
7//!
8//! # Examples
9//!
10//! ```
11//! use ailloli_ui_fs_runtime::FileTreeReconcileScheduler;
12//! let scheduler = FileTreeReconcileScheduler::new(false);
13//! assert!(!scheduler.uses_native_watch());
14//! assert!(scheduler.is_empty());
15//! ```
16
17use std::collections::{HashMap, HashSet};
18use std::num::NonZeroUsize;
19use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
20use std::sync::mpsc::{self, SyncSender, TrySendError};
21use std::sync::{Arc, Mutex};
22use std::thread::{self, JoinHandle};
23use std::time::{Duration, Instant};
24
25use ailloli_ui_fs::{
26 DirectoryLoadRequest, FileEntry, FileError, FileIdentity, FileKind, FileMetadata,
27 FileTreeNodeId, FileTreeSource, FileTreeSourceFactory, FileTreeStore, FileTreeStoreDelta,
28 FileTreeStoreError, FileUri, WatchEvent,
29};
30use ailloli_ui_runtime::{UiInbox, UiInboxStats, UiSendError, UiSender, UiWake, UiWakeError};
31
32/// Maximum number of queued worker requests and undrained UI responses.
33///
34/// The value is an item count, not bytes. Non-blocking request methods return
35/// [`FileTreeRuntimeError::QueueFull`] when the request side reaches 256.
36///
37/// # Examples
38///
39/// ```
40/// use ailloli_ui_fs_runtime::FILE_TREE_QUEUE_CAPACITY;
41/// assert_eq!(FILE_TREE_QUEUE_CAPACITY, 256);
42/// ```
43pub const FILE_TREE_QUEUE_CAPACITY: usize = 256;
44/// Maximum worker responses applied by one UI drain.
45///
46/// Watch polling limits are also clamped to this 256-item budget. A zero watch
47/// limit remains zero; draining always uses the nonzero constant itself.
48///
49/// # Examples
50///
51/// ```
52/// use ailloli_ui_fs_runtime::FILE_TREE_UI_DRAIN_BUDGET;
53/// assert_eq!(FILE_TREE_UI_DRAIN_BUDGET, 256);
54/// ```
55pub const FILE_TREE_UI_DRAIN_BUDGET: usize = 256;
56/// Maximum graceful worker-shutdown wait: two seconds.
57///
58/// # Examples
59///
60/// ```
61/// use std::time::Duration;
62/// use ailloli_ui_fs_runtime::FILE_TREE_FINISH_TIMEOUT;
63/// assert_eq!(FILE_TREE_FINISH_TIMEOUT, Duration::from_secs(2));
64/// ```
65pub const FILE_TREE_FINISH_TIMEOUT: Duration = Duration::from_secs(2);
66/// Initial and successful remote reconciliation interval: two seconds.
67///
68/// # Examples
69///
70/// ```
71/// use std::time::Duration;
72/// use ailloli_ui_fs_runtime::FILE_TREE_REMOTE_POLL_INTERVAL;
73/// assert_eq!(FILE_TREE_REMOTE_POLL_INTERVAL, Duration::from_secs(2));
74/// ```
75pub const FILE_TREE_REMOTE_POLL_INTERVAL: Duration = Duration::from_secs(2);
76/// Inclusive ceiling for exponential remote-poll error backoff: 30 seconds.
77///
78/// # Examples
79///
80/// ```
81/// use std::time::Duration;
82/// use ailloli_ui_fs_runtime::FILE_TREE_REMOTE_POLL_MAX_BACKOFF;
83/// assert_eq!(FILE_TREE_REMOTE_POLL_MAX_BACKOFF, Duration::from_secs(30));
84/// ```
85pub const FILE_TREE_REMOTE_POLL_MAX_BACKOFF: Duration = Duration::from_secs(30);
86
87/// Result of submitting logically deduplicated work to the worker.
88///
89/// # Examples
90///
91/// ```
92/// use ailloli_ui_fs_runtime::FileTreeEnqueueOutcome;
93/// assert_ne!(FileTreeEnqueueOutcome::Enqueued, FileTreeEnqueueOutcome::Coalesced);
94/// ```
95#[derive(Debug, Clone, Copy, PartialEq, Eq)]
96pub enum FileTreeEnqueueOutcome {
97 /// A new request occupied one bounded-queue slot.
98 Enqueued,
99 /// An equivalent already-active request covers the requested work.
100 Coalesced,
101}
102
103/// Provider mutation executed serially on the filesystem worker.
104///
105/// URIs are provider-neutral identifiers. `recursive` is forwarded without
106/// reinterpretation, and `CreateEntry` accepts any [`FileKind`] at the type
107/// level although the worker supports only file and directory creation.
108///
109/// # Examples
110///
111/// ```
112/// use ailloli_ui_fs::{FileKind, FileMetadata, FileTreeStore, FileUri};
113/// use ailloli_ui_fs_runtime::FileTreeMutation;
114/// let store = FileTreeStore::new(FileUri::parse("file:///")?, FileMetadata::new(FileKind::Directory))?;
115/// let mutation = FileTreeMutation::Remove {
116/// node_id: store.root(),
117/// uri: FileUri::parse("file:///")?,
118/// recursive: true,
119/// };
120/// assert_eq!(mutation.target_node_id(), store.root());
121/// # Ok::<(), Box<dyn std::error::Error>>(())
122/// ```
123#[non_exhaustive]
124#[derive(Debug, Clone, PartialEq, Eq)]
125pub enum FileTreeMutation {
126 /// Creates a directory and later inserts the attested result under `parent`.
127 CreateDirectory {
128 /// Existing store node that receives the new child.
129 parent: FileTreeNodeId,
130 /// Exact provider URI to create.
131 uri: FileUri,
132 },
133 /// Creates a file or directory with an identity reserved by the UI store.
134 CreateEntry {
135 /// Existing store node that receives the new child.
136 parent: FileTreeNodeId,
137 /// Store-local ID reserved before enqueueing the operation.
138 node_id: FileTreeNodeId,
139 /// Exact provider URI to create.
140 uri: FileUri,
141 /// Entry kind; only [`FileKind::File`] and [`FileKind::Directory`] work.
142 kind: FileKind,
143 },
144 /// Moves an existing entry and preserves its store-local node ID.
145 Move {
146 /// Existing store-local ID to update after provider attestation.
147 node_id: FileTreeNodeId,
148 /// Exact current provider URI.
149 from: FileUri,
150 /// Exact destination provider URI.
151 to: FileUri,
152 },
153 /// Removes an existing entry after provider confirmation.
154 Remove {
155 /// Existing store-local ID to remove.
156 node_id: FileTreeNodeId,
157 /// Exact provider URI to remove.
158 uri: FileUri,
159 /// Whether provider-defined recursive removal is permitted.
160 recursive: bool,
161 },
162}
163
164/// Accessors for store ownership and optional reserved identity.
165impl FileTreeMutation {
166 /// Returns the node serialized against other active mutations.
167 ///
168 /// Create operations target their parent; move/remove operations target the
169 /// affected node.
170 ///
171 /// # Examples
172 ///
173 /// ```
174 /// use ailloli_ui_fs::{FileKind, FileMetadata, FileTreeStore, FileUri};
175 /// use ailloli_ui_fs_runtime::FileTreeMutation;
176 /// let store = FileTreeStore::new(FileUri::parse("file:///")?, FileMetadata::new(FileKind::Directory))?;
177 /// let mutation = FileTreeMutation::CreateDirectory { parent: store.root(), uri: FileUri::parse("file:///new")? };
178 /// assert_eq!(mutation.target_node_id(), store.root());
179 /// # Ok::<(), Box<dyn std::error::Error>>(())
180 /// ```
181 pub const fn target_node_id(&self) -> FileTreeNodeId {
182 match self {
183 Self::CreateDirectory { parent, .. } | Self::CreateEntry { parent, .. } => *parent,
184 Self::Move { node_id, .. } | Self::Remove { node_id, .. } => *node_id,
185 }
186 }
187
188 /// Returns the preallocated ID only for [`FileTreeMutation::CreateEntry`].
189 ///
190 /// # Examples
191 ///
192 /// ```
193 /// use ailloli_ui_fs::{FileKind, FileMetadata, FileTreeStore, FileUri};
194 /// use ailloli_ui_fs_runtime::FileTreeMutation;
195 /// let mut store = FileTreeStore::new(FileUri::parse("file:///")?, FileMetadata::new(FileKind::Directory))?;
196 /// let reserved = store.reserve_node_id()?;
197 /// let mutation = FileTreeMutation::CreateEntry { parent: store.root(), node_id: reserved, uri: FileUri::parse("file:///new.txt")?, kind: FileKind::File };
198 /// assert_eq!(mutation.reserved_node_id(), Some(reserved));
199 /// # Ok::<(), Box<dyn std::error::Error>>(())
200 /// ```
201 pub const fn reserved_node_id(&self) -> Option<FileTreeNodeId> {
202 match self {
203 Self::CreateEntry { node_id, .. } => Some(*node_id),
204 _ => None,
205 }
206 }
207}
208
209/// Worker mutation envelope with a monotonically allocated request ID.
210///
211/// IDs start at one per runtime and never use zero. Values are observable in
212/// mutation responses but cannot be constructed directly by consumers.
213///
214/// # Examples
215///
216/// ```
217/// use ailloli_ui_fs_runtime::FileTreeMutationRequest;
218/// fn inspect(request: &FileTreeMutationRequest) {
219/// assert!(request.request_id() > 0);
220/// }
221/// ```
222#[derive(Debug, Clone, PartialEq, Eq)]
223pub struct FileTreeMutationRequest {
224 /// Runtime-local nonzero request sequence.
225 request_id: u64,
226 /// Exact mutation sent to the provider worker.
227 mutation: FileTreeMutation,
228}
229
230/// Read-only access to a worker mutation envelope.
231impl FileTreeMutationRequest {
232 /// Returns the runtime-local nonzero request sequence.
233 ///
234 /// # Examples
235 ///
236 /// ```
237 /// use ailloli_ui_fs_runtime::FileTreeMutationRequest;
238 /// fn inspect(request: &FileTreeMutationRequest) {
239 /// let id: u64 = request.request_id();
240 /// assert_ne!(id, 0);
241 /// }
242 /// ```
243 pub const fn request_id(&self) -> u64 {
244 self.request_id
245 }
246
247 /// Borrows the exact submitted mutation.
248 ///
249 /// # Examples
250 ///
251 /// ```
252 /// use ailloli_ui_fs_runtime::{FileTreeMutation, FileTreeMutationRequest};
253 /// fn inspect(request: &FileTreeMutationRequest) {
254 /// let _: &FileTreeMutation = request.mutation();
255 /// }
256 /// ```
257 pub const fn mutation(&self) -> &FileTreeMutation {
258 &self.mutation
259 }
260}
261
262/// Immediate UI-side effects of accepting or coalescing a mutation.
263///
264/// # Examples
265///
266/// ```
267/// use ailloli_ui_fs_runtime::{FileTreeEnqueueOutcome, FileTreeMutationEnqueue};
268/// fn inspect(result: &FileTreeMutationEnqueue) {
269/// assert!(matches!(result.outcome, FileTreeEnqueueOutcome::Enqueued | FileTreeEnqueueOutcome::Coalesced));
270/// }
271/// ```
272#[derive(Debug)]
273pub struct FileTreeMutationEnqueue {
274 /// Whether new worker work was queued or an identical mutation coalesced.
275 pub outcome: FileTreeEnqueueOutcome,
276 /// Store delta that pins the target as having a pending operation.
277 pub pending_delta: FileTreeStoreDelta,
278}
279
280/// One provider result delivered from the worker to the UI thread.
281///
282/// Directory and mutation responses retain their request envelope so stale
283/// generations and reserved IDs can be resolved deterministically.
284///
285/// # Examples
286///
287/// ```
288/// use ailloli_ui_fs_runtime::FileTreeWorkerResponse;
289/// let response = FileTreeWorkerResponse::Watch { events: Ok(Vec::new()) };
290/// assert!(response.directory_request().is_none());
291/// ```
292#[non_exhaustive]
293#[derive(Debug)]
294pub enum FileTreeWorkerResponse {
295 /// Directory entries and best-effort identities for one load request.
296 Directory {
297 /// Original store generation/request identity.
298 request: DirectoryLoadRequest,
299 /// Provider result; each identity may be absent or fail independently.
300 result: Result<Vec<(FileEntry, Option<FileIdentity>)>, FileError>,
301 },
302 /// Native or explicitly polled watch events.
303 Watch {
304 /// Ordered events, an empty successful batch, or one provider error.
305 events: Result<Vec<WatchEvent>, FileError>,
306 },
307 /// Result of enabling or disabling provider watch for one URI.
308 WatchConfigured {
309 /// Exact directory URI supplied by the UI.
310 uri: FileUri,
311 /// `true` for watch and `false` for unwatch.
312 enabled: bool,
313 /// Provider acknowledgement or typed error.
314 result: Result<(), FileError>,
315 },
316 /// Provider mutation result and optional resulting identity.
317 Mutation {
318 /// Original mutation envelope.
319 request: FileTreeMutationRequest,
320 /// Attested identity, `None` when unavailable/removing, or an error.
321 result: Result<Option<FileIdentity>, FileError>,
322 },
323}
324
325/// Variant inspection helpers for worker responses.
326impl FileTreeWorkerResponse {
327 /// Returns the directory request only for [`Self::Directory`].
328 ///
329 /// # Examples
330 ///
331 /// ```
332 /// use ailloli_ui_fs_runtime::FileTreeWorkerResponse;
333 /// let response = FileTreeWorkerResponse::Watch { events: Ok(Vec::new()) };
334 /// assert_eq!(response.directory_request(), None);
335 /// ```
336 pub fn directory_request(&self) -> Option<&DirectoryLoadRequest> {
337 match self {
338 Self::Directory { request, .. } => Some(request),
339 _ => None,
340 }
341 }
342}
343
344/// Point-in-time, monotonically accumulated worker metrics.
345///
346/// Counters saturate only by normal atomic integer wraparound; they are
347/// diagnostics, not synchronization. Depths are item counts sampled with
348/// relaxed ordering and may change immediately after the snapshot.
349///
350/// # Examples
351///
352/// ```
353/// use ailloli_ui_fs_runtime::FileTreeWorkerStats;
354/// let stats = FileTreeWorkerStats::default();
355/// assert_eq!(stats.requests_enqueued, 0);
356/// assert_eq!(stats.request_queue_depth, 0);
357/// ```
358#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
359pub struct FileTreeWorkerStats {
360 /// Successfully enqueued requests since runtime creation.
361 pub requests_enqueued: u64,
362 /// Equivalent directory, watch, or mutation requests not enqueued again.
363 pub requests_coalesced: u64,
364 /// Directory reads begun by the worker.
365 pub directory_reads: u64,
366 /// Native or explicit provider watch polls begun by the worker.
367 pub watch_polls: u64,
368 /// Responses offered to the UI inbox, including wake-failed enqueue success.
369 pub responses: u64,
370 /// Directory responses rejected by the store as stale.
371 pub stale_responses: u64,
372 /// Mutation calls begun by the worker.
373 pub mutations: u64,
374 /// Requests reserved or queued but not yet received by the worker.
375 pub request_queue_depth: usize,
376 /// Maximum observed request depth since runtime creation.
377 pub request_queue_max_depth: usize,
378 /// Distinct directory nodes with an owned current request.
379 pub active_directory_requests: usize,
380 /// Successfully configured provider watch URIs tracked by the worker.
381 pub watched_directories: usize,
382}
383
384/// Atomic backing storage shared by the UI handle and worker.
385#[derive(Default)]
386struct AtomicWorkerStats {
387 /// Atomic backing for `requests_enqueued`.
388 requests_enqueued: AtomicU64,
389 /// Atomic backing for `requests_coalesced`.
390 requests_coalesced: AtomicU64,
391 /// Atomic backing for `directory_reads`.
392 directory_reads: AtomicU64,
393 /// Atomic backing for `watch_polls`.
394 watch_polls: AtomicU64,
395 /// Atomic backing for `responses`.
396 responses: AtomicU64,
397 /// Atomic backing for `stale_responses`.
398 stale_responses: AtomicU64,
399 /// Atomic backing for `mutations`.
400 mutations: AtomicU64,
401 /// Current reserved/queued request count.
402 request_queue_depth: std::sync::atomic::AtomicUsize,
403 /// Maximum observed reserved/queued request count.
404 request_queue_max_depth: std::sync::atomic::AtomicUsize,
405 /// Current distinct directory request ownership count.
406 active_directory_requests: std::sync::atomic::AtomicUsize,
407 /// Current successfully watched URI count.
408 watched_directories: std::sync::atomic::AtomicUsize,
409}
410
411/// Atomic metrics operations used across the UI/worker boundary.
412impl AtomicWorkerStats {
413 /// Loads one relaxed point-in-time metrics snapshot.
414 fn snapshot(&self) -> FileTreeWorkerStats {
415 FileTreeWorkerStats {
416 requests_enqueued: self.requests_enqueued.load(Ordering::Relaxed),
417 requests_coalesced: self.requests_coalesced.load(Ordering::Relaxed),
418 directory_reads: self.directory_reads.load(Ordering::Relaxed),
419 watch_polls: self.watch_polls.load(Ordering::Relaxed),
420 responses: self.responses.load(Ordering::Relaxed),
421 stale_responses: self.stale_responses.load(Ordering::Relaxed),
422 mutations: self.mutations.load(Ordering::Relaxed),
423 request_queue_depth: self.request_queue_depth.load(Ordering::Relaxed),
424 request_queue_max_depth: self.request_queue_max_depth.load(Ordering::Relaxed),
425 active_directory_requests: self.active_directory_requests.load(Ordering::Relaxed),
426 watched_directories: self.watched_directories.load(Ordering::Relaxed),
427 }
428 }
429
430 /// Reserves one request-depth slot before attempting a send.
431 fn request_reserved(&self) {
432 let depth = self.request_queue_depth.fetch_add(1, Ordering::AcqRel) + 1;
433 self.request_queue_max_depth
434 .fetch_max(depth, Ordering::Relaxed);
435 }
436
437 /// Rolls back a request reservation after a failed send.
438 fn request_reservation_cancelled(&self) {
439 self.request_queue_depth.fetch_sub(1, Ordering::AcqRel);
440 }
441
442 /// Releases a request-depth slot after the worker receives it.
443 fn request_received(&self) {
444 self.request_queue_depth.fetch_sub(1, Ordering::AcqRel);
445 }
446}
447
448/// Next poll instant and current exponential delay for one directory.
449#[derive(Debug, Clone, Copy)]
450struct ReconcileSchedule {
451 /// Earliest instant at which the node is due.
452 next_due: Instant,
453 /// Current delay, ranging from two through 30 seconds.
454 backoff: Duration,
455}
456
457/// UI-owned targeted polling policy for sources without native watch support.
458/// Only expanded directories are scheduled; rendering never calls it.
459///
460/// # Examples
461///
462/// ```
463/// use ailloli_ui_fs_runtime::FileTreeReconcileScheduler;
464/// assert!(FileTreeReconcileScheduler::new(false).is_empty());
465/// assert!(FileTreeReconcileScheduler::new(true).uses_native_watch());
466/// ```
467#[derive(Debug)]
468pub struct FileTreeReconcileScheduler {
469 /// Whether provider notifications make polling unnecessary.
470 native_watch: bool,
471 /// One polling schedule per expanded store node.
472 scheduled: HashMap<FileTreeNodeId, ReconcileSchedule>,
473}
474
475/// Targeted expansion-driven polling operations.
476impl FileTreeReconcileScheduler {
477 /// Creates an empty scheduler for the provider's watch capability.
478 ///
479 /// A `true` value permanently disables schedule creation.
480 ///
481 /// # Examples
482 ///
483 /// ```
484 /// use ailloli_ui_fs_runtime::FileTreeReconcileScheduler;
485 /// let scheduler = FileTreeReconcileScheduler::new(true);
486 /// assert!(scheduler.uses_native_watch());
487 /// assert_eq!(scheduler.len(), 0);
488 /// ```
489 pub fn new(native_watch: bool) -> Self {
490 Self {
491 native_watch,
492 scheduled: HashMap::new(),
493 }
494 }
495
496 /// Reports the immutable provider watch mode selected at construction.
497 ///
498 /// # Examples
499 ///
500 /// ```
501 /// use ailloli_ui_fs_runtime::FileTreeReconcileScheduler;
502 /// assert!(!FileTreeReconcileScheduler::new(false).uses_native_watch());
503 /// ```
504 pub const fn uses_native_watch(&self) -> bool {
505 self.native_watch
506 }
507
508 /// Adds or removes polling for a directory's expansion state.
509 ///
510 /// Expanding an already scheduled node preserves its current deadline and
511 /// backoff. New non-native entries first become due two seconds after
512 /// `now`; collapsing removes them. Native-watch schedulers always stay empty.
513 ///
514 /// # Examples
515 ///
516 /// ```
517 /// use std::time::Instant;
518 /// use ailloli_ui_fs::{FileKind, FileMetadata, FileTreeStore, FileUri};
519 /// use ailloli_ui_fs_runtime::FileTreeReconcileScheduler;
520 /// let store = FileTreeStore::new(FileUri::parse("file:///")?, FileMetadata::new(FileKind::Directory))?;
521 /// let mut scheduler = FileTreeReconcileScheduler::new(false);
522 /// scheduler.set_expanded(store.root(), true, Instant::now());
523 /// assert_eq!(scheduler.len(), 1);
524 /// # Ok::<(), Box<dyn std::error::Error>>(())
525 /// ```
526 pub fn set_expanded(&mut self, node_id: FileTreeNodeId, expanded: bool, now: Instant) {
527 if self.native_watch || !expanded {
528 self.scheduled.remove(&node_id);
529 return;
530 }
531 self.scheduled.entry(node_id).or_insert(ReconcileSchedule {
532 next_due: now + FILE_TREE_REMOTE_POLL_INTERVAL,
533 backoff: FILE_TREE_REMOTE_POLL_INTERVAL,
534 });
535 }
536
537 /// Returns at most `limit` due node IDs in ascending raw-ID order.
538 ///
539 /// `limit == 0` returns an empty vector without changing schedules.
540 ///
541 /// # Examples
542 ///
543 /// ```
544 /// use std::time::Instant;
545 /// use ailloli_ui_fs_runtime::FileTreeReconcileScheduler;
546 /// let scheduler = FileTreeReconcileScheduler::new(false);
547 /// assert!(scheduler.due(Instant::now(), 0).is_empty());
548 /// ```
549 pub fn due(&self, now: Instant, limit: usize) -> Vec<FileTreeNodeId> {
550 let mut due = self
551 .scheduled
552 .iter()
553 .filter_map(|(id, schedule)| (schedule.next_due <= now).then_some(*id))
554 .collect::<Vec<_>>();
555 due.sort_by_key(|id| id.get());
556 due.truncate(limit);
557 due
558 }
559
560 /// Earliest instant at which a non-native expanded directory must be
561 /// reconciled. Hosts use this to arm one targeted UI timer instead of
562 /// polling from every frame.
563 ///
564 /// Returns `None` when no directory is scheduled.
565 ///
566 /// # Examples
567 ///
568 /// ```
569 /// use ailloli_ui_fs_runtime::FileTreeReconcileScheduler;
570 /// assert_eq!(FileTreeReconcileScheduler::new(false).next_due(), None);
571 /// ```
572 pub fn next_due(&self) -> Option<Instant> {
573 self.scheduled
574 .values()
575 .map(|schedule| schedule.next_due)
576 .min()
577 }
578
579 /// Resets an existing node to the two-second base interval after success.
580 ///
581 /// Unknown IDs are ignored.
582 ///
583 /// # Examples
584 ///
585 /// ```
586 /// use std::time::Instant;
587 /// use ailloli_ui_fs::{FileKind, FileMetadata, FileTreeStore, FileUri};
588 /// use ailloli_ui_fs_runtime::FileTreeReconcileScheduler;
589 /// let store = FileTreeStore::new(FileUri::parse("file:///")?, FileMetadata::new(FileKind::Directory))?;
590 /// let mut scheduler = FileTreeReconcileScheduler::new(false);
591 /// scheduler.note_success(store.root(), Instant::now());
592 /// assert!(scheduler.is_empty());
593 /// # Ok::<(), Box<dyn std::error::Error>>(())
594 /// ```
595 pub fn note_success(&mut self, node_id: FileTreeNodeId, now: Instant) {
596 if let Some(schedule) = self.scheduled.get_mut(&node_id) {
597 schedule.backoff = FILE_TREE_REMOTE_POLL_INTERVAL;
598 schedule.next_due = now + schedule.backoff;
599 }
600 }
601
602 /// Doubles an existing node's delay, saturating and clamping at 30 seconds.
603 ///
604 /// Unknown IDs are ignored; overflow cannot exceed the public maximum.
605 ///
606 /// # Examples
607 ///
608 /// ```
609 /// use std::time::Instant;
610 /// use ailloli_ui_fs::{FileKind, FileMetadata, FileTreeStore, FileUri};
611 /// use ailloli_ui_fs_runtime::FileTreeReconcileScheduler;
612 /// let store = FileTreeStore::new(FileUri::parse("file:///")?, FileMetadata::new(FileKind::Directory))?;
613 /// let mut scheduler = FileTreeReconcileScheduler::new(false);
614 /// scheduler.note_error(store.root(), Instant::now());
615 /// assert_eq!(scheduler.len(), 0);
616 /// # Ok::<(), Box<dyn std::error::Error>>(())
617 /// ```
618 pub fn note_error(&mut self, node_id: FileTreeNodeId, now: Instant) {
619 if let Some(schedule) = self.scheduled.get_mut(&node_id) {
620 schedule.backoff = schedule
621 .backoff
622 .saturating_mul(2)
623 .min(FILE_TREE_REMOTE_POLL_MAX_BACKOFF);
624 schedule.next_due = now + schedule.backoff;
625 }
626 }
627
628 /// Returns the number of expanded nodes scheduled for polling.
629 ///
630 /// # Examples
631 ///
632 /// ```
633 /// use ailloli_ui_fs_runtime::FileTreeReconcileScheduler;
634 /// assert_eq!(FileTreeReconcileScheduler::new(false).len(), 0);
635 /// ```
636 pub fn len(&self) -> usize {
637 self.scheduled.len()
638 }
639
640 /// Reports whether no directory is scheduled for polling.
641 ///
642 /// # Examples
643 ///
644 /// ```
645 /// use ailloli_ui_fs_runtime::FileTreeReconcileScheduler;
646 /// assert!(FileTreeReconcileScheduler::new(true).is_empty());
647 /// ```
648 pub fn is_empty(&self) -> bool {
649 self.scheduled.is_empty()
650 }
651}
652
653/// Failure reported while starting, enqueueing, applying, waking, or stopping.
654///
655/// Queue saturation and closure are recoverable request outcomes. Store errors
656/// may occur after a provider response and can leave earlier responses from the
657/// same drain already applied.
658///
659/// # Examples
660///
661/// ```
662/// use ailloli_ui_fs_runtime::FileTreeRuntimeError;
663/// assert_eq!(FileTreeRuntimeError::QueueFull.to_string(), "filesystem worker request queue is full");
664/// ```
665#[derive(Debug, thiserror::Error)]
666pub enum FileTreeRuntimeError {
667 #[error("failed to spawn filesystem worker: {0}")]
668 /// The operating system refused to start the named worker thread.
669 Spawn(std::io::Error),
670 #[error("filesystem source initialization failed: {0}")]
671 /// The factory failed inside the worker before the runtime became usable.
672 Source(FileError),
673 #[error("filesystem worker request queue is full")]
674 /// A non-blocking send found all 256 request slots occupied.
675 QueueFull,
676 #[error("filesystem worker is closed")]
677 /// The worker receiver disconnected before accepting a request.
678 Closed,
679 #[error("another filesystem mutation is active for node {0:?}")]
680 /// A different mutation already owns the same target node.
681 MutationBusy(FileTreeNodeId),
682 #[error("filesystem mutation request identifier space is exhausted")]
683 /// Incrementing the runtime-local `u64` mutation ID would overflow.
684 MutationIdentifierExhausted,
685 #[error("filesystem UI wake failed: {0}")]
686 /// Work was delivered or drained but the installed UI wake failed.
687 Wake(UiWakeError),
688 #[error("filesystem worker did not stop within {0:?}")]
689 /// Graceful shutdown exceeded the contained duration (normally two seconds).
690 FinishTimeout(Duration),
691 #[error("filesystem worker panicked")]
692 /// Source initialization or worker execution unwound across the thread boundary.
693 ThreadPanicked,
694 #[error(transparent)]
695 /// Applying UI-side state violated a [`FileTreeStore`] invariant.
696 Store(#[from] FileTreeStoreError),
697}
698
699/// Commands sent over the bounded worker request channel.
700enum WorkerRequest {
701 /// Reads a directory for one store generation/request.
702 Directory(DirectoryLoadRequest),
703 /// Executes a serialized provider mutation.
704 Mutation(FileTreeMutationRequest),
705 /// Enables or disables watch for an exact provider URI.
706 ConfigureWatch {
707 /// Exact directory URI.
708 uri: FileUri,
709 /// `true` to watch; `false` to unwatch.
710 enabled: bool,
711 },
712 /// Requests one explicit provider watch poll.
713 Watch {
714 /// Maximum events requested after clamping to the UI budget.
715 limit: usize,
716 },
717 /// Requests graceful worker-loop termination.
718 Shutdown,
719}
720
721/// One bounded UI drain before any results are applied to a store.
722///
723/// # Examples
724///
725/// ```
726/// use ailloli_ui_fs_runtime::FileTreeRuntimeDrain;
727/// let drain = FileTreeRuntimeDrain { responses: Vec::new(), remaining: false };
728/// assert!(drain.responses.is_empty() && !drain.remaining);
729/// ```
730#[derive(Debug)]
731pub struct FileTreeRuntimeDrain {
732 /// Up to [`FILE_TREE_UI_DRAIN_BUDGET`] ordered worker responses.
733 pub responses: Vec<FileTreeWorkerResponse>,
734 /// Whether the UI inbox still held responses after this batch.
735 pub remaining: bool,
736}
737
738/// Store effects and provider-side events produced by one applied drain.
739///
740/// Empty vectors mean that category produced no result. Mutation/provider watch
741/// failures are collected instead of aborting the remaining batch; store
742/// invariant failures are returned directly from `drain_into_store`.
743///
744/// # Examples
745///
746/// ```
747/// use ailloli_ui_fs_runtime::FileTreeApplyReport;
748/// let report = FileTreeApplyReport::default();
749/// assert!(report.deltas.is_empty());
750/// assert_eq!(report.stale_responses, 0);
751/// assert!(!report.remaining);
752/// ```
753#[derive(Debug, Default)]
754pub struct FileTreeApplyReport {
755 /// Ordered UI-store deltas successfully applied in this batch.
756 pub deltas: Vec<FileTreeStoreDelta>,
757 /// Successfully received watch events, not yet interpreted by the caller.
758 pub watch_events: Vec<WatchEvent>,
759 /// Provider watch-poll errors collected without aborting the batch.
760 pub watch_errors: Vec<FileError>,
761 /// Ordered `(uri, enabled, result)` watch-configuration acknowledgements.
762 pub watch_configuration: Vec<(FileUri, bool, Result<(), FileError>)>,
763 /// Directory responses rejected because their store generation was stale.
764 pub stale_responses: usize,
765 /// Mutation requests and provider errors that prevented store attestation.
766 pub mutation_errors: Vec<(FileTreeMutationRequest, FileError)>,
767 /// Whether additional inbox responses await a later bounded drain.
768 pub remaining: bool,
769}
770
771/// UI-owned handle for one provider worker.
772///
773/// The source is constructed and used only on the named worker thread. Request
774/// and response channels are each bounded to 256 items. Dropping the handle
775/// merely attempts a non-blocking shutdown enqueue; call [`Self::finish`] when
776/// confirmed termination is required.
777///
778/// # Examples
779///
780/// ```
781/// use ailloli_ui_fs_runtime::{FileTreeRuntime, FileTreeWorkerStats};
782/// fn inspect(runtime: &FileTreeRuntime) {
783/// let _: FileTreeWorkerStats = runtime.stats();
784/// }
785/// ```
786pub struct FileTreeRuntime {
787 /// Bounded non-blocking request sender.
788 requests: SyncSender<WorkerRequest>,
789 /// Bounded wake-aware response inbox drained by the UI thread.
790 responses: UiInbox<FileTreeWorkerResponse>,
791 /// Latest request ID owned for each directory node.
792 active_directories: Arc<Mutex<HashMap<FileTreeNodeId, u64>>>,
793 /// Mutation currently serialized for each target node.
794 active_mutations: Arc<Mutex<HashMap<FileTreeNodeId, FileTreeMutation>>>,
795 /// Next nonzero mutation request ID; exhaustion is an explicit error.
796 next_mutation_request_id: AtomicU64,
797 /// Coalescing bit for explicit watch polls.
798 watch_pending: Arc<AtomicBool>,
799 /// Metrics shared with the worker.
800 stats: Arc<AtomicWorkerStats>,
801 /// Provider capability reported during synchronous startup.
802 native_watch: bool,
803 /// Join handle retained until `finish`; present for every successful spawn.
804 thread: Option<JoinHandle<()>>,
805}
806
807/// Worker lifecycle, request, drain, and diagnostic operations.
808impl FileTreeRuntime {
809 /// Starts a named worker and constructs its provider source on that thread.
810 ///
811 /// This call waits synchronously for factory initialization. Both channels
812 /// have [`FILE_TREE_QUEUE_CAPACITY`] items. The first mutation ID is one;
813 /// native-watch capability is fixed from the created source.
814 ///
815 /// # Errors
816 ///
817 /// Returns [`FileTreeRuntimeError::Spawn`] if thread creation fails,
818 /// [`FileTreeRuntimeError::Source`] if the factory fails, or
819 /// [`FileTreeRuntimeError::ThreadPanicked`] if initialization disconnects.
820 ///
821 /// # Panics
822 ///
823 /// Panics only if [`FILE_TREE_QUEUE_CAPACITY`] is changed to zero.
824 ///
825 /// # Examples
826 ///
827 /// ```
828 /// use std::sync::Arc;
829 /// use ailloli_ui_fs::FileTreeSourceFactory;
830 /// use ailloli_ui_fs_runtime::{FileTreeRuntime, FileTreeRuntimeError};
831 /// let _spawn: fn(Arc<dyn FileTreeSourceFactory>) -> Result<FileTreeRuntime, FileTreeRuntimeError> = FileTreeRuntime::spawn;
832 /// ```
833 pub fn spawn(factory: Arc<dyn FileTreeSourceFactory>) -> Result<Self, FileTreeRuntimeError> {
834 let capacity = NonZeroUsize::new(FILE_TREE_QUEUE_CAPACITY).expect("non-zero capacity");
835 let (request_sender, request_receiver) = mpsc::sync_channel(capacity.get());
836 let (response_sender, responses) = UiInbox::channel(capacity);
837 let active_directories = Arc::new(Mutex::new(HashMap::new()));
838 let active_mutations = Arc::new(Mutex::new(HashMap::new()));
839 let watch_pending = Arc::new(AtomicBool::new(false));
840 let stats = Arc::new(AtomicWorkerStats::default());
841 let worker_stats = stats.clone();
842 let (ready_sender, ready_receiver) = mpsc::sync_channel(1);
843 let thread = thread::Builder::new()
844 .name("ailloli-fs-worker".into())
845 .spawn(move || {
846 let source = factory.create();
847 match source {
848 Ok(mut source) => {
849 let native_watch = source.supports_native_watch();
850 if ready_sender.send(Ok(native_watch)).is_ok() {
851 worker_loop(
852 source.as_mut(),
853 &request_receiver,
854 &response_sender,
855 &worker_stats,
856 );
857 }
858 }
859 Err(error) => {
860 let _ = ready_sender.send(Err(error));
861 }
862 }
863 })
864 .map_err(FileTreeRuntimeError::Spawn)?;
865 match ready_receiver.recv() {
866 Ok(Ok(native_watch)) => Ok(Self {
867 requests: request_sender,
868 responses,
869 active_directories,
870 active_mutations,
871 next_mutation_request_id: AtomicU64::new(1),
872 watch_pending,
873 stats,
874 native_watch,
875 thread: Some(thread),
876 }),
877 Ok(Err(error)) => {
878 let _ = thread.join();
879 Err(FileTreeRuntimeError::Source(error))
880 }
881 Err(_) => {
882 let _ = thread.join();
883 Err(FileTreeRuntimeError::ThreadPanicked)
884 }
885 }
886 }
887
888 /// Installs or replaces the UI-thread wake target for pending responses.
889 ///
890 /// The wake object must be thread-safe because the provider worker invokes
891 /// it. Already queued responses remain queued if waking fails.
892 ///
893 /// # Errors
894 ///
895 /// Returns [`FileTreeRuntimeError::Wake`] when late binding cannot notify the
896 /// installed target about already queued work.
897 ///
898 /// # Examples
899 ///
900 /// ```
901 /// use std::sync::Arc;
902 /// use ailloli_ui_runtime::UiWake;
903 /// use ailloli_ui_fs_runtime::{FileTreeRuntime, FileTreeRuntimeError};
904 /// fn install(runtime: &FileTreeRuntime, wake: Arc<dyn UiWake>) -> Result<(), FileTreeRuntimeError> {
905 /// runtime.install_wake(wake)
906 /// }
907 /// ```
908 pub fn install_wake(&self, wake: Arc<dyn UiWake>) -> Result<(), FileTreeRuntimeError> {
909 self.responses
910 .install_wake(wake)
911 .map_err(FileTreeRuntimeError::Wake)
912 }
913
914 /// Removes the current wake target without discarding queued responses.
915 ///
916 /// # Examples
917 ///
918 /// ```
919 /// use ailloli_ui_fs_runtime::FileTreeRuntime;
920 /// fn detach(runtime: &FileTreeRuntime) { runtime.detach_wake(); }
921 /// ```
922 pub fn detach_wake(&self) {
923 self.responses.detach_wake();
924 }
925
926 /// Non-blockingly enqueues one generation-tagged directory load.
927 ///
928 /// An identical active `(node_id, request_id)` returns `Coalesced`; a newer
929 /// request for the same node supersedes ownership of the older response.
930 /// Queue failure restores the previous ownership record.
931 ///
932 /// # Errors
933 ///
934 /// Returns [`FileTreeRuntimeError::QueueFull`] or
935 /// [`FileTreeRuntimeError::Closed`]. Mutex poisoning is recovered.
936 ///
937 /// # Examples
938 ///
939 /// ```
940 /// use ailloli_ui_fs::DirectoryLoadRequest;
941 /// use ailloli_ui_fs_runtime::{FileTreeEnqueueOutcome, FileTreeRuntime, FileTreeRuntimeError};
942 /// fn enqueue(runtime: &FileTreeRuntime, request: DirectoryLoadRequest) -> Result<FileTreeEnqueueOutcome, FileTreeRuntimeError> {
943 /// runtime.request_directory(request)
944 /// }
945 /// ```
946 pub fn request_directory(
947 &self,
948 request: DirectoryLoadRequest,
949 ) -> Result<FileTreeEnqueueOutcome, FileTreeRuntimeError> {
950 let node_id = request.node_id();
951 let request_id = request.request_id();
952 let previous_request = {
953 let mut active = self
954 .active_directories
955 .lock()
956 .unwrap_or_else(|error| error.into_inner());
957 if active.get(&node_id) == Some(&request_id) {
958 self.stats
959 .requests_coalesced
960 .fetch_add(1, Ordering::Relaxed);
961 return Ok(FileTreeEnqueueOutcome::Coalesced);
962 }
963 let previous = active.insert(node_id, request_id);
964 if previous.is_none() {
965 self.stats
966 .active_directory_requests
967 .fetch_add(1, Ordering::Relaxed);
968 }
969 previous
970 };
971 self.stats.request_reserved();
972 if let Err(error) = self.requests.try_send(WorkerRequest::Directory(request)) {
973 self.stats.request_reservation_cancelled();
974 let mut active = self
975 .active_directories
976 .lock()
977 .unwrap_or_else(|poison| poison.into_inner());
978 if active.get(&node_id) == Some(&request_id) {
979 match previous_request {
980 Some(previous_request) => {
981 active.insert(node_id, previous_request);
982 }
983 None => {
984 active.remove(&node_id);
985 self.stats
986 .active_directory_requests
987 .fetch_sub(1, Ordering::Relaxed);
988 }
989 }
990 }
991 return Err(map_request_send_error(error));
992 }
993 self.stats.requests_enqueued.fetch_add(1, Ordering::Relaxed);
994 Ok(FileTreeEnqueueOutcome::Enqueued)
995 }
996
997 /// Non-blockingly requests one provider watch poll.
998 ///
999 /// At most one explicit poll is pending. Repeated calls coalesce, and
1000 /// `limit` is clamped to 256 (while zero remains zero). Native-watch sources
1001 /// may also poll automatically every 50 milliseconds while idle.
1002 ///
1003 /// # Errors
1004 ///
1005 /// Returns [`FileTreeRuntimeError::QueueFull`] or
1006 /// [`FileTreeRuntimeError::Closed`]; failure reopens the coalescing slot.
1007 ///
1008 /// # Examples
1009 ///
1010 /// ```
1011 /// use ailloli_ui_fs_runtime::{FileTreeEnqueueOutcome, FileTreeRuntime, FileTreeRuntimeError};
1012 /// fn poll(runtime: &FileTreeRuntime) -> Result<FileTreeEnqueueOutcome, FileTreeRuntimeError> {
1013 /// runtime.request_watch(32)
1014 /// }
1015 /// ```
1016 pub fn request_watch(
1017 &self,
1018 limit: usize,
1019 ) -> Result<FileTreeEnqueueOutcome, FileTreeRuntimeError> {
1020 if self.watch_pending.swap(true, Ordering::AcqRel) {
1021 self.stats
1022 .requests_coalesced
1023 .fetch_add(1, Ordering::Relaxed);
1024 return Ok(FileTreeEnqueueOutcome::Coalesced);
1025 }
1026 self.stats.request_reserved();
1027 if let Err(error) = self.requests.try_send(WorkerRequest::Watch {
1028 limit: limit.min(FILE_TREE_UI_DRAIN_BUDGET),
1029 }) {
1030 self.stats.request_reservation_cancelled();
1031 self.watch_pending.store(false, Ordering::Release);
1032 return Err(map_request_send_error(error));
1033 }
1034 self.stats.requests_enqueued.fetch_add(1, Ordering::Relaxed);
1035 Ok(FileTreeEnqueueOutcome::Enqueued)
1036 }
1037
1038 /// Marks a store node pending and non-blockingly queues a provider mutation.
1039 ///
1040 /// Mutations serialize by [`FileTreeMutation::target_node_id`]. An identical
1041 /// active mutation coalesces but still returns the current pending-state
1042 /// delta. A different active mutation returns `MutationBusy`. Send/store/ID
1043 /// failures remove ownership and best-effort undo pending/reserved state.
1044 ///
1045 /// # Errors
1046 ///
1047 /// Returns queue/closure, busy-target, request-ID exhaustion, or store
1048 /// errors. Provider failures arrive later in [`FileTreeApplyReport`].
1049 ///
1050 /// # Examples
1051 ///
1052 /// ```
1053 /// use ailloli_ui_fs::FileTreeStore;
1054 /// use ailloli_ui_fs_runtime::{FileTreeMutation, FileTreeMutationEnqueue, FileTreeRuntime, FileTreeRuntimeError};
1055 /// fn enqueue(runtime: &FileTreeRuntime, store: &mut FileTreeStore, mutation: FileTreeMutation) -> Result<FileTreeMutationEnqueue, FileTreeRuntimeError> {
1056 /// runtime.request_mutation(store, mutation)
1057 /// }
1058 /// ```
1059 pub fn request_mutation(
1060 &self,
1061 store: &mut FileTreeStore,
1062 mutation: FileTreeMutation,
1063 ) -> Result<FileTreeMutationEnqueue, FileTreeRuntimeError> {
1064 let target = mutation.target_node_id();
1065 let reserved_node_id = mutation.reserved_node_id();
1066 {
1067 let mut active = self
1068 .active_mutations
1069 .lock()
1070 .unwrap_or_else(|error| error.into_inner());
1071 if active.get(&target) == Some(&mutation) {
1072 self.stats
1073 .requests_coalesced
1074 .fetch_add(1, Ordering::Relaxed);
1075 return Ok(FileTreeMutationEnqueue {
1076 outcome: FileTreeEnqueueOutcome::Coalesced,
1077 pending_delta: store.set_pending_operation(target, true)?,
1078 });
1079 }
1080 if active.contains_key(&target) {
1081 return Err(FileTreeRuntimeError::MutationBusy(target));
1082 }
1083 active.insert(target, mutation.clone());
1084 }
1085 let request_id = match self.next_mutation_request_id.fetch_update(
1086 Ordering::AcqRel,
1087 Ordering::Relaxed,
1088 |current| current.checked_add(1),
1089 ) {
1090 Ok(request_id) => request_id,
1091 Err(_) => {
1092 self.active_mutations
1093 .lock()
1094 .unwrap_or_else(|error| error.into_inner())
1095 .remove(&target);
1096 return Err(FileTreeRuntimeError::MutationIdentifierExhausted);
1097 }
1098 };
1099 let pending_delta = match store.set_pending_operation(target, true) {
1100 Ok(delta) => delta,
1101 Err(error) => {
1102 self.active_mutations
1103 .lock()
1104 .unwrap_or_else(|poison| poison.into_inner())
1105 .remove(&target);
1106 return Err(FileTreeRuntimeError::Store(error));
1107 }
1108 };
1109 let request = FileTreeMutationRequest {
1110 request_id,
1111 mutation,
1112 };
1113 self.stats.request_reserved();
1114 if let Err(error) = self.requests.try_send(WorkerRequest::Mutation(request)) {
1115 self.stats.request_reservation_cancelled();
1116 self.active_mutations
1117 .lock()
1118 .unwrap_or_else(|poison| poison.into_inner())
1119 .remove(&target);
1120 let _ = store.set_pending_operation(target, false);
1121 if let Some(node_id) = reserved_node_id {
1122 let _ = store.discard_reserved_node_id(node_id);
1123 }
1124 return Err(map_request_send_error(error));
1125 }
1126 self.stats.requests_enqueued.fetch_add(1, Ordering::Relaxed);
1127 Ok(FileTreeMutationEnqueue {
1128 outcome: FileTreeEnqueueOutcome::Enqueued,
1129 pending_delta,
1130 })
1131 }
1132
1133 /// Non-blockingly requests provider watch enablement for `uri`.
1134 ///
1135 /// Configuration requests are not coalesced; acknowledgement is returned in
1136 /// [`FileTreeApplyReport::watch_configuration`].
1137 ///
1138 /// # Errors
1139 ///
1140 /// Returns [`FileTreeRuntimeError::QueueFull`] or `Closed`.
1141 ///
1142 /// # Examples
1143 ///
1144 /// ```
1145 /// use ailloli_ui_fs::FileUri;
1146 /// use ailloli_ui_fs_runtime::{FileTreeRuntime, FileTreeRuntimeError};
1147 /// fn watch(runtime: &FileTreeRuntime, uri: FileUri) -> Result<(), FileTreeRuntimeError> {
1148 /// runtime.watch_directory(uri)
1149 /// }
1150 /// ```
1151 pub fn watch_directory(&self, uri: FileUri) -> Result<(), FileTreeRuntimeError> {
1152 self.enqueue_watch_configuration(uri, true)
1153 }
1154
1155 /// Non-blockingly requests provider watch removal for `uri`.
1156 ///
1157 /// Removing an unknown URI is delegated to the provider. Successful worker
1158 /// bookkeeping decrements `watched_directories` only when it was known.
1159 ///
1160 /// # Errors
1161 ///
1162 /// Returns [`FileTreeRuntimeError::QueueFull`] or `Closed`.
1163 ///
1164 /// # Examples
1165 ///
1166 /// ```
1167 /// use ailloli_ui_fs::FileUri;
1168 /// use ailloli_ui_fs_runtime::{FileTreeRuntime, FileTreeRuntimeError};
1169 /// fn unwatch(runtime: &FileTreeRuntime, uri: FileUri) -> Result<(), FileTreeRuntimeError> {
1170 /// runtime.unwatch_directory(uri)
1171 /// }
1172 /// ```
1173 pub fn unwatch_directory(&self, uri: FileUri) -> Result<(), FileTreeRuntimeError> {
1174 self.enqueue_watch_configuration(uri, false)
1175 }
1176
1177 /// Enqueues a watch configuration after reserving a diagnostics depth slot.
1178 ///
1179 /// # Errors
1180 ///
1181 /// Returns [`FileTreeRuntimeError::QueueFull`] when the worker queue is at
1182 /// capacity, or [`FileTreeRuntimeError::Closed`] after worker disconnection.
1183 /// The reserved diagnostics depth is cancelled on either failure.
1184 fn enqueue_watch_configuration(
1185 &self,
1186 uri: FileUri,
1187 enabled: bool,
1188 ) -> Result<(), FileTreeRuntimeError> {
1189 self.stats.request_reserved();
1190 if let Err(error) = self
1191 .requests
1192 .try_send(WorkerRequest::ConfigureWatch { uri, enabled })
1193 {
1194 self.stats.request_reservation_cancelled();
1195 return Err(map_request_send_error(error));
1196 }
1197 self.stats.requests_enqueued.fetch_add(1, Ordering::Relaxed);
1198 Ok(())
1199 }
1200
1201 /// Drains at most 256 ordered responses without applying store effects.
1202 ///
1203 /// Current directory ownership is cleared only by its matching request ID;
1204 /// any watch response reopens explicit watch polling, and mutation responses
1205 /// release target ownership. `remaining` requests another UI drain.
1206 ///
1207 /// # Errors
1208 ///
1209 /// Returns [`FileTreeRuntimeError::Wake`] if inbox drain rearming fails.
1210 /// The drained response batch is not returned in that case.
1211 ///
1212 /// # Panics
1213 ///
1214 /// Panics only if [`FILE_TREE_UI_DRAIN_BUDGET`] is changed to zero.
1215 ///
1216 /// # Examples
1217 ///
1218 /// ```
1219 /// use ailloli_ui_fs_runtime::{FileTreeRuntime, FileTreeRuntimeDrain, FileTreeRuntimeError};
1220 /// fn drain(runtime: &mut FileTreeRuntime) -> Result<FileTreeRuntimeDrain, FileTreeRuntimeError> {
1221 /// runtime.drain()
1222 /// }
1223 /// ```
1224 pub fn drain(&mut self) -> Result<FileTreeRuntimeDrain, FileTreeRuntimeError> {
1225 let budget = NonZeroUsize::new(FILE_TREE_UI_DRAIN_BUDGET).expect("non-zero budget");
1226 let drain = self
1227 .responses
1228 .drain(budget)
1229 .map_err(FileTreeRuntimeError::Wake)?;
1230 for response in &drain.messages {
1231 match response {
1232 FileTreeWorkerResponse::Directory { request, .. } => {
1233 let mut active = self
1234 .active_directories
1235 .lock()
1236 .unwrap_or_else(|error| error.into_inner());
1237 let removed = if active.get(&request.node_id()) == Some(&request.request_id()) {
1238 active.remove(&request.node_id());
1239 true
1240 } else {
1241 false
1242 };
1243 if removed {
1244 self.stats
1245 .active_directory_requests
1246 .fetch_sub(1, Ordering::Relaxed);
1247 }
1248 }
1249 FileTreeWorkerResponse::Watch { .. } => {
1250 self.watch_pending.store(false, Ordering::Release);
1251 }
1252 FileTreeWorkerResponse::WatchConfigured { .. } => {}
1253 FileTreeWorkerResponse::Mutation { request, .. } => {
1254 self.active_mutations
1255 .lock()
1256 .unwrap_or_else(|error| error.into_inner())
1257 .remove(&request.mutation.target_node_id());
1258 }
1259 }
1260 }
1261 Ok(FileTreeRuntimeDrain {
1262 responses: drain.messages,
1263 remaining: drain.remaining,
1264 })
1265 }
1266
1267 /// Drains responses and applies directory/mutation results to `store`.
1268 ///
1269 /// Stale directory results are counted and skipped. Watch and mutation
1270 /// provider errors are collected. Successful mutations clear pending state
1271 /// before applying attested inserts, moves, or removals in response order.
1272 ///
1273 /// # Errors
1274 ///
1275 /// Returns wake or store errors. Earlier effects from the same batch are not
1276 /// rolled back if a later store operation fails.
1277 ///
1278 /// # Examples
1279 ///
1280 /// ```
1281 /// use ailloli_ui_fs::FileTreeStore;
1282 /// use ailloli_ui_fs_runtime::{FileTreeApplyReport, FileTreeRuntime, FileTreeRuntimeError};
1283 /// fn apply(runtime: &mut FileTreeRuntime, store: &mut FileTreeStore) -> Result<FileTreeApplyReport, FileTreeRuntimeError> {
1284 /// runtime.drain_into_store(store)
1285 /// }
1286 /// ```
1287 pub fn drain_into_store(
1288 &mut self,
1289 store: &mut FileTreeStore,
1290 ) -> Result<FileTreeApplyReport, FileTreeRuntimeError> {
1291 let drain = self.drain()?;
1292 let mut report = FileTreeApplyReport {
1293 remaining: drain.remaining,
1294 ..FileTreeApplyReport::default()
1295 };
1296 for response in drain.responses {
1297 match response {
1298 FileTreeWorkerResponse::Directory { request, result } => {
1299 match store.apply_directory_result(&request, result) {
1300 Ok(delta) => report.deltas.push(delta),
1301 Err(FileTreeStoreError::StaleResponse { .. }) => {
1302 report.stale_responses += 1;
1303 self.stats.stale_responses.fetch_add(1, Ordering::Relaxed);
1304 }
1305 Err(error) => return Err(FileTreeRuntimeError::Store(error)),
1306 }
1307 }
1308 FileTreeWorkerResponse::Watch { events } => match events {
1309 Ok(events) => report.watch_events.extend(events),
1310 Err(error) => report.watch_errors.push(error),
1311 },
1312 FileTreeWorkerResponse::WatchConfigured {
1313 uri,
1314 enabled,
1315 result,
1316 } => report.watch_configuration.push((uri, enabled, result)),
1317 FileTreeWorkerResponse::Mutation { request, result } => {
1318 let target = request.mutation.target_node_id();
1319 if store.node(target).is_some() {
1320 report
1321 .deltas
1322 .push(store.set_pending_operation(target, false)?);
1323 }
1324 match result {
1325 Ok(identity) => {
1326 let delta = match &request.mutation {
1327 FileTreeMutation::CreateDirectory { parent, uri } => store
1328 .apply_attested_insert(
1329 *parent,
1330 FileEntry::new(
1331 uri.clone(),
1332 FileMetadata::new(FileKind::Directory),
1333 ),
1334 identity,
1335 )?,
1336 FileTreeMutation::CreateEntry {
1337 parent,
1338 node_id,
1339 uri,
1340 kind,
1341 } => store.apply_attested_insert_reserved(
1342 *parent,
1343 *node_id,
1344 FileEntry::new(uri.clone(), FileMetadata::new(*kind)),
1345 identity,
1346 )?,
1347 FileTreeMutation::Move { node_id, to, .. } => {
1348 store.apply_attested_move(*node_id, to.clone(), identity)?
1349 }
1350 FileTreeMutation::Remove { node_id, .. } => {
1351 store.apply_attested_remove(*node_id)?
1352 }
1353 };
1354 report.deltas.push(delta);
1355 }
1356 Err(error) => {
1357 if let Some(node_id) = request.mutation.reserved_node_id() {
1358 let _ = store.discard_reserved_node_id(node_id);
1359 }
1360 report.mutation_errors.push((request, error));
1361 }
1362 }
1363 }
1364 }
1365 }
1366 Ok(report)
1367 }
1368
1369 /// Returns a non-transactional point-in-time worker metrics snapshot.
1370 ///
1371 /// # Examples
1372 ///
1373 /// ```
1374 /// use ailloli_ui_fs_runtime::{FileTreeRuntime, FileTreeWorkerStats};
1375 /// fn inspect(runtime: &FileTreeRuntime) {
1376 /// let stats: FileTreeWorkerStats = runtime.stats();
1377 /// assert!(stats.request_queue_max_depth >= stats.request_queue_depth);
1378 /// }
1379 /// ```
1380 pub fn stats(&self) -> FileTreeWorkerStats {
1381 self.stats.snapshot()
1382 }
1383
1384 /// Returns the source capability captured during worker initialization.
1385 ///
1386 /// # Examples
1387 ///
1388 /// ```
1389 /// use ailloli_ui_fs_runtime::FileTreeRuntime;
1390 /// fn inspect(runtime: &FileTreeRuntime) {
1391 /// let supports_watch: bool = runtime.supports_native_watch();
1392 /// let _ = supports_watch;
1393 /// }
1394 /// ```
1395 pub const fn supports_native_watch(&self) -> bool {
1396 self.native_watch
1397 }
1398
1399 /// Creates a fresh empty UI scheduler matching the provider watch mode.
1400 ///
1401 /// No expansion state is shared with prior scheduler values.
1402 ///
1403 /// # Examples
1404 ///
1405 /// ```
1406 /// use ailloli_ui_fs_runtime::{FileTreeReconcileScheduler, FileTreeRuntime};
1407 /// fn scheduler(runtime: &FileTreeRuntime) -> FileTreeReconcileScheduler {
1408 /// runtime.reconcile_scheduler()
1409 /// }
1410 /// ```
1411 pub fn reconcile_scheduler(&self) -> FileTreeReconcileScheduler {
1412 FileTreeReconcileScheduler::new(self.native_watch)
1413 }
1414
1415 /// Snapshots response-inbox depth, capacity, wake, and delivery counters.
1416 ///
1417 /// # Examples
1418 ///
1419 /// ```
1420 /// use ailloli_ui_fs_runtime::FileTreeRuntime;
1421 /// use ailloli_ui_runtime::UiInboxStats;
1422 /// fn inspect(runtime: &FileTreeRuntime) {
1423 /// let _: UiInboxStats = runtime.inbox_stats();
1424 /// }
1425 /// ```
1426 pub fn inbox_stats(&self) -> UiInboxStats {
1427 self.responses.stats()
1428 }
1429
1430 /// Requests shutdown, drains responses, and joins within two seconds.
1431 ///
1432 /// A full request queue is retried every five milliseconds while response
1433 /// draining prevents a blocked worker send. A disconnected request channel
1434 /// is treated as shutdown sent. Unapplied drained responses are discarded.
1435 ///
1436 /// # Errors
1437 ///
1438 /// Returns `ThreadPanicked` when join observes a panic, or `FinishTimeout`
1439 /// after two seconds. On timeout the consumed value is dropped and its join
1440 /// handle detaches; shutdown is attempted once more by [`Drop`].
1441 ///
1442 /// # Panics
1443 ///
1444 /// Panics if [`FILE_TREE_UI_DRAIN_BUDGET`] is zero or if the worker join
1445 /// handle is absent after it reports that it is finished.
1446 ///
1447 /// # Examples
1448 ///
1449 /// ```
1450 /// use ailloli_ui_fs_runtime::{FileTreeRuntime, FileTreeRuntimeError};
1451 /// let _finish: fn(FileTreeRuntime) -> Result<(), FileTreeRuntimeError> = FileTreeRuntime::finish;
1452 /// ```
1453 pub fn finish(mut self) -> Result<(), FileTreeRuntimeError> {
1454 let deadline = Instant::now() + FILE_TREE_FINISH_TIMEOUT;
1455 let mut shutdown_sent = false;
1456 while Instant::now() < deadline {
1457 if !shutdown_sent {
1458 self.stats.request_reserved();
1459 match self.requests.try_send(WorkerRequest::Shutdown) {
1460 Ok(()) => {
1461 shutdown_sent = true;
1462 }
1463 Err(TrySendError::Full(_)) => {
1464 self.stats.request_reservation_cancelled();
1465 }
1466 Err(TrySendError::Disconnected(_)) => {
1467 self.stats.request_reservation_cancelled();
1468 shutdown_sent = true;
1469 }
1470 }
1471 }
1472 let _ = self
1473 .responses
1474 .drain(NonZeroUsize::new(FILE_TREE_UI_DRAIN_BUDGET).expect("non-zero budget"));
1475 if self.thread.as_ref().is_none_or(JoinHandle::is_finished) {
1476 let thread = self.thread.take().expect("thread present");
1477 return thread
1478 .join()
1479 .map_err(|_| FileTreeRuntimeError::ThreadPanicked);
1480 }
1481 thread::sleep(Duration::from_millis(5));
1482 }
1483 Err(FileTreeRuntimeError::FinishTimeout(
1484 FILE_TREE_FINISH_TIMEOUT,
1485 ))
1486 }
1487}
1488
1489/// Performs a best-effort non-blocking shutdown request on handle destruction.
1490impl Drop for FileTreeRuntime {
1491 /// Enqueues shutdown if capacity permits; never waits or joins.
1492 fn drop(&mut self) {
1493 self.stats.request_reserved();
1494 if self.requests.try_send(WorkerRequest::Shutdown).is_ok() {
1495 return;
1496 }
1497 self.stats.request_reservation_cancelled();
1498 }
1499}
1500
1501/// Owns all source calls and serially translates requests into UI responses.
1502fn worker_loop(
1503 source: &mut dyn FileTreeSource,
1504 requests: &mpsc::Receiver<WorkerRequest>,
1505 responses: &UiSender<FileTreeWorkerResponse>,
1506 stats: &AtomicWorkerStats,
1507) {
1508 let mut watched = HashSet::new();
1509 loop {
1510 let request = match requests.recv_timeout(Duration::from_millis(50)) {
1511 Ok(request) => {
1512 stats.request_received();
1513 request
1514 }
1515 Err(mpsc::RecvTimeoutError::Timeout) => {
1516 if source.supports_native_watch() {
1517 stats.watch_polls.fetch_add(1, Ordering::Relaxed);
1518 match source.poll_watch(FILE_TREE_UI_DRAIN_BUDGET) {
1519 Ok(events) if events.is_empty() => {}
1520 events => {
1521 if !send_worker_response(
1522 responses,
1523 stats,
1524 FileTreeWorkerResponse::Watch { events },
1525 ) {
1526 break;
1527 }
1528 }
1529 }
1530 }
1531 continue;
1532 }
1533 Err(mpsc::RecvTimeoutError::Disconnected) => break,
1534 };
1535 let response = match request {
1536 WorkerRequest::Directory(request) => {
1537 stats.directory_reads.fetch_add(1, Ordering::Relaxed);
1538 let result = source.read_dir(request.uri()).map(|entries| {
1539 entries
1540 .into_iter()
1541 .map(|entry| {
1542 let identity = source.identity(&entry.uri).ok().flatten();
1543 (entry, identity)
1544 })
1545 .collect()
1546 });
1547 FileTreeWorkerResponse::Directory { request, result }
1548 }
1549 WorkerRequest::Mutation(request) => {
1550 stats.mutations.fetch_add(1, Ordering::Relaxed);
1551 let result = match request.mutation() {
1552 FileTreeMutation::CreateDirectory { uri, .. } => source
1553 .create_directory(uri)
1554 .and_then(|()| source.identity(uri)),
1555 FileTreeMutation::CreateEntry { uri, kind, .. } => match kind {
1556 FileKind::Directory => source.create_directory(uri),
1557 FileKind::File => source.create_file(uri),
1558 _ => Err(FileError::Unsupported(format!(
1559 "creating {kind:?} filesystem entries is not supported"
1560 ))),
1561 }
1562 .and_then(|()| source.identity(uri)),
1563 FileTreeMutation::Move { from, to, .. } => source
1564 .move_entry(from, to)
1565 .and_then(|()| source.identity(to)),
1566 FileTreeMutation::Remove { uri, recursive, .. } => {
1567 source.remove_entry(uri, *recursive).map(|()| None)
1568 }
1569 };
1570 FileTreeWorkerResponse::Mutation { request, result }
1571 }
1572 WorkerRequest::Watch { limit } => {
1573 stats.watch_polls.fetch_add(1, Ordering::Relaxed);
1574 FileTreeWorkerResponse::Watch {
1575 events: source.poll_watch(limit),
1576 }
1577 }
1578 WorkerRequest::ConfigureWatch { uri, enabled } => {
1579 let result = if enabled {
1580 source.watch_directory(&uri)
1581 } else {
1582 source.unwatch_directory(&uri)
1583 };
1584 if result.is_ok() {
1585 if enabled {
1586 watched.insert(uri.clone());
1587 } else {
1588 watched.remove(&uri);
1589 }
1590 stats
1591 .watched_directories
1592 .store(watched.len(), Ordering::Relaxed);
1593 }
1594 FileTreeWorkerResponse::WatchConfigured {
1595 uri,
1596 enabled,
1597 result,
1598 }
1599 }
1600 WorkerRequest::Shutdown => break,
1601 };
1602 if !send_worker_response(responses, stats, response) {
1603 break;
1604 }
1605 }
1606}
1607
1608/// Delivers a response with backpressure until inbox space is available.
1609///
1610/// A wake failure still means the message was enqueued and processing may
1611/// continue; closure is the only response-send outcome that stops the worker.
1612fn send_worker_response(
1613 responses: &UiSender<FileTreeWorkerResponse>,
1614 stats: &AtomicWorkerStats,
1615 response: FileTreeWorkerResponse,
1616) -> bool {
1617 stats.responses.fetch_add(1, Ordering::Relaxed);
1618 matches!(
1619 responses.send_blocking(response),
1620 Ok(()) | Err(UiSendError::EnqueuedButWakeFailed(_))
1621 )
1622}
1623
1624/// Maps bounded-channel saturation and disconnect without exposing requests.
1625fn map_request_send_error(error: TrySendError<WorkerRequest>) -> FileTreeRuntimeError {
1626 match error {
1627 TrySendError::Full(_) => FileTreeRuntimeError::QueueFull,
1628 TrySendError::Disconnected(_) => FileTreeRuntimeError::Closed,
1629 }
1630}