ferrum_interfaces/vnext/resource/checkpoint/
maintenance.rs

1//! Optional checkpoint pool growth, separate from capture submission and claims.
2
3use super::*;
4use crate::vnext::{
5    CheckpointCapacityPolicy, DeviceCapacityPressure, DynamicPoolGrowthBatchReceipt,
6    DynamicPoolResidentPressure, ExecutionPlan, SequenceCheckpointBytePlan,
7};
8
9/// One plan-authenticated attempt to make compact checkpoint storage available.
10/// This owner neither reserves a sequence slot nor authorizes a state copy.
11/// After maintenance the caller must retry capture and its ordinary claims.
12#[must_use = "checkpoint maintenance must be attempted or dropped"]
13pub struct CheckpointCapacityMaintenance<R: DeviceRuntime> {
14    binding: TrustedPlanRuntimeBinding<R>,
15    requests: CheckpointBackingRequests,
16    policy: Option<CheckpointCapacityPolicy>,
17}
18
19#[derive(Debug)]
20pub enum CheckpointCapacityMaintenanceSkipReason {
21    Retention(CheckpointRetentionSkipReason),
22    DeviceCapacity(DeviceCapacityPressure),
23    PoolResident(DynamicPoolResidentPressure),
24}
25
26#[derive(Debug)]
27pub enum CheckpointCapacityMaintenanceOutcome {
28    /// Space was already available or growth was published. A concurrent claim
29    /// may consume it, so this receipt is not an allocation guarantee.
30    Ready(DynamicPoolGrowthBatchReceipt),
31    Skipped(CheckpointCapacityMaintenanceSkipReason),
32}
33
34#[derive(Clone, Copy)]
35enum CheckpointMaintenanceMode {
36    AvailableCapacityOnly,
37    ReclaimIdleNonTargets,
38}
39
40impl<R: DeviceRuntime> PlanRuntimeResources<R> {
41    /// Explicitly permits one idle-chunk reclaim attempt after ordinary
42    /// checkpoint growth fails at the device budget. All capture target pools
43    /// remain protected, as do committed occupancy and resident minima. This
44    /// does not reserve capacity for future, unadmitted foreground work.
45    ///
46    /// The owner must belong to this exact resource root. No sequence is
47    /// evicted or waited on, and a Ready receipt still requires a fresh claim.
48    pub fn try_maintain_checkpoint_with_idle_reclaim(
49        self: &Arc<Self>,
50        maintenance: CheckpointCapacityMaintenance<R>,
51    ) -> Result<CheckpointCapacityMaintenanceOutcome, VNextError> {
52        if !Arc::ptr_eq(self, &maintenance.binding.resources) {
53            return Err(invalid_resource(
54                "checkpoint maintenance belongs to another resource root",
55            ));
56        }
57        maintenance.maintain_with_mode(CheckpointMaintenanceMode::ReclaimIdleNonTargets)
58    }
59}
60
61impl<R: DeviceRuntime> TrustedPlanRuntimeBinding<R> {
62    pub fn prepare_checkpoint_capacity_maintenance(
63        &self,
64        plan: &ExecutionPlan,
65        byte_plan: &SequenceCheckpointBytePlan,
66    ) -> Result<CheckpointCapacityMaintenance<R>, VNextError> {
67        let _lifecycle = self
68            .resources
69            .read_lifecycle("prepare checkpoint maintenance")?;
70        if plan.plan_hash() != self.plan_hash()
71            || byte_plan.plan_hash() != self.plan_hash()
72            || plan.checkpoint_byte_plan(byte_plan.boundary())? != *byte_plan
73        {
74            return Err(invalid_resource(
75                "checkpoint maintenance requires this plan's certified byte layout",
76            ));
77        }
78        let requests = byte_plan.backing_requests()?;
79        self.evaluate_checkpoint_backing(&requests)?;
80        Ok(CheckpointCapacityMaintenance {
81            binding: TrustedPlanRuntimeBinding {
82                resources: Arc::clone(&self.resources),
83            },
84            requests,
85            policy: plan.payload().memory().checkpoint_capacity().copied(),
86        })
87    }
88}
89
90impl<R: DeviceRuntime> CheckpointCapacityMaintenance<R> {
91    /// Recomputes free-space and contiguous packing requirements against the
92    /// actual pools. Only presently available device capacity may be used:
93    /// no cross-pool reclamation, waiting, or foreground admission mutation.
94    pub fn try_maintain(self) -> Result<CheckpointCapacityMaintenanceOutcome, VNextError> {
95        self.maintain_with_mode(CheckpointMaintenanceMode::AvailableCapacityOnly)
96    }
97
98    fn maintain_with_mode(
99        self,
100        mode: CheckpointMaintenanceMode,
101    ) -> Result<CheckpointCapacityMaintenanceOutcome, VNextError> {
102        let _lifecycle = self
103            .binding
104            .resources
105            .read_lifecycle("maintain checkpoint capacity")?;
106        let Some(policy) = self.policy else {
107            return Ok(CheckpointCapacityMaintenanceOutcome::Skipped(
108                CheckpointCapacityMaintenanceSkipReason::Retention(
109                    CheckpointRetentionSkipReason::Disabled,
110                ),
111            ));
112        };
113        let evaluated = self.binding.evaluate_checkpoint_backing(&self.requests)?;
114        let retained = self
115            .binding
116            .logical_admission()
117            .checkpoint_retained_bytes()?;
118        let maximum = policy.maximum_retained_bytes();
119        let remaining = maximum.checked_sub(retained).ok_or_else(|| {
120            invalid_resource("checkpoint retained bytes exceed the bound plan policy")
121        })?;
122        if evaluated.extent_bytes > remaining {
123            return Ok(CheckpointCapacityMaintenanceOutcome::Skipped(
124                CheckpointCapacityMaintenanceSkipReason::Retention(
125                    CheckpointRetentionSkipReason::Capacity {
126                        requested_bytes: evaluated.extent_bytes,
127                        retained_bytes: retained,
128                        maximum_bytes: maximum,
129                    },
130                ),
131            ));
132        }
133        let pools = self.binding.dynamic_pools();
134        let result = match mode {
135            CheckpointMaintenanceMode::AvailableCapacityOnly => {
136                pools.maintain_checkpoint_capacity(&evaluated.slices)
137            }
138            CheckpointMaintenanceMode::ReclaimIdleNonTargets => {
139                pools.maintain_checkpoint_capacity_with_idle_reclaim(&evaluated.slices)
140            }
141        };
142        match result {
143            Ok(receipt) => Ok(CheckpointCapacityMaintenanceOutcome::Ready(receipt)),
144            Err(VNextError::DeviceCapacityUnavailable(pressure)) => {
145                Ok(CheckpointCapacityMaintenanceOutcome::Skipped(
146                    CheckpointCapacityMaintenanceSkipReason::DeviceCapacity(pressure),
147                ))
148            }
149            Err(VNextError::DynamicPoolResidentUnavailable(pressure)) => {
150                Ok(CheckpointCapacityMaintenanceOutcome::Skipped(
151                    CheckpointCapacityMaintenanceSkipReason::PoolResident(pressure),
152                ))
153            }
154            Err(error) => Err(error),
155        }
156    }
157}
ferrum_interfaces/vnext/resource/checkpoint/maintenance.rs

ferrum_interfaces/vnext/resource/checkpoint/
maintenance.rs
