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, sync::Arc, 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    #[error("query HTTP client setup failed: {0}")]
49    Client(#[source] reqwest::Error),
50}
51
52/// Connection pool and one IO worker shared by a command's cloned query contexts.
53/// Contains no signer, network authority, Agent or query response cache.
54#[derive(Debug)]
55pub(super) struct QueryTransport {
56    client: reqwest::Client,
57    runtime: Option<tokio::runtime::Runtime>,
58}
59
60impl QueryTransport {
61    fn new() -> Result<Self, IcpQueryError> {
62        // Match ic-agent's native client defaults; query deadlines remain tighter.
63        let client = reqwest::Client::builder()
64            .use_rustls_tls()
65            .timeout(Duration::from_secs(360))
66            .build()
67            .map_err(IcpQueryError::Client)?;
68        let runtime = tokio::runtime::Builder::new_multi_thread()
69            .worker_threads(1)
70            .thread_name("canic-query")
71            .enable_all()
72            .build()
73            .map_err(IcpQueryError::Runtime)?;
74        Ok(Self {
75            client,
76            runtime: Some(runtime),
77        })
78    }
79
80    fn block_on<F: Future>(&self, future: F) -> F::Output {
81        self.runtime
82            .as_ref()
83            .expect("live query transport")
84            .block_on(future)
85    }
86}
87
88impl Drop for QueryTransport {
89    fn drop(&mut self) {
90        // IcpCli may be owned by an async caller. No query can still borrow this
91        // final Arc; shut down idle IO without blocking that caller's runtime.
92        if let Some(runtime) = self.runtime.take() {
93            runtime.shutdown_background();
94        }
95    }
96}
97
98impl IcpCli {
99    fn query_transport(&self) -> Result<Arc<QueryTransport>, IcpQueryError> {
100        let mut retained = self.query_transport.lock().expect("query transport lock");
101        if let Some(transport) = retained.as_ref() {
102            return Ok(Arc::clone(transport));
103        }
104        let transport = Arc::new(QueryTransport::new()?);
105        *retained = Some(Arc::clone(&transport));
106        drop(retained);
107        Ok(transport)
108    }
109
110    /// Query one immutable request with at most three logical attempts in 30 seconds
111    /// after local identity/network resolution. Agent-internal HTTP retries and
112    /// certificate reads share that deadline but are not separate logical attempts.
113    /// The same verified signer, network, target, method and argument bind every attempt.
114    pub(crate) fn query_candid_readonly<I: CandidType, O: CandidType + DeserializeOwned>(
115        &self,
116        canister: Principal,
117        method: &str,
118        input: &I,
119    ) -> Result<O, IcpQueryError> {
120        self.measure_request(
121            crate::icp::IcpRequestKind::AgentQuery,
122            Some(&canister.to_text()),
123            Some(method),
124            || {
125                let argument = candid::encode_one(input).map_err(IcpQueryError::Encode)?;
126                let transport = self.query_transport()?;
127                let agent = self
128                    .build_authenticated_agent(
129                        Agent::builder()
130                            .with_http_client(transport.client.clone())
131                            .with_max_response_body_size(RESPONSE_BYTES),
132                    )
133                    .map_err(|error| IcpQueryError::Authority(Box::new(error)))?;
134                let bytes = query_bytes(self, &agent, canister, method, &argument)?;
135                let mut config = candid::de::DecoderConfig::new();
136                config.set_decoding_quota(RESPONSE_BYTES * 64);
137                config.set_skipping_quota(RESPONSE_BYTES);
138                candid::utils::decode_one_with_config(&bytes, &config)
139                    .map_err(IcpQueryError::Decode)
140            },
141        )
142    }
143}
144
145fn query_bytes(
146    icp: &IcpCli,
147    agent: &Agent,
148    canister: Principal,
149    method: &str,
150    argument: &[u8],
151) -> Result<Vec<u8>, IcpQueryError> {
152    icp.query_transport()?.block_on(async {
153        tokio::time::timeout(
154            TOTAL_TIMEOUT,
155            retry_query(|| async {
156                icp.record_remote_call();
157                tokio::time::timeout(
158                    ATTEMPT_TIMEOUT,
159                    agent.query(&canister, method).with_arg(argument).call(),
160                )
161                .await
162                .map_err(|_| IcpQueryError::Deadline)?
163                .map_err(|error| IcpQueryError::Agent(Box::new(error)))
164            }),
165        )
166        .await
167        .map_err(|_| IcpQueryError::Deadline)?
168    })
169}
170
171async fn retry_query<T, F, Fut>(mut query: F) -> Result<T, IcpQueryError>
172where
173    F: FnMut() -> Fut,
174    Fut: Future<Output = Result<T, IcpQueryError>>,
175{
176    for attempt in 0..ATTEMPTS {
177        match query().await {
178            Err(error) if attempt + 1 < ATTEMPTS && transient(&error) => {
179                tokio::time::sleep(Duration::from_millis(250 << attempt)).await;
180            }
181            result => return result,
182        }
183    }
184    unreachable!("the last attempt always returns")
185}
186
187fn transient(error: &IcpQueryError) -> bool {
188    match error {
189        IcpQueryError::Deadline => true,
190        IcpQueryError::Agent(error) => match error.as_ref() {
191            AgentError::TimeoutWaitingForResponse() => true,
192            AgentError::HttpError(error) => matches!(error.status, 408 | 429 | 502 | 503 | 504),
193            AgentError::TransportError(TransportError::Reqwest(error)) => {
194                error.is_timeout() || error.is_connect() || error.is_body()
195            }
196            _ => false,
197        },
198        _ => false,
199    }
200}
canic_host/icp/query/mod.rs

canic_host/icp/query/
mod.rs
