hopr_chain_connector/connector/
sequencer.rs

1use std::{
2    sync::atomic::AtomicU64,
3    time::{Duration, Instant},
4};
5
6use blokli_client::api::{BlokliQueryClient, BlokliTransactionClient};
7use futures::{FutureExt, SinkExt, StreamExt, TryFutureExt};
8use hopr_api::{
9    Address,
10    types::{
11        chain::prelude::{GasEstimation, SignableTransaction},
12        crypto::prelude::*,
13    },
14};
15
16use crate::{
17    errors::{self, ConnectorError},
18    utils::model_to_chain_info,
19};
20
21type TxRequest<T> = (
22    T,
23    Option<ChainKeypair>,
24    futures::channel::oneshot::Sender<errors::Result<blokli_client::api::TxId>>,
25);
26
27/// Takes care of sequencing transactions, nonce management, and submitting the transactions to the blockchain in-order.
28///
29/// This object is meant to be a singleton and cannot be cloned.
30pub struct TransactionSequencer<C, R> {
31    sender: futures::channel::mpsc::Sender<TxRequest<R>>,
32    client: std::sync::Arc<C>,
33}
34
35const TX_QUEUE_CAPACITY: usize = 2048;
36
37// How long until nonces for other signers (than node's own) expire
38const OTHER_SIGNER_NONCE_EXPIRATION: Duration = Duration::from_mins(5);
39
40struct FixedTti {
41    fixed: Address,
42    tti: std::time::Duration,
43}
44
45impl FixedTti {
46    #[inline]
47    fn duration_for(&self, key: &Address) -> Option<Duration> {
48        if key == &self.fixed {
49            None // expiration cleared => never expires by time
50        } else {
51            Some(self.tti) // standard time-to-idle
52        }
53    }
54}
55
56impl moka::Expiry<Address, std::sync::Arc<AtomicU64>> for FixedTti {
57    fn expire_after_create(&self, key: &Address, _: &std::sync::Arc<AtomicU64>, _: Instant) -> Option<Duration> {
58        self.duration_for(key)
59    }
60
61    fn expire_after_read(
62        &self,
63        key: &Address,
64        _: &std::sync::Arc<AtomicU64>,
65        _: Instant,
66        _: Option<Duration>,
67        _: Instant,
68    ) -> Option<Duration> {
69        self.duration_for(key)
70    }
71
72    fn expire_after_update(
73        &self,
74        key: &Address,
75        _: &std::sync::Arc<AtomicU64>,
76        _: Instant,
77        _: Option<Duration>,
78    ) -> Option<Duration> {
79        self.duration_for(key)
80    }
81}
82
83impl<C, R> TransactionSequencer<C, R>
84where
85    C: BlokliQueryClient + BlokliTransactionClient + Send + Sync + 'static,
86    R: SignableTransaction + Send + Sync + 'static,
87{
88    pub fn new(signer: ChainKeypair, client: std::sync::Arc<C>) -> Self {
89        tracing::debug!(signer = %signer.public().to_address(), "starting transaction sequencer");
90
91        let client_clone = client.clone();
92        let (sender, receiver) = futures::channel::mpsc::channel::<TxRequest<R>>(TX_QUEUE_CAPACITY);
93
94        // Nonce of our node signer never expires, all other signers expire after a few minutes
95        let current_nonce = moka::sync::CacheBuilder::new(1024)
96            .expire_after(FixedTti {
97                fixed: signer.public().to_address(),
98                tti: OTHER_SIGNER_NONCE_EXPIRATION,
99            })
100            .build();
101
102        let current_nonce_clone = current_nonce.clone();
103        hopr_utils::runtime::prelude::spawn(
104            receiver
105                .then(move |(tx, tx_signer, notifier): (R, _, _)| {
106                    let client = client_clone.clone();
107                    let signer = tx_signer.unwrap_or_else(|| signer.clone());
108                    let signer_addr = signer.public().to_address();
109                    let current_nonce = current_nonce.clone();
110                    async move {
111                        let chain_info = match client.query_chain_info().map_err(ConnectorError::from).await {
112                            Ok(chain_info) => {
113                                tracing::debug!(chain_id = chain_info.chain_id, "chain info retrieved for tx");
114                                chain_info
115                            }
116                            Err(e) => return (Err(e), signer_addr, notifier),
117                        };
118
119                        let parsed_chain_info = match model_to_chain_info(chain_info) {
120                            Ok(parsed_chain_info) => parsed_chain_info,
121                            Err(error) => return (Err(error), signer_addr, notifier),
122                        };
123                        let chain_id = parsed_chain_info.info.chain_id;
124                        let gas_estimation = GasEstimation::from(parsed_chain_info);
125                        tracing::debug!(?gas_estimation, "gas estimation used for tx");
126
127                        // We always query the transaction count for the signer and use
128                        // the maximum between this value and the local counter
129                        match client
130                            .query_transaction_count(&signer_addr.into())
131                            .map_err(ConnectorError::from)
132                            .await
133                        {
134                            Ok(tx_count) => {
135                                let prev_nonce = current_nonce
136                                    .entry(signer_addr)
137                                    .or_default()
138                                    .value()
139                                    .fetch_max(tx_count, std::sync::atomic::Ordering::Relaxed);
140
141                                tracing::debug!(prev_nonce, tx_count, "transaction count retrieved");
142                            }
143                            Err(e) => return (Err(e), signer_addr, notifier),
144                        }
145
146                        // At this point the value must be set, so we can load it
147                        let nonce = current_nonce
148                            .entry(signer_addr)
149                            .or_default()
150                            .value()
151                            .load(std::sync::atomic::Ordering::Relaxed);
152                        tracing::debug!(nonce, signer = %signer_addr, "nonce used for the tx");
153
154                        tx.sign_and_encode_to_eip2718(nonce, chain_id, gas_estimation.into(), &signer)
155                            .map_err(ConnectorError::from)
156                            .and_then(move |tx| {
157                                tracing::debug!(nonce, signer = %signer_addr, "submitting transaction");
158                                let client = client.clone();
159                                async move {
160                                    client
161                                        .submit_and_track_transaction(&tx)
162                                        .map_err(ConnectorError::from)
163                                        .await
164                                }
165                            })
166                            .map(|res| (res, signer_addr, notifier))
167                            .await
168                    }
169                })
170                .for_each(move |(res, signer_addr, notifier)| {
171                    // The nonce is incremented when the transaction succeeded or failed due to on-chain
172                    // rejection.
173                    if res.is_ok()
174                        || res
175                            .as_ref()
176                            .is_err_and(|error| error.as_transaction_rejection_error().is_some())
177                    {
178                        // Increment the nonce
179                        let prev_nonce = current_nonce_clone
180                            .entry(signer_addr)
181                            .or_default()
182                            .value()
183                            .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
184
185                        tracing::debug!(prev_nonce, signer = %signer_addr, ?res, "nonce incremented due to tx success or rejection");
186                    } else {
187                        tracing::warn!(?res, signer = %signer_addr, "nonce not incremented due to tx failure other than rejection");
188                    }
189
190                    if notifier.send(res).is_err() {
191                        tracing::debug!(
192                            "failed to notify transaction result - the caller may not want to await the result \
193                             anymore."
194                        );
195                    }
196                    futures::future::ready(())
197                })
198                .inspect(|_| tracing::warn!("transaction sequencer queue stopped")),
199        );
200
201        Self { sender, client }
202    }
203}
204
205impl<C, R> TransactionSequencer<C, R>
206where
207    C: BlokliTransactionClient + Send + Sync + 'static,
208{
209    /// Adds the transaction to the [`TransactionSequencer`] queue.
210    ///
211    /// The `timeout_until_finalized` is a total time until the TX is submitted and either confirmed or rejected.
212    pub async fn enqueue_transaction(
213        &self,
214        transaction: R,
215        timeout_until_finalized: std::time::Duration,
216        custom_signer: Option<ChainKeypair>,
217    ) -> errors::Result<impl Future<Output = errors::Result<blokli_client::api::types::Transaction>>> {
218        let (notifier_tx, notifier_rx) = futures::channel::oneshot::channel();
219
220        self.sender
221            .clone()
222            .send((transaction, custom_signer, notifier_tx))
223            .await
224            .map_err(|_| ConnectorError::InvalidState("transaction queue dropped"))?;
225
226        Ok(notifier_rx
227            .inspect_ok(|res| tracing::debug!(?res, "transaction tracking id received"))
228            .map(move |result| {
229                result
230                    .map_err(|_| ConnectorError::InvalidState("transaction notifier dropped"))
231                    .and_then(|tx_res| tx_res.map(|id| (id, timeout_until_finalized)))
232            })
233            .and_then(|(tx_id, timeout)| {
234                self.client
235                    .track_transaction(tx_id, timeout)
236                    .map_err(ConnectorError::from)
237                    .inspect(|res| tracing::debug!(?res, "transaction tracking done"))
238            }))
239    }
240}
241
242impl<C, R> Drop for TransactionSequencer<C, R> {
243    fn drop(&mut self) {
244        // Causes the internally spawned task that drives the queue to terminate
245        self.sender.close_channel();
246    }
247}
hopr_chain_connector/connector/sequencer.rs

hopr_chain_connector/connector/
sequencer.rs
