1use super::allocation::{AllocationError, AllocationLedger};
4use super::probe::{capture_directory_name_state, capture_file_record_state};
5use super::{
6 AuthenticatedProbeError, DecodeLimits, Dependency, DependencyRegion, DependencyState, FileKind,
7 FilePayload, FileRecord, FileRecordReadError, LogicalName, NamespacePath, NamespacePathError,
8 TreeReadError, lookup_file_record_async, lookup_file_records_async, lookup_tree_entries_async,
9 lookup_tree_entry_async,
10};
11use crate::async_storage::{self, AsyncObjectStore};
12use crate::cancellation::CancellationToken;
13use crate::model::{FilesystemProfile, VolumeConfig, VolumeConfigError};
14use crate::performance::{OperationFailure, WorkBudget, WorkCounters, WorkError};
15use crate::storage::{
16 ObjectFailure, ObjectId, ObjectRead, ObjectReadRequest, ObjectReadRetention, ObjectReceipt,
17 ObjectResult, ObjectStoreError,
18};
19use bytes::Bytes;
20use std::cmp::Ordering;
21use std::mem::size_of;
22use std::sync::{Mutex, MutexGuard};
23use thiserror::Error;
24
25struct CachedObject {
26 object_id: ObjectId,
27 bytes: Bytes,
28}
29
30struct CacheState {
31 slots: Vec<Option<CachedObject>>,
32 entry_count: usize,
33 retained_owned_bytes: u64,
34 external_resident_bytes: u64,
35}
36
37struct OperationReadCache<'a, S> {
38 store: &'a S,
39 state: Mutex<CacheState>,
40 maximum_entries: usize,
41 metadata_bytes: u64,
42}
43
44impl<'a, S> OperationReadCache<'a, S> {
45 fn new(
46 store: &'a S,
47 maximum_entries: usize,
48 budget: WorkBudget,
49 ) -> Result<(Self, WorkCounters), PathLookupFailure> {
50 let slot_count = maximum_entries
51 .checked_mul(2)
52 .and_then(usize::checked_next_power_of_two)
53 .ok_or_else(|| {
54 OperationFailure::before_work(PathLookupError::Work(WorkError::Overflow))
55 })?;
56 let requested = u64::try_from(slot_count)
57 .unwrap_or(u64::MAX)
58 .checked_mul(u64::try_from(size_of::<Option<CachedObject>>()).unwrap_or(u64::MAX))
59 .ok_or_else(|| {
60 OperationFailure::before_work(PathLookupError::Work(WorkError::Overflow))
61 })?;
62 WorkCounters {
63 allocation_operations: 1,
64 peak_allocation_bytes: requested,
65 ..WorkCounters::default()
66 }
67 .verify(budget)
68 .map_err(|error| OperationFailure::before_work(error.into()))?;
69 let mut slots = Vec::new();
70 slots
71 .try_reserve_exact(slot_count)
72 .map_err(|_| OperationFailure::before_work(PathLookupError::AllocationFailed))?;
73 slots.resize_with(slot_count, || None);
74 let metadata_bytes = u64::try_from(slots.capacity())
75 .unwrap_or(u64::MAX)
76 .checked_mul(u64::try_from(size_of::<Option<CachedObject>>()).unwrap_or(u64::MAX))
77 .ok_or_else(|| {
78 OperationFailure::before_work(PathLookupError::Work(WorkError::Overflow))
79 })?;
80 let work = WorkCounters {
81 allocation_operations: 1,
82 peak_allocation_bytes: metadata_bytes,
83 ..WorkCounters::default()
84 };
85 work.verify(budget)
86 .map_err(|error| OperationFailure::new(error.into(), work))?;
87 Ok((
88 Self {
89 store,
90 state: Mutex::new(CacheState {
91 slots,
92 entry_count: 0,
93 retained_owned_bytes: 0,
94 external_resident_bytes: 0,
95 }),
96 maximum_entries,
97 metadata_bytes,
98 },
99 work,
100 ))
101 }
102
103 fn probe(
104 &self,
105 object_id: ObjectId,
106 ) -> Result<(Option<ObjectRead>, usize, u64), ObjectFailure> {
107 let state = self.lock_state();
108 let mask = state.slots.len() - 1;
109 let mut index = object_hash(object_id) & mask;
110 let mut examined = 0_u64;
111 for _ in 0..state.slots.len() {
112 examined = examined.saturating_add(1);
113 #[allow(
114 clippy::indexing_slicing,
115 reason = "OperationReadCache::new computes slot_count via checked_next_power_of_two, so state.slots.len() is always a power of two; index is initialized as `object_hash(object_id) & mask` and updated only as `(index + 1) & mask` where `mask = state.slots.len() - 1`, so index is always < state.slots.len()"
116 )]
117 match &state.slots[index] {
118 Some(entry) if entry.object_id == object_id => {
119 return Ok((
120 Some(ObjectRead {
121 bytes: entry.bytes.clone(),
122 retention: ObjectReadRetention::Shared,
123 }),
124 index,
125 examined,
126 ));
127 }
128 Some(_) => index = (index + 1) & mask,
129 None => return Ok((None, index, examined)),
130 }
131 }
132 Err(ObjectFailure::new(
133 ObjectStoreError::Corrupt,
134 WorkCounters {
135 items_examined: examined,
136 ..WorkCounters::default()
137 },
138 ))
139 }
140
141 fn resident_bytes(&self) -> Result<u64, ObjectFailure> {
142 let state = self.lock_state();
143 self.metadata_bytes
144 .checked_add(state.external_resident_bytes)
145 .ok_or_else(|| ObjectFailure::before_work(ObjectStoreError::Work(WorkError::Overflow)))?
146 .checked_add(state.retained_owned_bytes)
147 .ok_or_else(|| ObjectFailure::before_work(ObjectStoreError::Work(WorkError::Overflow)))
148 }
149
150 fn add_external_resident_bytes(&self, bytes: u64) -> Result<u64, WorkError> {
151 let mut state = self.lock_state();
152 state.external_resident_bytes = state
153 .external_resident_bytes
154 .checked_add(bytes)
155 .ok_or(WorkError::Overflow)?;
156 Ok(state.external_resident_bytes)
157 }
158
159 fn lock_state(&self) -> MutexGuard<'_, CacheState> {
160 match self.state.lock() {
161 Ok(state) => state,
162 Err(poisoned) => poisoned.into_inner(),
163 }
164 }
165}
166
167fn object_hash(object_id: ObjectId) -> usize {
168 let mut hash = usize::from(object_id.kind.canonical_tag());
169 for byte in object_id.digest.as_bytes() {
170 hash ^= usize::from(*byte);
171 hash = hash.wrapping_mul(16_777_619);
172 }
173 hash
174}
175
176fn compare_object_id(left: ObjectId, right: ObjectId) -> Ordering {
177 left.kind
178 .canonical_tag()
179 .cmp(&right.kind.canonical_tag())
180 .then_with(|| left.digest.cmp(&right.digest))
181}
182
183fn budget_after_resident(
184 mut budget: WorkBudget,
185 resident: u64,
186) -> Result<WorkBudget, ObjectFailure> {
187 budget.peak_allocation_bytes = budget
188 .peak_allocation_bytes
189 .checked_sub(resident)
190 .ok_or_else(|| {
191 ObjectFailure::before_work(ObjectStoreError::Work(WorkError::BudgetExceeded {
192 counter: "peak_allocation_bytes",
193 observed: resident,
194 maximum: budget.peak_allocation_bytes,
195 }))
196 })?;
197 Ok(budget)
198}
199
200fn merge_backend_peak(
201 prior: WorkCounters,
202 mut backend: WorkCounters,
203 resident: u64,
204 budget: WorkBudget,
205) -> Result<WorkCounters, ObjectFailure> {
206 let simultaneous_peak = resident
207 .checked_add(backend.peak_allocation_bytes)
208 .ok_or_else(|| ObjectFailure::new(ObjectStoreError::Work(WorkError::Overflow), prior))?;
209 backend.peak_allocation_bytes = 0;
210 let mut work = prior
211 .checked_add(backend)
212 .map_err(|error| ObjectFailure::new(error.into(), prior))?;
213 work.peak_allocation_bytes = work.peak_allocation_bytes.max(simultaneous_peak);
214 work.verify(budget)
215 .map_err(|error| ObjectFailure::new(error.into(), work))?;
216 Ok(work)
217}
218
219fn merge_backend_failure(
220 prior: WorkCounters,
221 mut backend: WorkCounters,
222 resident: u64,
223 error: ObjectStoreError,
224) -> ObjectFailure {
225 let Some(simultaneous_peak) = resident.checked_add(backend.peak_allocation_bytes) else {
226 return ObjectFailure::new(ObjectStoreError::Work(WorkError::Overflow), prior);
227 };
228 backend.peak_allocation_bytes = 0;
229 let Ok(mut work) = prior.checked_add(backend) else {
230 return ObjectFailure::new(ObjectStoreError::Work(WorkError::Overflow), prior);
231 };
232 work.peak_allocation_bytes = work.peak_allocation_bytes.max(simultaneous_peak);
233 ObjectFailure::new(error, work)
234}
235
236impl<S: AsyncObjectStore> AsyncObjectStore for OperationReadCache<'_, S> {
237 fn decoded_cache_get(
238 &self,
239 key: async_storage::DecodedCacheKey,
240 ) -> Result<Option<async_storage::DecodedCacheValue>, ObjectStoreError> {
241 self.store.decoded_cache_get(key)
242 }
243
244 fn decoded_cache_admit(
245 &self,
246 key: async_storage::DecodedCacheKey,
247 value: async_storage::DecodedCacheValue,
248 ) -> Result<async_storage::DecodedCacheAdmission, ObjectStoreError> {
249 self.store.decoded_cache_admit(key, value)
250 }
251
252 async fn put(
253 &self,
254 object_id: ObjectId,
255 bytes: Bytes,
256 budget: WorkBudget,
257 cancellation: &CancellationToken,
258 ) -> ObjectResult<()> {
259 let resident = self.resident_bytes()?;
260 let backend_budget = budget_after_resident(budget, resident)?;
261 let receipt =
262 AsyncObjectStore::put(self.store, object_id, bytes, backend_budget, cancellation)
263 .await
264 .map_err(|failure| {
265 merge_backend_failure(
266 WorkCounters::default(),
267 *failure.work,
268 resident,
269 failure.error,
270 )
271 })?;
272 let work = merge_backend_peak(WorkCounters::default(), receipt.work, resident, budget)?;
273 Ok(ObjectReceipt { value: (), work })
274 }
275
276 async fn read(
277 &self,
278 object_id: ObjectId,
279 maximum_bytes: u64,
280 budget: WorkBudget,
281 cancellation: &CancellationToken,
282 ) -> ObjectResult<ObjectRead> {
283 let (hit, vacant_slot, examined) = self.probe(object_id)?;
284 let hit_work = WorkCounters {
285 items_examined: examined,
286 ..WorkCounters::default()
287 };
288 if let Some(value) = hit {
289 let observed = u64::try_from(value.bytes.len()).unwrap_or(u64::MAX);
290 if observed > maximum_bytes {
291 return Err(ObjectFailure::new(
292 ObjectStoreError::TooLarge {
293 observed,
294 maximum: maximum_bytes,
295 },
296 hit_work,
297 ));
298 }
299 hit_work
300 .verify(budget)
301 .map_err(|error| ObjectFailure::new(error.into(), hit_work))?;
302 return Ok(ObjectReceipt {
303 value,
304 work: hit_work,
305 });
306 }
307
308 let resident = self.resident_bytes()?;
309 let backend_budget = budget_after_resident(budget, resident)?;
310 let receipt = AsyncObjectStore::read(
311 self.store,
312 object_id,
313 maximum_bytes,
314 backend_budget,
315 cancellation,
316 )
317 .await
318 .map_err(|failure| {
319 merge_backend_failure(hit_work, *failure.work, resident, failure.error)
320 })?;
321 let owned_bytes = match receipt.value.retention {
322 ObjectReadRetention::Shared => 0,
323 ObjectReadRetention::Owned { logical_bytes } => logical_bytes,
324 };
325 let work = merge_backend_peak(hit_work, receipt.work, resident, budget)?;
326 let mut state = self.lock_state();
327 #[allow(
328 clippy::indexing_slicing,
329 reason = "vacant_slot is the index returned by probe(), which only ever returns an index bounded by its own `& mask` invariant (see probe's match on state.slots[index]); state.slots is created once in OperationReadCache::new and never resized afterward, so that bound still holds against this (possibly different) MutexGuard borrow of the same Vec"
330 )]
331 if state.entry_count < self.maximum_entries && state.slots[vacant_slot].is_none() {
332 state.slots[vacant_slot] = Some(CachedObject {
333 object_id,
334 bytes: receipt.value.bytes.clone(),
335 });
336 state.entry_count += 1;
337 state.retained_owned_bytes = state
338 .retained_owned_bytes
339 .checked_add(owned_bytes)
340 .ok_or_else(|| {
341 ObjectFailure::new(ObjectStoreError::Work(WorkError::Overflow), work)
342 })?;
343 }
344 Ok(ObjectReceipt {
345 value: receipt.value,
346 work,
347 })
348 }
349
350 async fn read_many(
351 &self,
352 requests: &[ObjectReadRequest],
353 budget: WorkBudget,
354 cancellation: &CancellationToken,
355 ) -> ObjectResult<Vec<ObjectRead>> {
356 async_storage::read_many_sequential_async(self, requests, budget, cancellation).await
357 }
358
359 async fn contains(
360 &self,
361 object_id: ObjectId,
362 budget: WorkBudget,
363 cancellation: &CancellationToken,
364 ) -> ObjectResult<bool> {
365 let (hit, _, examined) = self.probe(object_id)?;
366 if hit.is_some() {
367 let work = WorkCounters {
368 items_examined: examined,
369 ..WorkCounters::default()
370 };
371 work.verify(budget)
372 .map_err(|error| ObjectFailure::before_work(error.into()))?;
373 return Ok(ObjectReceipt { value: true, work });
374 }
375 let prior = WorkCounters {
376 items_examined: examined,
377 ..WorkCounters::default()
378 };
379 let resident = self.resident_bytes()?;
380 let remaining = prior
381 .remaining(budget)
382 .map_err(|error| ObjectFailure::before_work(error.into()))?;
383 let backend_budget = budget_after_resident(remaining, resident)?;
384 let receipt =
385 AsyncObjectStore::contains(self.store, object_id, backend_budget, cancellation)
386 .await
387 .map_err(|failure| {
388 merge_backend_failure(prior, *failure.work, resident, failure.error)
389 })?;
390 let work = merge_backend_peak(prior, receipt.work, resident, budget)?;
391 Ok(ObjectReceipt {
392 value: receipt.value,
393 work,
394 })
395 }
396}
397
398#[derive(Clone, Copy, Debug, Eq, PartialEq)]
400pub struct PathLookup {
401 pub record: Option<FileRecord>,
403 pub parent: Option<FileRecord>,
405 pub resolved_components: u16,
407 pub work: WorkCounters,
409}
410
411#[derive(Clone, Debug, Eq, PartialEq)]
413pub struct ObservedPathLookup {
414 pub lookup: PathLookup,
416 pub dependencies: Vec<Dependency>,
418}
419
420#[derive(Clone, Copy, Debug, Eq, PartialEq)]
422pub struct PathBatchEntry {
423 pub record: Option<FileRecord>,
425 pub parent: Option<FileRecord>,
427 pub resolved_components: u16,
429}
430
431#[derive(Clone, Debug, Eq, PartialEq)]
433pub struct PathBatchLookup {
434 pub entries: Vec<PathBatchEntry>,
436 pub retained_allocation_bytes: u64,
438 pub work: WorkCounters,
440}
441
442#[derive(Clone, Copy)]
443struct PendingBinding {
444 path_index: usize,
445 file_id: crate::foundation::FileId,
446 kind: FileKind,
447}
448
449#[derive(Clone, Copy)]
450struct ActivePath {
451 path_index: usize,
452 directory: ObjectId,
453}
454
455#[derive(Clone, Copy)]
456enum PathQueries<'a> {
457 Owned(&'a [NamespacePath]),
458 Borrowed(&'a [&'a NamespacePath]),
459}
460
461impl<'a> PathQueries<'a> {
462 const fn len(self) -> usize {
463 match self {
464 Self::Owned(paths) => paths.len(),
465 Self::Borrowed(paths) => paths.len(),
466 }
467 }
468
469 const fn is_empty(self) -> bool {
470 self.len() == 0
471 }
472
473 #[allow(
474 clippy::indexing_slicing,
475 reason = "PathQueries is a private helper and every call site in this module indexes with a value already bounded by `0..self.len()` (loop bounds `for index in 0..paths.len()`, or path_index/query_indices values that were themselves drawn from such a loop and threaded through fixed-length, never-resized scratch vectors), matching the array-of-two-slice contract of Mutation::paths elsewhere in this crate"
476 )]
477 fn get(self, index: usize) -> &'a NamespacePath {
478 match self {
479 Self::Owned(paths) => &paths[index],
480 Self::Borrowed(paths) => paths[index],
481 }
482 }
483}
484
485pub type PathLookupFailure = OperationFailure<PathLookupError>;
487
488#[derive(Debug, Error)]
490pub enum PathLookupError {
491 #[error(transparent)]
493 Config(#[from] VolumeConfigError),
494 #[error("path lookup batch is empty")]
496 EmptyBatch,
497 #[error("path lookup batch exceeds the volume limit")]
499 TooManyPaths,
500 #[error("namespace path exceeds the volume's admitted bounds")]
502 PathBounds,
503 #[error("namespace component encoding is unsupported by the volume profile")]
505 UnsupportedNameEncoding,
506 #[error("path lookup acceleration allocation failed")]
508 AllocationFailed,
509 #[error("generation root file record is missing")]
511 MissingRootRecord,
512 #[error("path traversal encountered a non-directory")]
514 NotDirectory,
515 #[error("namespace binding kind does not match its file record")]
517 KindMismatch,
518 #[error(transparent)]
520 FileTable(#[from] FileRecordReadError),
521 #[error(transparent)]
523 Tree(#[from] TreeReadError),
524 #[error(transparent)]
526 NamespacePath(#[from] NamespacePathError),
527 #[error(transparent)]
529 Dependency(#[from] AuthenticatedProbeError),
530 #[error(transparent)]
532 Work(#[from] WorkError),
533}
534
535pub fn lookup_path<S: crate::ImmediateObjectStore>(
548 store: &S,
549 generation: &super::GenerationRoot,
550 path: &NamespacePath,
551 config: VolumeConfig,
552 budget: WorkBudget,
553) -> Result<PathLookup, PathLookupFailure> {
554 let cancellation = CancellationToken::new();
555 async_storage::poll_immediate(lookup_path_async(
556 store,
557 generation,
558 path,
559 config,
560 budget,
561 &cancellation,
562 ))
563}
564
565pub async fn lookup_path_async<S: AsyncObjectStore>(
572 store: &S,
573 generation: &super::GenerationRoot,
574 path: &NamespacePath,
575 config: VolumeConfig,
576 budget: WorkBudget,
577 cancellation: &CancellationToken,
578) -> Result<PathLookup, PathLookupFailure> {
579 cancellation.check().map_err(|_| {
580 OperationFailure::before_work(PathLookupError::Tree(TreeReadError::Cancelled))
581 })?;
582 validate_path(path, config)?;
583 let limits = decode_limits(config);
584 let maximum_cache_entries = maximum_cache_entries(path, config)?;
585 let (cache, mut work) = OperationReadCache::new(store, maximum_cache_entries, budget)?;
586 let root = lookup_file_record_async(
587 &cache,
588 generation.file_table,
589 generation.root_file_id,
590 limits,
591 remaining(work, budget)?,
592 cancellation,
593 )
594 .await
595 .map_err(|failure| failure.map_with_prior_work(work, PathLookupError::FileTable))?;
596 work = add(work, root.work)?;
597 let mut current = root
598 .record
599 .ok_or_else(|| OperationFailure::new(PathLookupError::MissingRootRecord, work))?;
600 if path.is_root() {
601 return Ok(PathLookup {
602 record: Some(current),
603 parent: None,
604 resolved_components: 0,
605 work,
606 });
607 }
608
609 let mut parent = None;
610 for (index, component) in path.components().iter().enumerate() {
611 parent = Some(current);
612 let FilePayload::Directory { entries } = current.payload else {
613 return Err(OperationFailure::new(PathLookupError::NotDirectory, work));
614 };
615 if current.kind != FileKind::Directory {
616 return Err(OperationFailure::new(PathLookupError::KindMismatch, work));
617 }
618 let binding = lookup_tree_entry_async(
619 &cache,
620 entries,
621 component,
622 limits,
623 remaining(work, budget)?,
624 cancellation,
625 )
626 .await
627 .map_err(|failure| failure.map_with_prior_work(work, PathLookupError::Tree))?;
628 work = add(work, binding.work)?;
629 let Some(binding) = binding.entry else {
630 return Ok(PathLookup {
631 record: None,
632 parent,
633 resolved_components: u16::try_from(index).unwrap_or(u16::MAX),
634 work,
635 });
636 };
637 let record = lookup_file_record_async(
638 &cache,
639 generation.file_table,
640 binding.file_id,
641 limits,
642 remaining(work, budget)?,
643 cancellation,
644 )
645 .await
646 .map_err(|failure| failure.map_with_prior_work(work, PathLookupError::FileTable))?;
647 work = add(work, record.work)?;
648 current = record
649 .record
650 .ok_or_else(|| OperationFailure::new(PathLookupError::KindMismatch, work))?;
651 if current.kind != binding.kind {
652 return Err(OperationFailure::new(PathLookupError::KindMismatch, work));
653 }
654 }
655 Ok(PathLookup {
656 record: Some(current),
657 parent,
658 resolved_components: u16::try_from(path.depth()).unwrap_or(u16::MAX),
659 work,
660 })
661}
662
663pub async fn observe_path_async<S: AsyncObjectStore>(
676 store: &S,
677 generation: &super::GenerationRoot,
678 path: &NamespacePath,
679 config: VolumeConfig,
680 budget: WorkBudget,
681 cancellation: &CancellationToken,
682) -> Result<ObservedPathLookup, PathLookupFailure> {
683 observe_path_with_terminal_async(store, generation, path, config, budget, cancellation, true)
684 .await
685}
686
687pub(crate) async fn observe_path_edges_async<S: AsyncObjectStore>(
695 store: &S,
696 generation: &super::GenerationRoot,
697 path: &NamespacePath,
698 config: VolumeConfig,
699 budget: WorkBudget,
700 cancellation: &CancellationToken,
701) -> Result<ObservedPathLookup, PathLookupFailure> {
702 observe_path_with_terminal_async(store, generation, path, config, budget, cancellation, false)
703 .await
704}
705
706async fn observe_path_with_terminal_async<S: AsyncObjectStore>(
707 store: &S,
708 generation: &super::GenerationRoot,
709 path: &NamespacePath,
710 config: VolumeConfig,
711 budget: WorkBudget,
712 cancellation: &CancellationToken,
713 capture_terminal: bool,
714) -> Result<ObservedPathLookup, PathLookupFailure> {
715 cancellation.check().map_err(|_| {
716 OperationFailure::before_work(PathLookupError::Tree(TreeReadError::Cancelled))
717 })?;
718 validate_path(path, config)?;
719
720 let limits = decode_limits(config);
721 let maximum_cache_entries = maximum_cache_entries(path, config)?;
722 let (cache, mut work) = OperationReadCache::new(store, maximum_cache_entries, budget)?;
723 let mut allocations = AllocationLedger::default();
724 let mut dependencies = reserve_fixed::<Dependency>(
725 path.depth().saturating_add(1),
726 &mut allocations,
727 &mut work,
728 budget,
729 )?;
730 let dependency_vector_bytes = allocations.live_bytes();
731 cache
732 .add_external_resident_bytes(dependency_vector_bytes)
733 .map_err(|error| OperationFailure::new(error.into(), work))?;
734 let combined_resident = cache
735 .resident_bytes()
736 .map_err(|failure| OperationFailure::new(map_cache_error(failure.error), work))?;
737 work.peak_allocation_bytes = work.peak_allocation_bytes.max(combined_resident);
738 work.verify(budget)
739 .map_err(|error| OperationFailure::new(error.into(), work))?;
740
741 let root = lookup_file_record_async(
742 &cache,
743 generation.file_table,
744 generation.root_file_id,
745 limits,
746 remaining(work, budget)?,
747 cancellation,
748 )
749 .await
750 .map_err(|failure| failure.map_with_prior_work(work, PathLookupError::FileTable))?;
751 work = add(work, root.work)?;
752 let current = root
753 .record
754 .ok_or_else(|| OperationFailure::new(PathLookupError::MissingRootRecord, work))?;
755 if path.is_root() {
756 if !capture_terminal {
757 return Ok(ObservedPathLookup {
758 lookup: PathLookup {
759 record: Some(current),
760 parent: None,
761 resolved_components: 0,
762 work,
763 },
764 dependencies,
765 });
766 }
767 let state =
768 capture_file_record_state(current, remaining(work, budget)?, WorkCounters::default())
769 .map_err(|failure| failure.map_with_prior_work(work, PathLookupError::Dependency))?;
770 work = add(work, state.work)?;
771 dependencies.push(Dependency {
772 region: DependencyRegion::FileRecord(current.file_id),
773 expected: state.value,
774 });
775 return Ok(ObservedPathLookup {
776 lookup: PathLookup {
777 record: Some(current),
778 parent: None,
779 resolved_components: 0,
780 work,
781 },
782 dependencies,
783 });
784 }
785
786 observe_descendants(
787 ObserveContext {
788 cache: &cache,
789 generation,
790 path,
791 limits,
792 budget,
793 cancellation,
794 capture_terminal,
795 },
796 current,
797 dependencies,
798 dependency_vector_bytes,
799 work,
800 )
801 .await
802}
803
804struct ObserveContext<'a, S> {
805 cache: &'a OperationReadCache<'a, S>,
806 generation: &'a super::GenerationRoot,
807 path: &'a NamespacePath,
808 limits: DecodeLimits,
809 budget: WorkBudget,
810 cancellation: &'a CancellationToken,
811 capture_terminal: bool,
812}
813
814#[allow(clippy::too_many_lines)]
815async fn observe_descendants<S: AsyncObjectStore>(
816 context: ObserveContext<'_, S>,
817 mut current: FileRecord,
818 mut dependencies: Vec<Dependency>,
819 mut external_resident_bytes: u64,
820 mut work: WorkCounters,
821) -> Result<ObservedPathLookup, PathLookupFailure> {
822 let mut parent = None;
823 for (index, component) in context.path.components().iter().enumerate() {
824 let copied = copy_observation_name(
825 context.cache,
826 component,
827 context.limits.maximum_name_bytes,
828 external_resident_bytes,
829 work,
830 context.budget,
831 )?;
832 let dependency_name = copied.name;
833 external_resident_bytes = copied.external_resident_bytes;
834 work = copied.work;
835 parent = Some(current);
836 let FilePayload::Directory { entries } = current.payload else {
837 if !context.capture_terminal {
838 return Ok(ObservedPathLookup {
843 lookup: PathLookup {
844 record: None,
845 parent,
846 resolved_components: u16::try_from(index).unwrap_or(u16::MAX),
847 work,
848 },
849 dependencies,
850 });
851 }
852 return Err(OperationFailure::new(PathLookupError::NotDirectory, work));
853 };
854 if current.kind != FileKind::Directory {
855 return Err(OperationFailure::new(PathLookupError::KindMismatch, work));
856 }
857 let binding = lookup_tree_entry_async(
858 context.cache,
859 entries,
860 component,
861 context.limits,
862 remaining(work, context.budget)?,
863 context.cancellation,
864 )
865 .await
866 .map_err(|failure| failure.map_with_prior_work(work, PathLookupError::Tree))?;
867 work = add(work, binding.work)?;
868 let Some(binding) = binding.entry else {
869 dependencies.push(Dependency {
870 region: DependencyRegion::DirectoryName {
871 directory_id: current.file_id,
872 name: dependency_name,
873 },
874 expected: DependencyState::Absent,
875 });
876 return Ok(ObservedPathLookup {
877 lookup: PathLookup {
878 record: None,
879 parent,
880 resolved_components: u16::try_from(index).unwrap_or(u16::MAX),
881 work,
882 },
883 dependencies,
884 });
885 };
886 let state = capture_directory_name_state(
887 &binding,
888 remaining(work, context.budget)?,
889 WorkCounters::default(),
890 )
891 .map_err(|failure| failure.map_with_prior_work(work, PathLookupError::Dependency))?;
892 work = add(work, state.work)?;
893 let binding_file_id = binding.file_id;
894 let binding_kind = binding.kind;
895 dependencies.push(Dependency {
896 region: DependencyRegion::DirectoryName {
897 directory_id: current.file_id,
898 name: dependency_name,
899 },
900 expected: state.value,
901 });
902 let record = lookup_file_record_async(
903 context.cache,
904 context.generation.file_table,
905 binding_file_id,
906 context.limits,
907 remaining(work, context.budget)?,
908 context.cancellation,
909 )
910 .await
911 .map_err(|failure| failure.map_with_prior_work(work, PathLookupError::FileTable))?;
912 work = add(work, record.work)?;
913 current = record
914 .record
915 .ok_or_else(|| OperationFailure::new(PathLookupError::KindMismatch, work))?;
916 if current.kind != binding_kind {
917 return Err(OperationFailure::new(PathLookupError::KindMismatch, work));
918 }
919 }
920 if context.capture_terminal {
921 let state = capture_file_record_state(
922 current,
923 remaining(work, context.budget)?,
924 WorkCounters::default(),
925 )
926 .map_err(|failure| failure.map_with_prior_work(work, PathLookupError::Dependency))?;
927 work = add(work, state.work)?;
928 dependencies.push(Dependency {
929 region: DependencyRegion::FileRecord(current.file_id),
930 expected: state.value,
931 });
932 }
933 Ok(ObservedPathLookup {
934 lookup: PathLookup {
935 record: Some(current),
936 parent,
937 resolved_components: u16::try_from(context.path.depth()).unwrap_or(u16::MAX),
938 work,
939 },
940 dependencies,
941 })
942}
943
944pub fn lookup_paths<S: crate::ImmediateObjectStore>(
955 store: &S,
956 generation: &super::GenerationRoot,
957 paths: &[NamespacePath],
958 config: VolumeConfig,
959 budget: WorkBudget,
960) -> Result<PathBatchLookup, PathLookupFailure> {
961 let cancellation = CancellationToken::new();
962 async_storage::poll_immediate(lookup_paths_async(
963 store,
964 generation,
965 paths,
966 config,
967 budget,
968 &cancellation,
969 ))
970}
971
972pub async fn lookup_paths_async<S: AsyncObjectStore>(
979 store: &S,
980 generation: &super::GenerationRoot,
981 paths: &[NamespacePath],
982 config: VolumeConfig,
983 budget: WorkBudget,
984 cancellation: &CancellationToken,
985) -> Result<PathBatchLookup, PathLookupFailure> {
986 lookup_path_queries_async(
987 store,
988 generation,
989 PathQueries::Owned(paths),
990 config,
991 budget,
992 cancellation,
993 )
994 .await
995}
996
997pub async fn lookup_path_refs_async<S: AsyncObjectStore>(
1006 store: &S,
1007 generation: &super::GenerationRoot,
1008 paths: &[&NamespacePath],
1009 config: VolumeConfig,
1010 budget: WorkBudget,
1011 cancellation: &CancellationToken,
1012) -> Result<PathBatchLookup, PathLookupFailure> {
1013 lookup_path_queries_async(
1014 store,
1015 generation,
1016 PathQueries::Borrowed(paths),
1017 config,
1018 budget,
1019 cancellation,
1020 )
1021 .await
1022}
1023
1024pub fn lookup_path_refs<S: crate::ImmediateObjectStore>(
1030 store: &S,
1031 generation: &super::GenerationRoot,
1032 paths: &[&NamespacePath],
1033 config: VolumeConfig,
1034 budget: WorkBudget,
1035) -> Result<PathBatchLookup, PathLookupFailure> {
1036 async_storage::poll_immediate(lookup_path_refs_async(
1037 store,
1038 generation,
1039 paths,
1040 config,
1041 budget,
1042 &CancellationToken::new(),
1043 ))
1044}
1045
1046#[allow(clippy::too_many_lines)]
1047async fn lookup_path_queries_async<S: AsyncObjectStore>(
1048 store: &S,
1049 generation: &super::GenerationRoot,
1050 paths: PathQueries<'_>,
1051 config: VolumeConfig,
1052 budget: WorkBudget,
1053 cancellation: &CancellationToken,
1054) -> Result<PathBatchLookup, PathLookupFailure> {
1055 cancellation.check().map_err(|_| {
1056 OperationFailure::before_work(PathLookupError::Tree(TreeReadError::Cancelled))
1057 })?;
1058 config
1059 .validate()
1060 .map_err(|error| OperationFailure::before_work(error.into()))?;
1061 if paths.is_empty() {
1062 return Err(OperationFailure::before_work(PathLookupError::EmptyBatch));
1063 }
1064 if u32::try_from(paths.len()).unwrap_or(u32::MAX) > config.limits.maximum_paths_per_batch {
1065 return Err(OperationFailure::before_work(PathLookupError::TooManyPaths));
1066 }
1067 for index in 0..paths.len() {
1068 validate_path(paths.get(index), config)?;
1069 }
1070
1071 let total_components = (0..paths.len())
1072 .try_fold(0_usize, |total, index| {
1073 total
1074 .checked_add(paths.get(index).depth())
1075 .ok_or(PathLookupError::Work(WorkError::Overflow))
1076 })
1077 .map_err(OperationFailure::before_work)?;
1078 const MAXIMUM_BATCH_CACHE_ENTRIES: usize = 4_096;
1083 let cache_entries = maximum_cache_entries_for_components(total_components, config)?
1084 .min(MAXIMUM_BATCH_CACHE_ENTRIES);
1085 let (cache, mut work) = OperationReadCache::new(store, cache_entries, budget)?;
1086 let mut allocations = AllocationLedger::default();
1087 allocations
1088 .claim_bytes(cache.metadata_bytes, 0, &mut work, budget)
1089 .map_err(|error| allocation_failure(error, work))?;
1090
1091 let count = paths.len();
1092 let mut entries = reserve_fixed::<PathBatchEntry>(count, &mut allocations, &mut work, budget)?;
1093 let mut current =
1094 reserve_fixed::<Option<FileRecord>>(count, &mut allocations, &mut work, budget)?;
1095 let mut done = reserve_fixed::<u8>(count, &mut allocations, &mut work, budget)?;
1096 let mut active = reserve_fixed::<ActivePath>(count, &mut allocations, &mut work, budget)?;
1097 let mut names = reserve_fixed::<LogicalName>(count, &mut allocations, &mut work, budget)?;
1098 let mut query_indices = reserve_fixed::<usize>(count, &mut allocations, &mut work, budget)?;
1099 let mut pending = reserve_fixed::<PendingBinding>(count, &mut allocations, &mut work, budget)?;
1100 let mut file_ids =
1101 reserve_fixed::<crate::foundation::FileId>(count, &mut allocations, &mut work, budget)?;
1102 entries.resize(
1103 count,
1104 PathBatchEntry {
1105 record: None,
1106 parent: None,
1107 resolved_components: 0,
1108 },
1109 );
1110 current.resize(count, None);
1111 done.resize(count, 0);
1112 let limits = decode_limits(config);
1113 let root = lookup_file_record_async(
1114 &cache,
1115 generation.file_table,
1116 generation.root_file_id,
1117 limits,
1118 batch_sub_budget(work, budget, orchestration_live(&cache, allocations)?)?,
1119 cancellation,
1120 )
1121 .await
1122 .map_err(|failure| {
1123 merge_batch_failure(
1124 work,
1125 *failure.work,
1126 orchestration_live(&cache, allocations).unwrap_or(u64::MAX),
1127 PathLookupError::FileTable(failure.error),
1128 )
1129 })?;
1130 work = merge_batch_work(
1131 work,
1132 root.work,
1133 orchestration_live(&cache, allocations)?,
1134 budget,
1135 )?;
1136 let root_record = root
1137 .record
1138 .ok_or_else(|| OperationFailure::new(PathLookupError::MissingRootRecord, work))?;
1139 #[allow(
1140 clippy::indexing_slicing,
1141 reason = "entries, current, and done are each reserved and resized to exactly `count == paths.len()` above (reserve_fixed + entries.resize(count, ..)/current.resize(count, None)/done.resize(count, 0)) and are never resized again, so any index in 0..paths.len() is in bounds for all three"
1142 )]
1143 for index in 0..paths.len() {
1144 let path = paths.get(index);
1145 if path.is_root() {
1146 entries[index].record = Some(root_record);
1147 done[index] = 1;
1148 } else {
1149 current[index] = Some(root_record);
1150 }
1151 }
1152
1153 let maximum_depth = (0..paths.len())
1154 .map(|index| paths.get(index).depth())
1155 .max()
1156 .unwrap_or(0);
1157 for depth in 0..maximum_depth {
1158 pending.clear();
1159 active.clear();
1160 #[allow(
1161 clippy::indexing_slicing,
1162 reason = "done and current are each resized to exactly `count == paths.len()` once, before this loop, and never resized again, so any index in 0..paths.len() is in bounds for both"
1163 )]
1164 for index in 0..paths.len() {
1165 if done[index] != 0 || paths.get(index).depth() <= depth {
1166 continue;
1167 }
1168 let FilePayload::Directory { entries: directory } = current[index]
1169 .ok_or_else(|| OperationFailure::new(PathLookupError::KindMismatch, work))?
1170 .payload
1171 else {
1172 return Err(OperationFailure::new(PathLookupError::NotDirectory, work));
1173 };
1174 active.push(ActivePath {
1175 path_index: index,
1176 directory,
1177 });
1178 }
1179 let comparisons = std::cell::Cell::new(0_u64);
1180 active.sort_unstable_by(|left, right| {
1181 comparisons.set(comparisons.get().saturating_add(1));
1182 compare_object_id(left.directory, right.directory)
1183 .then_with(|| left.path_index.cmp(&right.path_index))
1184 });
1185 work = work
1186 .checked_add(WorkCounters {
1187 items_examined: comparisons.get(),
1188 ..WorkCounters::default()
1189 })
1190 .map_err(|error| OperationFailure::new(error.into(), work))?;
1191 work.verify(budget)
1192 .map_err(|error| OperationFailure::new(error.into(), work))?;
1193 let mut cursor = 0_usize;
1194 while cursor < active.len() {
1195 #[allow(
1196 clippy::indexing_slicing,
1197 reason = "guarded by the enclosing `while cursor < active.len()` condition"
1198 )]
1199 let directory = active[cursor].directory;
1200 let group_start = cursor;
1201 cursor += 1;
1202 #[allow(
1203 clippy::indexing_slicing,
1204 reason = "short-circuit: `cursor < active.len()` is checked first in the same `&&` expression before active[cursor] is evaluated"
1205 )]
1206 while cursor < active.len() && active[cursor].directory == directory {
1207 cursor += 1;
1208 }
1209 names.clear();
1210 query_indices.clear();
1211 #[allow(
1212 clippy::indexing_slicing,
1213 reason = "group_start starts each outer iteration equal to the prior cursor (<= active.len()) and only cursor advances (via the bounded while loops above) before this slice is taken, so group_start <= cursor <= active.len()"
1214 )]
1215 for item in &active[group_start..cursor] {
1216 query_indices.push(item.path_index);
1217 }
1218 let mut nested_bytes = 0_u64;
1219 #[allow(
1220 clippy::indexing_slicing,
1221 reason = "index is a path_index drawn from `active`, whose entries were pushed only when `paths.get(index).depth() > depth` in the loop above (the `if done[index] != 0 || paths.get(index).depth() <= depth { continue; }` guard), so depth < paths.get(*index).depth() == components().len()"
1222 )]
1223 for index in &query_indices {
1224 let source = &paths.get(*index).components()[depth];
1225 let (name, owned_bytes) = copy_batch_name(
1226 source,
1227 config.limits.maximum_component_bytes,
1228 &mut allocations,
1229 &mut work,
1230 budget,
1231 )?;
1232 nested_bytes = nested_bytes.checked_add(owned_bytes).ok_or_else(|| {
1233 OperationFailure::new(PathLookupError::Work(WorkError::Overflow), work)
1234 })?;
1235 names.push(name);
1236 }
1237 let looked_up = lookup_tree_entries_async(
1238 &cache,
1239 directory,
1240 &names,
1241 config.limits.maximum_paths_per_batch,
1242 limits,
1243 batch_sub_budget(work, budget, orchestration_live(&cache, allocations)?)?,
1244 cancellation,
1245 )
1246 .await
1247 .map_err(|failure| {
1248 merge_batch_failure(
1249 work,
1250 *failure.work,
1251 orchestration_live(&cache, allocations).unwrap_or(u64::MAX),
1252 PathLookupError::Tree(failure.error),
1253 )
1254 })?;
1255 work = merge_batch_work(
1256 work,
1257 looked_up.work,
1258 orchestration_live(&cache, allocations)?,
1259 budget,
1260 )?;
1261 #[allow(
1262 clippy::indexing_slicing,
1263 reason = "index is drawn from query_indices, whose entries are path_index values sourced from the earlier `0..paths.len()` loop; entries, current, and done are each resized to exactly `count == paths.len()` once and never resized again, so *index is always in bounds for all three"
1264 )]
1265 for (index, binding) in query_indices.iter().zip(looked_up.entries) {
1266 if let Some(binding) = binding {
1267 pending.push(PendingBinding {
1268 path_index: *index,
1269 file_id: binding.file_id,
1270 kind: binding.kind,
1271 });
1272 } else {
1273 entries[*index] = PathBatchEntry {
1274 record: None,
1275 parent: current[*index],
1276 resolved_components: u16::try_from(depth).unwrap_or(u16::MAX),
1277 };
1278 done[*index] = 1;
1279 }
1280 }
1281 names.clear();
1282 allocations
1283 .release(nested_bytes)
1284 .map_err(|error| allocation_failure(error, work))?;
1285 }
1286 if pending.is_empty() {
1287 continue;
1288 }
1289 file_ids.clear();
1290 file_ids.extend(pending.iter().map(|binding| binding.file_id));
1291 let records = lookup_file_records_async(
1292 &cache,
1293 generation.file_table,
1294 &file_ids,
1295 config.limits.maximum_paths_per_batch,
1296 limits,
1297 batch_sub_budget(work, budget, orchestration_live(&cache, allocations)?)?,
1298 cancellation,
1299 )
1300 .await
1301 .map_err(|failure| {
1302 merge_batch_failure(
1303 work,
1304 *failure.work,
1305 orchestration_live(&cache, allocations).unwrap_or(u64::MAX),
1306 PathLookupError::FileTable(failure.error),
1307 )
1308 })?;
1309 work = merge_batch_work(
1310 work,
1311 records.work,
1312 orchestration_live(&cache, allocations)?,
1313 budget,
1314 )?;
1315 #[allow(
1316 clippy::indexing_slicing,
1317 reason = "binding.path_index is a PendingBinding.path_index, which is always constructed from query_indices entries (see `pending.push(PendingBinding { path_index: *index, .. })` above), themselves sourced from the `0..paths.len()` loop; entries, current, and done are each resized to exactly `count == paths.len()` once and never resized again, so binding.path_index is always in bounds for all three"
1318 )]
1319 for (binding, record) in pending.iter().zip(records.records) {
1320 let record =
1321 record.ok_or_else(|| OperationFailure::new(PathLookupError::KindMismatch, work))?;
1322 if record.kind != binding.kind {
1323 return Err(OperationFailure::new(PathLookupError::KindMismatch, work));
1324 }
1325 let next_depth = depth + 1;
1326 if next_depth == paths.get(binding.path_index).depth() {
1327 entries[binding.path_index] = PathBatchEntry {
1328 record: Some(record),
1329 parent: current[binding.path_index],
1330 resolved_components: u16::try_from(next_depth).unwrap_or(u16::MAX),
1331 };
1332 done[binding.path_index] = 1;
1333 current[binding.path_index] = None;
1334 } else {
1335 current[binding.path_index] = Some(record);
1336 }
1337 }
1338 }
1339 if done.contains(&0) {
1340 return Err(OperationFailure::new(PathLookupError::KindMismatch, work));
1341 }
1342 let retained_allocation_bytes = u64::try_from(entries.capacity())
1343 .unwrap_or(u64::MAX)
1344 .checked_mul(u64::try_from(size_of::<PathBatchEntry>()).unwrap_or(u64::MAX))
1345 .ok_or_else(|| OperationFailure::new(PathLookupError::Work(WorkError::Overflow), work))?;
1346 Ok(PathBatchLookup {
1347 entries,
1348 retained_allocation_bytes,
1349 work,
1350 })
1351}
1352
1353fn validate_path(path: &NamespacePath, config: VolumeConfig) -> Result<(), PathLookupFailure> {
1354 config
1355 .validate()
1356 .map_err(|error| OperationFailure::before_work(error.into()))?;
1357 if path.encoded_bytes() > config.limits.maximum_path_bytes
1358 || path.depth() > usize::from(config.limits.maximum_path_depth)
1359 {
1360 return Err(OperationFailure::before_work(PathLookupError::PathBounds));
1361 }
1362 let maximum_component_bytes =
1363 usize::try_from(config.limits.maximum_component_bytes).unwrap_or(usize::MAX);
1364 for component in path.components() {
1365 if component.as_bytes().len() > maximum_component_bytes {
1366 return Err(OperationFailure::before_work(PathLookupError::PathBounds));
1367 }
1368 if !matches!(
1369 (config.profile, component.encoding()),
1370 (
1371 FilesystemProfile::Portable | FilesystemProfile::Browser,
1372 super::NameEncoding::Utf8
1373 ) | (
1374 FilesystemProfile::Posix,
1375 super::NameEncoding::Utf8 | super::NameEncoding::PosixBytes
1376 ) | (
1377 FilesystemProfile::Windows,
1378 super::NameEncoding::Utf8 | super::NameEncoding::WindowsUtf16Le
1379 )
1380 ) {
1381 return Err(OperationFailure::before_work(
1382 PathLookupError::UnsupportedNameEncoding,
1383 ));
1384 }
1385 }
1386 Ok(())
1387}
1388
1389fn decode_limits(config: VolumeConfig) -> DecodeLimits {
1390 DecodeLimits {
1391 maximum_object_bytes: config.limits.maximum_object_bytes,
1392 maximum_name_bytes: config.limits.maximum_component_bytes,
1393 maximum_page_items: config.limits.maximum_directory_page_entries,
1394 maximum_page_bytes: u32::try_from(config.limits.maximum_object_bytes).unwrap_or(u32::MAX),
1395 maximum_page_height: config.limits.maximum_page_height,
1396 maximum_visited_pages: u32::try_from(config.limits.maximum_objects_per_generation)
1397 .unwrap_or(u32::MAX),
1398 }
1399}
1400
1401fn reserve_fixed<T>(
1402 count: usize,
1403 allocations: &mut AllocationLedger,
1404 work: &mut WorkCounters,
1405 budget: WorkBudget,
1406) -> Result<Vec<T>, PathLookupFailure> {
1407 let bytes = allocations
1408 .claim_elements::<T>(count, work, budget)
1409 .map_err(|error| allocation_failure(error, *work))?;
1410 let mut values = Vec::new();
1411 if values.try_reserve_exact(count).is_err() {
1412 allocations
1413 .release(bytes)
1414 .map_err(|error| allocation_failure(error, *work))?;
1415 return Err(OperationFailure::new(
1416 PathLookupError::AllocationFailed,
1417 *work,
1418 ));
1419 }
1420 Ok(values)
1421}
1422
1423fn copy_batch_name(
1424 source: &LogicalName,
1425 maximum_bytes: u32,
1426 allocations: &mut AllocationLedger,
1427 work: &mut WorkCounters,
1428 budget: WorkBudget,
1429) -> Result<(LogicalName, u64), PathLookupFailure> {
1430 let bytes = crate::foundation::usize_to_u64(source.as_bytes().len());
1431 let attempted = work
1432 .checked_add(WorkCounters {
1433 allocation_operations: 1,
1434 ..WorkCounters::default()
1435 })
1436 .map_err(|error| OperationFailure::new(error.into(), *work))?;
1437 attempted
1438 .verify(budget)
1439 .map_err(|error| OperationFailure::new(error.into(), *work))?;
1440 *work = attempted;
1441
1442 let mut copied = Vec::new();
1443 copied
1444 .try_reserve_exact(source.as_bytes().len())
1445 .map_err(|_| OperationFailure::new(PathLookupError::AllocationFailed, *work))?;
1446 allocations
1447 .claim_bytes(bytes, 0, work, budget)
1448 .map_err(|error| allocation_failure(error, *work))?;
1449 let copied_work = work
1450 .checked_add(WorkCounters {
1451 bytes_copied: bytes,
1452 ..WorkCounters::default()
1453 })
1454 .map_err(|error| OperationFailure::new(error.into(), *work))?;
1455 copied_work
1456 .verify(budget)
1457 .map_err(|error| OperationFailure::new(error.into(), *work))?;
1458 copied.extend_from_slice(source.as_bytes());
1459 *work = copied_work;
1460 let name = LogicalName::new(source.encoding(), copied, maximum_bytes)
1461 .map_err(|_| OperationFailure::new(PathLookupError::KindMismatch, *work))?;
1462 Ok((name, bytes))
1463}
1464
1465struct CopiedObservationName {
1466 name: LogicalName,
1467 external_resident_bytes: u64,
1468 work: WorkCounters,
1469}
1470
1471fn copy_observation_name<S>(
1472 cache: &OperationReadCache<'_, S>,
1473 source: &LogicalName,
1474 maximum_bytes: u32,
1475 external_resident_bytes: u64,
1476 work: WorkCounters,
1477 budget: WorkBudget,
1478) -> Result<CopiedObservationName, PathLookupFailure> {
1479 let bytes = crate::foundation::usize_to_u64(source.as_bytes().len());
1480 let next_external = external_resident_bytes
1481 .checked_add(bytes)
1482 .ok_or_else(|| OperationFailure::new(PathLookupError::Work(WorkError::Overflow), work))?;
1483 let resident_before = cache
1484 .resident_bytes()
1485 .map_err(|failure| OperationFailure::new(map_cache_error(failure.error), work))?;
1486 let simultaneous = resident_before
1487 .checked_add(bytes)
1488 .ok_or_else(|| OperationFailure::new(PathLookupError::Work(WorkError::Overflow), work))?;
1489 let allocation_attempt = work
1490 .checked_add(WorkCounters {
1491 allocation_operations: 1,
1492 ..WorkCounters::default()
1493 })
1494 .map_err(|error| OperationFailure::new(error.into(), work))?;
1495 let mut admitted = allocation_attempt;
1496 admitted.peak_allocation_bytes = admitted.peak_allocation_bytes.max(simultaneous);
1497 admitted
1498 .verify(budget)
1499 .map_err(|error| OperationFailure::new(error.into(), work))?;
1500
1501 let mut copied = Vec::new();
1502 copied
1503 .try_reserve_exact(source.as_bytes().len())
1504 .map_err(|_| {
1505 OperationFailure::new(PathLookupError::AllocationFailed, allocation_attempt)
1506 })?;
1507 let copied_work = admitted
1508 .checked_add(WorkCounters {
1509 bytes_copied: bytes,
1510 ..WorkCounters::default()
1511 })
1512 .map_err(|error| OperationFailure::new(error.into(), admitted))?;
1513 copied_work
1514 .verify(budget)
1515 .map_err(|error| OperationFailure::new(error.into(), admitted))?;
1516 copied.extend_from_slice(source.as_bytes());
1517 let name = LogicalName::new(source.encoding(), copied, maximum_bytes)
1518 .map_err(|_| OperationFailure::new(PathLookupError::KindMismatch, copied_work))?;
1519 let retained = cache
1520 .add_external_resident_bytes(bytes)
1521 .map_err(|error| OperationFailure::new(error.into(), copied_work))?;
1522 if retained != next_external {
1523 return Err(OperationFailure::new(
1524 PathLookupError::Work(WorkError::Overflow),
1525 copied_work,
1526 ));
1527 }
1528 Ok(CopiedObservationName {
1529 name,
1530 external_resident_bytes: retained,
1531 work: copied_work,
1532 })
1533}
1534
1535fn allocation_failure(error: AllocationError, work: WorkCounters) -> PathLookupFailure {
1536 match error {
1537 AllocationError::Work(error) => OperationFailure::new(error.into(), work),
1538 AllocationError::Overflow | AllocationError::ReleaseInvariant => {
1539 OperationFailure::new(PathLookupError::Work(WorkError::Overflow), work)
1540 }
1541 AllocationError::InvalidCapacity
1542 | AllocationError::CapacityExceeded
1543 | AllocationError::AllocationFailed => {
1544 OperationFailure::new(PathLookupError::AllocationFailed, work)
1545 }
1546 }
1547}
1548
1549fn map_cache_error(error: ObjectStoreError) -> PathLookupError {
1550 match error {
1551 ObjectStoreError::Work(error) => PathLookupError::Work(error),
1552 error => PathLookupError::Tree(TreeReadError::Storage(error)),
1553 }
1554}
1555
1556fn orchestration_live<S>(
1557 cache: &OperationReadCache<'_, S>,
1558 allocations: AllocationLedger,
1559) -> Result<u64, PathLookupFailure> {
1560 allocations
1561 .live_bytes()
1562 .checked_sub(cache.metadata_bytes)
1563 .ok_or_else(|| OperationFailure::before_work(PathLookupError::Work(WorkError::Overflow)))
1564}
1565
1566fn batch_sub_budget(
1567 work: WorkCounters,
1568 budget: WorkBudget,
1569 orchestration_live: u64,
1570) -> Result<WorkBudget, PathLookupFailure> {
1571 let mut remaining = remaining(work, budget)?;
1572 remaining.peak_allocation_bytes = remaining
1573 .peak_allocation_bytes
1574 .checked_sub(orchestration_live)
1575 .ok_or_else(|| {
1576 OperationFailure::new(
1577 PathLookupError::Work(WorkError::BudgetExceeded {
1578 counter: "peak_allocation_bytes",
1579 observed: orchestration_live,
1580 maximum: remaining.peak_allocation_bytes,
1581 }),
1582 work,
1583 )
1584 })?;
1585 Ok(remaining)
1586}
1587
1588fn merge_batch_work(
1589 prior: WorkCounters,
1590 mut nested: WorkCounters,
1591 orchestration_live: u64,
1592 budget: WorkBudget,
1593) -> Result<WorkCounters, PathLookupFailure> {
1594 let simultaneous_peak = orchestration_live
1595 .checked_add(nested.peak_allocation_bytes)
1596 .ok_or_else(|| OperationFailure::new(PathLookupError::Work(WorkError::Overflow), prior))?;
1597 nested.peak_allocation_bytes = 0;
1598 let mut merged = prior
1599 .checked_add(nested)
1600 .map_err(|error| OperationFailure::new(error.into(), prior))?;
1601 merged.peak_allocation_bytes = merged.peak_allocation_bytes.max(simultaneous_peak);
1602 merged
1603 .verify(budget)
1604 .map_err(|error| OperationFailure::new(error.into(), merged))?;
1605 Ok(merged)
1606}
1607
1608fn merge_batch_failure(
1609 prior: WorkCounters,
1610 mut nested: WorkCounters,
1611 orchestration_live: u64,
1612 error: PathLookupError,
1613) -> PathLookupFailure {
1614 let Some(simultaneous_peak) = orchestration_live.checked_add(nested.peak_allocation_bytes)
1615 else {
1616 return OperationFailure::new(PathLookupError::Work(WorkError::Overflow), prior);
1617 };
1618 nested.peak_allocation_bytes = 0;
1619 let Ok(mut merged) = prior.checked_add(nested) else {
1620 return OperationFailure::new(PathLookupError::Work(WorkError::Overflow), prior);
1621 };
1622 merged.peak_allocation_bytes = merged.peak_allocation_bytes.max(simultaneous_peak);
1623 OperationFailure::new(error, merged)
1624}
1625
1626fn maximum_cache_entries(
1627 path: &NamespacePath,
1628 config: VolumeConfig,
1629) -> Result<usize, PathLookupFailure> {
1630 maximum_cache_entries_for_components(path.depth(), config)
1631}
1632
1633fn maximum_cache_entries_for_components(
1634 components: usize,
1635 config: VolumeConfig,
1636) -> Result<usize, PathLookupFailure> {
1637 let component_frontiers = components
1638 .checked_mul(2)
1639 .and_then(|value| value.checked_add(1))
1640 .ok_or_else(|| OperationFailure::before_work(PathLookupError::Work(WorkError::Overflow)))?;
1641 let maximum = component_frontiers
1642 .checked_mul(usize::from(config.limits.maximum_page_height))
1643 .ok_or_else(|| OperationFailure::before_work(PathLookupError::Work(WorkError::Overflow)))?;
1644 Ok(maximum
1645 .min(usize::try_from(config.limits.maximum_objects_per_generation).unwrap_or(usize::MAX)))
1646}
1647
1648fn add(left: WorkCounters, right: WorkCounters) -> Result<WorkCounters, PathLookupFailure> {
1649 left.checked_add(right)
1650 .map_err(|error| OperationFailure::new(error.into(), left))
1651}
1652
1653fn remaining(work: WorkCounters, budget: WorkBudget) -> Result<WorkBudget, PathLookupFailure> {
1654 work.remaining(budget)
1655 .map_err(|error| OperationFailure::new(error.into(), work))
1656}
1657
1658#[cfg(all(test, feature = "memory"))]
1659#[path = "tests/path_access.rs"]
1660mod tests;