fix: serialize Goat transactions per signer and stop client panics and silent failures

crates/client/src/btc_chain/esplora_bitcoin_adaptor.rs

@@ -8,8 +8,15 @@ use std::future::Future;
 use std::time::Duration;
 use tracing::warn;
 
-const TEST_URL: &str = "https://mempool.space/testnet/api";
 const MAIN_URL: &str = "https://mempool.space/api";
+const TESTNET4_URL: &str = "https://mempool.space/testnet4/api";
+const SIGNET_URL: &str = "https://mempool.space/signet/api";
+const TESTNET3_URL: &str = "https://mempool.space/testnet/api";
+
+/// Retries esplora-client makes for 429 and 5xx answers. Its backoff starts at 256 ms
+/// and doubles, so three retries wait about 1.8 s in all and fit in the request
+/// timeout; its default of six waits about 16 s, longer than the default timeout.
+const ESPLORA_MAX_RETRIES: usize = 3;
 
 #[derive(Debug)]
 pub struct BtcRpcTimeoutError {
@@ -32,7 +39,10 @@ impl std::error::Error for BtcRpcTimeoutError {}
 pub fn get_esplora_url(network: Network) -> &'static str {
     match network {
         Network::Bitcoin => MAIN_URL,
-        _ => TEST_URL,
+        Network::Testnet4 => TESTNET4_URL,
+        Network::Signet => SIGNET_URL,
+        // testnet3; regtest has no public explorer and needs an explicit URL
+        _ => TESTNET3_URL,
     }
 }
 
