liminal_protocol/lifecycle/operations/live_frontier/
checkpoint_restore.rs

1//! Restores executable frontier ownership only from a validated lifecycle capsule.
2
3use alloc::vec::Vec;
4
5use crate::algebra::{ResourceVector, WideResourceVector};
6use crate::lifecycle::{
7    ClosureAccounting, OrdinaryProjectionError, ParticipantConversation, RestoredLifecycleParts,
8    RestoredParticipantLifecycle, RetainedRecordCharge,
9};
10
11use super::LiveFrontierOwner;
12
13/// Participant values moved from the validated 32-byte-digest lifecycle capsule.
14pub type CheckpointParticipant =
15    RestoredParticipantLifecycle<[u8; 32], [u8; 32], [u8; 32], [u8; 32]>;
16
17/// The original shell and participants beside their restored executable frontier.
18#[derive(Debug)]
19#[non_exhaustive]
20pub struct RestoredLiveFrontierOwner {
21    /// The capsule's original conversation shell.
22    pub conversation: ParticipantConversation,
23    /// The capsule's original participant values and tombstones.
24    pub participants: Vec<CheckpointParticipant>,
25    /// The coupled frontier, accounting and exact keyed retained charges.
26    pub owner: LiveFrontierOwner,
27    /// The exact saved cursor facts, present only for owed closure.
28    pub cursor_episode: Option<crate::lifecycle::NonzeroDebtCursorEpisode>,
29    pub(in crate::lifecycle) observer_progress: u64,
30}
31
32/// Named checkpoint omissions and joint-state validation failures.
33#[derive(Debug, PartialEq, Eq)]
34pub enum LiveFrontierRestoreError {
35    /// The checkpoint contains no enrolled lifecycle.
36    MissingLifecycle,
37    /// The checkpoint omits closure accounting.
38    MissingClosureAccounting,
39    /// The checkpoint omits keyed retained charges.
40    MissingRetainedCharges,
41    /// Owed closure omits its saved cursor episode.
42    MissingCursorEpisode,
43    /// Clear closure unexpectedly carries a cursor episode.
44    UnexpectedCursorEpisode,
45    /// The original accounting reader refused the source.
46    AccountingCodec(crate::lifecycle::FrontierAccountingDecodeError),
47    /// The original retained-charge reader refused the source.
48    RetainedChargeCodec(crate::lifecycle::CapsuleDecodeError),
49    /// The original joint cursor-episode validator refused the source.
50    CursorEpisode(crate::lifecycle::StorageRestoreError),
51    /// The original saved-episode writer refused canonical encoding.
52    CursorEpisodeEncoding(crate::lifecycle::CursorFactEncodeError),
53    /// The shell and restored frontier belong to different conversations.
54    ForeignConversation,
55    /// The capsule's closure and decoded accounting disagree.
56    ClosureMismatch,
57    /// Retained rows exceed their configured limit.
58    RetainedRecordLimit,
59    /// Native retained-row, marker or baseline validation refused.
60    Accounting(OrdinaryProjectionError),
61}
62
63impl LiveFrontierOwner {
64    /// Restores coupled ownership from the validated capsule and decoded saved values.
65    ///
66    /// No participant, frontier, closure, charge or accounting field is regenerated.
67    ///
68    /// # Errors
69    ///
70    /// Names every absent saved field and refuses inconsistent closure, retained keys,
71    /// marker ledgers, baseline or retention limit.
72    fn restore_checkpoint_parts(
73        parts: RestoredLifecycleParts,
74        accounting: ClosureAccounting,
75        charges: Vec<RetainedRecordCharge>,
76        cursor_episode: Option<crate::lifecycle::NonzeroDebtCursorEpisode>,
77        saved: FrontierCheckpointSource<'_>,
78    ) -> Result<RestoredLiveFrontierOwner, LiveFrontierRestoreError> {
79        let lifecycle = parts
80            .lifecycle
81            .ok_or(LiveFrontierRestoreError::MissingLifecycle)?;
82        if lifecycle.frontiers().conversation_id() != parts.conversation.conversation_id() {
83            return Err(LiveFrontierRestoreError::ForeignConversation);
84        }
85        if lifecycle.closure() != accounting.state() {
86            return Err(LiveFrontierRestoreError::ClosureMismatch);
87        }
88        if lifecycle.frontiers().retained_records().len() as u128
89            > u128::from(saved.retained_record_limit)
90        {
91            return Err(LiveFrontierRestoreError::RetainedRecordLimit);
92        }
93        validate_saved_accounting(
94            lifecycle.frontiers(),
95            &accounting,
96            &charges,
97            saved.marker_max,
98        )
99        .map_err(LiveFrontierRestoreError::Accounting)?;
100        let (participants, frontiers, _) = lifecycle.into_parts();
101        Ok(RestoredLiveFrontierOwner {
102            conversation: parts.conversation,
103            participants,
104            cursor_episode,
105            observer_progress: saved.observer_progress,
106            owner: Self {
107                frontiers,
108                closure_accounting: accounting,
109                retained_charges: charges,
110                retained_record_limit: saved.retained_record_limit,
111            },
112        })
113    }
114}
115
116fn validate_saved_accounting(
117    frontiers: &crate::lifecycle::ClaimFrontiers,
118    accounting: &ClosureAccounting,
119    charges: &[RetainedRecordCharge],
120    marker_max: ResourceVector,
121) -> Result<(), OrdinaryProjectionError> {
122    let records = frontiers.retained_records();
123    let high_end = u128::from(frontiers.sequence().ledger().high_watermark()) + 1;
124    if high_end.checked_sub(frontiers.retained_floor()) != Some(records.len() as u128)
125        || records.len() != charges.len()
126    {
127        return Err(OrdinaryProjectionError::RetainedSuffix);
128    }
129    let mut retained = WideResourceVector::default();
130    for (index, (record, charge)) in records.iter().zip(charges).enumerate() {
131        if u128::from(record.delivery_seq) != frontiers.retained_floor() + index as u128
132            || charge.delivery_seq() != record.delivery_seq
133            || charge.admission_order() != record.admission_order
134        {
135            return Err(OrdinaryProjectionError::RetainedChargeKey {
136                index: u64::try_from(index).map_err(|_| OrdinaryProjectionError::RetainedSuffix)?,
137            });
138        }
139        let resource = charge.encoded_charge();
140        if resource.entries != 1 {
141            return Err(OrdinaryProjectionError::RetainedEntryCharge {
142                delivery_seq: charge.delivery_seq(),
143                entries: resource.entries,
144            });
145        }
146        retained =
147            super::super::ordinary_record_projection::checked_add_resource(retained, resource)?;
148    }
149    let pending = frontiers.pending_marker_candidate_count();
150    let retained_credits = u64::try_from(frontiers.retained_marker_records().len())
151        .map_err(|_| OrdinaryProjectionError::MarkerCredit)?;
152    let credits = retained_credits
153        .checked_add(pending)
154        .ok_or(OrdinaryProjectionError::MarkerCredit)?;
155    if credits != accounting.marker_capacity_credits() {
156        return Err(OrdinaryProjectionError::MarkerCreditAccounting {
157            derived: credits,
158            stored: accounting.marker_capacity_credits(),
159        });
160    }
161    let anchors = frontiers
162        .unaccepted_marker_anchor_count()
163        .checked_add(pending)
164        .ok_or(OrdinaryProjectionError::MarkerAnchor)?;
165    if anchors != accounting.marker_anchors() {
166        return Err(OrdinaryProjectionError::MarkerAnchorAccounting {
167            derived: anchors,
168            stored: accounting.marker_anchors(),
169        });
170    }
171    let baseline = super::super::ordinary_record_projection::retained_baseline_wide(
172        retained,
173        frontiers.identity_slot_limit(),
174        retained_credits,
175        marker_max,
176    )?;
177    if baseline != accounting.baseline() {
178        return Err(OrdinaryProjectionError::BaselineAccounting {
179            derived: baseline,
180            stored: accounting.baseline(),
181        });
182    }
183    Ok(())
184}
185
186impl LiveFrontierOwner {
187    /// Decodes every saved coupled field against a validated lifecycle capsule.
188    ///
189    /// # Errors
190    ///
191    /// Names absent fields, retains the original codec failures, and rejects
192    /// mismatched cursor facts, keyed charges, closure, marker ledgers or baseline.
193    pub fn restore_checkpoint(
194        parts: RestoredLifecycleParts,
195        saved: FrontierCheckpointSource<'_>,
196    ) -> Result<RestoredLiveFrontierOwner, LiveFrontierRestoreError> {
197        let lifecycle = parts
198            .lifecycle
199            .as_ref()
200            .ok_or(LiveFrontierRestoreError::MissingLifecycle)?;
201        let accounting = Self::decode_closure_accounting_source(
202            lifecycle.closure(),
203            saved
204                .closure_accounting
205                .ok_or(LiveFrontierRestoreError::MissingClosureAccounting)?,
206        )
207        .map_err(LiveFrontierRestoreError::AccountingCodec)?;
208        let charges = Self::decode_retained_charge_source(
209            saved
210                .retained_charges
211                .ok_or(LiveFrontierRestoreError::MissingRetainedCharges)?,
212        )
213        .map_err(LiveFrontierRestoreError::RetainedChargeCodec)?;
214        let episode = match lifecycle.closure() {
215            crate::lifecycle::ClosureState::Clear => {
216                if saved.cursor_episode.is_some() {
217                    return Err(LiveFrontierRestoreError::UnexpectedCursorEpisode);
218                }
219                None
220            }
221            crate::lifecycle::ClosureState::Owed { debt, .. } => {
222                let bytes = saved
223                    .cursor_episode
224                    .ok_or(LiveFrontierRestoreError::MissingCursorEpisode)?;
225                Some(
226                    crate::lifecycle::NonzeroDebtCursorEpisode::decode_checkpoint(
227                        bytes,
228                        lifecycle.frontiers(),
229                        debt,
230                        saved.observer_progress,
231                    )
232                    .map_err(LiveFrontierRestoreError::CursorEpisode)?,
233                )
234            }
235        };
236        Self::restore_checkpoint_parts(parts, accounting, charges, episode, saved)
237    }
238}
239
240/// Borrowed canonical saved fields and their required configured bounds.
241#[derive(Clone, Copy, Debug)]
242pub struct FrontierCheckpointSource<'a> {
243    /// Canonical closure-accounting scalar bytes; absence is a named refusal.
244    pub closure_accounting: Option<&'a [u8]>,
245    /// Canonical keyed retained-charge bytes; absence is a named refusal.
246    pub retained_charges: Option<&'a [u8]>,
247    /// Exact saved cursor episode, required for owed closure and absent for clear.
248    pub cursor_episode: Option<&'a [u8]>,
249    /// Saved observer progress used to validate the episode.
250    pub observer_progress: u64,
251    /// The configured bound on retained physical rows.
252    pub retained_record_limit: u64,
253    /// The configured maximum marker charge used by native baseline accounting.
254    pub marker_max: ResourceVector,
255}
liminal_protocol/lifecycle/operations/live_frontier/checkpoint_restore.rs

liminal_protocol/lifecycle/operations/live_frontier/
checkpoint_restore.rs
