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

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