fix(driver): fall back to another EOA when a submission account is unusable

crates/driver/src/domain/competition/mod.rs

@@ -2,7 +2,7 @@ use {
     self::solution::settlement,
     super::{
         Mempools,
-        mempools::SubmissionMode,
+        mempools::{self, SubmissionMode},
         time::{self, Remaining},
     },
     crate::{
@@ -35,7 +35,7 @@ use {
         cmp::Reverse,
         collections::{BTreeMap, HashMap, HashSet, VecDeque},
         sync::{Arc, Mutex},
-        time::Instant,
+        time::{Duration, Instant},
     },
     tokio::{sync::mpsc, task},
     tracing::{Instrument, instrument},
@@ -64,6 +64,17 @@ type Balances = HashMap<BalanceGroup, order::SellAmount>;
 /// auction still find its settlement.
 const MAX_CONCURRENT_AUCTIONS: usize = 5;
 
+/// How long an EIP-7702 submission account is benched after it fails to
+/// broadcast for an account-specific reason (no gas for the tx, a stuck nonce,
+/// a pending tx that can't be replaced). While benched, the account is held out
+/// of the selection pool, so the driver stops assigning settlements to a stuck
+/// lane until it recovers. The account is retried automatically once the
+/// cooldown elapses; a still-broken account is simply re-benched after a single
+/// failed attempt.
+// ponytail: a fixed cooldown is enough; lift to driver config only if operators
+// need per-network tuning.
+const UNHEALTHY_ACCOUNT_COOLDOWN: Duration = Duration::from_secs(30);
+
 /// An ongoing competition. There is one competition going on per solver at any
 /// time. The competition stores settlements to solutions generated by the
 /// driver, and allows them to be executed onchain when requested later. The
@@ -132,6 +143,7 @@ impl SubmitterPool {
             return Some(SubmitterGuard {
                 inner: GuardInner::Direct(permit),
                 solver_address: self.solver_address,
+                quarantine: None,
             });
         }
 
@@ -170,6 +182,30 @@ impl SubmitterPool {
         Some(SubmitterGuard {
             inner,
             solver_address: self.solver_address,
+            quarantine: None,
+        })
+    }
+
+    /// Non-blocking variant of [`acquire`]. Returns `None` when no slot is
+    /// immediately free. Used to fall back to another submission account on
+    /// retry: acquiring without blocking (while the failed account's slot is
+    /// still held) avoids deadlocking against other concurrent settlements.
+    fn try_acquire(&self) -> Option<SubmitterGuard> {
+        let inner = if let Ok(permit) = Arc::clone(&self.direct_slot).try_acquire_owned() {
+            GuardInner::Direct(permit)
+        } else {
+            let delegated = self.delegated.as_ref()?;
+            let mut channel = delegated.acquire.try_lock().ok()?;
+            let account = channel.try_recv().ok()?;
+            GuardInner::Delegated {
+                account,
+                release: delegated.release.clone(),
+            }
+        };
+        Some(SubmitterGuard {
+            inner,
+            solver_address: self.solver_address,
+            quarantine: None,
         })
     }
 
@@ -184,6 +220,10 @@ impl SubmitterPool {
 struct SubmitterGuard {
     inner: GuardInner,
     solver_address: eth::Address,
+    /// When set, a delegated account is returned to the pool only after this
+    /// delay instead of immediately, benching an unhealthy EOA (see
+    /// [`SubmitterGuard::quarantine`]). Always `None` for the direct slot.
+    quarantine: Option<Duration>,
 }
 
 enum GuardInner {
@@ -209,13 +249,32 @@ impl SubmitterGuard {
             GuardInner::Dropped => unreachable!(),
         }
     }
+
+    /// Bench the underlying delegated account: once this guard drops, the
+    /// account is withheld from the pool for `cooldown` so concurrent and
+    /// subsequent settlements stop selecting a stuck or underfunded lane until
+    /// it recovers. No-op for the direct solver EOA, which is never benched.
+    fn quarantine(&mut self, cooldown: Duration) {
+        if matches!(self.inner, GuardInner::Delegated { .. }) {
+            self.quarantine = Some(cooldown);
+        }
+    }
 }
 
 impl Drop for SubmitterGuard {
     fn drop(&mut self) {
         let inner = std::mem::replace(&mut self.inner, GuardInner::Dropped);
         if let GuardInner::Delegated { account, release } = inner {
+            let cooldown = self.quarantine;
             tokio::spawn(async move {
+                if let Some(cooldown) = cooldown {
+                    tracing::warn!(
+                        submitter = ?account.address(),
+                        ?cooldown,
+                        "benching unhealthy submission account before returning it to the pool"
+                    );
+                    tokio::time::sleep(cooldown).await;
+                }
                 if release.send(account).await.is_err() {
                     tracing::error!("failed to return submission account to pool: channel closed");
                 }
@@ -912,17 +971,51 @@ impl Competition {
         // Acquire a submission slot. The pool prefers the direct solver EOA
         // (no forwarding overhead); falls back to a delegated EIP-7702
         // submission account when the solver EOA is busy.
-        let guard = self
+        //
+        // If submission fails because the chosen account is unusable (no gas,
+        // stale nonce, a pending tx that can't be replaced), bench that account
+        // and retry the settlement from another one while the deadline still
+        // holds. Benching withholds the account from the pool for a cooldown so
+        // concurrent and subsequent settlements stop selecting a stuck lane
+        // until it recovers. Accounts spent on this request are also held until
+        // it finishes, so each retry picks a different one. This is safe against
+        // double-submission: a settlement that actually lands always returns
+        // `Ok`, so only failures (where nothing was broadcast) are ever retried.
+        let mut guard = self
             .submitter_pool
             .acquire()
             .await
             .ok_or(Error::SubmissionError)?;
-        let mode = guard.submission_mode();
-
-        let executed = self
-            .mempools
-            .execute(&settlement, submission_deadline, &mode)
-            .await;
+        let mut spent = Vec::new();
+        let executed = loop {
+            let mode = guard.submission_mode();
+            let executed = self
+                .mempools
+                .execute(&settlement, submission_deadline, &mode)
+                .await;
+
+            if !matches!(executed, Err(mempools::Error::SubmitterUnusable(_))) {
+                break executed;
+            }
+            // The chosen account could not broadcast: bench it regardless of
+            // whether we can retry, so it stops being selected while unhealthy.
+            guard.quarantine(UNHEALTHY_ACCOUNT_COOLDOWN);
+
+            let deadline_reached =
+                self.eth.current_block().borrow().number >= submission_deadline.0;
+            if deadline_reached {
+                break executed;
+            }
+            let Some(next) = self.submitter_pool.try_acquire() else {
+                // No other account currently free; give up and report the error.
+                break executed;
+            };
+            tracing::warn!(
+                ?mode,
+                "submission account unusable, retrying from another account"
+            );
+            spent.push(std::mem::replace(&mut guard, next));
+        };
 
         notify::executed(
             &self.solver,
@@ -1102,3 +1195,84 @@ pub enum Error {
     #[error("could not parse the request")]
     MalformedRequest,
 }
+
+#[cfg(test)]
+mod submitter_pool_tests {
+    use {
+        super::*,
+        alloy::primitives::{Address, address},
+    };
+
+    const SOLVER: eth::Address = address!("0000000000000000000000000000000000000099");
+
+    fn delegated_address(guard: &SubmitterGuard) -> eth::Address {
+        match guard.submission_mode() {
+            SubmissionMode::Delegated { submitter_eoa, .. } => submitter_eoa,
+            other => panic!("expected a delegated slot, got {other:?}"),
+        }
+    }
+
+    /// An account benched via [`SubmitterGuard::quarantine`] must not be handed
+    /// back out while it cools down, whereas a normally-released account returns
+    /// to the pool immediately. This is the core of "stop assigning settlements
+    /// to a stuck lane until it recovers" from issue #4541.
+    #[tokio::test]
+    async fn quarantined_account_is_withheld_from_the_pool() {
+        let pool = SubmitterPool::new(
+            SOLVER,
+            vec![
+                Account::Address(Address::with_last_byte(1)),
+                Account::Address(Address::with_last_byte(2)),
+            ],
+            0,
+        );
+
+        // Direct slot first, then both delegated accounts; pool now drained.
+        let direct = pool.acquire().await.expect("direct slot");
+        let mut first = pool.try_acquire().expect("first delegated account");
+        let second = pool.try_acquire().expect("second delegated account");
+        assert!(pool.try_acquire().is_none(), "pool should be drained");
+
+        let benched = delegated_address(&first);
+        let released = delegated_address(&second);
+
+        // Bench `first` for a cooldown long enough to outlast the test; release
+        // `second` the normal way.
+        first.quarantine(Duration::from_secs(3600));
+        drop(first);
+        drop(second);
+
+        // Let the (non-benched) release task return `second` to the pool.
+        tokio::time::sleep(Duration::from_millis(50)).await;
+
+        let reacquired = pool.try_acquire().expect("released account returns");
+        assert_eq!(delegated_address(&reacquired), released);
+        assert_ne!(delegated_address(&reacquired), benched);
+        assert!(
+            pool.try_acquire().is_none(),
+            "benched account must stay out of the pool while cooling down"
+        );
+
+        drop(direct);
+        drop(reacquired);
+    }
+
+    /// The direct solver EOA is the primary submitter and must never be benched,
+    /// even if it is the account that failed.
+    #[test]
+    fn direct_slot_is_never_quarantined() {
+        let permit = Arc::new(tokio::sync::Semaphore::new(1))
+            .try_acquire_owned()
+            .unwrap();
+        let mut guard = SubmitterGuard {
+            inner: GuardInner::Direct(permit),
+            solver_address: SOLVER,
+            quarantine: None,
+        };
+        guard.quarantine(Duration::from_secs(3600));
+        assert!(
+            guard.quarantine.is_none(),
+            "direct slot must not be quarantined"
+        );
+    }
+}