canic_host/fleet_ensure/ops/operator_mint/transport/
mod.rs

1//! Module: fleet_ensure::ops::operator_mint::transport
2//!
3//! Responsibility: bounded exact calls and authenticated receipt acquisition.
4//! Boundary: workflow must retain approval before any payment or notification.
5
6use crate::{
7    fleet_ensure::{
8        model::operator_mint::{OperatorMintIntentRecord, OperatorMintNotificationOutcomeRecord},
9        ops::operator_mint::{
10            receipts,
11            receipts::{
12                CyclesLedgerBlock, DepositVerificationError, ReceiptVerificationLimits,
13                VerifiedCyclesDeposit, icp,
14                icp::{
15                    IcpReadError, IcpReadLimits, IcpReadOutcome, IcpTransferRead,
16                    VerifiedIcpTransfer,
17                },
18            },
19        },
20        view::operator_mint::OperatorMintRateQuote,
21    },
22    icp::{IcpCli, IcpManagementCallError},
23};
24use candid::{CandidType, Nat, Principal};
25use ic_agent::{Agent, AgentError, agent::RequestStatusResponse};
26use icrc_ledger_types::icrc3::blocks::{GetBlocksRequest, ICRC3DataCertificate};
27use serde::{Deserialize, de::DeserializeOwned};
28use std::time::Duration;
29use thiserror::Error;
30
31#[derive(CandidType, Deserialize)]
32struct Rate {
33    timestamp_seconds: u64,
34    xdr_permyriad_per_icp: u64,
35}
36
37#[derive(CandidType, Deserialize)]
38struct RateReply {
39    data: Rate,
40}
41
42// These are bounded observation budgets, not a payment validity window.
43const RESPONSE_BYTES: usize = 4 * 1024 * 1024;
44const CALL_TIMEOUT: Duration = Duration::from_secs(120);
45const DEPOSIT_LIMITS: ReceiptVerificationLimits = ReceiptVerificationLimits {
46    certificate_bytes: 256 * 1024,
47    blocks: 1024,
48    value_nodes: 100_000,
49    value_bytes: RESPONSE_BYTES,
50    depth: 32,
51};
52const ICP_LIMITS: IcpReadLimits = IcpReadLimits {
53    certificate_bytes: RESPONSE_BYTES,
54    certificate_depth: 32,
55    reply_bytes: 1024 * 1024,
56    decoding_quota: 2_000_000,
57    skipping_quota: 100_000,
58};
59
60///
61/// OperatorMintTransportError
62///
63/// Ops-owned bounded observation failure; none permits a replacement payment.
64///
65#[derive(Debug, Error)]
66pub enum OperatorMintTransportError {
67    #[error(transparent)]
68    Agent(#[from] AgentError),
69    #[error("operator mint response exceeded its observation budget")]
70    BudgetExceeded,
71    #[error("operator mint response encoding or decoding failed")]
72    Candid(#[from] candid::Error),
73    #[error(transparent)]
74    Deposit(#[from] DepositVerificationError),
75    #[error(transparent)]
76    Identity(Box<IcpManagementCallError>),
77    #[error(transparent)]
78    Icp(Box<IcpReadError>),
79    #[error("operator mint receipt remains unavailable; retain the original intent")]
80    ReceiptUnavailable,
81    #[error("operator mint async runtime could not start")]
82    Runtime(#[from] std::io::Error),
83    #[error("operator mint selected network or signer differs from the retained review")]
84    ReaderMismatch,
85    #[error("operator mint response could not be serialized for verification")]
86    Serialization,
87    #[error("operator mint call timed out; its outcome remains unresolved")]
88    Timeout,
89    #[error(transparent)]
90    Wire(#[from] crate::fleet_ensure::ops::operator_mint::OperatorMintWireError),
91}
92
93impl From<IcpManagementCallError> for OperatorMintTransportError {
94    fn from(error: IcpManagementCallError) -> Self {
95        Self::Identity(Box::new(error))
96    }
97}
98
99impl From<IcpReadError> for OperatorMintTransportError {
100    fn from(error: IcpReadError) -> Self {
101        Self::Icp(Box::new(error))
102    }
103}
104
105///
106/// OperatorMintTransport
107///
108/// Ops-owned Agent fixed to the selected signer, network and HTTP byte budget.
109///
110pub struct OperatorMintTransport {
111    agent: Agent,
112}
113
114#[derive(CandidType, Deserialize)]
115struct BlocksReply {
116    blocks: Vec<CyclesLedgerBlock>,
117}
118
119impl OperatorMintTransport {
120    /// Read only the selected operator's default Cycles Ledger account.
121    pub fn operator_balance(&self, ledger: Principal) -> Result<u128, OperatorMintTransportError> {
122        tokio::runtime::Runtime::new()?.block_on(self.read_operator_balance(ledger))
123    }
124
125    /// Async observation of the same default account; not receipt evidence.
126    pub async fn read_operator_balance(
127        &self,
128        ledger: Principal,
129    ) -> Result<u128, OperatorMintTransportError> {
130        let account = icrc_ledger_types::icrc1::account::Account {
131            owner: self.operator()?,
132            subaccount: None,
133        };
134        let amount: Nat = self
135            .query(ledger, "icrc1_balance_of", candid::encode_one(account)?)
136            .await?;
137        (&amount.0)
138            .try_into()
139            .map_err(|_| OperatorMintTransportError::BudgetExceeded)
140    }
141    /// Read the advisory quote for a synchronous host command.
142    pub fn quote_blocking(
143        &self,
144        icp_ledger: Principal,
145        cmc: Principal,
146        cycles_ledger: Principal,
147    ) -> Result<OperatorMintRateQuote, OperatorMintTransportError> {
148        tokio::runtime::Runtime::new()?.block_on(self.quote(icp_ledger, cmc, cycles_ledger))
149    }
150    #[cfg(test)]
151    pub(in crate::fleet_ensure) const fn from_test_agent(agent: Agent) -> Self {
152        Self { agent }
153    }
154
155    /// Observe fees and an advisory CMC rate without submitting any payment.
156    pub async fn quote(
157        &self,
158        icp_ledger: Principal,
159        cmc: Principal,
160        cycles_ledger: Principal,
161    ) -> Result<OperatorMintRateQuote, OperatorMintTransportError> {
162        let rate: RateReply = self
163            .query(cmc, "get_icp_xdr_conversion_rate", candid::encode_args(())?)
164            .await?;
165        let transfer: Nat = self
166            .query(icp_ledger, "icrc1_fee", candid::encode_args(())?)
167            .await?;
168        let deposit: Nat = self
169            .query(cycles_ledger, "icrc1_fee", candid::encode_args(())?)
170            .await?;
171        Ok(OperatorMintRateQuote {
172            rate_timestamp_seconds: rate.data.timestamp_seconds,
173            xdr_permyriad_per_icp: rate.data.xdr_permyriad_per_icp,
174            transfer_fee_e8s: (&transfer.0)
175                .try_into()
176                .map_err(|_| OperatorMintTransportError::BudgetExceeded)?,
177            estimated_deposit_fee_cycles: (&deposit.0)
178                .try_into()
179                .map_err(|_| OperatorMintTransportError::BudgetExceeded)?,
180        })
181    }
182    /// Resolve the existing ICP identity and trusted network; no payment occurs.
183    pub fn from_icp(icp: &IcpCli) -> Result<Self, OperatorMintTransportError> {
184        Ok(Self {
185            agent: icp.authenticated_agent_with_response_limit(RESPONSE_BYTES)?,
186        })
187    }
188
189    /// Selected trust-anchor identity for the effect-free review.
190    #[must_use]
191    pub fn network_identity(&self) -> [u8; 32] {
192        receipts::network_identity_sha256(&self.agent.read_root_key())
193    }
194
195    /// Canonical identity of the same trust anchor used by this transport.
196    pub fn canonical_network_id(
197        &self,
198    ) -> Result<canic_contracts::ids::CanonicalNetworkId, OperatorMintTransportError> {
199        canic_contracts::ids::CanonicalNetworkId::from_der_root_trust_anchor(
200            &self.agent.read_root_key(),
201        )
202        .map_err(|_| OperatorMintTransportError::ReaderMismatch)
203    }
204
205    /// Selected signer, checked against the desired operator before reviewing.
206    pub fn operator(&self) -> Result<Principal, OperatorMintTransportError> {
207        self.agent
208            .get_principal()
209            .map_err(|_| OperatorMintTransportError::ReaderMismatch)
210    }
211
212    /// Submit one exact, already-approved ICP transfer argument.
213    pub async fn transfer(
214        &self,
215        intent: &OperatorMintIntentRecord,
216        argument: &[u8],
217    ) -> Result<Vec<u8>, OperatorMintTransportError> {
218        self.verify_reader(intent)?;
219        if argument != crate::fleet_ensure::ops::operator_mint::transfer_argument(intent)? {
220            return Err(OperatorMintTransportError::ReaderMismatch);
221        }
222        self.update(intent.authority.icp_ledger, "transfer", argument)
223            .await
224    }
225
226    /// Notify only for an authenticated ICP block and already-retained arguments.
227    pub async fn notify(
228        &self,
229        transfer: &VerifiedIcpTransfer,
230        argument: &[u8],
231    ) -> Result<Vec<u8>, OperatorMintTransportError> {
232        self.verify_reader(transfer.intent())?;
233        if argument
234            != crate::fleet_ensure::ops::operator_mint::notification_argument(
235                transfer.intent(),
236                transfer.block_index(),
237            )?
238        {
239            return Err(OperatorMintTransportError::ReaderMismatch);
240        }
241        self.update(
242            transfer.intent().authority.cmc,
243            "notify_mint_cycles",
244            argument,
245        )
246        .await
247    }
248
249    async fn update(
250        &self,
251        canister: Principal,
252        method: &str,
253        argument: &[u8],
254    ) -> Result<Vec<u8>, OperatorMintTransportError> {
255        Ok(tokio::time::timeout(
256            CALL_TIMEOUT,
257            self.agent
258                .update(&canister, method)
259                .with_arg(argument)
260                .call_and_wait(),
261        )
262        .await
263        .map_err(|_| OperatorMintTransportError::Timeout)??)
264    }
265
266    fn verify_reader(
267        &self,
268        intent: &OperatorMintIntentRecord,
269    ) -> Result<(), OperatorMintTransportError> {
270        if self.operator()? != intent.authority.operator
271            || self.network_identity() != intent.authority.network_identity_sha256
272        {
273            return Err(OperatorMintTransportError::ReaderMismatch);
274        }
275        Ok(())
276    }
277
278    /// Authenticate one ICP block with at most one Ledger-authorized archive hop.
279    pub async fn read_transfer(
280        &self,
281        intent: &OperatorMintIntentRecord,
282        block: u64,
283    ) -> Result<VerifiedIcpTransfer, OperatorMintTransportError> {
284        self.verify_reader(intent)?;
285        let read = icp::prepare_read(&self.agent, intent, block)?;
286        match self.read_icp(&read).await? {
287            IcpReadOutcome::Transfer(transfer) => Ok(transfer),
288            IcpReadOutcome::Archive(archive) => {
289                let read = icp::prepare_archive_read(&self.agent, &archive)?;
290                match self.read_icp(&read).await? {
291                    IcpReadOutcome::Transfer(transfer) => Ok(transfer),
292                    IcpReadOutcome::Archive(_) => {
293                        Err(OperatorMintTransportError::ReceiptUnavailable)
294                    }
295                }
296            }
297        }
298    }
299
300    async fn read_icp(
301        &self,
302        read: &IcpTransferRead,
303    ) -> Result<IcpReadOutcome, OperatorMintTransportError> {
304        tokio::time::timeout(CALL_TIMEOUT, async {
305            self.agent
306                .update_signed(read.canister(), read.signed_request().to_vec())
307                .await?;
308            loop {
309                let (status, certificate) = self
310                    .agent
311                    .request_status_raw(read.request_id(), read.canister())
312                    .await?;
313                match status {
314                    RequestStatusResponse::Replied(_) => {
315                        let mut bytes = Vec::new();
316                        ciborium::into_writer(&certificate, &mut bytes)
317                            .map_err(|_| OperatorMintTransportError::Serialization)?;
318                        return Ok(icp::verify_read(&self.agent, read, &bytes, ICP_LIMITS)?);
319                    }
320                    RequestStatusResponse::Rejected(_) | RequestStatusResponse::Done => {
321                        return Err(OperatorMintTransportError::ReceiptUnavailable);
322                    }
323                    _ => tokio::time::sleep(Duration::from_secs(1)).await,
324                }
325            }
326        })
327        .await
328        .map_err(|_| OperatorMintTransportError::Timeout)?
329    }
330
331    /// Fetch a bounded certified tip-to-deposit chain; incomplete history stays unresolved.
332    pub async fn read_deposit(
333        &self,
334        intent: &OperatorMintIntentRecord,
335        outcome: &OperatorMintNotificationOutcomeRecord,
336    ) -> Result<VerifiedCyclesDeposit, OperatorMintTransportError> {
337        self.verify_reader(intent)?;
338        let OperatorMintNotificationOutcomeRecord::Minted {
339            deposit_block_index,
340            ..
341        } = outcome
342        else {
343            return Err(OperatorMintTransportError::ReceiptUnavailable);
344        };
345        let ledger = intent.authority.cycles_ledger;
346        let evidence: Option<ICRC3DataCertificate> = self
347            .query(
348                ledger,
349                "icrc3_get_tip_certificate",
350                candid::encode_args(())?,
351            )
352            .await?;
353        let evidence = evidence.ok_or(OperatorMintTransportError::ReceiptUnavailable)?;
354        let tip = receipts::certified_tip_index(&self.agent, intent, &evidence, DEPOSIT_LIMITS)?;
355        let count = tip
356            .checked_sub(*deposit_block_index)
357            .and_then(|n| n.checked_add(1))
358            .filter(|count| *count <= DEPOSIT_LIMITS.blocks as u128)
359            .ok_or(OperatorMintTransportError::BudgetExceeded)?;
360        let reply: BlocksReply = self
361            .query(
362                ledger,
363                "icrc3_get_blocks",
364                candid::encode_one(vec![GetBlocksRequest {
365                    start: Nat::from(*deposit_block_index),
366                    length: Nat::from(count),
367                }])?,
368            )
369            .await?;
370        let mut blocks = reply.blocks;
371        blocks.sort_by(|a, b| b.id.cmp(&a.id));
372        Ok(receipts::verify_cycles_deposit(
373            &self.agent,
374            intent,
375            outcome,
376            &evidence,
377            &blocks,
378            DEPOSIT_LIMITS,
379        )?)
380    }
381
382    async fn query<T: CandidType + DeserializeOwned>(
383        &self,
384        canister: Principal,
385        method: &str,
386        args: Vec<u8>,
387    ) -> Result<T, OperatorMintTransportError> {
388        let bytes = tokio::time::timeout(
389            CALL_TIMEOUT,
390            self.agent.query(&canister, method).with_arg(args).call(),
391        )
392        .await
393        .map_err(|_| OperatorMintTransportError::Timeout)??;
394        let mut config = candid::de::DecoderConfig::new();
395        config
396            .set_decoding_quota(2_000_000)
397            .set_skipping_quota(100_000)
398            .set_max_type_len(10_000);
399        Ok(candid::utils::decode_one_with_config(&bytes, &config)?)
400    }
401}
canic_host/fleet_ensure/ops/operator_mint/transport/mod.rs

canic_host/fleet_ensure/ops/operator_mint/transport/
mod.rs