@@ -47,6 +57,8 @@ impl EsploraBitcoinAdaptor {
         let timeout_secs = get_btc_request_timeout_secs();
         EsploraBitcoinAdaptor {
             esplora: Builder::new(esplora_url.unwrap_or(get_esplora_url(network)))
+                .timeout(timeout_secs)
+                .max_retries(ESPLORA_MAX_RETRIES)
                 .build_async()
                 .expect("Could not build esplora client"),
             network,
@@ -142,3 +154,16 @@ impl BitcoinAdaptor for EsploraBitcoinAdaptor {
         self.with_timeout("get_header_by_hash", self.esplora.get_header_by_hash(block_hash)).await
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn default_explorer_matches_the_network() {
+        assert_eq!(get_esplora_url(Network::Bitcoin), MAIN_URL);
+        assert_eq!(get_esplora_url(Network::Testnet4), TESTNET4_URL);
+        assert_eq!(get_esplora_url(Network::Signet), SIGNET_URL);
+        assert_eq!(get_esplora_url(Network::Testnet), TESTNET3_URL);
+    }
+}

crates/client/src/btc_chain/mod.rs

@@ -10,7 +10,6 @@ pub mod bitcoin_adaptor;
 pub mod bitcoin_chain;
 mod esplora_bitcoin_adaptor;
 pub use esplora_bitcoin_adaptor::BtcRpcTimeoutError;
-pub mod mempool_v1_type;
 mod mock_bitcoin_adaptor;
 
 #[derive(Debug)]

crates/client/src/goat_chain/goat_adaptor.rs

@@ -29,7 +29,9 @@ use alloy::{
 };
 use anyhow::{bail, format_err};
 use async_trait::async_trait;
+use std::collections::HashMap;
 use std::str::FromStr;
+use std::sync::{Arc, LazyLock, Mutex};
 use std::time::Duration;
 use tokio::time;
 use uuid::Uuid;
@@ -469,6 +471,23 @@ type StakeManagementInstance = IStakeManagementInstance<
     >,
 >;
 
+/// How long a send waits for its receipt before reporting failure.
+const RECEIPT_WAIT_SECS: u64 = 30;
+
+/// One lock per sender address, shared by every adaptor in the process. It is held
+/// from reading the nonce until the receipt arrives or the wait times out, so two
+/// clients with the same key never sign with the same nonce. The nonce comes from
+/// the latest block on purpose: with sends serialized it equals the pending nonce,
+/// except after a timeout, where reusing it replaces the stuck transaction instead
+/// of queueing a duplicate behind it.
+static SENDER_LOCKS: LazyLock<Mutex<HashMap<Address, Arc<tokio::sync::Mutex<()>>>>> =
+    LazyLock::new(Default::default);
+
+fn sender_lock(sender: Address) -> Arc<tokio::sync::Mutex<()>> {
+    let mut locks = SENDER_LOCKS.lock().unwrap_or_else(std::sync::PoisonError::into_inner);
+    locks.entry(sender).or_default().clone()
+}
+
 pub struct GoatAdaptor {
     chain_id: ChainId,
     signer: EthereumWallet,
@@ -542,16 +561,18 @@ impl GoatAdaptor {
         &self,
         tx_request: TransactionRequest,
     ) -> anyhow::Result<TxHash> {
-        let (tx_hash, _) = self.handle_transaction_request_with_wait(tx_request, 10).await?;
+        let (tx_hash, _) =
+            self.handle_transaction_request_with_wait(tx_request, RECEIPT_WAIT_SECS).await?;
         Ok(tx_hash)
     }
 
     async fn handle_transaction_request_eip1559(
         &self,
         tx_request: TransactionRequest,
     ) -> anyhow::Result<TxHash> {
-        let (tx_hash, _) =
-            self.handle_transaction_request_with_wait_eip1559(tx_request, 10).await?;
+        let (tx_hash, _) = self
+            .handle_transaction_request_with_wait_eip1559(tx_request, RECEIPT_WAIT_SECS)
+            .await?;
         Ok(tx_hash)
     }
 
@@ -583,11 +604,13 @@ impl GoatAdaptor {
         mut tx_request: TransactionRequest,
         max_wait_secs: u64,
     ) -> anyhow::Result<(TxHash, TransactionReceipt)> {
+        let sender = tx_request.from.ok_or_else(|| format_err!("transaction has no sender"))?;
+        let sender_lock = sender_lock(sender);
+        let _sending = sender_lock.lock().await;
         // update  gas price nonce gas_limit
         tx_request.gas_price = Some(self.provider.clone().get_gas_price().await?);
         tracing::info!("gas price: {}", tx_request.gas_price.unwrap());
-        tx_request.nonce =
-            Some(self.provider.clone().get_transaction_count(tx_request.from.unwrap()).await?);
+        tx_request.nonce = Some(self.provider.clone().get_transaction_count(sender).await?);
         tracing::info!("tx: {:?}", tx_request);
         tx_request.gas = Some(self.provider.clone().estimate_gas(tx_request.clone()).await?);
         tracing::info!("estimated gas: {:?}", tx_request.gas);
@@ -639,14 +662,20 @@ impl GoatAdaptor {
                 }
             };
         }
-        bail!("tx_hash:{tx_hash} receipt not found within {max_wait_secs} seconds")
+        bail!(
+            "tx_hash:{tx_hash} receipt not found within {max_wait_secs} seconds; it may still be \
+             pending, and the next transaction from this sender reuses its nonce and replaces it"
+        )
     }
 
     async fn handle_transaction_request_with_wait_eip1559(
         &self,
         mut tx_request: TransactionRequest,
         max_wait_secs: u64,
     ) -> anyhow::Result<(TxHash, TransactionReceipt)> {
+        let sender = tx_request.from.ok_or_else(|| format_err!("transaction has no sender"))?;
+        let sender_lock = sender_lock(sender);
+        let _sending = sender_lock.lock().await;
         let (max_fee_per_gas, max_priority_fee_per_gas) = self.get_eip1559_fee_params().await?;
         tx_request.gas_price = None;
         tx_request.max_fee_per_gas = Some(max_fee_per_gas);
@@ -656,8 +685,7 @@ impl GoatAdaptor {
             max_fee_per_gas,
             max_priority_fee_per_gas
         );
-        tx_request.nonce =
-            Some(self.provider.clone().get_transaction_count(tx_request.from.unwrap()).await?);
+        tx_request.nonce = Some(self.provider.clone().get_transaction_count(sender).await?);
         tracing::info!("tx(type2): {:?}", tx_request);
         tx_request.gas = Some(self.provider.clone().estimate_gas(tx_request.clone()).await?);
         tracing::info!("estimated gas(type2): {:?}", tx_request.gas);
@@ -706,7 +734,10 @@ impl GoatAdaptor {
                 }
             };
         }
-        bail!("tx_hash:{tx_hash} receipt not found within {max_wait_secs} seconds")
+        bail!(
+            "tx_hash:{tx_hash} receipt not found within {max_wait_secs} seconds; it may still be \
+             pending, and the next transaction from this sender reuses its nonce and replaces it"
+        )
     }
 }
 
@@ -1726,6 +1757,22 @@ impl GoatAdaptor {
     }
 }
 
+#[cfg(test)]
+mod sender_lock_tests {
+    use super::*;
+
+    #[test]
+    fn adaptors_share_one_lock_per_sender() {
+        let alice = Address::repeat_byte(0xa1);
+        let bob = Address::repeat_byte(0xb0);
+        let first = sender_lock(alice);
+        let _sending = first.try_lock().unwrap();
+        // another adaptor with the same key waits; another key does not
+        assert!(sender_lock(alice).try_lock().is_err());
+        assert!(sender_lock(bob).try_lock().is_ok());
+    }
+}
+
 #[cfg(test)]
 mod digest_tests {
     use super::*;