canic_host/icp/query/
mod.rs

1//! Module: icp::query
2//!
3//! Responsibility: bounded authenticated reads with typed transient retries.
4//! Boundary: this transport exposes queries only; identity and network remain ICP-owned.
5
6#[cfg(test)]
7mod tests;
8
9use crate::icp::{IcpCli, IcpManagementCallError};
10use candid::{CandidType, Principal};
11use ic_agent::{Agent, AgentError, agent::agent_error::TransportError};
12use serde::de::DeserializeOwned;
13use std::{future::Future, time::Duration};
14use thiserror::Error;
15
16// Bound the transport envelope, not the logical contents of application storage.
17const RESPONSE_BYTES: usize = 8 * 1024 * 1024;
18const ATTEMPTS: usize = 3;
19const ATTEMPT_TIMEOUT: Duration = Duration::from_secs(10);
20const TOTAL_TIMEOUT: Duration = Duration::from_secs(30);
21
22///
23/// IcpQueryError
24///
25/// Transport-owned authenticated read failure. Application rejections and decode
26/// failures never authorize a retry.
27///
28#[derive(Debug, Error)]
29pub enum IcpQueryError {
30    #[error("authenticated query failed: {0}")]
31    Agent(#[source] Box<AgentError>),
32
33    #[error("query identity/network resolution failed: {0}")]
34    Authority(#[source] Box<IcpManagementCallError>),
35
36    #[error("authenticated query exceeded its bounded read deadline")]
37    Deadline,
38
39    #[error("query response decoding failed: {0}")]
40    Decode(#[source] candid::Error),
41
42    #[error("query argument encoding failed: {0}")]
43    Encode(#[source] candid::Error),
44
45    #[error("query runtime failed: {0}")]
46    Runtime(#[source] std::io::Error),
47}
48
49impl IcpCli {
50    /// Query one immutable request with at most three logical attempts in 30 seconds
51    /// after local identity/network resolution. Agent-internal HTTP retries and
52    /// certificate reads share that deadline but are not separate logical attempts.
53    /// The same verified signer, network, target, method and argument bind every attempt.
54    pub(crate) fn query_candid_readonly<I: CandidType, O: CandidType + DeserializeOwned>(
55        &self,
56        canister: Principal,
57        method: &str,
58        input: &I,
59    ) -> Result<O, IcpQueryError> {
60        self.measure_request(
61            crate::icp::IcpRequestKind::AgentQuery,
62            Some(&canister.to_text()),
63            Some(method),
64            || {
65                let argument = candid::encode_one(input).map_err(IcpQueryError::Encode)?;
66                let agent = self
67                    .authenticated_agent_with_response_limit(RESPONSE_BYTES)
68                    .map_err(|error| IcpQueryError::Authority(Box::new(error)))?;
69                let bytes = query_bytes(self, &agent, canister, method, &argument)?;
70                let mut config = candid::de::DecoderConfig::new();
71                config.set_decoding_quota(RESPONSE_BYTES * 64);
72                config.set_skipping_quota(RESPONSE_BYTES);
73                candid::utils::decode_one_with_config(&bytes, &config)
74                    .map_err(IcpQueryError::Decode)
75            },
76        )
77    }
78}
79
80fn query_bytes(
81    icp: &IcpCli,
82    agent: &Agent,
83    canister: Principal,
84    method: &str,
85    argument: &[u8],
86) -> Result<Vec<u8>, IcpQueryError> {
87    let runtime = tokio::runtime::Builder::new_current_thread()
88        .enable_all()
89        .build()
90        .map_err(IcpQueryError::Runtime)?;
91    runtime.block_on(async {
92        tokio::time::timeout(
93            TOTAL_TIMEOUT,
94            retry_query(|| async {
95                icp.record_remote_call();
96                tokio::time::timeout(
97                    ATTEMPT_TIMEOUT,
98                    agent.query(&canister, method).with_arg(argument).call(),
99                )
100                .await
101                .map_err(|_| IcpQueryError::Deadline)?
102                .map_err(|error| IcpQueryError::Agent(Box::new(error)))
103            }),
104        )
105        .await
106        .map_err(|_| IcpQueryError::Deadline)?
107    })
108}
109
110async fn retry_query<T, F, Fut>(mut query: F) -> Result<T, IcpQueryError>
111where
112    F: FnMut() -> Fut,
113    Fut: Future<Output = Result<T, IcpQueryError>>,
114{
115    for attempt in 0..ATTEMPTS {
116        match query().await {
117            Err(error) if attempt + 1 < ATTEMPTS && transient(&error) => {
118                tokio::time::sleep(Duration::from_millis(250 << attempt)).await;
119            }
120            result => return result,
121        }
122    }
123    unreachable!("the last attempt always returns")
124}
125
126fn transient(error: &IcpQueryError) -> bool {
127    match error {
128        IcpQueryError::Deadline => true,
129        IcpQueryError::Agent(error) => match error.as_ref() {
130            AgentError::TimeoutWaitingForResponse() => true,
131            AgentError::HttpError(error) => matches!(error.status, 408 | 429 | 502 | 503 | 504),
132            AgentError::TransportError(TransportError::Reqwest(error)) => {
133                error.is_timeout() || error.is_connect() || error.is_body()
134            }
135            _ => false,
136        },
137        _ => false,
138    }
139}
canic_host/icp/query/mod.rs

canic_host/icp/query/
mod.rs
