ferrum_interfaces/vnext/completion/checkpoint_access/
entry.rs

1use super::super::{CheckpointTimingPhase, StateTransferKind};
2use super::*;
3use crate::vnext::{
4    CheckpointBackingAllocationDecision, CheckpointPartitionNumerics, DeviceTimingMode,
5    ExecutionPlan, SequenceCheckpointCapability, SequenceCheckpointLayout,
6    SequenceStateTransferKind, SequenceStateTransferPreparation, TrustedPlanRuntimeBinding,
7};
8
9fn access_layout(
10    plan: &ExecutionPlan,
11) -> Result<&SequenceCheckpointLayout, CheckpointAccessSkipReason> {
12    if plan.payload().memory().checkpoint_capacity().is_none() {
13        return Err(CheckpointAccessSkipReason::Disabled);
14    }
15    let SequenceCheckpointCapability::Enabled(layout) = plan.sequence_checkpoint_capability()
16    else {
17        return Err(CheckpointAccessSkipReason::Unsupported);
18    };
19    if !layout.inputs().conditioning_inputs().is_empty() {
20        return Err(CheckpointAccessSkipReason::MissingConditioningEvidence);
21    }
22    if layout.providers().iter().any(|provider| {
23        provider.contract().partition_numerics() == CheckpointPartitionNumerics::SamePartitionOnly
24    }) {
25        return Err(CheckpointAccessSkipReason::MissingPartitionEvidence);
26    }
27    Ok(layout)
28}
29
30fn wrap_submission<R: DeviceRuntime>(
31    submission: Result<StateTransferSubmission<R>, VNextError>,
32    restore: Option<RestoreInput<R>>,
33) -> NativeCheckpointStart<R> {
34    match submission {
35        Ok(StateTransferSubmission::Submitted(handle)) => {
36            NativeCheckpointStart::Submitted(NativeCheckpointTransfer { handle, restore })
37        }
38        Ok(StateTransferSubmission::Indeterminate(handle)) => {
39            NativeCheckpointStart::Indeterminate(NativeCheckpointTransfer { handle, restore })
40        }
41        Ok(StateTransferSubmission::ContractAfterSubmission { error, handle }) => {
42            NativeCheckpointStart::ContractAfterSubmission {
43                error,
44                transfer: NativeCheckpointTransfer { handle, restore },
45            }
46        }
47        Err(error) => NativeCheckpointStart::NotSubmitted(error),
48    }
49}
50
51impl<R: DeviceRuntime> CompletionReaper<R> {
52    /// Capture only the source's current proven complete boundary. Optional
53    /// retention never grows backing or waits for foreground capacity here.
54    pub fn try_capture_sequence_checkpoint(
55        self: &Arc<Self>,
56        plan: &ExecutionPlan,
57        binding: &TrustedPlanRuntimeBinding<R>,
58        source: Arc<SequenceSession<R>>,
59        lane: Arc<ExecutionLane<R>>,
60    ) -> Result<NativeCheckpointStart<R>, VNextError> {
61        self.try_capture_sequence_checkpoint_with_timing(
62            plan,
63            binding,
64            source,
65            lane,
66            DeviceTimingMode::Off,
67        )
68    }
69
70    /// Same capture authority and lifecycle, with optional terminal device
71    /// elapsed time. Enabled modes use Completion only: no kernel counters or
72    /// changes to copy encoder boundaries are introduced.
73    pub fn try_capture_sequence_checkpoint_with_timing(
74        self: &Arc<Self>,
75        plan: &ExecutionPlan,
76        binding: &TrustedPlanRuntimeBinding<R>,
77        source: Arc<SequenceSession<R>>,
78        lane: Arc<ExecutionLane<R>>,
79        timing_mode: DeviceTimingMode,
80    ) -> Result<NativeCheckpointStart<R>, VNextError> {
81        let preparation = self.checkpoint_timings.start(
82            StateTransferKind::Capture,
83            CheckpointTimingPhase::PrepareClaim,
84        );
85        let layout = match access_layout(plan) {
86            Ok(layout) => layout,
87            Err(reason) => return Ok(NativeCheckpointStart::Skipped(reason)),
88        };
89        if plan.plan_hash() != binding.plan_hash()
90            || plan.plan_hash() != source.resources().plan_evidence().plan_hash()
91            || binding.coordinator_id() != source.resources().coordinator_id()
92        {
93            return Err(invalid_completion(
94                "checkpoint capture names another admitted plan",
95            ));
96        }
97        let generation = source.resources().backing_generation()?;
98        let guard = match source
99            .try_prepare_state_transfer(SequenceStateTransferKind::CaptureRead, generation)?
100        {
101            SequenceStateTransferPreparation::Prepared(guard) => guard,
102            SequenceStateTransferPreparation::Busy => {
103                return Ok(NativeCheckpointStart::Skipped(
104                    CheckpointAccessSkipReason::Busy,
105                ));
106            }
107            SequenceStateTransferPreparation::StaleBacking => {
108                return Ok(NativeCheckpointStart::Skipped(
109                    CheckpointAccessSkipReason::StaleBacking,
110                ));
111            }
112        };
113        let boundary = guard.completed_boundary()?;
114        let boundary_tokens = u64::try_from(boundary.completed_tokens())
115            .map_err(|_| invalid_completion("checkpoint boundary exceeds u64"))?;
116        let source_start = u64::try_from(boundary.capture_span_start())
117            .map_err(|_| invalid_completion("checkpoint span start exceeds u64"))?;
118        let input_length = u64::try_from(boundary.full_input().len())
119            .map_err(|_| invalid_completion("checkpoint input length exceeds u64"))?;
120        if !layout.permits_capture_from(source_start, boundary_tokens, input_length) {
121            return Ok(NativeCheckpointStart::Skipped(
122                CheckpointAccessSkipReason::BoundaryNotPermitted,
123            ));
124        }
125        let byte_plan = Arc::new(plan.checkpoint_byte_plan(boundary_tokens)?);
126        let owner = match binding.try_allocate_checkpoint_backing(&byte_plan.backing_requests()?)? {
127            CheckpointBackingAllocationDecision::Allocated(owner) => owner,
128            CheckpointBackingAllocationDecision::Skipped(reason) => {
129                return Ok(NativeCheckpointStart::Skipped(
130                    CheckpointAccessSkipReason::Retention(reason),
131                ));
132            }
133            CheckpointBackingAllocationDecision::Deferred(reason) => {
134                // Maintenance must never hold the source state gate. Its
135                // authenticated requests grant no permission to copy this
136                // boundary later; a fresh capture must validate it again.
137                drop(guard);
138                return Ok(NativeCheckpointStart::CapacityMaintenance {
139                    reason: CheckpointAccessSkipReason::CapacityDeferred(reason),
140                    maintenance: binding
141                        .prepare_checkpoint_capacity_maintenance(plan, &byte_plan)?,
142                });
143            }
144            CheckpointBackingAllocationDecision::BackingDeferred(reason) => {
145                drop(guard);
146                return Ok(NativeCheckpointStart::CapacityMaintenance {
147                    reason: CheckpointAccessSkipReason::BackingDeferred(reason),
148                    maintenance: binding
149                        .prepare_checkpoint_capacity_maintenance(plan, &byte_plan)?,
150                });
151            }
152            CheckpointBackingAllocationDecision::PermanentRejected(reason) => {
153                return Ok(NativeCheckpointStart::Skipped(
154                    CheckpointAccessSkipReason::PermanentRejected(reason),
155                ));
156            }
157        };
158        let permit = owner.try_reserve_capture()?;
159        drop(preparation);
160        Ok(wrap_submission(
161            self.submit_capture_with_timing(guard, permit, byte_plan, lane, timing_mode),
162            None,
163        ))
164    }
165
166    /// The target must already have sufficient admitted backing. This binds
167    /// its exact incarnation and actual complete input before device encoding;
168    /// neither can be replaced later when taking the restore result.
169    pub fn try_restore_sequence_checkpoint(
170        self: &Arc<Self>,
171        plan: &ExecutionPlan,
172        target: Arc<SequenceSession<R>>,
173        checkpoint: &SequenceCheckpoint<R>,
174        full_input: Arc<[u32]>,
175        lane: Arc<ExecutionLane<R>>,
176    ) -> Result<NativeCheckpointStart<R>, VNextError> {
177        self.try_restore_sequence_checkpoint_with_timing(
178            plan,
179            target,
180            checkpoint,
181            full_input,
182            lane,
183            DeviceTimingMode::Off,
184        )
185    }
186
187    /// Same restore and consumer-acknowledgement contract, with Completion
188    /// timing for the whole transfer submission (including initialization).
189    pub fn try_restore_sequence_checkpoint_with_timing(
190        self: &Arc<Self>,
191        plan: &ExecutionPlan,
192        target: Arc<SequenceSession<R>>,
193        checkpoint: &SequenceCheckpoint<R>,
194        full_input: Arc<[u32]>,
195        lane: Arc<ExecutionLane<R>>,
196        timing_mode: DeviceTimingMode,
197    ) -> Result<NativeCheckpointStart<R>, VNextError> {
198        let preparation = self.checkpoint_timings.start(
199            StateTransferKind::Restore,
200            CheckpointTimingPhase::PrepareClaim,
201        );
202        let layout = match access_layout(plan) {
203            Ok(layout) => layout,
204            Err(reason) => return Ok(NativeCheckpointStart::Skipped(reason)),
205        };
206        if plan.plan_hash() != target.resources().plan_evidence().plan_hash()
207            || plan.plan_hash() != checkpoint.plan_hash()
208            || full_input.get(..checkpoint.completed_tokens()) != Some(checkpoint.token_prefix())
209        {
210            return Err(invalid_completion(
211                "checkpoint restore differs from the target plan or token prefix",
212            ));
213        }
214        checkpoint
215            .inner
216            .validate_reuse_contract(layout, &full_input)?;
217        let generation = target.resources().backing_generation()?;
218        let guard = match target
219            .try_prepare_state_transfer(SequenceStateTransferKind::RestoreWrite, generation)?
220        {
221            SequenceStateTransferPreparation::Prepared(guard) => guard,
222            SequenceStateTransferPreparation::Busy => {
223                return Ok(NativeCheckpointStart::Skipped(
224                    CheckpointAccessSkipReason::Busy,
225                ));
226            }
227            SequenceStateTransferPreparation::StaleBacking => {
228                return Ok(NativeCheckpointStart::Skipped(
229                    CheckpointAccessSkipReason::StaleBacking,
230                ));
231            }
232        };
233        guard.validate_restore_input(Arc::clone(&full_input))?;
234        drop(preparation);
235        let submission = self.submit_restore_with_timing(
236            guard,
237            Arc::clone(&checkpoint.inner),
238            layout,
239            lane,
240            timing_mode,
241        );
242        Ok(wrap_submission(
243            submission,
244            Some(RestoreInput { target, full_input }),
245        ))
246    }
247}
ferrum_interfaces/vnext/completion/checkpoint_access/entry.rs

ferrum_interfaces/vnext/completion/checkpoint_access/
entry.rs
