ferrum_interfaces/vnext/execution/
storage.rs

1use super::{
2    canonical_fingerprint, invalid_plan, is_canonical_sha256, quantize_storage_bytes,
3    validate_active_sequence_ceiling, AllocationKind, AllocationLifetime, BTreeSet,
4    BlockedTensorPadding, BufferUsage, ContractVersion, Deserialize, Deserializer,
5    DynamicResourceDemand, DynamicResourceShape, DynamicStorageProfile, ElementType, NodeId,
6    ResolvedTensorLayout, ResourceId, ResourceWorkShape, Serialize, StateInitialization,
7    VNextError,
8};
9
10#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize)]
11#[serde(transparent)]
12pub struct DynamicBackingPoolId(String);
13
14impl DynamicBackingPoolId {
15    pub(super) fn from_compatibility(key: &PoolCompatibilityKey) -> Result<Self, VNextError> {
16        Ok(Self(format!(
17            "dynamic-pool/sha256/{}",
18            canonical_fingerprint(key, "fingerprint dynamic pool compatibility")?
19        )))
20    }
21
22    pub(super) fn validate(&self) -> Result<(), VNextError> {
23        let Some(hash) = self.0.strip_prefix("dynamic-pool/sha256/") else {
24            return Err(invalid_plan(
25                "dynamic backing pool id has an invalid prefix",
26            ));
27        };
28        if !is_canonical_sha256(hash) {
29            return Err(invalid_plan("dynamic backing pool id has an invalid hash"));
30        }
31        Ok(())
32    }
33
34    pub fn as_str(&self) -> &str {
35        &self.0
36    }
37}
38
39impl<'de> Deserialize<'de> for DynamicBackingPoolId {
40    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
41    where
42        D: Deserializer<'de>,
43    {
44        let id = Self(String::deserialize(deserializer)?);
45        id.validate().map_err(serde::de::Error::custom)?;
46        Ok(id)
47    }
48}
49
50#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
51pub struct DynamicStorageContract {
52    pub(super) profile: DynamicStorageProfile,
53    pub(super) logical_layout_fingerprint: String,
54}
55
56impl DynamicStorageContract {
57    pub(super) fn new(
58        profile: DynamicStorageProfile,
59        logical_layout_fingerprint: String,
60    ) -> Result<Self, VNextError> {
61        if !is_canonical_sha256(&logical_layout_fingerprint) {
62            return Err(invalid_plan(
63                "dynamic storage logical layout fingerprint is invalid",
64            ));
65        }
66        Ok(Self {
67            profile,
68            logical_layout_fingerprint,
69        })
70    }
71
72    #[cfg(test)]
73    pub(crate) fn resource_test_contract(
74        profile: DynamicStorageProfile,
75        logical_layout_fingerprint: String,
76    ) -> Result<Self, VNextError> {
77        Self::new(profile, logical_layout_fingerprint)
78    }
79
80    pub const fn profile(&self) -> DynamicStorageProfile {
81        self.profile
82    }
83
84    pub fn logical_layout_fingerprint(&self) -> &str {
85        &self.logical_layout_fingerprint
86    }
87}
88
89#[derive(Deserialize)]
90#[serde(deny_unknown_fields)]
91pub(super) struct DynamicStorageContractWire {
92    pub(super) profile: DynamicStorageProfile,
93    pub(super) logical_layout_fingerprint: String,
94}
95
96impl<'de> Deserialize<'de> for DynamicStorageContract {
97    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
98    where
99        D: Deserializer<'de>,
100    {
101        let wire = DynamicStorageContractWire::deserialize(deserializer)?;
102        Self::new(wire.profile, wire.logical_layout_fingerprint).map_err(serde::de::Error::custom)
103    }
104}
105
106#[derive(Serialize)]
107#[serde(rename_all = "snake_case")]
108pub(super) enum TensorStorageLayoutClass<'a> {
109    Contiguous,
110    Strided {
111        byte_strides: &'a [u64],
112    },
113    Blocked {
114        block: &'a [u64],
115        axis_order: &'a [u32],
116        padding: BlockedStoragePaddingClass,
117    },
118}
119
120#[derive(Serialize)]
121#[serde(rename_all = "snake_case")]
122pub(super) enum BlockedStoragePaddingClass {
123    Exact,
124    ZeroFill,
125}
126
127#[derive(Serialize)]
128#[serde(rename_all = "snake_case")]
129pub(super) enum WorkspaceStorageLayoutClass {
130    OpaqueBytesV1,
131}
132
133pub(super) fn tensor_storage_layout_fingerprint(
134    layout: &ResolvedTensorLayout,
135) -> Result<String, VNextError> {
136    let class = match layout {
137        ResolvedTensorLayout::Contiguous => TensorStorageLayoutClass::Contiguous,
138        ResolvedTensorLayout::Strided { byte_strides } => {
139            TensorStorageLayoutClass::Strided { byte_strides }
140        }
141        ResolvedTensorLayout::Blocked {
142            block,
143            axis_order,
144            padding,
145        } => TensorStorageLayoutClass::Blocked {
146            block,
147            axis_order,
148            padding: match padding {
149                BlockedTensorPadding::Exact => BlockedStoragePaddingClass::Exact,
150                BlockedTensorPadding::ZeroFill { .. } => BlockedStoragePaddingClass::ZeroFill,
151            },
152        },
153    };
154    canonical_fingerprint(&class, "fingerprint tensor storage layout class")
155}
156
157pub(super) fn workspace_storage_layout_fingerprint() -> Result<String, VNextError> {
158    canonical_fingerprint(
159        &WorkspaceStorageLayoutClass::OpaqueBytesV1,
160        "fingerprint workspace storage layout class",
161    )
162}
163
164pub(super) fn static_contiguous_storage_profile() -> Result<DynamicStorageProfile, VNextError> {
165    DynamicStorageProfile::new(
166        super::DynamicStorageAllocator::LinearArena,
167        super::DynamicStorageView::Contiguous,
168    )
169}
170
171#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
172pub struct PoolCompatibilityKey {
173    pub(super) version: ContractVersion,
174    pub(super) profile: DynamicStorageProfile,
175    pub(super) usage: BufferUsage,
176    pub(super) element_type: ElementType,
177    pub(super) logical_layout_fingerprint: String,
178    pub(super) alignment_bytes: u64,
179}
180
181impl PoolCompatibilityKey {
182    pub(super) fn new(
183        storage: &DynamicStorageContract,
184        usage: BufferUsage,
185        element_type: ElementType,
186        alignment_bytes: u64,
187    ) -> Result<Self, VNextError> {
188        if alignment_bytes == 0 || !alignment_bytes.is_power_of_two() {
189            return Err(invalid_plan(
190                "dynamic pool compatibility alignment is invalid",
191            ));
192        }
193        let key = Self {
194            version: ContractVersion::new(1, 0),
195            profile: storage.profile,
196            usage,
197            element_type,
198            logical_layout_fingerprint: storage.logical_layout_fingerprint.clone(),
199            alignment_bytes,
200        };
201        key.validate()?;
202        Ok(key)
203    }
204
205    pub(super) fn validate(&self) -> Result<(), VNextError> {
206        if self.version != ContractVersion::new(1, 0)
207            || !is_canonical_sha256(&self.logical_layout_fingerprint)
208            || self.alignment_bytes == 0
209            || !self.alignment_bytes.is_power_of_two()
210        {
211            return Err(invalid_plan("dynamic pool compatibility key is invalid"));
212        }
213        Ok(())
214    }
215
216    pub const fn profile(&self) -> DynamicStorageProfile {
217        self.profile
218    }
219
220    pub const fn usage(&self) -> BufferUsage {
221        self.usage
222    }
223
224    pub const fn element_type(&self) -> ElementType {
225        self.element_type
226    }
227
228    pub fn logical_layout_fingerprint(&self) -> &str {
229        &self.logical_layout_fingerprint
230    }
231
232    pub const fn alignment_bytes(&self) -> u64 {
233        self.alignment_bytes
234    }
235}
236
237#[derive(Deserialize)]
238#[serde(deny_unknown_fields)]
239pub(super) struct PoolCompatibilityKeyWire {
240    pub(super) version: ContractVersion,
241    pub(super) profile: DynamicStorageProfile,
242    pub(super) usage: BufferUsage,
243    pub(super) element_type: ElementType,
244    pub(super) logical_layout_fingerprint: String,
245    pub(super) alignment_bytes: u64,
246}
247
248impl<'de> Deserialize<'de> for PoolCompatibilityKey {
249    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
250    where
251        D: Deserializer<'de>,
252    {
253        let wire = PoolCompatibilityKeyWire::deserialize(deserializer)?;
254        let key = Self {
255            version: wire.version,
256            profile: wire.profile,
257            usage: wire.usage,
258            element_type: wire.element_type,
259            logical_layout_fingerprint: wire.logical_layout_fingerprint,
260            alignment_bytes: wire.alignment_bytes,
261        };
262        key.validate().map_err(serde::de::Error::custom)?;
263        Ok(key)
264    }
265}
266
267#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
268#[serde(rename_all = "snake_case")]
269pub enum DynamicPoolProvisioningMode {
270    DemandDrivenElastic,
271}
272
273/// Typed bounds for elastic residency. `minimum_resident_bytes` is the amount
274/// required to make one request runnable, not an initial reservation. Pools
275/// may grow on demand up to `maximum_resident_bytes`; the process-wide device
276/// account remains the authority when several pools compete for that memory.
277#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
278#[serde(deny_unknown_fields)]
279pub struct DynamicPoolProvisioningPolicy {
280    pub(super) mode: DynamicPoolProvisioningMode,
281    pub(super) minimum_resident_bytes: u64,
282    pub(super) maximum_resident_bytes: u64,
283}
284
285impl DynamicPoolProvisioningPolicy {
286    pub(super) fn demand_driven(
287        minimum_resident_bytes: u64,
288        maximum_resident_bytes: u64,
289    ) -> Result<Self, VNextError> {
290        let policy = Self {
291            mode: DynamicPoolProvisioningMode::DemandDrivenElastic,
292            minimum_resident_bytes,
293            maximum_resident_bytes,
294        };
295        policy.validate()?;
296        Ok(policy)
297    }
298
299    pub(super) fn validate(&self) -> Result<(), VNextError> {
300        if self.minimum_resident_bytes == 0
301            || self.maximum_resident_bytes < self.minimum_resident_bytes
302        {
303            return Err(invalid_plan("dynamic pool provisioning bounds are invalid"));
304        }
305        Ok(())
306    }
307
308    pub const fn mode(&self) -> DynamicPoolProvisioningMode {
309        self.mode
310    }
311
312    pub const fn minimum_resident_bytes(&self) -> u64 {
313        self.minimum_resident_bytes
314    }
315
316    pub const fn maximum_resident_bytes(&self) -> u64 {
317        self.maximum_resident_bytes
318    }
319}
320
321/// One self-contained physical-compatibility class for demand-driven backing.
322/// Membership, runnable minima, completion-order reuse evidence, and elastic
323/// bounds are canonical plan data, so a runtime does not have to rediscover
324/// pool structure by scanning unrelated descriptors.
325#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
326pub struct DynamicBackingPoolSpec {
327    pub(super) pool_id: DynamicBackingPoolId,
328    pub(super) compatibility: PoolCompatibilityKey,
329    pub(super) resource_ids: Vec<ResourceId>,
330    pub(super) minimum_request_bytes: u64,
331    pub(super) minimum_sequence_bytes: u64,
332    pub(super) minimum_step_bytes: u64,
333    pub(super) minimum_invocation_peak_bytes: u64,
334    pub(super) step_resource_slots: Vec<StepResourceSlot>,
335    pub(super) theoretical_ceiling_bytes: CanonicalU128,
336    pub(super) reusable_workspace_ceiling_bytes: u64,
337    pub(super) provisioning: DynamicPoolProvisioningPolicy,
338    pub(super) invocation_liveness_mode: InvocationLivenessMode,
339    pub(super) invocation_liveness: Vec<InvocationResourceLiveness>,
340}
341
342impl DynamicBackingPoolSpec {
343    #[allow(clippy::too_many_arguments)]
344    pub(super) fn from_core(
345        compatibility: PoolCompatibilityKey,
346        resource_ids: Vec<ResourceId>,
347        minimum_request_bytes: u64,
348        minimum_sequence_bytes: u64,
349        minimum_step_bytes: u64,
350        minimum_invocation_peak_bytes: u64,
351        step_resource_slots: Vec<StepResourceSlot>,
352        theoretical_ceiling_bytes: u128,
353        reusable_workspace_ceiling_bytes: u64,
354        dynamic_capacity_bytes: u64,
355        invocation_liveness_mode: InvocationLivenessMode,
356        invocation_liveness: Vec<InvocationResourceLiveness>,
357    ) -> Result<Self, VNextError> {
358        compatibility.validate()?;
359        let pool_id = DynamicBackingPoolId::from_compatibility(&compatibility)?;
360        let minimum_resident_bytes = minimum_request_bytes
361            .checked_add(minimum_sequence_bytes)
362            .and_then(|bytes| bytes.checked_add(minimum_step_bytes))
363            .and_then(|bytes| bytes.checked_add(minimum_invocation_peak_bytes))
364            .ok_or_else(|| invalid_plan("dynamic pool runnable minimum overflows u64"))?;
365        let combined_ceiling_bytes = theoretical_ceiling_bytes
366            .checked_add(u128::from(reusable_workspace_ceiling_bytes))
367            .ok_or_else(|| invalid_plan("dynamic pool combined ceiling overflows u128"))?;
368        let maximum_resident_bytes =
369            u64::try_from(combined_ceiling_bytes.min(u128::from(dynamic_capacity_bytes)))
370                .map_err(|_| invalid_plan("dynamic pool resident ceiling exceeds u64"))?;
371        let spec = Self {
372            pool_id,
373            compatibility,
374            resource_ids,
375            minimum_request_bytes,
376            minimum_sequence_bytes,
377            minimum_step_bytes,
378            minimum_invocation_peak_bytes,
379            step_resource_slots,
380            theoretical_ceiling_bytes: CanonicalU128::new(theoretical_ceiling_bytes),
381            reusable_workspace_ceiling_bytes,
382            provisioning: DynamicPoolProvisioningPolicy::demand_driven(
383                minimum_resident_bytes,
384                maximum_resident_bytes,
385            )?,
386            invocation_liveness_mode,
387            invocation_liveness,
388        };
389        spec.validate_local()?;
390        Ok(spec)
391    }
392
393    pub(super) fn validate_local(&self) -> Result<(), VNextError> {
394        self.pool_id.validate()?;
395        self.compatibility.validate()?;
396        self.provisioning.validate()?;
397        for slot in &self.step_resource_slots {
398            slot.validate()?;
399        }
400        let minimum_resident_bytes = self
401            .minimum_request_bytes
402            .checked_add(self.minimum_sequence_bytes)
403            .and_then(|bytes| bytes.checked_add(self.minimum_step_bytes))
404            .and_then(|bytes| bytes.checked_add(self.minimum_invocation_peak_bytes))
405            .ok_or_else(|| invalid_plan("dynamic pool runnable minimum overflows u64"))?;
406        if self.pool_id != DynamicBackingPoolId::from_compatibility(&self.compatibility)?
407            || self.resource_ids.is_empty()
408            || self.resource_ids.windows(2).any(|pair| pair[0] >= pair[1])
409            || minimum_resident_bytes != self.provisioning.minimum_resident_bytes
410            || u128::from(self.provisioning.maximum_resident_bytes)
411                > self
412                    .theoretical_ceiling_bytes
413                    .get()
414                    .checked_add(u128::from(self.reusable_workspace_ceiling_bytes))
415                    .ok_or_else(|| invalid_plan("dynamic pool combined ceiling overflows u128"))?
416            || self
417                .step_resource_slots
418                .windows(2)
419                .any(|pair| pair[0].resource_ids >= pair[1].resource_ids)
420            || self
421                .step_resource_slots
422                .iter()
423                .flat_map(|slot| slot.resource_ids.iter())
424                .collect::<BTreeSet<_>>()
425                .len()
426                != self
427                    .step_resource_slots
428                    .iter()
429                    .map(|slot| slot.resource_ids.len())
430                    .sum::<usize>()
431        {
432            return Err(invalid_plan(
433                "dynamic backing pool identity, membership, or bounds are invalid",
434            ));
435        }
436        match self.invocation_liveness_mode {
437            InvocationLivenessMode::NoInvocationResources => {
438                if self.minimum_invocation_peak_bytes != 0 || !self.invocation_liveness.is_empty() {
439                    return Err(invalid_plan(
440                        "non-invocation pool carries invocation liveness evidence",
441                    ));
442                }
443            }
444            InvocationLivenessMode::TotalOrderReuse
445            | InvocationLivenessMode::ConservativeConcurrent => {
446                if self.minimum_invocation_peak_bytes == 0
447                    || self.invocation_liveness.is_empty()
448                    || self
449                        .invocation_liveness
450                        .windows(2)
451                        .any(|pair| pair[0].node_id >= pair[1].node_id)
452                {
453                    return Err(invalid_plan(
454                        "invocation pool liveness evidence is empty or non-canonical",
455                    ));
456                }
457            }
458        }
459        Ok(())
460    }
461
462    pub fn pool_id(&self) -> &DynamicBackingPoolId {
463        &self.pool_id
464    }
465
466    pub fn compatibility(&self) -> &PoolCompatibilityKey {
467        &self.compatibility
468    }
469
470    pub fn resource_ids(&self) -> &[ResourceId] {
471        &self.resource_ids
472    }
473
474    pub const fn minimum_request_bytes(&self) -> u64 {
475        self.minimum_request_bytes
476    }
477
478    pub const fn minimum_sequence_bytes(&self) -> u64 {
479        self.minimum_sequence_bytes
480    }
481
482    pub const fn minimum_step_bytes(&self) -> u64 {
483        self.minimum_step_bytes
484    }
485
486    pub const fn minimum_invocation_peak_bytes(&self) -> u64 {
487        self.minimum_invocation_peak_bytes
488    }
489
490    pub fn step_resource_slots(&self) -> &[StepResourceSlot] {
491        &self.step_resource_slots
492    }
493
494    pub fn theoretical_ceiling_bytes(&self) -> u128 {
495        self.theoretical_ceiling_bytes.get()
496    }
497
498    pub const fn reusable_workspace_ceiling_bytes(&self) -> u64 {
499        self.reusable_workspace_ceiling_bytes
500    }
501
502    pub fn provisioning(&self) -> &DynamicPoolProvisioningPolicy {
503        &self.provisioning
504    }
505
506    pub const fn invocation_liveness_mode(&self) -> InvocationLivenessMode {
507        self.invocation_liveness_mode
508    }
509
510    pub fn invocation_liveness(&self) -> &[InvocationResourceLiveness] {
511        &self.invocation_liveness
512    }
513}
514
515#[derive(Deserialize)]
516#[serde(deny_unknown_fields)]
517pub(super) struct DynamicBackingPoolSpecWire {
518    pub(super) pool_id: DynamicBackingPoolId,
519    pub(super) compatibility: PoolCompatibilityKey,
520    pub(super) resource_ids: Vec<ResourceId>,
521    pub(super) minimum_request_bytes: u64,
522    pub(super) minimum_sequence_bytes: u64,
523    pub(super) minimum_step_bytes: u64,
524    pub(super) minimum_invocation_peak_bytes: u64,
525    pub(super) step_resource_slots: Vec<StepResourceSlot>,
526    pub(super) theoretical_ceiling_bytes: CanonicalU128,
527    pub(super) reusable_workspace_ceiling_bytes: u64,
528    pub(super) provisioning: DynamicPoolProvisioningPolicy,
529    pub(super) invocation_liveness_mode: InvocationLivenessMode,
530    pub(super) invocation_liveness: Vec<InvocationResourceLiveness>,
531}
532
533impl<'de> Deserialize<'de> for DynamicBackingPoolSpec {
534    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
535    where
536        D: Deserializer<'de>,
537    {
538        let wire = DynamicBackingPoolSpecWire::deserialize(deserializer)?;
539        let spec = Self {
540            pool_id: wire.pool_id,
541            compatibility: wire.compatibility,
542            resource_ids: wire.resource_ids,
543            minimum_request_bytes: wire.minimum_request_bytes,
544            minimum_sequence_bytes: wire.minimum_sequence_bytes,
545            minimum_step_bytes: wire.minimum_step_bytes,
546            minimum_invocation_peak_bytes: wire.minimum_invocation_peak_bytes,
547            step_resource_slots: wire.step_resource_slots,
548            theoretical_ceiling_bytes: wire.theoretical_ceiling_bytes,
549            reusable_workspace_ceiling_bytes: wire.reusable_workspace_ceiling_bytes,
550            provisioning: wire.provisioning,
551            invocation_liveness_mode: wire.invocation_liveness_mode,
552            invocation_liveness: wire.invocation_liveness,
553        };
554        spec.validate_local().map_err(serde::de::Error::custom)?;
555        Ok(spec)
556    }
557}
558
559#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
560pub struct DynamicResourceDescriptor {
561    pub(super) base_resource_id: ResourceId,
562    pub(super) demand: DynamicResourceDemand,
563    pub(super) alignment_bytes: u64,
564    pub(super) usage: BufferUsage,
565    pub(super) element_type: ElementType,
566    pub(super) lifetime: AllocationLifetime,
567    pub(super) kind: AllocationKind,
568    pub(super) storage: DynamicStorageContract,
569    pub(super) pool_id: DynamicBackingPoolId,
570    pub(super) initialization: StateInitialization,
571    /// Protocol-only ceiling used for checked evidence. No API may iterate,
572    /// reserve, allocate, or claim this many instances.
573    pub(super) theoretical_maximum_instances: u32,
574}
575
576impl DynamicResourceDescriptor {
577    #[allow(clippy::too_many_arguments)]
578    pub(super) fn new(
579        base_resource_id: ResourceId,
580        demand: DynamicResourceDemand,
581        alignment_bytes: u64,
582        usage: BufferUsage,
583        element_type: ElementType,
584        lifetime: AllocationLifetime,
585        kind: AllocationKind,
586        storage: DynamicStorageContract,
587        initialization: StateInitialization,
588        theoretical_maximum_instances: u32,
589    ) -> Result<Self, VNextError> {
590        validate_active_sequence_ceiling(theoretical_maximum_instances)?;
591        if alignment_bytes == 0
592            || !alignment_bytes.is_power_of_two()
593            || lifetime == AllocationLifetime::Plan
594        {
595            return Err(invalid_plan(
596                "dynamic resource descriptor has invalid alignment or static lifetime",
597            ));
598        }
599        let kind_valid = match &kind {
600            AllocationKind::InitializationScratch => false,
601            AllocationKind::Scratch { .. } => {
602                lifetime == AllocationLifetime::Invocation
603                    && usage == BufferUsage::Scratch
604                    && element_type == ElementType::U8
605            }
606            AllocationKind::Binding { .. } => {
607                lifetime == AllocationLifetime::Invocation
608                    && usage == BufferUsage::Binding
609                    && element_type == ElementType::U8
610            }
611            AllocationKind::Persistent { .. } => {
612                matches!(
613                    lifetime,
614                    AllocationLifetime::Request
615                        | AllocationLifetime::Sequence
616                        | AllocationLifetime::Step
617                ) && usage == BufferUsage::Persistent
618                    && element_type == ElementType::U8
619            }
620            AllocationKind::Value => usage != BufferUsage::Weights,
621        };
622        if !kind_valid {
623            return Err(invalid_plan(
624                "dynamic resource kind, lifetime, usage, or element type is inconsistent",
625            ));
626        }
627        if initialization == StateInitialization::Zero
628            && (kind != AllocationKind::Value
629                || usage != BufferUsage::State
630                || lifetime != AllocationLifetime::Sequence)
631        {
632            return Err(invalid_plan(
633                "zero initialization requires semantic Sequence state backing",
634            ));
635        }
636        demand.validate()?;
637        if matches!(
638            &demand,
639            DynamicResourceDemand::ActualSequences {
640                maximum_sequences,
641                ..
642            } if *maximum_sequences != theoretical_maximum_instances
643        ) {
644            return Err(invalid_plan(
645                "actual-sequence demand and descriptor instance ceilings differ",
646            ));
647        }
648        let pool_id = DynamicBackingPoolId::from_compatibility(&PoolCompatibilityKey::new(
649            &storage,
650            usage,
651            element_type,
652            alignment_bytes,
653        )?)?;
654        let descriptor = Self {
655            base_resource_id,
656            demand,
657            alignment_bytes,
658            usage,
659            element_type,
660            lifetime,
661            kind,
662            storage,
663            pool_id,
664            initialization,
665            theoretical_maximum_instances,
666        };
667        descriptor.evaluate_request_bytes_for_shape(descriptor.demand.minimum_shape())?;
668        descriptor
669            .evaluate_request_bytes_for_shape(descriptor.demand.theoretical_maximum_shape())?;
670        Ok(descriptor)
671    }
672
673    #[cfg(test)]
674    pub(crate) fn resource_test_binding(
675        base_resource_id: ResourceId,
676        demand: DynamicResourceDemand,
677        alignment_bytes: u64,
678        node_id: NodeId,
679        storage: DynamicStorageContract,
680        theoretical_maximum_instances: u32,
681    ) -> Result<Self, VNextError> {
682        Self::new(
683            base_resource_id,
684            demand,
685            alignment_bytes,
686            BufferUsage::Binding,
687            ElementType::U8,
688            AllocationLifetime::Invocation,
689            AllocationKind::Binding { node_id },
690            storage,
691            StateInitialization::None,
692            theoretical_maximum_instances,
693        )
694    }
695
696    pub fn base_resource_id(&self) -> &ResourceId {
697        &self.base_resource_id
698    }
699
700    pub fn demand(&self) -> &DynamicResourceDemand {
701        &self.demand
702    }
703
704    pub const fn theoretical_maximum_instances(&self) -> u32 {
705        self.theoretical_maximum_instances
706    }
707
708    pub fn evaluate_logical_request_bytes(
709        &self,
710        work: &ResourceWorkShape,
711    ) -> Result<u64, VNextError> {
712        self.demand.evaluate_bytes(work)
713    }
714
715    pub(crate) fn evaluate_logical_request_bytes_for_shape(
716        &self,
717        shape: DynamicResourceShape,
718    ) -> Result<u64, VNextError> {
719        self.demand.evaluate_shape_bytes(shape)
720    }
721
722    pub fn physical_allocation_quantum_bytes(&self) -> u64 {
723        match self.storage.profile().allocator() {
724            super::DynamicStorageAllocator::LinearArena => self.alignment_bytes,
725            super::DynamicStorageAllocator::FixedBlockArena { block_bytes } => {
726                block_bytes.max(self.alignment_bytes)
727            }
728        }
729    }
730
731    /// Exact physical claim for one logical shape. The semantic demand stays
732    /// unchanged in the plan; allocator geometry is applied only at this
733    /// boundary so admission and backing cannot under-count fixed blocks.
734    pub fn evaluate_request_bytes(&self, work: &ResourceWorkShape) -> Result<u64, VNextError> {
735        self.evaluate_request_bytes_for_shape(work.immediate_shape())
736    }
737
738    pub fn evaluate_fit_request_bytes(&self, work: &ResourceWorkShape) -> Result<u64, VNextError> {
739        self.evaluate_request_bytes_for_shape(work.fit_shape())
740    }
741
742    pub(crate) fn evaluate_request_bytes_for_shape(
743        &self,
744        shape: DynamicResourceShape,
745    ) -> Result<u64, VNextError> {
746        quantize_storage_bytes(
747            self.evaluate_logical_request_bytes_for_shape(shape)?,
748            self.alignment_bytes,
749            self.storage.profile(),
750        )
751    }
752
753    pub fn minimum_request_bytes(&self) -> Result<u64, VNextError> {
754        self.evaluate_request_bytes_for_shape(self.demand.minimum_shape())
755    }
756
757    pub fn theoretical_maximum_request_bytes(&self) -> Result<u64, VNextError> {
758        self.evaluate_request_bytes_for_shape(self.demand.theoretical_maximum_shape())
759    }
760
761    /// Conservative maximum physical residency for this resource across all
762    /// live instances. Sequence-shaped claims share the plan's global active
763    /// sequence ceiling, so multiplying a full-batch claim by that ceiling
764    /// would count the same participants twice. Splitting every participant
765    /// into its own claim is the physical-padding worst case.
766    pub(super) fn theoretical_maximum_resident_bytes(&self) -> Result<u128, VNextError> {
767        let per_instance_bytes = match self.demand {
768            DynamicResourceDemand::ActualSequences { .. } => self.minimum_request_bytes()?,
769            _ => self.theoretical_maximum_request_bytes()?,
770        };
771        Ok(u128::from(per_instance_bytes) * u128::from(self.theoretical_maximum_instances))
772    }
773
774    pub const fn alignment_bytes(&self) -> u64 {
775        self.alignment_bytes
776    }
777
778    pub const fn usage(&self) -> BufferUsage {
779        self.usage
780    }
781
782    pub const fn element_type(&self) -> ElementType {
783        self.element_type
784    }
785
786    pub const fn lifetime(&self) -> AllocationLifetime {
787        self.lifetime
788    }
789
790    pub fn kind(&self) -> &AllocationKind {
791        &self.kind
792    }
793
794    pub fn storage(&self) -> &DynamicStorageContract {
795        &self.storage
796    }
797
798    pub fn pool_id(&self) -> &DynamicBackingPoolId {
799        &self.pool_id
800    }
801
802    pub const fn initialization(&self) -> StateInitialization {
803        self.initialization
804    }
805}
806
807#[derive(Deserialize)]
808#[serde(deny_unknown_fields)]
809pub(super) struct DynamicResourceDescriptorWire {
810    pub(super) base_resource_id: ResourceId,
811    pub(super) demand: DynamicResourceDemand,
812    pub(super) alignment_bytes: u64,
813    pub(super) usage: BufferUsage,
814    pub(super) element_type: ElementType,
815    pub(super) lifetime: AllocationLifetime,
816    pub(super) kind: AllocationKind,
817    pub(super) storage: DynamicStorageContract,
818    pub(super) pool_id: DynamicBackingPoolId,
819    pub(super) initialization: StateInitialization,
820    pub(super) theoretical_maximum_instances: u32,
821}
822
823impl<'de> Deserialize<'de> for DynamicResourceDescriptor {
824    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
825    where
826        D: Deserializer<'de>,
827    {
828        let wire = DynamicResourceDescriptorWire::deserialize(deserializer)?;
829        let descriptor = Self::new(
830            wire.base_resource_id,
831            wire.demand,
832            wire.alignment_bytes,
833            wire.usage,
834            wire.element_type,
835            wire.lifetime,
836            wire.kind,
837            wire.storage,
838            wire.initialization,
839            wire.theoretical_maximum_instances,
840        )
841        .map_err(serde::de::Error::custom)?;
842        if descriptor.pool_id != wire.pool_id {
843            return Err(serde::de::Error::custom(
844                "dynamic resource pool id is not core-derived from compatibility",
845            ));
846        }
847        Ok(descriptor)
848    }
849}
850
851#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
852#[serde(transparent)]
853pub(super) struct CanonicalU128(String);
854
855impl CanonicalU128 {
856    pub(super) fn new(value: u128) -> Self {
857        Self(value.to_string())
858    }
859
860    pub(super) fn get(&self) -> u128 {
861        self.0
862            .parse()
863            .expect("canonical u128 is validated at construction or deserialization")
864    }
865}
866
867impl<'de> Deserialize<'de> for CanonicalU128 {
868    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
869    where
870        D: Deserializer<'de>,
871    {
872        let value = String::deserialize(deserializer)?;
873        let parsed = value.parse::<u128>().map_err(serde::de::Error::custom)?;
874        if parsed.to_string() != value {
875            return Err(serde::de::Error::custom(
876                "u128 evidence must be a canonical unsigned decimal string",
877            ));
878        }
879        Ok(Self(value))
880    }
881}
882
883#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
884#[serde(rename_all = "snake_case")]
885pub enum InvocationLivenessMode {
886    NoInvocationResources,
887    /// Every invocation row in this pool is ordered by a transitive node
888    /// completion dependency, so one runnable request needs only the maximum
889    /// row size rather than their sum.
890    TotalOrderReuse,
891    /// The plan cannot prove that every invocation row completes before the
892    /// next one starts. The runnable minimum therefore sums member resources.
893    ConservativeConcurrent,
894}
895
896/// A set of Step-scoped logical resources that may project onto one physical
897/// extent. Multi-resource slots are emitted only when plan dependencies prove
898/// that every member's final user completes before the next member starts.
899#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
900#[serde(deny_unknown_fields)]
901pub struct StepResourceSlot {
902    pub(super) kind: StepResourceSlotKind,
903    pub(super) resource_ids: Vec<ResourceId>,
904}
905
906impl StepResourceSlot {
907    pub(super) fn dedicated(resource_id: ResourceId) -> Self {
908        Self {
909            kind: StepResourceSlotKind::Dedicated,
910            resource_ids: vec![resource_id],
911        }
912    }
913
914    pub(super) fn ordered_single_fence_wave(
915        mut resource_ids: Vec<ResourceId>,
916    ) -> Result<Self, VNextError> {
917        resource_ids.sort();
918        if resource_ids.len() < 2 || resource_ids.windows(2).any(|pair| pair[0] == pair[1]) {
919            return Err(invalid_plan(
920                "ordered single-fence step slot requires at least two unique resources",
921            ));
922        }
923        Ok(Self {
924            kind: StepResourceSlotKind::OrderedSingleFenceStepWave,
925            resource_ids,
926        })
927    }
928
929    pub(super) fn validate(&self) -> Result<(), VNextError> {
930        if self.resource_ids.is_empty()
931            || self.resource_ids.windows(2).any(|pair| pair[0] >= pair[1])
932            || match self.kind {
933                StepResourceSlotKind::Dedicated => self.resource_ids.len() != 1,
934                StepResourceSlotKind::OrderedSingleFenceStepWave => self.resource_ids.len() < 2,
935            }
936        {
937            return Err(invalid_plan(
938                "step resource slot kind or members are invalid",
939            ));
940        }
941        Ok(())
942    }
943
944    pub const fn kind(&self) -> StepResourceSlotKind {
945        self.kind
946    }
947
948    pub fn resource_ids(&self) -> &[ResourceId] {
949        &self.resource_ids
950    }
951}
952
953#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
954#[serde(rename_all = "snake_case")]
955pub enum StepResourceSlotKind {
956    Dedicated,
957    /// Every member is an internal activation and the runtime must submit the
958    /// canonical plan order as one ordered command batch with one terminal
959    /// fence before it may consume this reuse proof.
960    OrderedSingleFenceStepWave,
961}
962
963#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
964#[serde(deny_unknown_fields)]
965pub struct InvocationResourceLiveness {
966    pub(super) node_id: NodeId,
967    pub(super) resource_ids: Vec<ResourceId>,
968}
969
970impl InvocationResourceLiveness {
971    pub fn node_id(&self) -> &NodeId {
972        &self.node_id
973    }
974
975    pub fn resource_ids(&self) -> &[ResourceId] {
976        &self.resource_ids
977    }
978}
ferrum_interfaces/vnext/execution/storage.rs

ferrum_interfaces/vnext/execution/
storage.rs
