acta_maker_sdk/chain/
confirmation.rs

1use solana_sdk::signature::Signature;
2
3use super::{ChainError, rpc::ChainClient};
4
5#[derive(Debug, Clone, PartialEq, Eq)]
6pub enum ConfirmationStatus {
7    Confirmed,
8    Failed(String),
9    /// Not observed at the requested commitment. Never proves failure or expiry.
10    Pending,
11}
12
13impl ChainClient {
14    /// One status read per chunk of at most 256 signatures, including history.
15    pub async fn signature_statuses(
16        &self,
17        signatures: &[Signature],
18    ) -> Result<Vec<ConfirmationStatus>, ChainError> {
19        let mut results = Vec::with_capacity(signatures.len());
20        for chunk in signatures.chunks(256) {
21            let statuses = self
22                .rpc()
23                .get_signature_statuses_with_history(chunk)
24                .await?
25                .value;
26            if statuses.len() != chunk.len() {
27                return Err(ChainError::InvalidAccountData);
28            }
29            results.extend(statuses.into_iter().map(|status| match status {
30                Some(status) if status.satisfies_commitment(self.rpc().commitment()) => {
31                    match status.err {
32                        Some(error) => ConfirmationStatus::Failed(error.to_string()),
33                        None => ConfirmationStatus::Confirmed,
34                    }
35                }
36                _ => ConfirmationStatus::Pending,
37            }));
38        }
39        Ok(results)
40    }
41
42    /// Wait on one Solana WS connection, then reconcile unresolved signatures in one
43    /// chunked status pass on disconnect, subscription error, or deadline. No polling.
44    #[cfg(feature = "ws-client")]
45    pub async fn confirm_signatures(
46        &self,
47        ws_url: &str,
48        signatures: &[Signature],
49        timeout: std::time::Duration,
50    ) -> Result<Vec<ConfirmationStatus>, ChainError> {
51        let mut results = vec![ConfirmationStatus::Pending; signatures.len()];
52        if signatures.is_empty() {
53            return Ok(results);
54        }
55        let _ = tokio::time::timeout(
56            timeout,
57            self.subscribe_signatures(ws_url, signatures, &mut results),
58        )
59        .await;
60        let unresolved: Vec<_> = results
61            .iter()
62            .enumerate()
63            .filter_map(|(i, status)| (*status == ConfirmationStatus::Pending).then_some(i))
64            .collect();
65        let pending: Vec<_> = unresolved.iter().map(|i| signatures[*i]).collect();
66        for (index, status) in unresolved
67            .into_iter()
68            .zip(self.signature_statuses(&pending).await?)
69        {
70            results[index] = status;
71        }
72        Ok(results)
73    }
74
75    #[cfg(feature = "ws-client")]
76    async fn subscribe_signatures(
77        &self,
78        ws_url: &str,
79        signatures: &[Signature],
80        results: &mut [ConfirmationStatus],
81    ) -> Result<(), ChainError> {
82        use futures_util::{SinkExt, StreamExt};
83        use serde_json::{Value, json};
84        use tokio_tungstenite::{connect_async, tungstenite::Message};
85        let transport = |error: tokio_tungstenite::tungstenite::Error| {
86            ChainError::Confirmation(error.to_string())
87        };
88        let (mut socket, _) = connect_async(ws_url).await.map_err(transport)?;
89        for (index, signature) in signatures.iter().enumerate() {
90            crate::io_observer::record(crate::io_observer::Event::Rpc {
91                method: "signatureSubscribe",
92            });
93            socket.send(Message::Text(json!({"jsonrpc":"2.0", "id":index, "method":"signatureSubscribe", "params":[signature.to_string(), {"commitment": self.rpc().commitment().commitment}]}).to_string().into())).await.map_err(transport)?;
94        }
95        let mut subscriptions = std::collections::BTreeMap::new();
96        let mut early = std::collections::BTreeMap::new();
97        while let Some(message) = socket.next().await {
98            let message = message.map_err(transport)?;
99            crate::io_observer::record(crate::io_observer::Event::Ws {
100                venue: "solana",
101                stream: "signatures",
102                bytes: message.len(),
103            });
104            let value: Value = match message {
105                Message::Text(text) => serde_json::from_str(&text)
106                    .map_err(|error| ChainError::Confirmation(error.to_string()))?,
107                Message::Ping(data) => {
108                    socket.send(Message::Pong(data)).await.map_err(transport)?;
109                    continue;
110                }
111                Message::Close(_) => break,
112                _ => continue,
113            };
114            if value.get("error").is_some() {
115                break;
116            }
117            if let (Some(id), Some(subscription)) = (value["id"].as_u64(), value["result"].as_u64())
118            {
119                let index = usize::try_from(id).map_err(|_| ChainError::InvalidAccountData)?;
120                if index >= signatures.len() {
121                    return Err(ChainError::InvalidAccountData);
122                }
123                if let Some(status) = early.remove(&subscription) {
124                    results[index] = status;
125                } else {
126                    subscriptions.insert(subscription, index);
127                }
128            } else if value["method"] == "signatureNotification" {
129                let subscription = value["params"]["subscription"]
130                    .as_u64()
131                    .ok_or(ChainError::InvalidAccountData)?;
132                let error = value["params"]["result"]["value"]
133                    .get("err")
134                    .ok_or(ChainError::InvalidAccountData)?;
135                let status = if error.is_null() {
136                    ConfirmationStatus::Confirmed
137                } else {
138                    ConfirmationStatus::Failed(error.to_string())
139                };
140                if let Some(index) = subscriptions.remove(&subscription) {
141                    results[index] = status;
142                } else {
143                    if early.len() >= signatures.len() {
144                        return Err(ChainError::InvalidAccountData);
145                    }
146                    early.insert(subscription, status);
147                }
148            }
149            if results
150                .iter()
151                .all(|status| *status != ConfirmationStatus::Pending)
152            {
153                break;
154            }
155        }
156        Ok(())
157    }
158}
159
160#[cfg(all(test, feature = "ws-client"))]
161#[path = "confirmation_tests.rs"]
162mod tests;
acta_maker_sdk/chain/confirmation.rs

acta_maker_sdk/chain/
confirmation.rs
