Skip to main content

ferrum_interfaces/vnext/operation/
buffer_view.rs

1use std::ops::Range;
2
3use super::super::{
4    AllocationLifetime, BatchWorkShape, BufferDescriptor, BufferUsage, DeviceBufferRetention,
5    DeviceRuntime, DynamicResourceDemand, DynamicResourceShape, LeasedBufferView,
6    LogicalBackingBufferView, LogicalBackingSegmentBinding, NodeWorkContract, ResourceId,
7    ResourceTransactionIdentity, VNextError,
8};
9use super::foundation::invalid_operation;
10use super::{DynamicStorageView, ResolvedStorageComponent, ResolvedValueBinding};
11
12#[derive(Debug, Clone, Copy, PartialEq, Eq)]
13pub(super) enum ValueBindingPhysicalCoverage {
14    CanonicalComponent,
15    RuntimeTokenView,
16}
17
18pub(super) fn validate_value_binding_physical_coverage(
19    work: &NodeWorkContract,
20    binding: &ResolvedValueBinding,
21    component: &ResolvedStorageComponent,
22    descriptor: &BufferDescriptor,
23    dynamic_demand: Option<&DynamicResourceDemand>,
24    value_alignment_bytes: u64,
25) -> Result<ValueBindingPhysicalCoverage, VNextError> {
26    let required_end = component
27        .offset_bytes()
28        .checked_add(component.length_bytes())
29        .ok_or_else(|| invalid_operation("bound component range overflows u64"))?;
30    if descriptor.usage != binding.usage()
31        || descriptor.element_type != component.element_type()
32        || descriptor.alignment_bytes < value_alignment_bytes
33        || descriptor.alignment_bytes % value_alignment_bytes != 0
34        || component.offset_bytes() % value_alignment_bytes != 0
35    {
36        return Err(invalid_operation(format!(
37            "resource `{}` differs from its value binding",
38            component.resource_id()
39        )));
40    }
41
42    let Some(projection) = work.token_projection(binding.role(), binding.ordinal()) else {
43        if required_end > descriptor.size_bytes {
44            return Err(invalid_operation(format!(
45                "resource `{}` differs from its value binding",
46                component.resource_id()
47            )));
48        }
49        return Ok(ValueBindingPhysicalCoverage::CanonicalComponent);
50    };
51
52    let axis = usize::try_from(projection.axis())
53        .map_err(|_| invalid_operation("token projection axis exceeds usize"))?;
54    let canonical_extent = projection.canonical_extent();
55    let bytes_per_token = component
56        .length_bytes()
57        .checked_div(canonical_extent)
58        .filter(|bytes| *bytes > 0)
59        .ok_or_else(|| invalid_operation("token projection has zero canonical extent"))?;
60    let demand_matches = matches!(
61        dynamic_demand,
62        Some(DynamicResourceDemand::Tokens {
63            bytes_per_token: planned_bytes_per_token,
64            ..
65        }) if *planned_bytes_per_token == bytes_per_token
66    );
67    if binding.usage() != BufferUsage::Activations
68        || component.offset_bytes() != 0
69        || component.length_bytes() % canonical_extent != 0
70        || usize::try_from(projection.rank()).ok() != Some(binding.tensor().dimensions().len())
71        || binding.tensor().dimensions().get(axis) != Some(&canonical_extent)
72        || !demand_matches
73        || descriptor.size_bytes < bytes_per_token
74    {
75        return Err(invalid_operation(format!(
76            "resource `{}` differs from its token-projected value binding",
77            component.resource_id()
78        )));
79    }
80
81    Ok(ValueBindingPhysicalCoverage::RuntimeTokenView)
82}
83
84#[derive(Debug, Clone, Copy, PartialEq, Eq)]
85enum StepParticipantRangeCoordinates {
86    SourceToken,
87    ParticipantLocal,
88}
89
90/// Maps source-token coordinates used by product input uploads into shared
91/// Step backing. Fixed participant values already use local coordinates.
92pub(crate) fn translate_step_participant_upload_range(
93    demand: &DynamicResourceDemand,
94    work_shape: &BatchWorkShape,
95    participant_index: usize,
96    semantic_range: Range<u64>,
97) -> Result<Range<u64>, VNextError> {
98    translate_step_participant_range(
99        demand,
100        work_shape,
101        participant_index,
102        semantic_range,
103        StepParticipantRangeCoordinates::SourceToken,
104    )
105}
106
107/// Maps participant-local completion coordinates into shared Step backing.
108pub(crate) fn translate_step_participant_readback_range(
109    demand: &DynamicResourceDemand,
110    work_shape: &BatchWorkShape,
111    participant_index: usize,
112    semantic_range: Range<u64>,
113) -> Result<Range<u64>, VNextError> {
114    translate_step_participant_range(
115        demand,
116        work_shape,
117        participant_index,
118        semantic_range,
119        StepParticipantRangeCoordinates::ParticipantLocal,
120    )
121}
122
123/// Converts one provider-visible packed token range back into the
124/// participant-local coordinates required by completion readback requests.
125/// The inverse is explicit so a packed range cannot be projected twice.
126pub(crate) fn packed_step_token_range_to_participant_local_readback(
127    demand: &DynamicResourceDemand,
128    work_shape: &BatchWorkShape,
129    participant_index: usize,
130    packed_range: Range<u64>,
131) -> Result<Range<u64>, VNextError> {
132    if packed_range.start >= packed_range.end {
133        return Err(invalid_operation(
134            "packed participant token readback range is empty",
135        ));
136    }
137    let token_range = work_shape
138        .participant_token_ranges()
139        .get(participant_index)
140        .ok_or_else(|| invalid_operation("packed participant token readback is out of range"))?;
141    let DynamicResourceDemand::Tokens {
142        bytes_per_token,
143        maximum_tokens,
144    } = demand
145    else {
146        return Err(invalid_operation(
147            "packed participant token readback requires token demand",
148        ));
149    };
150    if work_shape.immediate_tokens() > *maximum_tokens {
151        return Err(invalid_operation(
152            "packed participant token readback exceeds its planned ceiling",
153        ));
154    }
155    let participant_start = token_range
156        .immediate_token_range()
157        .start
158        .checked_mul(*bytes_per_token)
159        .ok_or_else(|| invalid_operation("packed participant token offset overflows u64"))?;
160    let participant_end = token_range
161        .immediate_token_range()
162        .end
163        .checked_mul(*bytes_per_token)
164        .ok_or_else(|| invalid_operation("packed participant token range overflows u64"))?;
165    if packed_range.start < participant_start || packed_range.end > participant_end {
166        return Err(invalid_operation(
167            "packed participant token readback is outside its immediate span",
168        ));
169    }
170    Ok(packed_range.start - participant_start..packed_range.end - participant_start)
171}
172
173fn translate_step_participant_range(
174    demand: &DynamicResourceDemand,
175    work_shape: &BatchWorkShape,
176    participant_index: usize,
177    semantic_range: Range<u64>,
178    coordinates: StepParticipantRangeCoordinates,
179) -> Result<Range<u64>, VNextError> {
180    if semantic_range.start >= semantic_range.end {
181        return Err(invalid_operation(
182            "participant resource projection has an empty semantic range",
183        ));
184    }
185    if participant_index >= work_shape.participant_token_ranges().len() {
186        return Err(invalid_operation(
187            "participant resource projection is out of range",
188        ));
189    }
190    match demand {
191        DynamicResourceDemand::ActualSequences {
192            bytes_per_sequence,
193            maximum_sequences,
194        } => {
195            if work_shape.immediate_sequences() > *maximum_sequences
196                || semantic_range.end > *bytes_per_sequence
197            {
198                return Err(invalid_operation(
199                    "participant fixed resource projection exceeds its planned stride",
200                ));
201            }
202            let base = bytes_per_sequence
203                .checked_mul(u64::try_from(participant_index).map_err(|_| {
204                    invalid_operation("participant resource projection exceeds u64")
205                })?)
206                .ok_or_else(|| {
207                    invalid_operation("participant fixed resource projection overflows u64")
208                })?;
209            let translated_start = base
210                .checked_add(semantic_range.start)
211                .ok_or_else(|| invalid_operation("participant fixed range overflows u64"))?;
212            let translated_end = base
213                .checked_add(semantic_range.end)
214                .ok_or_else(|| invalid_operation("participant fixed range overflows u64"))?;
215            Ok(translated_start..translated_end)
216        }
217        DynamicResourceDemand::Tokens {
218            bytes_per_token,
219            maximum_tokens,
220        } => {
221            if work_shape.immediate_tokens() > *maximum_tokens {
222                return Err(invalid_operation(
223                    "participant token resource projection exceeds its planned ceiling",
224                ));
225            }
226            let token_range = &work_shape.participant_token_ranges()[participant_index];
227            let source = token_range.source_token_range();
228            let packed = token_range.immediate_token_range();
229            let source_start = source
230                .start
231                .checked_mul(*bytes_per_token)
232                .ok_or_else(|| invalid_operation("source token byte offset overflows u64"))?;
233            let source_end = source
234                .end
235                .checked_mul(*bytes_per_token)
236                .ok_or_else(|| invalid_operation("source token byte range overflows u64"))?;
237            let packed_start = packed
238                .start
239                .checked_mul(*bytes_per_token)
240                .ok_or_else(|| invalid_operation("packed token byte offset overflows u64"))?;
241            let relative_range = match coordinates {
242                StepParticipantRangeCoordinates::SourceToken => {
243                    if semantic_range.start < source_start || semantic_range.end > source_end {
244                        return Err(invalid_operation(
245                            "participant token resource projection is outside its source span",
246                        ));
247                    }
248                    semantic_range.start - source_start..semantic_range.end - source_start
249                }
250                StepParticipantRangeCoordinates::ParticipantLocal => {
251                    let span_bytes = source_end - source_start;
252                    if semantic_range.end > span_bytes {
253                        return Err(invalid_operation(
254                            "participant token readback exceeds its immediate span",
255                        ));
256                    }
257                    semantic_range
258                }
259            };
260            let translated_start = packed_start
261                .checked_add(relative_range.start)
262                .ok_or_else(|| invalid_operation("packed token projection overflows u64"))?;
263            let translated_end = translated_start
264                .checked_add(relative_range.end - relative_range.start)
265                .ok_or_else(|| invalid_operation("packed token range overflows u64"))?;
266            Ok(translated_start..translated_end)
267        }
268        // Fixed/page-shaped internal activations are batch-wide resources. The
269        // planner gives product-fixed I/O an ActualSequences demand, so only
270        // that typed demand is participant-local here.
271        _ => Ok(semantic_range),
272    }
273}
274
275#[derive(Debug, Clone, Copy, PartialEq, Eq)]
276pub enum OperationBufferStorageKind {
277    StaticContiguous,
278    DynamicContiguous,
279    DynamicPaged,
280}
281
282enum OperationBufferSource<'a, B> {
283    Static {
284        view: LeasedBufferView<'a, B>,
285        retention: DeviceBufferRetention,
286    },
287    Backing(LogicalBackingBufferView<'a, B>),
288}
289
290#[derive(Debug, Clone, Copy, PartialEq, Eq)]
291enum OperationBufferCoverage {
292    Exact,
293    /// The operation sees an exact logical prefix while resource authority
294    /// retains wider physical capacity for a frontier or reusable bucket.
295    BackingWindow {
296        offset_bytes: u64,
297    },
298}
299
300enum OperationRegionSource<'a, B> {
301    Contiguous {
302        buffer: &'a B,
303        physical_base_offset_bytes: u64,
304        retention: DeviceBufferRetention,
305    },
306    Paged {
307        bindings: &'a [LogicalBackingSegmentBinding<B>],
308        logical_origin_bytes: u64,
309    },
310}
311
312mod copy;
313pub(crate) use copy::OperationBufferCopy;
314
315/// A checked logical range translated to physical device-buffer regions.
316/// Dynamic buffers never expose an arena buffer without its physical offsets.
317pub struct OperationBufferRegions<'a, B> {
318    storage_kind: OperationBufferStorageKind,
319    logical_offset_bytes: u64,
320    logical_length_bytes: u64,
321    source: OperationRegionSource<'a, B>,
322}
323
324impl<'a, B> OperationBufferRegions<'a, B> {
325    pub const fn storage_kind(&self) -> OperationBufferStorageKind {
326        self.storage_kind
327    }
328
329    pub const fn logical_offset_bytes(&self) -> u64 {
330        self.logical_offset_bytes
331    }
332
333    pub const fn logical_length_bytes(&self) -> u64 {
334        self.logical_length_bytes
335    }
336
337    pub fn iter(&self) -> OperationBufferRegionIter<'a, B> {
338        let logical_end_bytes = self
339            .logical_offset_bytes
340            .checked_add(self.logical_length_bytes)
341            .expect("validated operation logical range does not overflow");
342        match &self.source {
343            OperationRegionSource::Contiguous {
344                buffer,
345                physical_base_offset_bytes,
346                retention,
347            } => OperationBufferRegionIter {
348                state: OperationBufferRegionIterState::Contiguous(Some(OperationPhysicalRegion {
349                    buffer: *buffer,
350                    logical_offset_bytes: self.logical_offset_bytes,
351                    physical_offset_bytes: physical_base_offset_bytes
352                        .checked_add(self.logical_offset_bytes)
353                        .expect("validated contiguous physical range does not overflow"),
354                    length_bytes: self.logical_length_bytes,
355                    retention: retention.clone(),
356                })),
357            },
358            OperationRegionSource::Paged {
359                bindings,
360                logical_origin_bytes,
361            } => OperationBufferRegionIter {
362                state: OperationBufferRegionIterState::Paged {
363                    bindings: *bindings,
364                    logical_origin_bytes: *logical_origin_bytes,
365                    requested_start_bytes: logical_origin_bytes
366                        .checked_add(self.logical_offset_bytes)
367                        .expect("validated paged window start does not overflow"),
368                    requested_end_bytes: logical_origin_bytes
369                        .checked_add(logical_end_bytes)
370                        .expect("validated paged window end does not overflow"),
371                    next_segment: 0,
372                    next_segment_logical_offset_bytes: 0,
373                },
374            },
375        }
376    }
377}
378
379/// One indivisible physical region. The buffer reference is intentionally
380/// returned only together with the physical byte range.
381pub struct OperationPhysicalRegion<'a, B> {
382    buffer: &'a B,
383    logical_offset_bytes: u64,
384    physical_offset_bytes: u64,
385    length_bytes: u64,
386    retention: DeviceBufferRetention,
387}
388
389impl<'a, B> OperationPhysicalRegion<'a, B> {
390    pub const fn logical_offset_bytes(&self) -> u64 {
391        self.logical_offset_bytes
392    }
393
394    pub const fn length_bytes(&self) -> u64 {
395        self.length_bytes
396    }
397
398    pub fn buffer_and_physical_range(
399        &self,
400    ) -> (&'a B, std::ops::Range<u64>, DeviceBufferRetention) {
401        (
402            self.buffer,
403            self.physical_offset_bytes
404                ..self
405                    .physical_offset_bytes
406                    .checked_add(self.length_bytes)
407                    .expect("validated physical region does not overflow"),
408            self.retention.clone(),
409        )
410    }
411}
412
413pub struct OperationBufferRegionIter<'a, B> {
414    state: OperationBufferRegionIterState<'a, B>,
415}
416
417enum OperationBufferRegionIterState<'a, B> {
418    Contiguous(Option<OperationPhysicalRegion<'a, B>>),
419    Paged {
420        bindings: &'a [LogicalBackingSegmentBinding<B>],
421        logical_origin_bytes: u64,
422        requested_start_bytes: u64,
423        requested_end_bytes: u64,
424        next_segment: usize,
425        next_segment_logical_offset_bytes: u64,
426    },
427}
428
429impl<'a, B> Iterator for OperationBufferRegionIter<'a, B> {
430    type Item = OperationPhysicalRegion<'a, B>;
431
432    fn next(&mut self) -> Option<Self::Item> {
433        match &mut self.state {
434            OperationBufferRegionIterState::Contiguous(region) => region.take(),
435            OperationBufferRegionIterState::Paged {
436                bindings,
437                logical_origin_bytes,
438                requested_start_bytes,
439                requested_end_bytes,
440                next_segment,
441                next_segment_logical_offset_bytes,
442            } => {
443                while let Some(binding) = bindings.get(*next_segment) {
444                    *next_segment += 1;
445                    let (segment_logical_end, region) = translate_paged_segment(
446                        binding.buffer(),
447                        binding.retention(),
448                        binding.segment().offset_bytes(),
449                        binding.segment().length_bytes(),
450                        *next_segment_logical_offset_bytes,
451                        *requested_start_bytes,
452                        *requested_end_bytes,
453                    );
454                    *next_segment_logical_offset_bytes = segment_logical_end;
455                    if let Some(mut region) = region {
456                        region.logical_offset_bytes = region
457                            .logical_offset_bytes
458                            .checked_sub(*logical_origin_bytes)
459                            .expect("paged region is inside its validated backing window");
460                        return Some(region);
461                    }
462                }
463                None
464            }
465        }
466    }
467}
468
469fn translate_paged_segment<'a, B>(
470    buffer: &'a B,
471    retention: DeviceBufferRetention,
472    physical_offset_bytes: u64,
473    length_bytes: u64,
474    logical_start_bytes: u64,
475    requested_start_bytes: u64,
476    requested_end_bytes: u64,
477) -> (u64, Option<OperationPhysicalRegion<'a, B>>) {
478    let logical_end_bytes = logical_start_bytes
479        .checked_add(length_bytes)
480        .expect("validated backing segments do not overflow");
481    let translated_start = logical_start_bytes.max(requested_start_bytes);
482    let translated_end = logical_end_bytes.min(requested_end_bytes);
483    let region = (translated_start < translated_end).then(|| OperationPhysicalRegion {
484        buffer,
485        logical_offset_bytes: translated_start,
486        physical_offset_bytes: physical_offset_bytes
487            .checked_add(translated_start - logical_start_bytes)
488            .expect("validated paged physical range does not overflow"),
489        length_bytes: translated_end - translated_start,
490        retention,
491    });
492    (logical_end_bytes, region)
493}
494
495const fn operation_storage_kind(view: DynamicStorageView) -> OperationBufferStorageKind {
496    match view {
497        DynamicStorageView::Contiguous => OperationBufferStorageKind::DynamicContiguous,
498        DynamicStorageView::PagedRegions { .. } => OperationBufferStorageKind::DynamicPaged,
499    }
500}
501
502fn validate_dynamic_binding_layout(
503    storage_kind: OperationBufferStorageKind,
504    logical_size_bytes: u64,
505    mut binding_lengths: impl ExactSizeIterator<Item = u64>,
506    coverage: OperationBufferCoverage,
507) -> Result<(), VNextError> {
508    let binding_count = binding_lengths.len();
509    if storage_kind == OperationBufferStorageKind::StaticContiguous {
510        return Err(invalid_operation(
511            "dynamic backing cannot claim static storage kind",
512        ));
513    }
514    if binding_count == 0 {
515        return Err(invalid_operation(
516            "dynamic backing has no physical segment binding",
517        ));
518    }
519    if storage_kind == OperationBufferStorageKind::DynamicContiguous && binding_count != 1 {
520        return Err(invalid_operation(
521            "contiguous dynamic storage requires one physical segment binding",
522        ));
523    }
524    let covered = binding_lengths.try_fold(0_u64, |total, length_bytes| {
525        total
526            .checked_add(length_bytes)
527            .ok_or_else(|| invalid_operation("backing segment coverage overflows u64"))
528    })?;
529    let window_offset = match coverage {
530        OperationBufferCoverage::Exact => 0,
531        OperationBufferCoverage::BackingWindow { offset_bytes } => offset_bytes,
532    };
533    let required_end = window_offset
534        .checked_add(logical_size_bytes)
535        .ok_or_else(|| invalid_operation("operation backing window overflows u64"))?;
536    if covered < required_end
537        || (coverage == OperationBufferCoverage::Exact && covered != logical_size_bytes)
538    {
539        return Err(invalid_operation(
540            "dynamic backing segments do not cover the operation's exact logical view",
541        ));
542    }
543    Ok(())
544}
545
546pub(super) fn sequence_execution_shape(
547    committed: DynamicResourceShape,
548    source_end_tokens: u64,
549) -> Result<DynamicResourceShape, VNextError> {
550    if committed.sequences() != 1
551        || source_end_tokens == 0
552        || source_end_tokens > committed.tokens()
553    {
554        return Err(invalid_operation(
555            "sequence operation frontier is empty or exceeds committed backing",
556        ));
557    }
558    Ok(DynamicResourceShape::from_validated(
559        1,
560        source_end_tokens,
561        committed.pages(),
562    ))
563}
564
565pub struct OperationBufferView<'a, B> {
566    descriptor: BufferDescriptor,
567    source: OperationBufferSource<'a, B>,
568    coverage: OperationBufferCoverage,
569    allocation_lifetime: AllocationLifetime,
570    packed_batch_coordinates: bool,
571}
572
573impl<'a, B> OperationBufferView<'a, B> {
574    pub(super) fn from_static(
575        view: LeasedBufferView<'a, B>,
576        retention: DeviceBufferRetention,
577    ) -> Self {
578        Self {
579            descriptor: view.committed_descriptor().clone(),
580            source: OperationBufferSource::Static { view, retention },
581            coverage: OperationBufferCoverage::Exact,
582            allocation_lifetime: AllocationLifetime::Plan,
583            packed_batch_coordinates: false,
584        }
585    }
586
587    pub(super) fn from_backing_exact(
588        descriptor: BufferDescriptor,
589        backing: LogicalBackingBufferView<'a, B>,
590        allocation_lifetime: AllocationLifetime,
591    ) -> Self {
592        Self::from_backing(
593            descriptor,
594            backing,
595            OperationBufferCoverage::Exact,
596            allocation_lifetime,
597        )
598    }
599
600    pub(crate) fn from_backing_prefix(
601        descriptor: BufferDescriptor,
602        backing: LogicalBackingBufferView<'a, B>,
603        allocation_lifetime: AllocationLifetime,
604    ) -> Self {
605        Self::from_backing(
606            descriptor,
607            backing,
608            OperationBufferCoverage::BackingWindow { offset_bytes: 0 },
609            allocation_lifetime,
610        )
611    }
612
613    pub(super) fn from_backing_window(
614        descriptor: BufferDescriptor,
615        backing: LogicalBackingBufferView<'a, B>,
616        offset_bytes: u64,
617        allocation_lifetime: AllocationLifetime,
618    ) -> Self {
619        Self::from_backing(
620            descriptor,
621            backing,
622            OperationBufferCoverage::BackingWindow { offset_bytes },
623            allocation_lifetime,
624        )
625    }
626
627    fn from_backing(
628        descriptor: BufferDescriptor,
629        backing: LogicalBackingBufferView<'a, B>,
630        coverage: OperationBufferCoverage,
631        allocation_lifetime: AllocationLifetime,
632    ) -> Self {
633        Self {
634            descriptor,
635            source: OperationBufferSource::Backing(backing),
636            coverage,
637            allocation_lifetime,
638            packed_batch_coordinates: false,
639        }
640    }
641
642    pub(super) fn with_packed_batch_coordinates(mut self, packed: bool) -> Self {
643        self.packed_batch_coordinates = packed;
644        self
645    }
646
647    pub(super) fn validate_runtime<R>(
648        &self,
649        runtime: &R,
650        expected_static_identity: Option<&ResourceTransactionIdentity>,
651    ) -> Result<(), VNextError>
652    where
653        R: DeviceRuntime<Buffer = B>,
654    {
655        match &self.source {
656            OperationBufferSource::Static { view, .. } => {
657                let actual = runtime.buffer_descriptor(view.buffer());
658                if Some(view.identity()) != expected_static_identity
659                    || &actual != view.committed_descriptor()
660                    || view.generation() == 0
661                {
662                    return Err(invalid_operation(format!(
663                        "runtime descriptor differs from committed static resource `{}`",
664                        self.resource_id()
665                    )));
666                }
667            }
668            OperationBufferSource::Backing(backing_view) => {
669                let bindings = backing_view.segment_bindings();
670                if bindings.is_empty()
671                    || bindings.len() != backing_view.committed_evidence_segments().count()
672                    || bindings
673                        .iter()
674                        .zip(backing_view.committed_evidence_segments())
675                        .any(|(binding, evidence)| {
676                            let actual = runtime.buffer_descriptor(binding.buffer());
677                            binding.segment() != evidence
678                                || binding.chunk() != evidence.chunk()
679                                || &actual != binding.descriptor()
680                                || binding
681                                    .segment()
682                                    .offset_bytes()
683                                    .checked_add(binding.segment().length_bytes())
684                                    .is_none_or(|end| end > binding.descriptor().size_bytes)
685                        })
686                {
687                    return Err(invalid_operation(format!(
688                        "runtime descriptor differs from a committed backing chunk for `{}`",
689                        self.resource_id()
690                    )));
691                }
692            }
693        }
694        Ok(())
695    }
696
697    pub fn resource_id(&self) -> &ResourceId {
698        &self.descriptor.resource_id
699    }
700
701    pub fn descriptor(&self) -> &BufferDescriptor {
702        &self.descriptor
703    }
704
705    pub const fn allocation_lifetime(&self) -> AllocationLifetime {
706        self.allocation_lifetime
707    }
708
709    pub const fn uses_packed_batch_coordinates(&self) -> bool {
710        self.packed_batch_coordinates
711    }
712
713    pub fn storage_kind(&self) -> OperationBufferStorageKind {
714        match &self.source {
715            OperationBufferSource::Static { .. } => OperationBufferStorageKind::StaticContiguous,
716            OperationBufferSource::Backing(view) => {
717                operation_storage_kind(view.storage_profile().view())
718            }
719        }
720    }
721
722    pub fn translate(
723        &self,
724        logical_offset_bytes: u64,
725        logical_length_bytes: u64,
726    ) -> Result<OperationBufferRegions<'_, B>, VNextError> {
727        let logical_end_bytes = logical_offset_bytes
728            .checked_add(logical_length_bytes)
729            .ok_or_else(|| invalid_operation("operation logical buffer range overflows u64"))?;
730        if logical_length_bytes == 0 || logical_end_bytes > self.descriptor.size_bytes {
731            return Err(invalid_operation(
732                "operation logical buffer range is empty or outside its resource",
733            ));
734        }
735        match &self.source {
736            OperationBufferSource::Static { view, retention } => Ok(OperationBufferRegions {
737                storage_kind: OperationBufferStorageKind::StaticContiguous,
738                logical_offset_bytes,
739                logical_length_bytes,
740                source: OperationRegionSource::Contiguous {
741                    buffer: view.buffer(),
742                    physical_base_offset_bytes: 0,
743                    retention: retention.clone(),
744                },
745            }),
746            OperationBufferSource::Backing(view) => {
747                let bindings = view.segment_bindings();
748                if bindings.len() != view.committed_evidence_segments().count()
749                    || bindings.iter().zip(view.committed_evidence_segments()).any(
750                        |(binding, segment)| {
751                            binding.segment() != segment || binding.chunk() != segment.chunk()
752                        },
753                    )
754                {
755                    return Err(invalid_operation(
756                        "dynamic backing bindings differ from committed segment evidence",
757                    ));
758                }
759                let storage_kind = operation_storage_kind(view.storage_profile().view());
760                let backing_window_offset = match self.coverage {
761                    OperationBufferCoverage::Exact => 0,
762                    OperationBufferCoverage::BackingWindow { offset_bytes } => offset_bytes,
763                };
764                validate_dynamic_binding_layout(
765                    storage_kind,
766                    self.descriptor.size_bytes,
767                    bindings
768                        .iter()
769                        .map(|binding| binding.segment().length_bytes()),
770                    self.coverage,
771                )?;
772                let source = match storage_kind {
773                    OperationBufferStorageKind::DynamicContiguous => {
774                        let binding = &bindings[0];
775                        OperationRegionSource::Contiguous {
776                            buffer: binding.buffer(),
777                            physical_base_offset_bytes: binding
778                                .segment()
779                                .offset_bytes()
780                                .checked_add(backing_window_offset)
781                                .ok_or_else(|| {
782                                    invalid_operation(
783                                        "participant backing physical offset overflows u64",
784                                    )
785                                })?,
786                            retention: binding.retention(),
787                        }
788                    }
789                    OperationBufferStorageKind::DynamicPaged => OperationRegionSource::Paged {
790                        bindings,
791                        logical_origin_bytes: backing_window_offset,
792                    },
793                    OperationBufferStorageKind::StaticContiguous => unreachable!(
794                        "dynamic storage kind was validated before region construction"
795                    ),
796                };
797                Ok(OperationBufferRegions {
798                    storage_kind,
799                    logical_offset_bytes,
800                    logical_length_bytes,
801                    source,
802                })
803            }
804        }
805    }
806}
807
808#[cfg(test)]
809mod operation_buffer_region_tests {
810    use super::{
811        packed_step_token_range_to_participant_local_readback, sequence_execution_shape,
812        translate_paged_segment, translate_step_participant_readback_range,
813        translate_step_participant_upload_range, validate_dynamic_binding_layout,
814        validate_value_binding_physical_coverage, OperationBufferCoverage, OperationBufferRegions,
815        OperationBufferStorageKind, OperationRegionSource, ValueBindingPhysicalCoverage,
816    };
817    use crate::vnext::{
818        AliasPolicy, BatchWorkShape, BufferDescriptor, BufferUsage, DeviceBufferRetention,
819        DynamicResourceDemand, DynamicResourceShape, ElementType, NodeWorkContract, ProgramValueId,
820        ResolvedTensorLayout, ResolvedTensorSpec, ResolvedValueBinding, ResolvedValueRole,
821        ResolvedValueStorage, ResourceId, TensorAccess, TokenSpanWork,
822    };
823    use std::sync::atomic::Ordering;
824    use std::sync::Arc;
825
826    #[test]
827    fn token_projection_validates_runtime_view_instead_of_canonical_extent() {
828        let resource_id = ResourceId::new("resource.activation.token-ids").unwrap();
829        let binding = ResolvedValueBinding::new(
830            ProgramValueId::new("value.input.token-ids").unwrap(),
831            ResolvedValueRole::Input,
832            0,
833            ResolvedTensorSpec::new(
834                vec![128],
835                ElementType::U32,
836                ResolvedTensorLayout::Contiguous,
837            )
838            .unwrap(),
839            TensorAccess::Read,
840            AliasPolicy::NoAlias,
841            BufferUsage::Activations,
842            None,
843            ResolvedValueStorage::single(resource_id.clone(), 0, 512, ElementType::U32).unwrap(),
844        )
845        .unwrap();
846        let work: NodeWorkContract = serde_json::from_value(serde_json::json!({
847            "tokens": {
848                "source": {
849                    "value_id": "value.input.token-ids",
850                    "role": "input",
851                    "ordinal": 0,
852                    "axis": 0,
853                    "rank": 1,
854                    "canonical_extent": 128
855                },
856                "projections": [{
857                    "value_id": "value.input.token-ids",
858                    "role": "input",
859                    "ordinal": 0,
860                    "axis": 0,
861                    "rank": 1,
862                    "canonical_extent": 128
863                }]
864            }
865        }))
866        .unwrap();
867        let descriptor = BufferDescriptor {
868            resource_id,
869            size_bytes: 160,
870            alignment_bytes: 16,
871            usage: BufferUsage::Activations,
872            element_type: ElementType::U32,
873        };
874        let component = &binding.storage().components()[0];
875        let demand = DynamicResourceDemand::tokens(4, 128).unwrap();
876
877        assert_eq!(
878            validate_value_binding_physical_coverage(
879                &work,
880                &binding,
881                component,
882                &descriptor,
883                Some(&demand),
884                16,
885            )
886            .unwrap(),
887            ValueBindingPhysicalCoverage::RuntimeTokenView
888        );
889        assert!(validate_value_binding_physical_coverage(
890            &NodeWorkContract::Fixed,
891            &binding,
892            component,
893            &descriptor,
894            Some(&demand),
895            16,
896        )
897        .is_err());
898        assert!(validate_value_binding_physical_coverage(
899            &work,
900            &binding,
901            component,
902            &descriptor,
903            Some(&DynamicResourceDemand::tokens(8, 128).unwrap()),
904            16,
905        )
906        .is_err());
907    }
908
909    #[test]
910    fn paged_translation_uses_each_chunks_exact_buffer() {
911        struct MockBinding<'a> {
912            buffer: &'a u8,
913            physical_offset_bytes: u64,
914            length_bytes: u64,
915        }
916
917        let first_buffer = 7_u8;
918        let second_buffer = 11_u8;
919        let bindings = [
920            MockBinding {
921                buffer: &first_buffer,
922                physical_offset_bytes: 64,
923                length_bytes: 8,
924            },
925            MockBinding {
926                buffer: &second_buffer,
927                physical_offset_bytes: 200,
928                length_bytes: 12,
929            },
930        ];
931        let mut next_logical_offset = 0;
932        let translated = bindings
933            .iter()
934            .filter_map(|binding| {
935                let (logical_end, region) = translate_paged_segment(
936                    binding.buffer,
937                    DeviceBufferRetention::plan(Arc::new(())),
938                    binding.physical_offset_bytes,
939                    binding.length_bytes,
940                    next_logical_offset,
941                    6,
942                    16,
943                );
944                next_logical_offset = logical_end;
945                region
946            })
947            .collect::<Vec<_>>();
948
949        assert_eq!(translated.len(), 2);
950        let (first, first_physical, _first_retention) = translated[0].buffer_and_physical_range();
951        assert!(std::ptr::eq(first, &first_buffer));
952        assert_eq!(translated[0].logical_offset_bytes(), 6);
953        assert_eq!(first_physical, 70..72);
954        let (second, second_physical, _second_retention) =
955            translated[1].buffer_and_physical_range();
956        assert!(std::ptr::eq(second, &second_buffer));
957        assert_eq!(translated[1].logical_offset_bytes(), 8);
958        assert_eq!(second_physical, 200..208);
959    }
960
961    #[test]
962    fn contiguous_layout_rejects_cross_chunk_bindings() {
963        let first_buffer = 7_u8;
964        let second_buffer = 11_u8;
965        let bindings = [(&first_buffer, 8_u64), (&second_buffer, 12_u64)];
966
967        let error = validate_dynamic_binding_layout(
968            OperationBufferStorageKind::DynamicContiguous,
969            20,
970            bindings.iter().map(|(_, length_bytes)| *length_bytes),
971            OperationBufferCoverage::Exact,
972        )
973        .unwrap_err();
974
975        assert!(error
976            .to_string()
977            .contains("contiguous dynamic storage requires one physical segment binding"));
978    }
979
980    #[test]
981    fn operation_prefix_view_retains_wider_backing_coverage() {
982        validate_dynamic_binding_layout(
983            OperationBufferStorageKind::DynamicPaged,
984            64,
985            [64_u64, 64].into_iter(),
986            OperationBufferCoverage::BackingWindow { offset_bytes: 0 },
987        )
988        .unwrap();
989        validate_dynamic_binding_layout(
990            OperationBufferStorageKind::DynamicContiguous,
991            64,
992            [128_u64].into_iter(),
993            OperationBufferCoverage::BackingWindow { offset_bytes: 0 },
994        )
995        .unwrap();
996
997        assert!(validate_dynamic_binding_layout(
998            OperationBufferStorageKind::DynamicPaged,
999            64,
1000            [64_u64, 64].into_iter(),
1001            OperationBufferCoverage::Exact,
1002        )
1003        .is_err());
1004        assert!(validate_dynamic_binding_layout(
1005            OperationBufferStorageKind::DynamicPaged,
1006            128,
1007            [64_u64].into_iter(),
1008            OperationBufferCoverage::BackingWindow { offset_bytes: 0 },
1009        )
1010        .is_err());
1011    }
1012
1013    #[test]
1014    fn participant_step_projection_keeps_upload_and_readback_coordinate_spaces_distinct() {
1015        let first_tokens = vec![1_u32; 20];
1016        let second_tokens = vec![2_u32; 9];
1017        let work = BatchWorkShape::test_only(vec![
1018            TokenSpanWork::from_token_ids(&first_tokens, 17..18).unwrap(),
1019            TokenSpanWork::from_token_ids(&second_tokens, 7..9).unwrap(),
1020        ])
1021        .unwrap();
1022        let tokens = DynamicResourceDemand::tokens(4, 32).unwrap();
1023
1024        assert_eq!(
1025            translate_step_participant_upload_range(&tokens, &work, 0, 68..72).unwrap(),
1026            0..4
1027        );
1028        assert_eq!(
1029            translate_step_participant_upload_range(&tokens, &work, 1, 28..36).unwrap(),
1030            4..12
1031        );
1032        assert_eq!(
1033            translate_step_participant_readback_range(&tokens, &work, 1, 0..8).unwrap(),
1034            4..12
1035        );
1036        assert_eq!(
1037            packed_step_token_range_to_participant_local_readback(&tokens, &work, 1, 4..12)
1038                .unwrap(),
1039            0..8
1040        );
1041        assert!(translate_step_participant_readback_range(&tokens, &work, 1, 28..36).is_err());
1042        assert!(
1043            packed_step_token_range_to_participant_local_readback(&tokens, &work, 1, 28..36)
1044                .is_err()
1045        );
1046    }
1047
1048    #[test]
1049    fn fixed_participant_step_projection_uses_disjoint_aligned_strides() {
1050        let tokens = [1_u32];
1051        let work = BatchWorkShape::test_only(vec![
1052            TokenSpanWork::from_token_ids(&tokens, 0..1).unwrap(),
1053            TokenSpanWork::from_token_ids(&tokens, 0..1).unwrap(),
1054        ])
1055        .unwrap();
1056        let fixed = DynamicResourceDemand::actual_sequences(16, 4).unwrap();
1057
1058        assert_eq!(
1059            translate_step_participant_upload_range(&fixed, &work, 0, 0..4).unwrap(),
1060            0..4
1061        );
1062        assert_eq!(
1063            translate_step_participant_upload_range(&fixed, &work, 1, 0..4).unwrap(),
1064            16..20
1065        );
1066        assert_eq!(
1067            translate_step_participant_readback_range(&fixed, &work, 1, 0..4).unwrap(),
1068            16..20
1069        );
1070        assert!(translate_step_participant_upload_range(&fixed, &work, 1, 0..17).is_err());
1071        assert_eq!(
1072            translate_step_participant_upload_range(
1073                &DynamicResourceDemand::fixed(16).unwrap(),
1074                &work,
1075                1,
1076                0..4,
1077            )
1078            .unwrap(),
1079            0..4
1080        );
1081    }
1082
1083    #[test]
1084    fn sequence_execution_shape_uses_the_executed_source_frontier() {
1085        let committed = DynamicResourceShape::from_validated(1, 8, 3);
1086        let projected = sequence_execution_shape(committed, 4).unwrap();
1087
1088        assert_eq!(projected.sequences(), 1);
1089        assert_eq!(projected.tokens(), 4);
1090        assert_eq!(projected.pages(), 3);
1091        assert!(sequence_execution_shape(committed, 0).is_err());
1092        assert!(sequence_execution_shape(committed, 9).is_err());
1093        assert!(
1094            sequence_execution_shape(DynamicResourceShape::from_validated(2, 8, 3), 4).is_err()
1095        );
1096    }
1097
1098    #[test]
1099    fn contiguous_translation_applies_physical_base_offset() {
1100        let buffer = 9_u8;
1101        let regions = OperationBufferRegions {
1102            storage_kind: OperationBufferStorageKind::DynamicContiguous,
1103            logical_offset_bytes: 16,
1104            logical_length_bytes: 32,
1105            source: OperationRegionSource::Contiguous {
1106                buffer: &buffer,
1107                physical_base_offset_bytes: 4096,
1108                retention: DeviceBufferRetention::plan(Arc::new(())),
1109            },
1110        };
1111
1112        let translated = regions.iter().collect::<Vec<_>>();
1113        assert_eq!(translated.len(), 1);
1114        let (actual, physical, _retention) = translated[0].buffer_and_physical_range();
1115        assert_eq!(*actual, buffer);
1116        assert_eq!(translated[0].logical_offset_bytes(), 16);
1117        assert_eq!(physical, 4112..4144);
1118    }
1119
1120    #[test]
1121    fn physical_region_retains_opaque_owner_after_translation_source_drops() {
1122        struct DropOwner(Arc<std::sync::atomic::AtomicBool>);
1123
1124        impl Drop for DropOwner {
1125            fn drop(&mut self) {
1126                self.0.store(true, Ordering::Release);
1127            }
1128        }
1129
1130        let buffer = 9_u8;
1131        let dropped = Arc::new(std::sync::atomic::AtomicBool::new(false));
1132        let owner = Arc::new(DropOwner(Arc::clone(&dropped)));
1133        let regions = OperationBufferRegions {
1134            storage_kind: OperationBufferStorageKind::DynamicContiguous,
1135            logical_offset_bytes: 0,
1136            logical_length_bytes: 8,
1137            source: OperationRegionSource::Contiguous {
1138                buffer: &buffer,
1139                physical_base_offset_bytes: 64,
1140                retention: DeviceBufferRetention::plan(Arc::clone(&owner)),
1141            },
1142        };
1143        drop(owner);
1144
1145        let translated = regions.iter().collect::<Vec<_>>();
1146        drop(regions);
1147        assert!(!dropped.load(Ordering::Acquire));
1148        let (_, physical, retention) = translated[0].buffer_and_physical_range();
1149        assert_eq!(physical, 64..72);
1150        drop(translated);
1151        assert!(!dropped.load(Ordering::Acquire));
1152
1153        drop(retention);
1154        assert!(dropped.load(Ordering::Acquire));
1155    }
1156}