evm_oracle_state/pending/
ethereum.rs

1//! Ethereum public pending-transaction source using Alchemy's full-body stream.
2
3use std::{fmt, time::Duration};
4
5use alloy_consensus::Transaction as _;
6use alloy_eips::eip2718::Encodable2718;
7use alloy_network::TransactionResponse;
8use alloy_primitives::{Bytes, keccak256};
9use futures_util::{SinkExt, StreamExt};
10use serde_json::{Value, json};
11use tokio::sync::watch;
12use tokio_tungstenite::{connect_async, tungstenite::Message};
13
14use super::{
15    ETHEREUM_MAINNET_CHAIN_ID, PendingOracleCandidateSource, PendingOracleOrderingHandle,
16    PendingOracleSource, PendingOracleSourceDescriptor, PendingOracleSourceError,
17    PendingOracleSourceFuture, PendingOracleSourceId, PendingOracleSourceSink,
18    PendingOracleTransmissionId, PendingTransportCandidate,
19};
20
21/// Alchemy WebSocket source for full Ethereum pending transaction bodies.
22#[derive(Clone)]
23pub struct AlchemyPendingTransactionSource {
24    ws_url: String,
25    source_id: PendingOracleSourceId,
26    chain_id: u64,
27}
28
29impl AlchemyPendingTransactionSource {
30    /// Construct a source for an Alchemy WebSocket RPC URL.
31    pub fn new(ws_url: impl Into<String>) -> Self {
32        Self {
33            ws_url: ws_url.into(),
34            source_id: PendingOracleSourceId::new("alchemy-pending-transactions"),
35            chain_id: ETHEREUM_MAINNET_CHAIN_ID,
36        }
37    }
38
39    /// Override the stable source identifier used in health and provenance.
40    pub fn source_id(mut self, source_id: PendingOracleSourceId) -> Self {
41        self.source_id = source_id;
42        self
43    }
44
45    /// Set the declared chain id. It must match the pending runtime configuration.
46    pub fn chain_id(mut self, chain_id: u64) -> Self {
47        self.chain_id = chain_id;
48        self
49    }
50
51    async fn run_forever(
52        self,
53        sink: PendingOracleSourceSink,
54        mut shutdown: watch::Receiver<bool>,
55    ) -> Result<(), PendingOracleSourceError> {
56        let mut retry = Duration::from_secs(1);
57        loop {
58            if *shutdown.borrow() {
59                return Ok(());
60            }
61            match self.run_connection(&sink, &mut shutdown).await {
62                Ok(()) if *shutdown.borrow() => return Ok(()),
63                Ok(()) => sink.coverage_gap("Alchemy pending WebSocket ended"),
64                Err(error) => sink.coverage_gap(error.to_string()),
65            }
66            sink.reconnecting();
67            tokio::select! {
68                changed = shutdown.changed() => {
69                    if changed.is_err() || *shutdown.borrow() {
70                        return Ok(());
71                    }
72                }
73                () = tokio::time::sleep(retry) => {}
74            }
75            retry = (retry * 2).min(Duration::from_secs(30));
76        }
77    }
78
79    async fn run_connection(
80        &self,
81        sink: &PendingOracleSourceSink,
82        shutdown: &mut watch::Receiver<bool>,
83    ) -> Result<(), PendingOracleSourceError> {
84        let (socket, _) = connect_async(self.ws_url.as_str())
85            .await
86            .map_err(transport_error)?;
87        let (mut writer, mut reader) = socket.split();
88        writer
89            .send(Message::Text(
90                json!({
91                    "jsonrpc": "2.0",
92                    "id": 1,
93                    "method": "eth_subscribe",
94                    "params": ["alchemy_pendingTransactions", {"hashesOnly": false}],
95                })
96                .to_string()
97                .into(),
98            ))
99            .await
100            .map_err(transport_error)?;
101        writer
102            .send(Message::Text(
103                json!({
104                    "jsonrpc": "2.0",
105                    "id": 2,
106                    "method": "eth_subscribe",
107                    "params": ["newHeads"],
108                })
109                .to_string()
110                .into(),
111            ))
112            .await
113            .map_err(transport_error)?;
114
115        let mut pending_subscription = None;
116        let mut head_subscription = None;
117        let mut current_head = None;
118        while pending_subscription.is_none() || head_subscription.is_none() {
119            let message = tokio::time::timeout(Duration::from_secs(15), reader.next())
120                .await
121                .map_err(|_| {
122                    PendingOracleSourceError::Transport(
123                        "timed out waiting for Alchemy subscription acknowledgements".to_string(),
124                    )
125                })?
126                .ok_or_else(|| {
127                    PendingOracleSourceError::Transport(
128                        "Alchemy WebSocket closed during subscription".to_string(),
129                    )
130                })?
131                .map_err(transport_error)?;
132            let Some(value) = message_json(message)? else {
133                continue;
134            };
135            if let Some(error) = value.get("error") {
136                return Err(PendingOracleSourceError::Transport(format!(
137                    "Alchemy subscription failed: {error}"
138                )));
139            }
140            match value.get("id").and_then(Value::as_u64) {
141                Some(1) => pending_subscription = value_string(&value, "result"),
142                Some(2) => head_subscription = value_string(&value, "result"),
143                _ => {}
144            }
145        }
146        sink.ready();
147
148        loop {
149            tokio::select! {
150                changed = shutdown.changed() => {
151                    if changed.is_err() || *shutdown.borrow() {
152                        let _ = writer.close().await;
153                        return Ok(());
154                    }
155                }
156                message = reader.next() => {
157                    let Some(message) = message else {
158                        return Err(PendingOracleSourceError::Transport(
159                            "Alchemy pending WebSocket closed".to_string(),
160                        ));
161                    };
162                    let message = message.map_err(transport_error)?;
163                    let Some(value) = message_json(message)? else {
164                        continue;
165                    };
166                    sink.transport_message();
167                    let Some(params) = value.get("params") else {
168                        continue;
169                    };
170                    let subscription = params.get("subscription").and_then(Value::as_str);
171                    let result = params.get("result");
172                    if subscription == head_subscription.as_deref() {
173                        current_head = result
174                            .and_then(|header| header.get("number"))
175                            .and_then(Value::as_str)
176                            .and_then(parse_quantity);
177                    } else if subscription == pending_subscription.as_deref() {
178                        let Some(result) = result else {
179                            continue;
180                        };
181                        let Some(candidate) = self.transaction_candidate(result.clone(), current_head)
182                        else {
183                            continue;
184                        };
185                        if sink
186                            .runtime()
187                            .interests()
188                            .iter()
189                            .any(|interest| interest.matches(&candidate))
190                        {
191                            sink.candidate();
192                            match sink.runtime().observe_candidate(candidate) {
193                                Ok(report) => {
194                                    for failure in report.failures {
195                                        tracing::debug!(
196                                            adapter_id = %failure.adapter_id,
197                                            error = %failure.error,
198                                            "pending oracle adapter rejected Alchemy candidate"
199                                        );
200                                    }
201                                }
202                                Err(error) => {
203                                    tracing::debug!(%error, "pending oracle runtime rejected Alchemy candidate");
204                                }
205                            }
206                        }
207                    }
208                }
209            }
210        }
211    }
212
213    fn transaction_candidate(
214        &self,
215        value: Value,
216        observed_at_head: Option<u64>,
217    ) -> Option<PendingTransportCandidate> {
218        let transaction: alloy_rpc_types_eth::Transaction = serde_json::from_value(value).ok()?;
219        let to = transaction.to()?;
220        let calldata = transaction.input().clone();
221        if calldata.len() < 4 {
222            return None;
223        }
224        let tx_hash = transaction.tx_hash();
225        let signed_envelope = Bytes::from(transaction.inner.inner().encoded_2718());
226        let ordering = if keccak256(&signed_envelope) == tx_hash {
227            PendingOracleOrderingHandle::RawEthereumTransaction {
228                tx_hash,
229                signed_envelope,
230            }
231        } else {
232            PendingOracleOrderingHandle::TransactionHashOnly { tx_hash }
233        };
234        Some(PendingTransportCandidate::new(
235            self.chain_id,
236            PendingOracleTransmissionId::from_hash(tx_hash),
237            PendingOracleSource::PublicMempool,
238            self.source_id.clone(),
239            to,
240            calldata,
241            ordering,
242            observed_at_head,
243        ))
244    }
245}
246
247impl fmt::Debug for AlchemyPendingTransactionSource {
248    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
249        formatter
250            .debug_struct("AlchemyPendingTransactionSource")
251            .field("ws_url", &"<redacted>")
252            .field("source_id", &self.source_id)
253            .field("chain_id", &self.chain_id)
254            .finish()
255    }
256}
257
258impl PendingOracleCandidateSource for AlchemyPendingTransactionSource {
259    fn descriptor(&self) -> PendingOracleSourceDescriptor {
260        PendingOracleSourceDescriptor::new(
261            PendingOracleSource::PublicMempool,
262            self.source_id.clone(),
263        )
264    }
265
266    fn run(
267        self: Box<Self>,
268        sink: PendingOracleSourceSink,
269        shutdown: watch::Receiver<bool>,
270    ) -> PendingOracleSourceFuture {
271        Box::pin(async move { self.run_forever(sink, shutdown).await })
272    }
273}
274
275fn value_string(value: &Value, key: &str) -> Option<String> {
276    value.get(key).and_then(Value::as_str).map(str::to_owned)
277}
278
279fn parse_quantity(value: &str) -> Option<u64> {
280    u64::from_str_radix(value.strip_prefix("0x")?, 16).ok()
281}
282
283fn message_json(message: Message) -> Result<Option<Value>, PendingOracleSourceError> {
284    match message {
285        Message::Text(text) => serde_json::from_str(text.as_str())
286            .map(Some)
287            .map_err(transport_error),
288        Message::Binary(bytes) => serde_json::from_slice(&bytes)
289            .map(Some)
290            .map_err(transport_error),
291        Message::Close(_) => Err(PendingOracleSourceError::Transport(
292            "pending WebSocket closed".to_string(),
293        )),
294        Message::Ping(_) | Message::Pong(_) | Message::Frame(_) => Ok(None),
295    }
296}
297
298fn transport_error(error: impl fmt::Display) -> PendingOracleSourceError {
299    PendingOracleSourceError::Transport(error.to_string())
300}
301
302#[cfg(test)]
303mod tests {
304    use std::str::FromStr;
305
306    use alloy_primitives::B256;
307
308    use super::*;
309
310    #[test]
311    fn rpc_transaction_reencodes_to_its_declared_hash() {
312        let value = serde_json::from_str(
313            r#"{
314                "type":"0x2","chainId":"0x1","nonce":"0x84974","gas":"0x30d40",
315                "maxFeePerGas":"0x7a4e39a","maxPriorityFeePerGas":"0x1ef37be",
316                "to":"0xe73d53e3a982ab2750a0b76f9012e18b256cc243","value":"0x0",
317                "accessList":[],"input":"0x1249c58b",
318                "r":"0x7432676eb4f3b0f8e2c44044e6cf25e5421ac2d1d93412ccbec481e88b8102c1",
319                "s":"0x1800a3996abb76a262638c58c51693d410ca6b24b6a196851e36b1b4513953ab",
320                "yParity":"0x0","v":"0x0",
321                "hash":"0x483b92a2d30693c28bc812a1e6747ce2af8b04694ccf08c1c85ec67bb04962ca",
322                "blockHash":null,"blockNumber":null,"transactionIndex":null,
323                "from":"0x82a53178e7a7e454ab31eea6063fdca338418f74","gasPrice":"0x7a4e39a"
324            }"#,
325        )
326        .expect("transaction fixture JSON");
327        let candidate = AlchemyPendingTransactionSource::new("wss://redacted")
328            .transaction_candidate(value, Some(100))
329            .expect("full transaction candidate");
330        let expected =
331            B256::from_str("0x483b92a2d30693c28bc812a1e6747ce2af8b04694ccf08c1c85ec67bb04962ca")
332                .unwrap();
333        let PendingOracleOrderingHandle::RawEthereumTransaction {
334            tx_hash,
335            signed_envelope,
336        } = candidate.ordering
337        else {
338            panic!("expected exact signed transaction");
339        };
340        assert_eq!(tx_hash, expected);
341        assert_eq!(keccak256(signed_envelope), expected);
342    }
343}
evm_oracle_state/pending/ethereum.rs

evm_oracle_state/pending/
ethereum.rs
