ferrum_interfaces/vnext/execution/
checkpoint_capacity.rs

1//! Optional plan growth permission for authenticated Sequence checkpoints.
2//!
3//! These bounds neither reserve memory nor enable capture. A runtime must also
4//! enforce the aggregate retained-byte limit and foreground admission priority
5//! before using the permission; every allocation still uses the device budget.
6
7use std::num::NonZeroU64;
8
9use super::{
10    invalid_plan, AllocationKind, AllocationLifetime, BTreeMap, BTreeSet, BufferUsage, Deserialize,
11    DynamicBackingPoolId, DynamicResourceDescriptor, MemoryPlan, PlanNode, ResourceId,
12    SequenceCheckpointLayout, Serialize, VNextError,
13};
14
15/// One aggregate cap for independently retained, allocator-aligned checkpoint
16/// extents, including in-flight and index-evicted but still pinned owners.
17/// Per-pool growth ceilings are alternatives within this cap, not additive
18/// reservations. This policy does not add request or sequence slots.
19#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
20#[serde(deny_unknown_fields)]
21pub struct CheckpointCapacityPolicy {
22    maximum_retained_bytes: NonZeroU64,
23}
24
25impl CheckpointCapacityPolicy {
26    pub fn new(maximum_retained_bytes: u64) -> Result<Self, VNextError> {
27        Ok(Self {
28            maximum_retained_bytes: NonZeroU64::new(maximum_retained_bytes)
29                .ok_or_else(|| invalid_plan("checkpoint capacity must be non-zero"))?,
30        })
31    }
32
33    pub const fn maximum_retained_bytes(&self) -> u64 {
34        self.maximum_retained_bytes.get()
35    }
36
37    fn pool_ceiling(&self, descriptor: &DynamicResourceDescriptor) -> u64 {
38        let quantum = descriptor.physical_allocation_quantum_bytes();
39        // Round down: a partial quantum cannot authorize a physical extent.
40        self.maximum_retained_bytes() / quantum * quantum
41    }
42}
43
44/// Layout authority is required here. A Sequence/State descriptor alone is
45/// insufficient to authorize checkpoint storage, including for token-scaled
46/// demands. Aliased states and multiple resources in one pool add no quota.
47pub(super) fn derive_checkpoint_growth_ceilings(
48    policy: Option<&CheckpointCapacityPolicy>,
49    layout: Option<&SequenceCheckpointLayout>,
50    descriptors: &[DynamicResourceDescriptor],
51) -> Result<BTreeMap<DynamicBackingPoolId, u64>, VNextError> {
52    let (Some(policy), Some(layout)) = (policy, layout) else {
53        return Ok(BTreeMap::new());
54    };
55    let descriptors = descriptors
56        .iter()
57        .map(|descriptor| (descriptor.base_resource_id(), descriptor))
58        .collect::<BTreeMap<_, _>>();
59    let resources = layout
60        .states()
61        .iter()
62        .map(|state| state.resource_id())
63        .collect::<BTreeSet<_>>();
64    let mut ceilings = BTreeMap::new();
65    for resource in resources {
66        let descriptor = descriptors
67            .get(resource)
68            .ok_or_else(|| invalid_plan("checkpoint state has no memory descriptor"))?;
69        if descriptor.lifetime() != AllocationLifetime::Sequence
70            || descriptor.usage() != BufferUsage::State
71            || *descriptor.kind() != AllocationKind::Value
72        {
73            return Err(invalid_plan(
74                "checkpoint growth requires certified Sequence state value storage",
75            ));
76        }
77        let ceiling = policy.pool_ceiling(descriptor);
78        if ceiling != 0 {
79            if ceilings
80                .insert(descriptor.pool_id().clone(), ceiling)
81                .is_some_and(|previous| previous != ceiling)
82            {
83                return Err(invalid_plan(
84                    "checkpoint resources disagree on their physical pool quantum",
85                ));
86            }
87        }
88    }
89    Ok(ceilings)
90}
91
92impl MemoryPlan {
93    /// Rebuild only physical pool bounds after the complete semantic layout
94    /// has been authenticated. Normal descriptor demands and minima are kept.
95    pub(super) fn with_checkpoint_capacity(
96        mut self,
97        policy: Option<CheckpointCapacityPolicy>,
98        layout: Option<&SequenceCheckpointLayout>,
99        nodes: &[PlanNode],
100        retained_completion_resources: &BTreeSet<ResourceId>,
101    ) -> Result<Self, VNextError> {
102        if policy.is_none() && self.checkpoint_capacity.is_none() {
103            return Ok(self);
104        }
105        let dynamic_capacity = self
106            .usable_capacity_bytes
107            .checked_sub(self.static_bytes)
108            .ok_or_else(|| invalid_plan("static memory exceeds usable capacity"))?;
109        let ceilings =
110            derive_checkpoint_growth_ceilings(policy.as_ref(), layout, &self.dynamic_descriptors)?;
111        let reusable = self
112            .reusable_execution
113            .as_ref()
114            .map(|plan| plan.pool_workspace_ceilings())
115            .transpose()?
116            .unwrap_or_default();
117        self.dynamic_pools = Self::derive_dynamic_pools_with_checkpoint(
118            &self.dynamic_descriptors,
119            nodes,
120            dynamic_capacity,
121            &reusable,
122            retained_completion_resources,
123            &ceilings,
124        )?;
125        self.checkpoint_capacity = policy;
126        self.validate()?;
127        Ok(self)
128    }
129}
130
131/// Wire-local bounds checking is deliberately weaker than authority. The
132/// execution plan additionally rebuilds the exact allowed pool set from its
133/// certified layout, and external wire data must survive semantic rebuilding.
134pub(super) fn validate_checkpoint_pool_ceiling(
135    policy: Option<&CheckpointCapacityPolicy>,
136    members: &[&DynamicResourceDescriptor],
137    ceiling: u64,
138) -> Result<(), VNextError> {
139    if ceiling == 0 {
140        return Ok(());
141    }
142    let policy =
143        policy.ok_or_else(|| invalid_plan("checkpoint pool growth has no capacity policy"))?;
144    if !members.iter().any(|descriptor| {
145        descriptor.lifetime() == AllocationLifetime::Sequence
146            && descriptor.usage() == BufferUsage::State
147            && *descriptor.kind() == AllocationKind::Value
148            && policy.pool_ceiling(descriptor) == ceiling
149    }) {
150        return Err(invalid_plan(
151            "checkpoint pool growth differs from its State allocation quantum or cap",
152        ));
153    }
154    Ok(())
155}
ferrum_interfaces/vnext/execution/checkpoint_capacity.rs

ferrum_interfaces/vnext/execution/
checkpoint_capacity.rs
