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            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)?;
91        }
92        let mut subscriptions = std::collections::BTreeMap::new();
93        let mut early = std::collections::BTreeMap::new();
94        while let Some(message) = socket.next().await {
95            let message = message.map_err(transport)?;
96            let value: Value = match message {
97                Message::Text(text) => serde_json::from_str(&text)
98                    .map_err(|error| ChainError::Confirmation(error.to_string()))?,
99                Message::Ping(data) => {
100                    socket.send(Message::Pong(data)).await.map_err(transport)?;
101                    continue;
102                }
103                Message::Close(_) => break,
104                _ => continue,
105            };
106            if value.get("error").is_some() {
107                break;
108            }
109            if let (Some(id), Some(subscription)) = (value["id"].as_u64(), value["result"].as_u64())
110            {
111                let index = usize::try_from(id).map_err(|_| ChainError::InvalidAccountData)?;
112                if index >= signatures.len() {
113                    return Err(ChainError::InvalidAccountData);
114                }
115                if let Some(status) = early.remove(&subscription) {
116                    results[index] = status;
117                } else {
118                    subscriptions.insert(subscription, index);
119                }
120            } else if value["method"] == "signatureNotification" {
121                let subscription = value["params"]["subscription"]
122                    .as_u64()
123                    .ok_or(ChainError::InvalidAccountData)?;
124                let error = value["params"]["result"]["value"]
125                    .get("err")
126                    .ok_or(ChainError::InvalidAccountData)?;
127                let status = if error.is_null() {
128                    ConfirmationStatus::Confirmed
129                } else {
130                    ConfirmationStatus::Failed(error.to_string())
131                };
132                if let Some(index) = subscriptions.remove(&subscription) {
133                    results[index] = status;
134                } else {
135                    if early.len() >= signatures.len() {
136                        return Err(ChainError::InvalidAccountData);
137                    }
138                    early.insert(subscription, status);
139                }
140            }
141            if results
142                .iter()
143                .all(|status| *status != ConfirmationStatus::Pending)
144            {
145                break;
146            }
147        }
148        Ok(())
149    }
150}
151
152#[cfg(all(test, feature = "ws-client"))]
153#[path = "confirmation_tests.rs"]
154mod tests;
acta_maker_sdk/chain/confirmation.rs

acta_maker_sdk/chain/
confirmation.rs
