Track superseded mempool errors separately

crates/driver/src/domain/mempools.rs

@@ -5,11 +5,12 @@ use {
         infra::{self, Ethereum, observe},
     },
     alloy::{consensus::Transaction, eips::eip1559::Eip1559Estimation, sol_types::SolCall},
-    anyhow::Context,
+    anyhow::{Context, anyhow},
     contracts::CowSettlementForwarder::CowSettlementForwarder,
     eth_domain_types::{self as eth, BlockNo, TxId},
     ethrpc::block_stream::into_stream,
     futures::{FutureExt, StreamExt, future::select_ok},
+    itertools::Itertools,
     num::Saturating,
     thiserror::Error,
     tracing::Instrument,
@@ -48,6 +49,80 @@ pub struct Mempools {
     ethereum: Ethereum,
 }
 
+#[derive(Clone, Copy)]
+enum Outcome {
+    /// Future was dropped before completing — observed as `Superseded` at the
+    /// metric layer.
+    Pending,
+    Success {
+        blocks_passed: u64,
+    },
+    Failed {
+        reason: FailureReason,
+        blocks_passed: Option<u64>,
+    },
+    Disabled,
+}
+
+#[derive(Clone, Copy)]
+enum FailureReason {
+    Revert,
+    Expired,
+    Other,
+}
+
+impl FailureReason {
+    fn metric_label(self) -> &'static str {
+        match self {
+            FailureReason::Revert => "Revert",
+            FailureReason::Expired => "Expired",
+            FailureReason::Other => "Other",
+        }
+    }
+}
+
+impl Outcome {
+    fn metric_label(self) -> &'static str {
+        match self {
+            Outcome::Pending => "Superseded",
+            Outcome::Success { .. } => "Success",
+            Outcome::Failed { reason, .. } => reason.metric_label(),
+            Outcome::Disabled => "Disabled",
+        }
+    }
+
+    fn blocks_passed(self) -> Option<u64> {
+        match self {
+            Outcome::Pending | Outcome::Disabled => None,
+            Outcome::Success { blocks_passed } => Some(blocks_passed),
+            Outcome::Failed { blocks_passed, .. } => blocks_passed,
+        }
+    }
+}
+
+impl From<&Result<SubmissionSuccess, Error>> for Outcome {
+    fn from(result: &Result<SubmissionSuccess, Error>) -> Self {
+        match result {
+            Ok(s) => Outcome::Success {
+                blocks_passed: s.blocks_passed(),
+            },
+            Err(Error::Disabled) => Outcome::Disabled,
+            Err(err @ (Error::Revert { .. } | Error::SimulationRevert { .. })) => Outcome::Failed {
+                reason: FailureReason::Revert,
+                blocks_passed: err.blocks_passed(),
+            },
+            Err(err @ Error::Expired { .. }) => Outcome::Failed {
+                reason: FailureReason::Expired,
+                blocks_passed: err.blocks_passed(),
+            },
+            Err(Error::Other(_)) => Outcome::Failed {
+                reason: FailureReason::Other,
+                blocks_passed: None,
+            },
+        }
+    }
+}
+
 impl Mempools {
     pub fn try_new(mempools: Vec<infra::Mempool>, ethereum: Ethereum) -> Result<Self, NoMempools> {
         if mempools.is_empty() {
@@ -57,26 +132,53 @@ impl Mempools {
         }
     }
 
+    /// Race the enabled mempools concurrently; first success wins. Pending
+    /// submission futures are dropped at that point and every other mempool is
+    /// recorded as `Superseded`. If every mempool fails, return one of the
+    /// failure errors.
     pub async fn execute(
         &self,
         settlement: &Settlement,
         submission_deadline: BlockNo,
         mode: &SubmissionMode,
     ) -> Result<eth::TxId, Error> {
-        let (submission, _remaining_futures) = select_ok(self.mempools.iter().map(|mempool| {
-            async move {
-                let result = self
-                    .submit(mempool, settlement, submission_deadline, mode)
-                    .instrument(tracing::info_span!("mempool", kind = mempool.to_string()))
-                    .await;
-                observe::mempool_executed(mempool, settlement, &result);
-                result
-            }
-            .boxed()
-        }))
-        .await?;
+        let mut stats = vec![Outcome::Pending; self.mempools.len()];
+
+        let res = select_ok(self.mempools.iter().zip(stats.iter_mut()).map(
+            |(mempool, stat)| {
+                async move {
+                    let result = self
+                        .submit(mempool, settlement, submission_deadline, mode)
+                        .instrument(tracing::info_span!("mempool", kind = %mempool))
+                        .await;
+                    // Log inline so errors from mempools that later get superseded still surface;
+                    // metrics are emitted from `update_metrics` once the race outcome is known.
+                    observe::mempool_log(mempool, settlement, &result);
+                    *stat = Outcome::from(&result);
+                    result
+                }
+                .boxed()
+            },
+        ))
+        .await
+        // Drop the remaining futures (and the mutable borrow on `stats` they
+        // carry) so `update_metrics` can read `stats` below.
+        .map(|(success, _remaining)| success);
+
+        self.update_metrics(&stats);
 
-        Ok(submission.tx_hash)
+        Ok(res?.tx_hash)
+    }
+
+    /// A mempool is disabled if all of the following are true:
+    /// * the settlement may revert (see [`Settlement::may_revert`])
+    /// * the pool has revert protection enabled (see
+    ///   [`Self::revert_protection`])
+    /// * reverts can get mined (see [`infra::Mempool::reverts_can_get_mined`])
+    fn is_disabled(&self, mempool: &infra::Mempool, settlement: &Settlement) -> bool {
+        settlement.may_revert()
+            && matches!(self.revert_protection(), RevertProtection::Enabled)
+            && mempool.reverts_can_get_mined()
     }
 
     /// Defines if the mempools are configured in a way that guarantees that
@@ -99,12 +201,7 @@ impl Mempools {
         submission_deadline: BlockNo,
         mode: &SubmissionMode,
     ) -> Result<SubmissionSuccess, Error> {
-        // Don't submit risky transactions if revert protection is
-        // enabled and the settlement may revert in this mempool.
-        if settlement.may_revert()
-            && matches!(self.revert_protection(), RevertProtection::Enabled)
-            && mempool.reverts_can_get_mined()
-        {
+        if self.is_disabled(mempool, settlement) {
             return Err(Error::Disabled);
         }
 
@@ -262,7 +359,7 @@ impl Mempools {
                     }
                 }
             }
-            Err(Error::Other(anyhow::anyhow!(
+            Err(Error::Other(anyhow!(
                 "Block stream finished unexpectedly"
             )))
         }
@@ -375,6 +472,23 @@ impl Mempools {
             Some(pending_tx_gas_price.scaled_by_pct(GAS_PRICE_BUMP_PCT))
         }
     }
+
+    /// Update per-mempool metrics based on submission outcomes.
+    ///
+    /// When a winner exists, `Failed` outcomes are reclassified as `Superseded`
+    /// since errors are typically race-condition false-positives.
+    fn update_metrics(&self, stats: &[Outcome]) {
+        let winner_exists = stats.iter().any(|s| matches!(s, Outcome::Success { .. }));
+        // Using `zip_eq` to catch regressions in tests (sizes always match in
+        // practice).
+        for (mempool, &outcome) in self.mempools.iter().zip_eq(stats.iter()) {
+            let label = match outcome {
+                Outcome::Failed { .. } if winner_exists => "Superseded",
+                other => other.metric_label(),
+            };
+            observe::mempool_submission_result(mempool, label, outcome.blocks_passed());
+        }
+    }
 }
 
 /// Applies the solver's gas fee override if present. When a replacement
@@ -444,6 +558,15 @@ pub struct SubmissionSuccess {
     pub submitted_at_block: eth::BlockNo,
 }
 
+impl SubmissionSuccess {
+    /// Number of blocks between submission start and on-chain inclusion.
+    pub fn blocks_passed(&self) -> u64 {
+        self.included_in_block
+            .saturating_sub(self.submitted_at_block)
+            .0
+    }
+}
+
 #[derive(Debug, Error)]
 #[error("no mempools configured, cannot execute settlements")]
 pub struct NoMempools;
@@ -489,3 +612,28 @@ pub enum Error {
     #[error("Failed to submit: {0:?}")]
     Other(#[from] anyhow::Error),
 }
+
+impl Error {
+    /// Number of blocks between the first submission and when the error was
+    /// returned, if the error carries that timing.
+    pub fn blocks_passed(&self) -> Option<u64> {
+        let (start, end) = match self {
+            Self::Revert {
+                submitted_at_block,
+                reverted_at_block,
+                ..
+            }
+            | Self::SimulationRevert {
+                submitted_at_block,
+                reverted_at_block,
+            } => (*submitted_at_block, *reverted_at_block),
+            Self::Expired {
+                submitted_at_block,
+                submission_deadline,
+                ..
+            } => (*submitted_at_block, *submission_deadline),
+            Self::Disabled | Self::Other(_) => return None,
+        };
+        Some(end.saturating_sub(start).0)
+    }
+}

crates/driver/src/infra/observe/mod.rs

@@ -24,7 +24,6 @@ use {
     },
     eth_domain_types::{self as eth, Gas},
     ethrpc::block_stream::BlockInfo,
-    num::Saturating,
     std::{
         collections::{BTreeMap, HashSet},
         time::Duration,
@@ -360,80 +359,49 @@ pub fn solver_response(
         .observe(compute_time.as_secs_f64());
 }
 
-/// Observe the result of mempool transaction execution.
-pub fn mempool_executed(
+/// Log a single mempool submission attempt. Called inline from the racing
+/// task so that errors from mempools that later get superseded are still
+/// visible in logs. Metrics are emitted separately from `update_metrics`
+/// once the race outcome is known.
+pub fn mempool_log(
     mempool: &Mempool,
     settlement: &Settlement,
-    res: &Result<SubmissionSuccess, mempools::Error>,
+    result: &Result<SubmissionSuccess, mempools::Error>,
 ) {
-    match res {
-        Ok(submission) => {
-            tracing::info!(
-                txid = ?submission.tx_hash,
-                %mempool,
-                ?settlement,
-                "sending transaction via mempool succeeded",
-            );
-        }
-        Err(mempools::Error::Disabled) => {
-            tracing::debug!(
-                %mempool,
-                "sending transaction via mempool disabled",
-            );
-        }
-        Err(err) => {
-            tracing::warn!(
-                ?err,
-                %mempool,
-                ?settlement,
-                "sending transaction via mempool failed",
-            );
-        }
+    match result {
+        Ok(submission) => tracing::info!(
+            txid = ?submission.tx_hash,
+            %mempool,
+            ?settlement,
+            "sending transaction via mempool succeeded",
+        ),
+        Err(mempools::Error::Disabled) => tracing::debug!(
+            %mempool,
+            "mempool disabled, not sending transaction",
+        ),
+        Err(err) => tracing::warn!(
+            ?err,
+            %mempool,
+            ?settlement,
+            "sending transaction via mempool failed",
+        ),
     }
-    let result = match res {
-        Ok(_) => "Success",
-        Err(mempools::Error::Revert { .. } | mempools::Error::SimulationRevert { .. }) => "Revert",
-        Err(mempools::Error::Expired { .. }) => "Expired",
-        Err(mempools::Error::Other(_)) => "Other",
-        Err(mempools::Error::Disabled) => "Disabled",
-    };
+}
+
+/// Emit per-mempool race counters with the final, reclassified label
+/// (`Success` / `Revert` / `Expired` / `Other` / `Superseded` / `Disabled`).
+/// Called once per mempool after the race resolves.
+pub fn mempool_submission_result(mempool: &Mempool, label: &str, blocks_passed: Option<u64>) {
+    let name = mempool.to_string();
     metrics::get()
         .mempool_submission
-        .with_label_values(&[mempool.to_string().as_str(), result])
+        .with_label_values(&[name.as_str(), label])
         .inc();
-
-    // For some of the errors we are interested in observing the exact block numbers
-    // passed since the first submission.
-    let blocks_passed = match res {
-        Ok(SubmissionSuccess {
-            submitted_at_block,
-            included_in_block,
-            ..
-        }) => Some(("Success", submitted_at_block, included_in_block)),
-        Err(mempools::Error::Revert {
-            tx_id: _,
-            submitted_at_block,
-            reverted_at_block,
-        }) => Some(("Revert", submitted_at_block, reverted_at_block)),
-        Err(mempools::Error::SimulationRevert {
-            submitted_at_block,
-            reverted_at_block,
-        }) => Some(("Revert", submitted_at_block, reverted_at_block)),
-        Err(mempools::Error::Expired {
-            tx_id: _,
-            submitted_at_block,
-            submission_deadline,
-        }) => Some(("Expired", submitted_at_block, submission_deadline)),
-        Err(mempools::Error::Other(_)) => None,
-        Err(mempools::Error::Disabled) => None,
-    };
-
-    if let Some((label, start, end)) = blocks_passed {
-        let blocks_passed = end.saturating_sub(*start);
+    if let Some(blocks) = blocks_passed {
         metrics::get()
             .mempool_submission_results_blocks_passed
-            .with_label_values(&[mempool.to_string().as_str(), label])
-            .inc_by(blocks_passed.0);
+            .with_label_values(&[name.as_str(), label])
+            .inc_by(blocks);
     }
 }