Skip to main content

acyclic_fs/kernel/
range.rs

1//! Sparse extent planning that reads only page frontiers intersecting a request.
2
3use 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/// One clipped logical span required to satisfy a range read.
24#[derive(Clone, Debug, Eq, PartialEq)]
25pub struct ExtentSlice {
26    /// Inclusive logical file offset.
27    pub offset: u64,
28    /// Positive bytes in this slice.
29    pub length: u64,
30    /// Exclusive end of the complete canonical extent containing this slice.
31    pub source_end: u64,
32    /// Physical representation, with content offset adjusted to the slice.
33    pub kind: ExtentKind,
34}
35
36/// Bounded range plan and exact page-read evidence.
37#[derive(Clone, Debug, Eq, PartialEq)]
38pub struct ExtentPlan {
39    /// Ordered contiguous spans covering the requested range.
40    pub spans: Vec<ExtentSlice>,
41    /// Exact kernel-visible work performed.
42    pub work: WorkCounters,
43    /// Logical owned capacity retained by `spans` in this returned plan.
44    pub retained_allocation_bytes: u64,
45    /// Exact authenticated forward page exposed by this traversal, if any.
46    pub next_residency: Option<ResidencyHint>,
47}
48
49/// Native sparse seek class.
50#[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
51pub enum ExtentSeekTarget {
52    /// First content or physically allocated-zero byte.
53    Data,
54    /// First unallocated hole byte, including logical end-of-file.
55    Hole,
56}
57
58/// Complete admitted sparse seek request.
59#[derive(Clone, Copy, Debug, Eq, PartialEq)]
60pub struct ExtentSeekRequest {
61    /// Authenticated extent-tree root.
62    pub root: ObjectId,
63    /// Complete logical file size.
64    pub file_size: u64,
65    /// Inclusive logical starting offset.
66    pub offset: u64,
67    /// Sparse class to locate.
68    pub target: ExtentSeekTarget,
69    /// Canonical page and decode limits.
70    pub limits: DecodeLimits,
71    /// Exact admitted implementation work.
72    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/// Complete admitted request shared by synchronous and asynchronous range
111/// planners.
112#[derive(Clone, Copy, Debug, Eq, PartialEq)]
113pub struct ExtentRangeRequest {
114    /// Authenticated extent-tree root.
115    pub root: ObjectId,
116    /// Complete logical file size.
117    pub file_size: u64,
118    /// Exact logical range to cover.
119    pub range: ByteRange,
120    /// Maximum sparse spans returned.
121    pub maximum_spans: u32,
122    /// Canonical page and decode limits.
123    pub limits: DecodeLimits,
124    /// Exact admitted implementation work.
125    pub budget: WorkBudget,
126}
127
128/// Plans one exact file range without reading unrelated extent pages or blobs.
129///
130/// # Errors
131///
132/// Rejects invalid ranges, malformed page bounds, cycles, excessive height,
133/// backend failures, or work beyond the admitted budget.
134pub 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
156/// Asynchronously plans one exact sparse range through the same transition
157/// machine as [`plan_extent_range`].
158///
159/// # Errors
160///
161/// Returns the same typed planning failures, including cooperative cancellation.
162pub 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
188/// Finds one sparse data/hole boundary without restarting extent-tree
189/// traversal or reading file-content objects.
190///
191/// # Errors
192///
193/// Rejects invalid offsets, malformed authenticated pages, cycles, excessive
194/// height, cancellation, backend failures, or work beyond the admitted budget.
195pub 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
206/// Asynchronous sparse seek through the same monotonic transition machine as
207/// [`seek_extent`].
208///
209/// # Errors
210///
211/// Returns the same typed failures, including cooperative cancellation.
212pub 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/// Sparse extent range-planning failures.
854#[derive(Debug, Error)]
855pub enum ExtentReadError {
856    /// Cooperative cancellation occurred before the next storage boundary.
857    #[error("extent range planning was cancelled")]
858    Cancelled,
859    /// The private driver attempted an impossible transition.
860    #[error("extent range transition state is invalid")]
861    TraversalState,
862    /// Root object is not an authenticated extent page.
863    #[error("extent root is not an extent page")]
864    WrongRootKind,
865    /// Requested range exceeds the file or overflows.
866    #[error("requested file range is invalid")]
867    InvalidRange,
868    /// Requested output-span bound must be positive for a non-empty range.
869    #[error("extent span limit must be non-zero")]
870    InvalidSpanLimit,
871    /// Decode or traversal scratch could not be admitted or allocated.
872    #[error("extent planning scratch allocation failed")]
873    AllocationFailed,
874    /// Decode or traversal limits are not representable on this target.
875    #[error("extent planning limits are invalid")]
876    InvalidLimits,
877    /// Intersecting sparse spans exceed the caller's admitted output bound.
878    #[error("extent plan exceeds its admitted span count")]
879    TooManySpans,
880    /// Child graph references an ancestor or repeated page.
881    #[error("extent page graph contains a cycle or alias")]
882    Cycle,
883    /// Traversal exceeds its hard page-height bound.
884    #[error("extent page height exceeds its admitted bound")]
885    HeightExceeded,
886    /// Parent child range does not match the decoded child content.
887    #[error("extent child bounds do not match its page")]
888    ChildBoundsMismatch,
889    /// Planned spans do not cover the request exactly once.
890    #[error("extent plan does not cover the requested range exactly")]
891    IncompleteCoverage,
892    /// Stored page exceeds its decode/work bound.
893    #[error("extent page has {observed} bytes; maximum is {maximum}")]
894    PageTooLarge {
895        /// Observed canonical bytes.
896        observed: u64,
897        /// Admitted maximum.
898        maximum: u64,
899    },
900    /// Immutable-object backend failed.
901    #[error(transparent)]
902    Storage(#[from] ObjectStoreError),
903    /// Canonical page failed decoding or validation.
904    #[error(transparent)]
905    Decode(#[from] CanonicalDecodeError),
906    /// Exact work exceeded or overflowed its budget.
907    #[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
939/// Sparse extent-plan failure retaining exact spent work.
940pub 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;