BM-198: Adds TXN timeout config to finalize batch (github#37)

This PR adds a configurable timeout to wait for transactions to be
included, currently only configured for the submitter but we can add it
to all places we need additional time.

We hit a TXN Timeout while waiting for the fulfill batch on Sepolia. 

Also closes BM-207 by attempting to just trigger a supervisor recover
when the event listener fails.

---------

Co-authored-by: Nuke <[email protected]>

Author: mothran

broker.toml

@@ -13,10 +13,11 @@ max_file_size = 50_000_000
 status_poll_ms = 1000
 bonsai_r0_zkvm_ver = "1.1.1"
 req_retry_count = 3
-set_builder_guest_path = "./target/riscv-guest/riscv32im-risc0-zkvm-elf/release/set-builder-guest"
-assessor_set_guest_path = "./target/riscv-guest/riscv32im-risc0-zkvm-elf/release/assessor-guest"
+# set_builder_guest_path = "./target/riscv-guest/riscv32im-risc0-zkvm-elf/release/set-builder-guest"
+# assessor_set_guest_path = "./target/riscv-guest/riscv32im-risc0-zkvm-elf/release/assessor-guest"
 
 [batcher]
 batch_max_time = 1000
 batch_size = 1
 block_deadline_buffer_secs = 120
+# txn_timeout = 30

crates/broker/src/config.rs

@@ -116,6 +116,8 @@ pub struct BatcherConfig {
     /// Number of seconds before the lowest block deadline in the order batch
     /// to flush the batch. This should be approximately snark_proving_time * 2
     pub block_deadline_buffer_secs: u64,
+    /// Timeout, in seconds for transaction confirmations
+    pub txn_timeout: Option<u64>,
 }
 
 impl Default for BatcherConfig {
@@ -125,6 +127,7 @@ impl Default for BatcherConfig {
             batch_size: Some(2),
             batch_max_fees: None,
             block_deadline_buffer_secs: 120,
+            txn_timeout: None,
         }
     }
 }
@@ -330,7 +333,8 @@ req_retry_count = 0
 [batcher]
 batch_max_time = 300
 batch_size = 2
-block_deadline_buffer_secs = 120"#;
+block_deadline_buffer_secs = 120
+txn_timeout = 45"#;
 
     const BAD_CONFIG: &str = r#"
 [market]
@@ -372,6 +376,7 @@ error = ?"#;
         assert_eq!(config.batcher.batch_size, Some(2));
         assert_eq!(config.batcher.batch_max_fees, Some("0.1".into()));
         assert_eq!(config.batcher.block_deadline_buffer_secs, 120);
+        assert_eq!(config.batcher.txn_timeout, None);
     }
 
     #[tokio::test]
@@ -413,6 +418,7 @@ error = ?"#;
             assert_eq!(config.market.allow_client_addresses, Some(vec![Address::ZERO]));
             assert_eq!(config.prover.status_poll_ms, 1000);
             assert!(config.prover.bonsai_r0_zkvm_ver.is_none());
+            assert_eq!(config.batcher.txn_timeout, Some(45));
         }
         tracing::debug!("closing...");
     }

crates/broker/src/lib.rs

@@ -511,12 +511,13 @@ where
 
         let submitter = Arc::new(submitter::Submitter::new(
             self.db.clone(),
+            self.config_watcher.config.clone(),
             prover.clone(),
             self.provider.clone(),
             self.args.set_verifier_addr,
             self.args.proof_market_addr,
             set_builder_img_data.0,
-        ));
+        )?);
         supervisor_tasks.spawn(async move {
             task::supervisor(1, submitter).await.context("Failed to start submitter service")?;
             Ok(())

crates/broker/src/market_monitor.rs

@@ -188,7 +188,7 @@ where
             })
             .await;
 
-        Ok(())
+        anyhow::bail!("Event polling exited, polling failed (possible RPC error)");
     }
 }
 

crates/broker/src/submitter.rs

@@ -2,7 +2,7 @@
 //
 // All rights reserved.
 
-use std::sync::Arc;
+use std::{sync::Arc, time::Duration};
 
 use aggregation_set::{SetInclusionReceipt, SetInclusionReceiptVerifierParameters};
 use alloy::{
@@ -22,6 +22,7 @@ use risc0_zkvm::{
 };
 
 use crate::{
+    config::ConfigLock,
     db::DbObj,
     provers::ProverObj,
     task::{RetryRes, RetryTask, SupervisorErr},
@@ -45,24 +46,34 @@ where
     #[allow(clippy::too_many_arguments)]
     pub fn new(
         db: DbObj,
+        config: ConfigLock,
         prover: ProverObj,
         provider: Arc<P>,
         set_verifier_addr: Address,
         market_addr: Address,
         set_builder_img_id: Digest,
-    ) -> Self {
-        let market = ProofMarketService::new(
+    ) -> Result<Self> {
+        let txn_timeout_opt = {
+            let config = config.lock_all().context("Failed to read config")?;
+            config.batcher.txn_timeout
+        };
+
+        let mut market = ProofMarketService::new(
             market_addr,
             provider.clone(),
             provider.default_signer_address(),
         );
+        if let Some(txn_timeout) = txn_timeout_opt {
+            market = market.with_timeout(Duration::from_secs(txn_timeout));
+        }
+
         let set_verifier = SetVerifierService::new(
             set_verifier_addr,
             provider.clone(),
             provider.default_signer_address(),
         );
 
-        Self { db, prover, market, set_verifier, set_builder_img_id }
+        Ok(Self { db, prover, market, set_verifier, set_builder_img_id })
     }
 
     async fn fetch_encode_g16(&self, g16_proof_id: &str) -> Result<Vec<u8>> {
@@ -537,14 +548,17 @@ mod tests {
 
         market.lockin_request(&order.request, &client_sig.into(), None).await.unwrap();
 
+        let config = ConfigLock::default();
         let submitter = Submitter::new(
             db.clone(),
+            config,
             prover.clone(),
             provider.clone(),
             *set_verifier.address(),
             *proof_market.address(),
             set_builder_id,
-        );
+        )
+        .unwrap();
 
         assert!(submitter.process_next_batch().await.unwrap());