ferrum_interfaces/vnext/completion/
checkpoint_access.rs

1//! Public, owner-preserving access to the existing native state-transfer path.
2//! Raw writable backing, capture permits and terminal seals remain private.
3
4use super::{
5    invalid_completion, CapturedCheckpoint, CompletionReaper, CompletionSlotId, ExecutionLane,
6    RestoreFrontierPublication, StateTransferFailureReason, StateTransferHandle,
7    StateTransferObservation, StateTransferResult, StateTransferSubmission,
8};
9use crate::vnext::{
10    AdmissionDeferred, AdmissionRejected, CheckpointAuthorityId, CheckpointCapacityMaintenance,
11    CheckpointRetentionSkipReason, DeviceErrorReport, DeviceRuntime, DynamicBackingDeferred,
12    PlanHash, SequenceAuthorityId, SequenceSession, SequenceSessionEpoch, VNextError,
13};
14use std::fmt;
15use std::sync::Arc;
16
17mod entry;
18mod recovery;
19
20/// Immutable copied state with independent physical/logical ownership. Clones
21/// pin the same accounted extents; they do not retain the source execution slot.
22pub struct SequenceCheckpoint<R: DeviceRuntime> {
23    inner: Arc<CapturedCheckpoint<R>>,
24}
25
26impl<R: DeviceRuntime> Clone for SequenceCheckpoint<R> {
27    fn clone(&self) -> Self {
28        Self {
29            inner: Arc::clone(&self.inner),
30        }
31    }
32}
33
34impl<R: DeviceRuntime> fmt::Debug for SequenceCheckpoint<R> {
35    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
36        formatter
37            .debug_struct("SequenceCheckpoint")
38            .field("authority", &self.authority())
39            .field("completed_tokens", &self.completed_tokens())
40            .field("retained_bytes", &self.retained_bytes())
41            .finish_non_exhaustive()
42    }
43}
44
45impl<R: DeviceRuntime> SequenceCheckpoint<R> {
46    pub fn authority(&self) -> CheckpointAuthorityId {
47        self.inner.backing().authority()
48    }
49
50    pub fn plan_hash(&self) -> &PlanHash {
51        self.inner.byte_plan().plan_hash()
52    }
53
54    pub fn layout_fingerprint(&self) -> &str {
55        self.inner.byte_plan().layout_fingerprint()
56    }
57
58    pub fn completed_tokens(&self) -> usize {
59        self.inner.boundary().completed_tokens()
60    }
61
62    pub fn token_prefix(&self) -> &[u32] {
63        self.inner.boundary().token_prefix()
64    }
65
66    /// Original complete token input, including any not-yet-computed suffix.
67    pub fn full_input(&self) -> &[u32] {
68        self.inner.boundary().full_input()
69    }
70
71    pub fn logical_bytes(&self) -> u64 {
72        self.inner.backing().logical_bytes()
73    }
74
75    /// Exclusive allocator-aligned extents charged to the existing plan ledger.
76    pub fn retained_bytes(&self) -> u64 {
77        self.inner.backing().extent_bytes()
78    }
79}
80
81#[derive(Debug)]
82pub enum CheckpointAccessSkipReason {
83    Disabled,
84    Unsupported,
85    MissingConditioningEvidence,
86    MissingPartitionEvidence,
87    BoundaryNotPermitted,
88    Busy,
89    StaleBacking,
90    Retention(CheckpointRetentionSkipReason),
91    CapacityDeferred(AdmissionDeferred),
92    BackingDeferred(DynamicBackingDeferred),
93    PermanentRejected(AdmissionRejected),
94}
95
96/// No device work has been submitted for `Skipped`, `CapacityMaintenance`, or
97/// `NotSubmitted`. Maintenance does not reserve the source: after consuming its
98/// authority the caller must repeat capture and all boundary/ownership checks.
99/// Every possibly-submitted outcome retains a handle in the same native reaper.
100#[must_use = "possibly-submitted transfers must reach a terminal or recovery outcome"]
101pub enum NativeCheckpointStart<R: DeviceRuntime> {
102    Skipped(CheckpointAccessSkipReason),
103    CapacityMaintenance {
104        reason: CheckpointAccessSkipReason,
105        maintenance: CheckpointCapacityMaintenance<R>,
106    },
107    NotSubmitted(VNextError),
108    Submitted(NativeCheckpointTransfer<R>),
109    Indeterminate(NativeCheckpointTransfer<R>),
110    ContractAfterSubmission {
111        error: VNextError,
112        transfer: NativeCheckpointTransfer<R>,
113    },
114}
115
116#[derive(Debug, Clone, Copy, PartialEq, Eq)]
117pub enum NativeCheckpointObservation {
118    Pending,
119    Indeterminate,
120    Quarantined,
121    Ready,
122}
123
124impl From<StateTransferObservation> for NativeCheckpointObservation {
125    fn from(value: StateTransferObservation) -> Self {
126        match value {
127            StateTransferObservation::Pending => Self::Pending,
128            StateTransferObservation::Indeterminate => Self::Indeterminate,
129            StateTransferObservation::Quarantined => Self::Quarantined,
130            StateTransferObservation::Ready => Self::Ready,
131        }
132    }
133}
134
135#[derive(Debug, Clone)]
136pub enum NativeCheckpointFailure {
137    FailedButQuiescent(DeviceErrorReport),
138    ContractFailedButQuiescent(String),
139    AbandonedAfterDrain,
140}
141
142impl From<&StateTransferFailureReason> for NativeCheckpointFailure {
143    fn from(value: &StateTransferFailureReason) -> Self {
144        match value {
145            StateTransferFailureReason::FailedButQuiescent(report) => {
146                Self::FailedButQuiescent(report.clone())
147            }
148            StateTransferFailureReason::ContractFailedButQuiescent(reason) => {
149                Self::ContractFailedButQuiescent(reason.clone())
150            }
151            StateTransferFailureReason::AbandonedAfterDrain => Self::AbandonedAfterDrain,
152        }
153    }
154}
155
156#[must_use = "restored state stays gated until its publication is acknowledged"]
157pub enum NativeCheckpointResult<R: DeviceRuntime> {
158    Captured(SequenceCheckpoint<R>),
159    Restored(CheckpointRestorePublication<R>),
160    Failed(NativeCheckpointFailure),
161}
162
163struct RestoreInput<R: DeviceRuntime> {
164    target: Arc<SequenceSession<R>>,
165    full_input: Arc<[u32]>,
166}
167
168/// One native reaper slot. Polling errors are not proof of quiescence. Keep the
169/// handle for blocking recovery or lane drain. Dropping it never releases an
170/// unknown write: it marks the exact result abandoned for the existing reaper's
171/// sweep/recovery path, which continues owning all native resources meanwhile.
172#[must_use = "retain the transfer through terminal delivery or recovery"]
173pub struct NativeCheckpointTransfer<R: DeviceRuntime> {
174    handle: StateTransferHandle<R>,
175    restore: Option<RestoreInput<R>>,
176}
177
178impl<R: DeviceRuntime> NativeCheckpointTransfer<R> {
179    pub fn slot_id(&self) -> CompletionSlotId {
180        self.handle.slot_id()
181    }
182
183    pub fn poll(&self) -> Result<NativeCheckpointObservation, VNextError> {
184        self.handle.poll().map(Into::into)
185    }
186
187    /// Blocking recovery; callers should use their existing completion worker.
188    pub fn wait_for_recovery(&self) -> Result<NativeCheckpointObservation, VNextError> {
189        self.handle.wait_for_recovery().map(Into::into)
190    }
191
192    pub fn recover_by_draining_lane(&self) -> Result<NativeCheckpointObservation, VNextError> {
193        self.handle.recover_by_draining_lane().map(Into::into)
194    }
195
196    /// Takes a terminal result exactly once. A successful restore installs the
197    /// core frontier for the target/input bound at submission, but keeps its
198    /// gate closed while the caller publishes executor/scheduler progress.
199    pub fn take_result(&mut self) -> Result<Option<NativeCheckpointResult<R>>, VNextError> {
200        let Some(result) = self.handle.take()? else {
201            return Ok(None);
202        };
203        match result {
204            StateTransferResult::Captured(inner) => {
205                if self.restore.is_some() {
206                    return Err(invalid_completion(
207                        "restore access received a capture result",
208                    ));
209                }
210                Ok(Some(NativeCheckpointResult::Captured(SequenceCheckpoint {
211                    inner,
212                })))
213            }
214            StateTransferResult::RestoreReady(pending) => {
215                let input = self.restore.take().ok_or_else(|| {
216                    invalid_completion("restore result lost its bound target input")
217                })?;
218                let inner = pending.install_frontier(&input.target, input.full_input)?;
219                Ok(Some(NativeCheckpointResult::Restored(
220                    CheckpointRestorePublication {
221                        inner,
222                        target: input.target,
223                    },
224                )))
225            }
226            StateTransferResult::Failed(failure) => {
227                // The native terminal path already cancelled a failed restore.
228                self.restore.take();
229                Ok(Some(NativeCheckpointResult::Failed(
230                    failure.reason().into(),
231                )))
232            }
233        }
234    }
235}
236
237impl<R: DeviceRuntime> Drop for NativeCheckpointTransfer<R> {
238    fn drop(&mut self) {
239        if let Some(input) = &self.restore {
240            // Cancellation never releases the native transfer's retained gate.
241            // The same reaper must still prove quiescence before freeing bytes.
242            let _ = input.target.request_cancel();
243        }
244        self.handle.abandon_consumer();
245    }
246}
247
248/// Non-cloneable outer publication owner. Dropping or rejecting it cancels the
249/// exact target before its state gate opens. It exposes no writable resources.
250#[must_use = "acknowledge only after executor and scheduler progress is published"]
251pub struct CheckpointRestorePublication<R: DeviceRuntime> {
252    inner: RestoreFrontierPublication<R>,
253    target: Arc<SequenceSession<R>>,
254}
255
256impl<R: DeviceRuntime> CheckpointRestorePublication<R> {
257    pub fn completed_tokens(&self) -> usize {
258        self.inner.completed_tokens()
259    }
260
261    pub fn token_prefix(&self) -> &[u32] {
262        self.inner.token_prefix()
263    }
264
265    pub fn target_sequence_authority(&self) -> SequenceAuthorityId {
266        self.target.sequence_authority()
267    }
268
269    pub fn target_epoch(&self) -> SequenceSessionEpoch {
270        self.target.epoch()
271    }
272
273    pub fn matches_target(&self, target: &Arc<SequenceSession<R>>) -> bool {
274        Arc::ptr_eq(&self.target, target)
275    }
276
277    pub fn acknowledge(self) -> Result<(), VNextError> {
278        self.inner.acknowledge().map(|_| ())
279    }
280}
ferrum_interfaces/vnext/completion/checkpoint_access.rs

ferrum_interfaces/vnext/completion/
checkpoint_access.rs
