1use super::allocation::{AllocationError, AllocationLedger, LogicalVecCapacity, VisitedObjectSet};
4use super::codec::DecodedPageKind;
5use super::extent::extent_page_decode_shape;
6use super::frontier;
7use super::frontier::Machine as _;
8use super::{CanonicalDecodeError, DecodeLimits, ExtentKind, ExtentPage, decode_extent_page};
9use crate::async_storage::{
10 AsyncObjectStore, DecodedCacheAdmission, DecodedCacheKey, DecodedCacheValue,
11};
12use crate::cancellation::CancellationToken;
13use crate::performance::{OperationFailure, OperationReceipt, WorkBudget, WorkCounters, WorkError};
14use crate::speculation::{ResidencyHint, ResidencyReason};
15use crate::storage::{
16 ByteRange, ObjectId, ObjectKind, ObjectRead, ObjectReadRequest, ObjectReadRetention,
17 ObjectReceipt, ObjectStore, ObjectStoreError,
18};
19use std::mem::size_of;
20use std::sync::Arc;
21use thiserror::Error;
22
23#[derive(Clone, Debug, Eq, PartialEq)]
25pub struct ExtentSlice {
26 pub offset: u64,
28 pub length: u64,
30 pub source_end: u64,
32 pub kind: ExtentKind,
34}
35
36#[derive(Clone, Debug, Eq, PartialEq)]
38pub struct ExtentPlan {
39 pub spans: Vec<ExtentSlice>,
41 pub work: WorkCounters,
43 pub retained_allocation_bytes: u64,
45 pub next_residency: Option<ResidencyHint>,
47}
48
49#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
51pub enum ExtentSeekTarget {
52 Data,
54 Hole,
56}
57
58#[derive(Clone, Copy, Debug, Eq, PartialEq)]
60pub struct ExtentSeekRequest {
61 pub root: ObjectId,
63 pub file_size: u64,
65 pub offset: u64,
67 pub target: ExtentSeekTarget,
69 pub limits: DecodeLimits,
71 pub budget: WorkBudget,
73}
74
75#[derive(Clone, Copy)]
76enum TraversalMode {
77 Plan,
78 Seek(ExtentSeekTarget),
79}
80
81enum RangeOutput {
82 Plan(ExtentPlan),
83 Seek(OperationReceipt<Option<u64>>),
84}
85
86struct PendingPage {
87 page: ObjectId,
88 first_offset: u64,
89 end_offset: u64,
90 height: u16,
91}
92
93#[derive(Clone, Copy)]
94struct PlanInput {
95 file_size: u64,
96 range: ByteRange,
97 maximum_spans: u32,
98}
99
100struct SpanCollector<'a> {
101 input: PlanInput,
102 range_end: u64,
103 budget: WorkBudget,
104 spans: &'a mut Vec<ExtentSlice>,
105 work: &'a mut WorkCounters,
106 allocations: &'a mut AllocationLedger,
107 spans_capacity: &'a mut LogicalVecCapacity,
108}
109
110#[derive(Clone, Copy, Debug, Eq, PartialEq)]
113pub struct ExtentRangeRequest {
114 pub root: ObjectId,
116 pub file_size: u64,
118 pub range: ByteRange,
120 pub maximum_spans: u32,
122 pub limits: DecodeLimits,
124 pub budget: WorkBudget,
126}
127
128pub fn plan_extent_range<S: ObjectStore>(
135 store: &S,
136 request: ExtentRangeRequest,
137) -> Result<ExtentPlan, ExtentReadFailure> {
138 let input = PlanInput {
139 file_size: request.file_size,
140 range: request.range,
141 maximum_spans: request.maximum_spans,
142 };
143 let mut machine = RangeMachine::new(
144 request.root,
145 input,
146 TraversalMode::Plan,
147 request.limits,
148 request.budget,
149 )?;
150 match frontier::drive_sync(store, &mut machine)? {
151 RangeOutput::Plan(plan) => Ok(plan),
152 RangeOutput::Seek(_) => Err(failed(ExtentReadError::TraversalState, machine.work)),
153 }
154}
155
156pub async fn plan_extent_range_async<S: AsyncObjectStore>(
163 store: &S,
164 request: ExtentRangeRequest,
165 cancellation: &CancellationToken,
166) -> Result<ExtentPlan, ExtentReadFailure> {
167 if cancellation.is_cancelled() {
168 return Err(failed(ExtentReadError::Cancelled, WorkCounters::default()));
169 }
170 let input = PlanInput {
171 file_size: request.file_size,
172 range: request.range,
173 maximum_spans: request.maximum_spans,
174 };
175 let mut machine = RangeMachine::new(
176 request.root,
177 input,
178 TraversalMode::Plan,
179 request.limits,
180 request.budget,
181 )?;
182 match drive_range_async(store, &mut machine, cancellation).await? {
183 RangeOutput::Plan(plan) => Ok(plan),
184 RangeOutput::Seek(_) => Err(failed(ExtentReadError::TraversalState, machine.work)),
185 }
186}
187
188pub fn seek_extent<S: ObjectStore>(
196 store: &S,
197 request: ExtentSeekRequest,
198) -> Result<OperationReceipt<Option<u64>>, ExtentReadFailure> {
199 let mut machine = seek_machine(&request)?;
200 match frontier::drive_sync(store, &mut machine)? {
201 RangeOutput::Seek(receipt) => Ok(receipt),
202 RangeOutput::Plan(_) => Err(failed(ExtentReadError::TraversalState, machine.work)),
203 }
204}
205
206pub async fn seek_extent_async<S: AsyncObjectStore>(
213 store: &S,
214 request: ExtentSeekRequest,
215 cancellation: &CancellationToken,
216) -> Result<OperationReceipt<Option<u64>>, ExtentReadFailure> {
217 if cancellation.is_cancelled() {
218 return Err(failed(ExtentReadError::Cancelled, WorkCounters::default()));
219 }
220 let mut machine = seek_machine(&request)?;
221 match drive_range_async(store, &mut machine, cancellation).await? {
222 RangeOutput::Seek(receipt) => Ok(receipt),
223 RangeOutput::Plan(_) => Err(failed(ExtentReadError::TraversalState, machine.work)),
224 }
225}
226
227async fn drive_range_async<S: AsyncObjectStore>(
228 store: &S,
229 machine: &mut RangeMachine,
230 cancellation: &CancellationToken,
231) -> Result<RangeOutput, ExtentReadFailure> {
232 loop {
233 if cancellation.is_cancelled() {
234 return Err(machine.cancelled());
235 }
236 if let Some(output) = machine.complete()? {
237 return Ok(output);
238 }
239 let request = machine.prepare_read()?;
240 let key = DecodedCacheKey::new::<ExtentPage>(request.page, machine.limits);
241 let cached = store
242 .decoded_cache_get(key)
243 .map_err(|error| failed(error.into(), machine.work))?;
244 if let Some(cached) = cached {
245 machine.accept_cached_page(request.prospective, cached)?;
246 continue;
247 }
248 let receipt = AsyncObjectStore::read(
249 store,
250 request.page,
251 request.maximum_bytes,
252 request.remaining,
253 cancellation,
254 )
255 .await
256 .map_err(|failure| machine.storage_failure(request.prospective, failure))?;
257 machine.accept_page_with_cache(store, key, request.prospective, &receipt)?;
258 }
259}
260
261fn seek_machine(request: &ExtentSeekRequest) -> Result<RangeMachine, ExtentReadFailure> {
262 if request.offset > request.file_size {
263 return Err(failed(
264 ExtentReadError::InvalidRange,
265 WorkCounters::default(),
266 ));
267 }
268 RangeMachine::new(
269 request.root,
270 PlanInput {
271 file_size: request.file_size,
272 range: ByteRange {
273 offset: request.offset,
274 length: request.file_size - request.offset,
275 },
276 maximum_spans: 1,
277 },
278 TraversalMode::Seek(request.target),
279 request.limits,
280 request.budget,
281 )
282}
283
284struct RangeMachine {
285 input: PlanInput,
286 mode: TraversalMode,
287 seek_result: Option<u64>,
288 range_end: u64,
289 limits: DecodeLimits,
290 budget: WorkBudget,
291 pending: Vec<PendingPage>,
292 pending_capacity: LogicalVecCapacity,
293 awaiting: Option<PendingPage>,
294 visited: VisitedObjectSet,
295 spans: Vec<ExtentSlice>,
296 spans_capacity: LogicalVecCapacity,
297 work: WorkCounters,
298 allocations: AllocationLedger,
299 successor_page: Option<(u64, ResidencyHint)>,
300}
301
302impl RangeMachine {
303 fn new(
304 root: ObjectId,
305 input: PlanInput,
306 mode: TraversalMode,
307 limits: DecodeLimits,
308 budget: WorkBudget,
309 ) -> Result<Self, ExtentReadFailure> {
310 if root.kind != ObjectKind::ExtentPage {
311 return Err(failed(
312 ExtentReadError::WrongRootKind,
313 WorkCounters::default(),
314 ));
315 }
316 let range_end = input
317 .range
318 .offset
319 .checked_add(input.range.length)
320 .ok_or_else(|| failed(ExtentReadError::InvalidRange, WorkCounters::default()))?;
321 if range_end > input.file_size {
322 return Err(failed(
323 ExtentReadError::InvalidRange,
324 WorkCounters::default(),
325 ));
326 }
327 if input.range.length != 0 && input.maximum_spans == 0 {
328 return Err(failed(
329 ExtentReadError::InvalidSpanLimit,
330 WorkCounters::default(),
331 ));
332 }
333 if !limits.page_limits_valid(1) {
334 return Err(failed(
335 ExtentReadError::HeightExceeded,
336 WorkCounters::default(),
337 ));
338 }
339 let maximum_visited = usize::try_from(limits.maximum_visited_pages)
340 .map_err(|_| failed(ExtentReadError::InvalidLimits, WorkCounters::default()))?;
341 let mut allocations = AllocationLedger::default();
342 let mut work = WorkCounters::default();
343 let visited = VisitedObjectSet::new(maximum_visited, &mut allocations, &mut work, budget)
344 .map_err(|error| failed(error.into(), work))?;
345 let mut pending = Vec::new();
346 let mut pending_capacity = LogicalVecCapacity::default();
347 if input.range.length != 0 {
348 pending_capacity
349 .ensure_for_push(
350 &mut pending,
351 maximum_visited,
352 &mut allocations,
353 &mut work,
354 budget,
355 )
356 .map_err(|error| failed(error.into(), work))?;
357 pending.push(PendingPage {
358 page: root,
359 first_offset: 0,
360 end_offset: input.file_size,
361 height: 1,
362 });
363 }
364 Ok(Self {
365 input,
366 mode,
367 seek_result: (input.range.length == 0
368 && matches!(mode, TraversalMode::Seek(ExtentSeekTarget::Hole)))
369 .then_some(input.file_size),
370 range_end,
371 limits,
372 budget,
373 pending,
374 pending_capacity,
375 awaiting: None,
376 visited,
377 spans: Vec::new(),
378 spans_capacity: LogicalVecCapacity::default(),
379 work,
380 allocations,
381 successor_page: None,
382 })
383 }
384
385 fn finish(&mut self) -> Result<RangeOutput, ExtentReadFailure> {
386 match self.mode {
387 TraversalMode::Plan => {
388 if self.input.range.length != 0
389 && (self.spans.first().map(|span| span.offset) != Some(self.input.range.offset)
390 || self
391 .spans
392 .last()
393 .and_then(|span| span.offset.checked_add(span.length))
394 != Some(self.range_end)
395 || self.spans.windows(2).any(|pair| {
396 let [first, second] = pair else {
397 return false;
398 };
399 first.offset.checked_add(first.length) != Some(second.offset)
400 }))
401 {
402 return Err(failed(ExtentReadError::IncompleteCoverage, self.work));
403 }
404 let retained_allocation_bytes = self.spans_capacity.logical_bytes();
405 Ok(RangeOutput::Plan(ExtentPlan {
406 spans: std::mem::take(&mut self.spans),
407 work: self.work,
408 retained_allocation_bytes,
409 next_residency: self.successor_page.map(|(_, hint)| hint),
410 }))
411 }
412 TraversalMode::Seek(target) => Ok(RangeOutput::Seek(OperationReceipt {
413 value: self
414 .seek_result
415 .or_else(|| (target == ExtentSeekTarget::Hole).then_some(self.input.file_size)),
416 work: self.work,
417 })),
418 }
419 }
420
421 fn prepare_read(&mut self) -> Result<frontier::ReadRequest, ExtentReadFailure> {
422 let expected = self
423 .pending
424 .pop()
425 .ok_or_else(|| failed(ExtentReadError::TraversalState, self.work))?;
426 if expected.height > self.limits.maximum_page_height {
427 return Err(failed(ExtentReadError::HeightExceeded, self.work));
428 }
429 let visited = self
430 .visited
431 .insert(
432 expected.page,
433 &mut self.allocations,
434 &mut self.work,
435 self.budget,
436 )
437 .map_err(|error| failed(error.into(), self.work))?;
438 if !visited.inserted {
439 return Err(failed(ExtentReadError::Cycle, self.work));
440 }
441 let prospective = self
442 .work
443 .checked_add(WorkCounters {
444 page_reads: 1,
445 ..WorkCounters::default()
446 })
447 .map_err(|error| failed(error.into(), self.work))?;
448 let mut remaining = prospective
449 .remaining(self.budget)
450 .map_err(|error| failed(error.into(), self.work))?;
451 remaining.peak_allocation_bytes = self
452 .budget
453 .peak_allocation_bytes
454 .checked_sub(self.allocations.live_bytes())
455 .ok_or_else(|| failed(ExtentReadError::Work(WorkError::Overflow), self.work))?;
456 let page = expected.page;
457 self.awaiting = Some(expected);
458 Ok(frontier::ReadRequest {
459 page,
460 maximum_bytes: self.limits.maximum_page_object_bytes(),
461 remaining,
462 prospective,
463 })
464 }
465
466 fn accept_page(
467 &mut self,
468 prospective: WorkCounters,
469 receipt: &ObjectReceipt<ObjectRead>,
470 ) -> Result<Option<RangeOutput>, ExtentReadFailure> {
471 let expected = self
472 .awaiting
473 .take()
474 .ok_or_else(|| failed(ExtentReadError::TraversalState, self.work))?;
475 self.work = merge_backend_work(prospective, receipt.work, self.allocations.live_bytes())
476 .map_err(|error| failed(error.into(), prospective))?;
477 self.work
478 .verify(self.budget)
479 .map_err(|error| failed(error.into(), self.work))?;
480 let (decoded, decoded_bytes, retained_bytes) = self.decode_page(receipt)?;
481 let result = self.accept_decoded(&decoded, &expected);
482 self.allocations
483 .release(decoded_bytes)
484 .map_err(|error| failed(error.into(), self.work))?;
485 self.allocations
486 .release(retained_bytes)
487 .map_err(|error| failed(error.into(), self.work))?;
488 result?;
489 Ok(None)
490 }
491
492 fn decode_page(
493 &mut self,
494 receipt: &ObjectReceipt<ObjectRead>,
495 ) -> Result<(ExtentPage, u64, u64), ExtentReadFailure> {
496 let retained_bytes = match receipt.value.retention {
497 ObjectReadRetention::Shared => 0,
498 ObjectReadRetention::Owned { logical_bytes } => logical_bytes,
499 };
500 self.allocations
501 .claim_bytes(retained_bytes, 0, &mut self.work, self.budget)
502 .map_err(|error| failed(error.into(), self.work))?;
503 let shape = extent_page_decode_shape(&receipt.value, self.limits)
504 .map_err(|error| failed(error.into(), self.work))?;
505 self.add_work(WorkCounters {
506 items_examined: u64::try_from(shape.items).unwrap_or(u64::MAX),
507 ..WorkCounters::default()
508 })?;
509 let item_bytes = match shape.kind {
510 DecodedPageKind::Leaf => size_of::<super::Extent>(),
511 DecodedPageKind::Internal => size_of::<super::ExtentChild>(),
512 };
513 let decoded_bytes = shape
514 .items
515 .checked_mul(item_bytes)
516 .map(crate::foundation::usize_to_u64)
517 .ok_or_else(|| failed(ExtentReadError::AllocationFailed, self.work))?;
518 self.allocations
519 .claim_bytes(
520 decoded_bytes,
521 u64::from(decoded_bytes != 0),
522 &mut self.work,
523 self.budget,
524 )
525 .map_err(|error| failed(error.into(), self.work))?;
526 let decoded = decode_extent_page(&receipt.value, self.limits)
527 .map_err(|error| failed(error.into(), self.work))?;
528 Ok((decoded, decoded_bytes, retained_bytes))
529 }
530
531 fn accept_cached_page(
532 &mut self,
533 prospective: WorkCounters,
534 cached: DecodedCacheValue,
535 ) -> Result<(), ExtentReadFailure> {
536 let expected = self
537 .awaiting
538 .take()
539 .ok_or_else(|| failed(ExtentReadError::TraversalState, self.work))?;
540 self.work = prospective;
541 let page = cached
542 .value
543 .downcast::<ExtentPage>()
544 .map_err(|_| failed(ObjectStoreError::Corrupt.into(), self.work))?;
545 self.accept_decoded(&page, &expected)
546 }
547
548 fn accept_page_with_cache<S: AsyncObjectStore>(
549 &mut self,
550 store: &S,
551 key: DecodedCacheKey,
552 prospective: WorkCounters,
553 receipt: &ObjectReceipt<ObjectRead>,
554 ) -> Result<(), ExtentReadFailure> {
555 let expected = self
556 .awaiting
557 .take()
558 .ok_or_else(|| failed(ExtentReadError::TraversalState, self.work))?;
559 self.work = merge_backend_work(prospective, receipt.work, self.allocations.live_bytes())
560 .map_err(|error| failed(error.into(), prospective))?;
561 self.work
562 .verify(self.budget)
563 .map_err(|error| failed(error.into(), self.work))?;
564 let (decoded, decoded_bytes, retained_bytes) = self.decode_page(receipt)?;
565 let admitted = store.decoded_cache_admit(
566 key,
567 DecodedCacheValue {
568 value: Arc::new(decoded),
569 logical_bytes: decoded_bytes,
570 },
571 );
572 let result = match admitted {
573 Ok(DecodedCacheAdmission::Shared(value)) => {
574 self.allocations
575 .release(decoded_bytes)
576 .map_err(|error| failed(error.into(), self.work))?;
577 let page = value
578 .value
579 .downcast::<ExtentPage>()
580 .map_err(|_| failed(ObjectStoreError::Corrupt.into(), self.work))?;
581 self.accept_decoded(&page, &expected)
582 }
583 Ok(DecodedCacheAdmission::Uncached(value)) => {
584 let page = value
585 .value
586 .downcast::<ExtentPage>()
587 .map_err(|_| failed(ObjectStoreError::Corrupt.into(), self.work))?;
588 let result = self.accept_decoded(&page, &expected);
589 self.allocations
590 .release(decoded_bytes)
591 .map_err(|error| failed(error.into(), self.work))?;
592 result
593 }
594 Err(error) => {
595 self.allocations
596 .release(decoded_bytes)
597 .map_err(|release| failed(release.into(), self.work))?;
598 Err(failed(error.into(), self.work))
599 }
600 };
601 self.allocations
602 .release(retained_bytes)
603 .map_err(|error| failed(error.into(), self.work))?;
604 result
605 }
606
607 fn accept_decoded(
608 &mut self,
609 decoded: &ExtentPage,
610 expected: &PendingPage,
611 ) -> Result<(), ExtentReadFailure> {
612 match decoded {
613 ExtentPage::Leaf(extents) => match self.mode {
614 TraversalMode::Plan => {
615 SpanCollector {
616 input: self.input,
617 range_end: self.range_end,
618 budget: self.budget,
619 spans: &mut self.spans,
620 work: &mut self.work,
621 allocations: &mut self.allocations,
622 spans_capacity: &mut self.spans_capacity,
623 }
624 .collect(extents, expected)
625 .map_err(|error| failed(error, self.work))?;
626 }
627 TraversalMode::Seek(target) => {
628 self.accept_seek_leaf(extents, expected, target)?;
629 }
630 },
631 ExtentPage::Internal(children) => {
632 if children.first().map(|child| child.first_offset) != Some(expected.first_offset)
633 || children.last().map(|child| child.end_offset) != Some(expected.end_offset)
634 {
635 return Err(failed(ExtentReadError::ChildBoundsMismatch, self.work));
636 }
637 for child in children.iter().rev() {
638 self.work = self
639 .work
640 .checked_add(WorkCounters {
641 items_examined: 1,
642 ..WorkCounters::default()
643 })
644 .map_err(|error| failed(error.into(), self.work))?;
645 self.work
646 .verify(self.budget)
647 .map_err(|error| failed(error.into(), self.work))?;
648 if child.first_offset < self.range_end
649 && child.end_offset > self.input.range.offset
650 {
651 self.pending_capacity
652 .ensure_for_push(
653 &mut self.pending,
654 usize::try_from(self.limits.maximum_visited_pages)
655 .unwrap_or(usize::MAX),
656 &mut self.allocations,
657 &mut self.work,
658 self.budget,
659 )
660 .map_err(|error| failed(error.into(), self.work))?;
661 self.pending.push(PendingPage {
662 page: child.page,
663 first_offset: child.first_offset,
664 end_offset: child.end_offset,
665 height: expected.height.checked_add(1).ok_or_else(|| {
666 failed(ExtentReadError::HeightExceeded, self.work)
667 })?,
668 });
669 } else if self.input.range.length != 0
670 && child.first_offset >= self.range_end
671 && self
672 .successor_page
673 .is_none_or(|(offset, _)| child.first_offset < offset)
674 {
675 self.successor_page = Some((
676 child.first_offset,
677 ResidencyHint {
678 request: ObjectReadRequest {
679 object_id: child.page,
680 maximum_bytes: self.limits.maximum_page_object_bytes(),
681 },
682 reason: ResidencyReason::SequentialRange,
683 },
684 ));
685 }
686 }
687 }
688 }
689 Ok(())
690 }
691
692 fn accept_seek_leaf(
693 &mut self,
694 extents: &[super::Extent],
695 expected: &PendingPage,
696 target: ExtentSeekTarget,
697 ) -> Result<(), ExtentReadFailure> {
698 if extents.first().map(|extent| extent.offset) != Some(expected.first_offset)
699 || extents
700 .last()
701 .and_then(|extent| extent.offset.checked_add(extent.length))
702 != Some(expected.end_offset)
703 {
704 return Err(failed(ExtentReadError::ChildBoundsMismatch, self.work));
705 }
706 for extent in extents {
707 self.add_work(WorkCounters {
708 items_examined: 1,
709 ..WorkCounters::default()
710 })?;
711 let end = extent
712 .offset
713 .checked_add(extent.length)
714 .ok_or_else(|| failed(ExtentReadError::InvalidRange, self.work))?;
715 if end <= self.input.range.offset {
716 continue;
717 }
718 let matches = match target {
719 ExtentSeekTarget::Data => !matches!(extent.kind, ExtentKind::Hole),
720 ExtentSeekTarget::Hole => matches!(extent.kind, ExtentKind::Hole),
721 };
722 if matches {
723 self.seek_result = Some(extent.offset.max(self.input.range.offset));
724 self.pending.clear();
725 return Ok(());
726 }
727 }
728 Ok(())
729 }
730
731 fn add_work(&mut self, delta: WorkCounters) -> Result<(), ExtentReadFailure> {
732 let prospective = self
733 .work
734 .checked_add(delta)
735 .map_err(|error| failed(error.into(), self.work))?;
736 prospective
737 .verify(self.budget)
738 .map_err(|error| failed(error.into(), self.work))?;
739 self.work = prospective;
740 Ok(())
741 }
742}
743
744impl frontier::Machine for RangeMachine {
745 type Output = RangeOutput;
746 type Failure = ExtentReadFailure;
747
748 fn complete(&mut self) -> Result<Option<Self::Output>, Self::Failure> {
749 if self.pending.is_empty() && self.awaiting.is_none() {
750 return self.finish().map(Some);
751 }
752 Ok(None)
753 }
754
755 fn prepare_read(&mut self) -> Result<frontier::ReadRequest, Self::Failure> {
756 RangeMachine::prepare_read(self)
757 }
758
759 fn accept(
760 &mut self,
761 prospective: WorkCounters,
762 receipt: &ObjectReceipt<ObjectRead>,
763 ) -> Result<Option<Self::Output>, Self::Failure> {
764 self.accept_page(prospective, receipt)
765 }
766
767 fn storage_failure(
768 &self,
769 prospective: WorkCounters,
770 failure: crate::storage::ObjectFailure,
771 ) -> Self::Failure {
772 match prospective.checked_add(*failure.work) {
773 Ok(combined) => failed(failure.error.into(), combined),
774 Err(error) => failed(error.into(), prospective),
775 }
776 }
777
778 fn cancelled(&self) -> Self::Failure {
779 failed(ExtentReadError::Cancelled, self.work)
780 }
781}
782
783impl SpanCollector<'_> {
784 fn collect(
785 &mut self,
786 extents: &[super::Extent],
787 expected: &PendingPage,
788 ) -> Result<(), ExtentReadError> {
789 if extents.first().map(|extent| extent.offset) != Some(expected.first_offset)
790 || extents
791 .last()
792 .and_then(|extent| extent.offset.checked_add(extent.length))
793 != Some(expected.end_offset)
794 {
795 return Err(ExtentReadError::ChildBoundsMismatch);
796 }
797 for extent in extents {
798 *self.work = self.work.checked_add(WorkCounters {
799 items_examined: 1,
800 ..WorkCounters::default()
801 })?;
802 self.work.verify(self.budget)?;
803 let extent_end = extent
804 .offset
805 .checked_add(extent.length)
806 .ok_or(ExtentReadError::InvalidRange)?;
807 let start = extent.offset.max(self.input.range.offset);
808 let end = extent_end.min(self.range_end);
809 if start < end {
810 if u32::try_from(self.spans.len()).unwrap_or(u32::MAX) >= self.input.maximum_spans {
811 return Err(ExtentReadError::TooManySpans);
812 }
813 self.spans_capacity.ensure_for_push(
814 self.spans,
815 usize::try_from(self.input.maximum_spans)
816 .map_err(|_| ExtentReadError::InvalidSpanLimit)?,
817 self.allocations,
818 self.work,
819 self.budget,
820 )?;
821 self.spans.push(ExtentSlice {
822 offset: start,
823 length: end - start,
824 source_end: extent_end,
825 kind: clip_kind(extent.kind.clone(), start - extent.offset)?,
826 });
827 *self.work = self.work.checked_add(WorkCounters {
828 items_returned: 1,
829 ..WorkCounters::default()
830 })?;
831 self.work.verify(self.budget)?;
832 }
833 }
834 Ok(())
835 }
836}
837
838fn clip_kind(kind: ExtentKind, delta: u64) -> Result<ExtentKind, ExtentReadError> {
839 match kind {
840 ExtentKind::Content {
841 object,
842 object_offset,
843 } => Ok(ExtentKind::Content {
844 object,
845 object_offset: object_offset
846 .checked_add(delta)
847 .ok_or(ExtentReadError::InvalidRange)?,
848 }),
849 other => Ok(other),
850 }
851}
852
853#[derive(Debug, Error)]
855pub enum ExtentReadError {
856 #[error("extent range planning was cancelled")]
858 Cancelled,
859 #[error("extent range transition state is invalid")]
861 TraversalState,
862 #[error("extent root is not an extent page")]
864 WrongRootKind,
865 #[error("requested file range is invalid")]
867 InvalidRange,
868 #[error("extent span limit must be non-zero")]
870 InvalidSpanLimit,
871 #[error("extent planning scratch allocation failed")]
873 AllocationFailed,
874 #[error("extent planning limits are invalid")]
876 InvalidLimits,
877 #[error("extent plan exceeds its admitted span count")]
879 TooManySpans,
880 #[error("extent page graph contains a cycle or alias")]
882 Cycle,
883 #[error("extent page height exceeds its admitted bound")]
885 HeightExceeded,
886 #[error("extent child bounds do not match its page")]
888 ChildBoundsMismatch,
889 #[error("extent plan does not cover the requested range exactly")]
891 IncompleteCoverage,
892 #[error("extent page has {observed} bytes; maximum is {maximum}")]
894 PageTooLarge {
895 observed: u64,
897 maximum: u64,
899 },
900 #[error(transparent)]
902 Storage(#[from] ObjectStoreError),
903 #[error(transparent)]
905 Decode(#[from] CanonicalDecodeError),
906 #[error(transparent)]
908 Work(#[from] WorkError),
909}
910
911impl From<AllocationError> for ExtentReadError {
912 fn from(error: AllocationError) -> Self {
913 match error {
914 AllocationError::Work(error) => Self::Work(error),
915 AllocationError::Overflow | AllocationError::ReleaseInvariant => {
916 Self::Work(WorkError::Overflow)
917 }
918 AllocationError::InvalidCapacity
919 | AllocationError::CapacityExceeded
920 | AllocationError::AllocationFailed => Self::AllocationFailed,
921 }
922 }
923}
924
925fn merge_backend_work(
926 prior: WorkCounters,
927 mut backend: WorkCounters,
928 live_bytes: u64,
929) -> Result<WorkCounters, WorkError> {
930 let simultaneous_peak = live_bytes
931 .checked_add(backend.peak_allocation_bytes)
932 .ok_or(WorkError::Overflow)?;
933 backend.peak_allocation_bytes = 0;
934 let mut merged = prior.checked_add(backend)?;
935 merged.peak_allocation_bytes = merged.peak_allocation_bytes.max(simultaneous_peak);
936 Ok(merged)
937}
938
939pub type ExtentReadFailure = OperationFailure<ExtentReadError>;
941
942fn failed(error: ExtentReadError, work: WorkCounters) -> ExtentReadFailure {
943 OperationFailure::new(error, work)
944}
945
946#[cfg(all(test, feature = "memory"))]
947#[path = "tests/range.rs"]
948mod tests;