canic_host/fleet_ensure/workflow/operator_mint/execution/
mod.rs1use crate::fleet_ensure::{
7 model::operator_mint::{
8 OperatorMintAuthority, OperatorMintNotificationOutcomeRecord, OperatorMintReviewRecord,
9 },
10 ops::{
11 EnsurePaths, lock_operation,
12 operator_mint::transport::{OperatorMintTransport, OperatorMintTransportError},
13 },
14 policy::operator_mint::progress,
15 workflow::{
16 EnsureWorkflowError,
17 operator_mint::{self, load, selected},
18 },
19};
20use std::convert::Infallible;
21use thiserror::Error;
22
23#[derive(Debug, Error)]
29pub enum OperatorMintExecutionError {
30 #[error("operator mint async runtime could not start")]
31 Runtime(#[from] std::io::Error),
32 #[error(transparent)]
33 Transport(Box<OperatorMintTransportError>),
34 #[error(transparent)]
35 Workflow(Box<EnsureWorkflowError<Infallible>>),
36}
37
38impl From<OperatorMintTransportError> for OperatorMintExecutionError {
39 fn from(error: OperatorMintTransportError) -> Self {
40 Self::Transport(Box::new(error))
41 }
42}
43
44impl From<EnsureWorkflowError<Infallible>> for OperatorMintExecutionError {
45 fn from(error: EnsureWorkflowError<Infallible>) -> Self {
46 Self::Workflow(Box::new(error))
47 }
48}
49
50pub fn apply_blocking(
52 paths: &EnsurePaths,
53 authority: &OperatorMintAuthority,
54 digest: &str,
55 transport: &OperatorMintTransport,
56) -> Result<OperatorMintReviewRecord, OperatorMintExecutionError> {
57 tokio::runtime::Runtime::new()?.block_on(apply(paths, authority, digest, transport))
58}
59
60pub async fn apply(
66 paths: &EnsurePaths,
67 authority: &OperatorMintAuthority,
68 digest: &str,
69 transport: &OperatorMintTransport,
70) -> Result<OperatorMintReviewRecord, OperatorMintExecutionError> {
71 let (retained, source, maximum_debit) = {
72 let _lock = lock_operation(paths).map_err(EnsureWorkflowError::from)?;
73 let (plan, journal) = load(paths)?;
74 let source = crate::fleet_ensure::policy::operator_mint::operator_source(&journal)
75 .ok_or(EnsureWorkflowError::JournalIntegrity)?;
76 let maximum = crate::fleet_ensure::workflow::funding_plan::<Infallible>(&plan, &journal)?
77 .conservation
78 .maximum_operator_debit_cycles;
79 (
80 selected(&journal, authority, digest)?.clone(),
81 source,
82 maximum,
83 )
84 };
85 if retained.receipt.is_some() {
86 return Ok(retained);
87 }
88 if transport.operator()? != authority.operator
89 || transport.network_identity() != authority.network_identity_sha256
90 {
91 return Err(OperatorMintTransportError::ReaderMismatch.into());
92 }
93 if retained.transfer_argument.is_none() {
94 let observed = transport
95 .read_operator_balance(authority.cycles_ledger)
96 .await?;
97 if source
98 .checked_sub(observed)
99 .is_none_or(|debit| debit > maximum_debit)
100 {
101 return Err(EnsureWorkflowError::DriftedBeforeApply.into());
102 }
103 }
104 let mut retained = operator_mint::approve(paths, authority, digest)?;
105 if retained
106 .transfer_outcome
107 .as_ref()
108 .and_then(progress::transfer_block)
109 .is_none()
110 {
111 let argument = retained
112 .transfer_argument
113 .as_ref()
114 .ok_or(EnsureWorkflowError::OperatorMintReviewConflict)?;
115 let reply = transport.transfer(&retained.intent, argument).await?;
116 retained = operator_mint::record_transfer_reply(paths, authority, digest, &reply)?;
117 }
118 let Some(block) = retained
119 .transfer_outcome
120 .as_ref()
121 .and_then(progress::transfer_block)
122 else {
123 return Ok(retained);
124 };
125 let transfer = transport.read_transfer(&retained.intent, block).await?;
126 retained = operator_mint::prepare_notification(paths, authority, digest)?;
127 let notification = retained
128 .notification
129 .as_ref()
130 .ok_or(EnsureWorkflowError::OperatorMintReviewConflict)?;
131 if !matches!(
132 notification.outcome,
133 Some(
134 OperatorMintNotificationOutcomeRecord::Minted { .. }
135 | OperatorMintNotificationOutcomeRecord::Refunded { .. }
136 | OperatorMintNotificationOutcomeRecord::TransactionTooOld { .. }
137 | OperatorMintNotificationOutcomeRecord::InvalidTransaction { .. }
138 )
139 ) {
140 let reply = transport.notify(&transfer, ¬ification.argument).await?;
141 retained = operator_mint::record_notification_reply(paths, authority, digest, &reply)?;
142 }
143 let Some(outcome @ OperatorMintNotificationOutcomeRecord::Minted { .. }) = retained
144 .notification
145 .as_ref()
146 .and_then(|n| n.outcome.as_ref())
147 else {
148 return Ok(retained);
149 };
150 let deposit = transport.read_deposit(&retained.intent, outcome).await?;
151 Ok(operator_mint::admit_credit(
152 paths, authority, digest, &transfer, &deposit,
153 )?)
154}