canic_host/fleet_ensure/ops/capacity_import/journal/
mod.rs

1//! Construct and validate capacity-import handoff records using Fleet effect records.
2//!
3//! Workflow must retain each changed journal under the shared operation lock before effects.
4//! These helpers never dispatch an IC request or infer success from an unavailable reply.
5
6pub mod retirement;
7mod store;
8#[cfg(test)]
9mod tests;
10
11pub use store::CapacityImportJournalStore;
12pub(in crate::fleet_ensure::ops) use store::require_no_approved_import;
13
14use crate::fleet_ensure::{
15    model::{
16        EffectRecord, EffectState, EnsureAction,
17        capacity_import::{
18            CapacityImportHandoffRecord, CapacityImportHandoffRequestRecord,
19            CapacityImportJournalRecord, CapacityImportPlanRecord, CapacityImportReservationRecord,
20        },
21    },
22    ops::{
23        action_sha256,
24        capacity_import::{
25            CapacityImportReviewError,
26            transport::{CompletedHandoff, validate_request},
27            verify_review,
28        },
29    },
30    policy::capacity_import::{
31        CapacityImportPolicyError, admit_handoffs, admit_source_handoff, retained_source_debit,
32        source_total,
33    },
34    view::capacity_import::{CapacityImportDestinationView, CapacityImportSourceView},
35};
36use candid::Principal;
37use thiserror::Error;
38
39/// Typed phase identifying which bounded inventory observation failed.
40#[derive(Clone, Copy, Debug, Eq, PartialEq)]
41pub enum CapacityImportInventoryStage {
42    Decode,
43    Membership,
44    Page,
45    Policy,
46    Query,
47    Summary,
48}
49
50/// Journal rejection preserves the original operation and all issued intent.
51#[derive(Debug, Error)]
52pub enum CapacityImportJournalError {
53    #[error(
54        "no original capacity observation is retained for {canister}; keep the original review inputs when declaring additional funding, or review this source before declaring a credit"
55    )]
56    FundingBaselineMissing { canister: Principal },
57    #[error(
58        "capacity import requires a completed current Fleet or receipted bootstrap setup with exact installed infrastructure records"
59    )]
60    InfrastructureRequired,
61    #[error("capacity import infrastructure module, controllers, placement or Registry changed")]
62    InfrastructureChanged,
63    #[error(
64        "capacity import requires an exact operator disposition declaration for every candidate"
65    )]
66    DispositionInvalid,
67    #[error(
68        "capacity import inventory is incomplete, changed, duplicated or exceeds its observation bound"
69    )]
70    InventoryInvalid,
71    #[error("capacity import inventory observation of {canister} failed at {stage:?}")]
72    InventoryObservation {
73        canister: Principal,
74        stage: CapacityImportInventoryStage,
75    },
76    #[error(
77        "management observation failed for {canister}; verify operator controller access before retrying"
78    )]
79    ObservationUnavailable { canister: Principal },
80    #[error(transparent)]
81    Inspection(Box<crate::canister_protocol::CanisterProtocolError>),
82    #[error(transparent)]
83    Prerequisite(
84        #[from] crate::fleet_ensure::ops::capacity_import::CapacityImportPrerequisiteError,
85    ),
86    #[error(
87        "Coordinator {coordinator} is unavailable; verify its initialization, network and operator access"
88    )]
89    CoordinatorUnavailable { coordinator: Principal },
90    #[error("Coordinator rejected the import prerequisite query: {0}")]
91    CoordinatorRejected(canic_core::dto::error::Error),
92    #[error(transparent)]
93    Review(#[from] CapacityImportReviewError),
94    #[error(transparent)]
95    Policy(#[from] CapacityImportPolicyError),
96    #[error("capacity import journal does not match the reviewed handoff sequence")]
97    Integrity,
98    #[error("capacity import request differs from the exact reviewed controller handoff")]
99    RequestInvalid,
100    #[error("capacity import transport differs from the reviewed network or signer")]
101    ReaderMismatch,
102    #[error("Root rejected the capacity import request: {0}")]
103    RootRejected(canic_core::dto::error::Error),
104    #[error(
105        "capacity import budget cannot cover the complete operation: {paid_calls} calls, at least {minimum_calls} required; debit ceiling {maximum_debit_cycles}, required {required_debit_cycles}"
106    )]
107    InsufficientRootBudget {
108        paid_calls: u32,
109        minimum_calls: u32,
110        maximum_debit_cycles: u128,
111        required_debit_cycles: u128,
112    },
113    #[error(
114        "Root capacity import hit a capacity limit in phase {phase:?}: {reserved_debit_cycles}/{maximum_debit_cycles} cycles reserved, {observed_debit_cycles} observed debit, {paid_calls}/{maximum_paid_calls} calls; preserve issued operation authority"
115    )]
116    RootCapacityLimit {
117        phase: canic_core::dto::pool_import::PoolImportPhase,
118        reserved_debit_cycles: u128,
119        maximum_debit_cycles: u128,
120        observed_debit_cycles: u128,
121        paid_calls: u32,
122        maximum_paid_calls: u32,
123    },
124    #[error("Root capacity import response has an invalid current-contract encoding")]
125    RootResponseInvalid,
126    #[error("capacity import requires exact approval and a protected Root reservation")]
127    ReservationRequired,
128    #[error("capacity import effect outcome is unresolved; retain its original intent")]
129    Unresolved,
130
131    #[error(
132        "capacity import exhausted its reviewed {step} attempts; preserve the original journal"
133    )]
134    BudgetExhausted { step: String },
135    #[error("capacity import inventory inputs or publication evidence changed")]
136    PublicationConflict,
137    #[error("capacity import generator inputs are invalid: {0}")]
138    InventoryProjection(
139        Box<crate::fleet_ensure::generate::capacity_import::CapacityImportInventoryError>,
140    ),
141    #[error("another capacity import already retains approval or effect intent")]
142    Conflict,
143    #[error(transparent)]
144    State(#[from] crate::fleet_ensure::ops::EnsureStateError),
145    #[error("capacity import journal file operation failed")]
146    Io(#[from] std::io::Error),
147    #[error("capacity import journal cannot be decoded or encoded")]
148    Encoding(#[from] serde_json::Error),
149}
150
151impl From<crate::canister_protocol::CanisterProtocolError> for CapacityImportJournalError {
152    fn from(error: crate::canister_protocol::CanisterProtocolError) -> Self {
153        Self::Inspection(Box::new(error))
154    }
155}
156
157/// Construct the initial journal without approval or effects.
158pub fn reviewed(
159    plan: CapacityImportPlanRecord,
160) -> Result<CapacityImportJournalRecord, CapacityImportJournalError> {
161    verify_review(&plan, plan.plan_sha256)?;
162    Ok(CapacityImportJournalRecord {
163        handoffs: plan
164            .sources
165            .iter()
166            .map(|source| CapacityImportHandoffRecord {
167                canister_id: source.binding.canister_id,
168                retirements: Vec::new(),
169                effect: None,
170                request: None,
171                before_reserved_cycles: None,
172                after_reserved_cycles: None,
173            })
174            .collect(),
175        plan,
176        operation: None,
177        approved: false,
178        reservation: None,
179    })
180}
181
182/// Bind explicit approval to fresh, complete authority and obligation observations.
183pub fn approve(
184    journal: &CapacityImportJournalRecord,
185    digest: [u8; 32],
186    destination: &CapacityImportDestinationView,
187    sources: &[CapacityImportSourceView],
188) -> Result<CapacityImportJournalRecord, CapacityImportJournalError> {
189    validate(journal)?;
190    verify_review(&journal.plan, digest)?;
191    if journal.approved {
192        return Ok(journal.clone());
193    }
194    admit_handoffs(&journal.plan, destination, sources)?;
195    let mut updated = journal.clone();
196    updated.approved = true;
197    Ok(updated)
198}
199
200/// Retain a reservation only after the transport authenticates the Root's protected reply.
201pub fn reserve(
202    journal: &CapacityImportJournalRecord,
203    reservation: CapacityImportReservationRecord,
204) -> Result<CapacityImportJournalRecord, CapacityImportJournalError> {
205    validate(journal)?;
206    if !journal.approved || !reservation_matches(&journal.plan, &reservation) {
207        return Err(CapacityImportJournalError::ReservationRequired);
208    }
209    if let Some(retained) = &journal.reservation {
210        if *retained != reservation {
211            return Err(CapacityImportJournalError::Integrity);
212        }
213        return Ok(journal.clone());
214    }
215    let mut updated = journal.clone();
216    updated.reservation = Some(reservation);
217    Ok(updated)
218}
219
220/// Retain one exact intent before issuing the controller update.
221/// A repeated call returns the existing intent; it grants no additional issuance.
222pub fn prepare_handoff(
223    journal: &CapacityImportJournalRecord,
224    observed: &CapacityImportSourceView,
225    request: CapacityImportHandoffRequestRecord,
226) -> Result<CapacityImportJournalRecord, CapacityImportJournalError> {
227    validate(journal)?;
228    if !journal.approved || journal.reservation.is_none() {
229        return Err(CapacityImportJournalError::ReservationRequired);
230    }
231    let index = source_index(journal, observed.binding.canister_id)?;
232    if journal.handoffs[index].effect.is_some() {
233        return Ok(journal.clone());
234    }
235    let source = &journal.plan.sources[index];
236    if !crate::fleet_ensure::policy::capacity_import::requires_handoff(&journal.plan, source) {
237        return Err(CapacityImportJournalError::Integrity);
238    }
239    admit_source_handoff(source, observed)?;
240    validate_request(&journal.plan, observed.binding.canister_id, &request)?;
241    let mut updated = journal.clone();
242    updated.handoffs[index].effect = Some(handoff_intent(&journal.plan, index, observed.cycles));
243    updated.handoffs[index].before_reserved_cycles = Some(observed.reserved_cycles);
244    updated.handoffs[index].request = Some(request);
245    Ok(updated)
246}
247
248fn handoff_intent(
249    plan: &CapacityImportPlanRecord,
250    index: usize,
251    observed_cycles: u128,
252) -> EffectRecord {
253    EffectRecord {
254        maintenance_attempts: 0,
255        publication_attempts: 0,
256        action_sha256: action_sha256(&handoff_action(plan, index)),
257        created_principal: None,
258        destination_post_cycles: None,
259        destination_pre_cycles: None,
260        post_cycles: None,
261        pre_cycles: Some(observed_cycles),
262        pre_canister_version: Some(plan.sources[index].binding.canister_version),
263        progress_identity: None,
264        receipt: None,
265        state: EffectState::Intent,
266    }
267}
268
269/// Record issuance before submitting the exact action. Issued intent cannot be reissued here.
270pub fn issue_handoff(
271    journal: &CapacityImportJournalRecord,
272    canister: Principal,
273) -> Result<CapacityImportJournalRecord, CapacityImportJournalError> {
274    validate(journal)?;
275    let index = source_index(journal, canister)?;
276    let mut updated = journal.clone();
277    let effect = updated.handoffs[index]
278        .effect
279        .as_mut()
280        .ok_or(CapacityImportJournalError::Integrity)?;
281    if effect.state != EffectState::Intent {
282        return Err(CapacityImportJournalError::Unresolved);
283    }
284    effect.state = EffectState::Issued;
285    Ok(updated)
286}
287
288/// Reconcile a lost update reply only against the exact post-handoff management identity.
289/// Original controllers, failed reads and foreign changes never establish non-execution.
290pub fn observe_handoff(
291    journal: &CapacityImportJournalRecord,
292    observed: &CapacityImportSourceView,
293    completion: &CompletedHandoff,
294) -> Result<CapacityImportJournalRecord, CapacityImportJournalError> {
295    validate(journal)?;
296    let index = source_index(journal, observed.binding.canister_id)?;
297    let source = &journal.plan.sources[index];
298    let request = journal.handoffs[index]
299        .request
300        .as_ref()
301        .ok_or(CapacityImportJournalError::Integrity)?;
302    if !completion.matches(&journal.plan, source.binding.canister_id, request) {
303        return Err(CapacityImportJournalError::Unresolved);
304    }
305    apply_observed_handoff(journal, observed)
306}
307
308// Called only after exact certified reply or retained terminal ingress evidence.
309fn apply_observed_handoff(
310    journal: &CapacityImportJournalRecord,
311    observed: &CapacityImportSourceView,
312) -> Result<CapacityImportJournalRecord, CapacityImportJournalError> {
313    validate(journal)?;
314    let index = source_index(journal, observed.binding.canister_id)?;
315    let effect = journal.handoffs[index]
316        .effect
317        .as_ref()
318        .ok_or(CapacityImportJournalError::Integrity)?;
319    if effect.state == EffectState::Applied {
320        return Ok(journal.clone());
321    }
322    if effect.state != EffectState::Issued {
323        return Err(CapacityImportJournalError::Integrity);
324    }
325    let source = &journal.plan.sources[index];
326    let mut expected = source.binding.clone();
327    expected
328        .controllers
329        .clone_from(&journal.plan.transitional_controllers);
330    expected.canister_version = expected
331        .canister_version
332        .checked_add(1)
333        .ok_or(CapacityImportJournalError::Integrity)?;
334    if observed.binding != expected {
335        return Err(CapacityImportJournalError::Unresolved);
336    }
337    retained_source_debit(source, observed.cycles, observed.reserved_cycles)?;
338    let before = source_total(
339        effect
340            .pre_cycles
341            .ok_or(CapacityImportJournalError::Integrity)?,
342        journal.handoffs[index]
343            .before_reserved_cycles
344            .ok_or(CapacityImportJournalError::Integrity)?,
345    )?;
346    if source_total(observed.cycles, observed.reserved_cycles)? > before {
347        return Err(CapacityImportJournalError::Unresolved);
348    }
349    let mut updated = journal.clone();
350    let effect = updated.handoffs[index]
351        .effect
352        .as_mut()
353        .ok_or(CapacityImportJournalError::Integrity)?;
354    effect.state = EffectState::Applied;
355    effect.post_cycles = Some(observed.cycles);
356    // The receipt records the observed version, not a synthetic IC request ID.
357    effect.receipt = Some(expected.canister_version.to_string());
358    updated.handoffs[index].after_reserved_cycles = Some(observed.reserved_cycles);
359    validate(&updated)?;
360    Ok(updated)
361}
362
363/// Reject malformed or substituted records before any workflow resumes them.
364pub fn validate(journal: &CapacityImportJournalRecord) -> Result<(), CapacityImportJournalError> {
365    verify_review(&journal.plan, journal.plan.plan_sha256)?;
366    crate::fleet_ensure::ops::capacity_import::publication::validate(journal)?;
367    if journal.handoffs.len() != journal.plan.sources.len() {
368        return Err(CapacityImportJournalError::Integrity);
369    }
370    if let Some(reservation) = &journal.reservation
371        && (!journal.approved || !reservation_matches(&journal.plan, reservation))
372    {
373        return Err(CapacityImportJournalError::Integrity);
374    }
375    for (index, handoff) in journal.handoffs.iter().enumerate() {
376        let source = &journal.plan.sources[index];
377        retirement::validate(&journal.plan, handoff)?;
378        if handoff.canister_id != source.binding.canister_id {
379            return Err(CapacityImportJournalError::Integrity);
380        }
381        let Some(effect) = &handoff.effect else {
382            if !handoff.retirements.is_empty()
383                || handoff.request.is_some()
384                || handoff.before_reserved_cycles.is_some()
385                || handoff.after_reserved_cycles.is_some()
386            {
387                return Err(CapacityImportJournalError::Integrity);
388            }
389            continue;
390        };
391        if !crate::fleet_ensure::policy::capacity_import::requires_handoff(&journal.plan, source) {
392            return Err(CapacityImportJournalError::Integrity);
393        }
394        validate_request(
395            &journal.plan,
396            handoff.canister_id,
397            handoff
398                .request
399                .as_ref()
400                .ok_or(CapacityImportJournalError::Integrity)?,
401        )?;
402        let action_bound =
403            effect.action_sha256 == action_sha256(&handoff_action(&journal.plan, index));
404        let authority_bound = journal.approved
405            && journal.reservation.is_some()
406            && effect.pre_canister_version == Some(source.binding.canister_version);
407        let counters_empty = effect.maintenance_attempts == 0 && effect.publication_attempts == 0;
408        let unrelated_empty = effect.created_principal.is_none()
409            && effect.destination_pre_cycles.is_none()
410            && effect.destination_post_cycles.is_none()
411            && effect.progress_identity.is_none();
412        if !action_bound || !authority_bound || !counters_empty || !unrelated_empty {
413            return Err(CapacityImportJournalError::Integrity);
414        }
415        let before = effect
416            .pre_cycles
417            .ok_or(CapacityImportJournalError::Integrity)?;
418        let before_reserved = handoff
419            .before_reserved_cycles
420            .ok_or(CapacityImportJournalError::Integrity)?;
421        retained_source_debit(source, before, before_reserved)?;
422        match effect.state {
423            EffectState::Intent | EffectState::Issued => {
424                if effect.post_cycles.is_some()
425                    || effect.receipt.is_some()
426                    || handoff.after_reserved_cycles.is_some()
427                {
428                    return Err(CapacityImportJournalError::Integrity);
429                }
430            }
431            EffectState::Applied => {
432                let after = effect
433                    .post_cycles
434                    .ok_or(CapacityImportJournalError::Integrity)?;
435                let version = source
436                    .binding
437                    .canister_version
438                    .checked_add(1)
439                    .ok_or(CapacityImportJournalError::Integrity)?;
440                let after_reserved = handoff
441                    .after_reserved_cycles
442                    .ok_or(CapacityImportJournalError::Integrity)?;
443                if source_total(after, after_reserved)? > source_total(before, before_reserved)?
444                    || effect.receipt.as_deref() != Some(version.to_string().as_str())
445                {
446                    return Err(CapacityImportJournalError::Integrity);
447                }
448                retained_source_debit(source, after, after_reserved)?;
449            }
450        }
451    }
452    Ok(())
453}
454
455/// A source already held by Root requires no host controller effect.
456/// All other sources require a certified completion retained by this journal owner.
457#[must_use]
458pub fn custody_ready(journal: &CapacityImportJournalRecord, canister: Principal) -> bool {
459    let Some(index) = journal
460        .plan
461        .sources
462        .iter()
463        .position(|source| source.binding.canister_id == canister)
464    else {
465        return false;
466    };
467    let Some(handoff) = journal.handoffs.get(index) else {
468        return false;
469    };
470    if handoff.canister_id != canister {
471        return false;
472    }
473    if crate::fleet_ensure::policy::capacity_import::requires_handoff(
474        &journal.plan,
475        &journal.plan.sources[index],
476    ) {
477        handoff
478            .effect
479            .as_ref()
480            .is_some_and(|effect| effect.state == EffectState::Applied)
481    } else {
482        handoff.effect.is_none()
483            && handoff.request.is_none()
484            && handoff.retirements.is_empty()
485            && handoff.before_reserved_cycles.is_none()
486            && handoff.after_reserved_cycles.is_none()
487    }
488}
489
490/// Whether every selected source has retained host completion or reviewed Root custody.
491#[must_use]
492pub fn all_custody_ready(journal: &CapacityImportJournalRecord) -> bool {
493    journal.handoffs.len() == journal.plan.sources.len()
494        && journal
495            .plan
496            .sources
497            .iter()
498            .all(|source| custody_ready(journal, source.binding.canister_id))
499}
500
501fn source_index(
502    journal: &CapacityImportJournalRecord,
503    canister: Principal,
504) -> Result<usize, CapacityImportJournalError> {
505    journal
506        .plan
507        .sources
508        .iter()
509        .position(|source| source.binding.canister_id == canister)
510        .ok_or(CapacityImportJournalError::Integrity)
511}
512
513fn handoff_action(plan: &CapacityImportPlanRecord, index: usize) -> EnsureAction {
514    let canister = plan.sources[index].binding.canister_id.to_text();
515    EnsureAction::SetControllers {
516        controller_canisters: Vec::new(),
517        controllers: plan
518            .transitional_controllers
519            .iter()
520            .map(Principal::to_text)
521            .collect(),
522        name: canister.clone(),
523        principal: canister,
524    }
525}
526
527fn reservation_matches(
528    plan: &CapacityImportPlanRecord,
529    reservation: &CapacityImportReservationRecord,
530) -> bool {
531    reservation.plan_sha256 == plan.plan_sha256
532        && reservation.authority == plan.authority
533        && reservation.sources
534            == plan
535                .sources
536                .iter()
537                .map(|source| source.binding.canister_id)
538                .collect::<Vec<_>>()
539}
canic_host/fleet_ensure/ops/capacity_import/journal/mod.rs

canic_host/fleet_ensure/ops/capacity_import/journal/
mod.rs
