relay: typed AlreadyProcessed error to skip retry on duplicate submits

process_deposit and process_refund now return Result<(), ProcessError>.
The Linera bridge contract reverts with "deposit already processed" or
"refund already processed" on a duplicate key; before this, the relayer
treated both as ordinary failures, burned retry budget, and emitted
refund_failed metrics when check_refund_completion hadn't yet caught up.

Classify the error chain in serve_loop and short-circuit in the pending
processors — mark the entry as completed instead of retrying.

Author: deuszx

linera-bridge/src/monitor/evm.rs

@@ -13,7 +13,11 @@ use tokio::sync::{Notify, RwLock};
 use super::{MonitorState, PendingDeposit, PendingRefund};
 use crate::{
     proof::{parse_burn_blocked_event, parse_deposit_event, DepositKey, ReceiptLog, RefundKey},
-    relay::{self, evm::EvmClient, linera::LineraClient},
+    relay::{
+        self,
+        evm::EvmClient,
+        linera::{LineraClient, ProcessError},
+    },
 };
 
 /// Background task that polls EVM for `DepositInitiated` events and checks
@@ -115,7 +119,16 @@ pub(crate) async fn process_pending_deposits<E: linera_core::environment::Enviro
                         tracing::info!(%tx_hash, "Deposit processed successfully");
                         relay::update_linera_balance_metric(linera_client).await;
                     }
-                    Err(e) => {
+                    Err(ProcessError::AlreadyProcessed) => {
+                        // The chain already recorded this key (e.g. another
+                        // relayer beat us to it). Skip the retry budget and
+                        // mark the deposit completed instead of waiting for
+                        // the periodic completion check.
+                        tracing::info!(%tx_hash, "Deposit already processed on-chain");
+                        monitor.write().await.complete_deposit(&pending.key).await;
+                        continue;
+                    }
+                    Err(ProcessError::Other(e)) => {
                         tracing::warn!(%tx_hash, "Deposit processing failed: {e}");
                     }
                 }
@@ -356,7 +369,16 @@ pub(crate) async fn process_pending_refunds<E: linera_core::environment::Environ
                     tracing::info!(%tx_hash, "Refund processed successfully");
                     relay::update_linera_balance_metric(linera_client).await;
                 }
-                Err(e) => {
+                Err(ProcessError::AlreadyProcessed) => {
+                    // Same short-circuit as the deposit path: the chain
+                    // already recorded this refund key, so the duplicate
+                    // submission isn't worth retrying — mark it completed
+                    // and skip the retry-counter bump.
+                    tracing::info!(%tx_hash, "Refund already processed on-chain");
+                    monitor.write().await.complete_refund(&pending.key).await;
+                    continue;
+                }
+                Err(ProcessError::Other(e)) => {
                     tracing::warn!(%tx_hash, "Refund processing failed: {e}");
                 }
             },

linera-bridge/src/relay/linera.rs

@@ -17,6 +17,41 @@ use tokio::sync::{mpsc, oneshot};
 
 use crate::proof::{DepositKey, RefundKey};
 
+/// Outcome of a `process_deposit` / `process_refund` submission.
+///
+/// The bridge contract reverts with `"deposit already processed"` or
+/// `"refund already processed"` when the chain has already recorded the key.
+/// In that case the on-chain state is already what the relayer wanted, so the
+/// duplicate submission is not a retry-worthy failure and the caller should
+/// short-circuit instead of burning its retry budget.
+#[derive(Debug, thiserror::Error)]
+pub enum ProcessError {
+    /// The chain already recorded this deposit/refund key. A duplicate
+    /// submission isn't a retry-worthy failure — the on-chain state is
+    /// already what we wanted.
+    #[error("already processed on-chain")]
+    AlreadyProcessed,
+    #[error(transparent)]
+    Other(#[from] anyhow::Error),
+}
+
+impl ProcessError {
+    /// Walks the error chain looking for the bridge contract's
+    /// `"already processed"` panic substring. Matches both
+    /// `"deposit already processed"` and `"refund already processed"` panics,
+    /// regardless of how many layers of context wrap the underlying error.
+    pub fn classify(err: anyhow::Error) -> Self {
+        if err
+            .chain()
+            .any(|c| c.to_string().contains("already processed"))
+        {
+            ProcessError::AlreadyProcessed
+        } else {
+            ProcessError::Other(err)
+        }
+    }
+}
+
 /// A write operation to be executed on the bridge chain.
 /// Sent to the main loop which serializes all chain mutations.
 #[derive(Debug)]
@@ -27,11 +62,11 @@ pub(crate) enum ChainOperation {
     },
     ProcessDeposit {
         proof: crate::proof::gen::DepositProof,
-        response: oneshot::Sender<Result<()>>,
+        response: oneshot::Sender<Result<(), ProcessError>>,
     },
     ProcessRefund {
         proof: crate::proof::gen::BurnBlockedProof,
-        response: oneshot::Sender<Result<()>>,
+        response: oneshot::Sender<Result<(), ProcessError>>,
     },
 }
 
@@ -125,7 +160,10 @@ impl<E: linera_core::environment::Environment> LineraClient<E> {
 
     // ── Write operations (sent to main loop via channel) ──
 
-    pub async fn process_deposit(&self, proof: crate::proof::gen::DepositProof) -> Result<()> {
+    pub async fn process_deposit(
+        &self,
+        proof: crate::proof::gen::DepositProof,
+    ) -> Result<(), ProcessError> {
         let (resp_tx, resp_rx) = oneshot::channel();
         self.op_tx
             .send(ChainOperation::ProcessDeposit {
@@ -137,7 +175,10 @@ impl<E: linera_core::environment::Environment> LineraClient<E> {
         resp_rx.await.with_context(|| "Response channel closed")?
     }
 
-    pub async fn process_refund(&self, proof: crate::proof::gen::BurnBlockedProof) -> Result<()> {
+    pub async fn process_refund(
+        &self,
+        proof: crate::proof::gen::BurnBlockedProof,
+    ) -> Result<(), ProcessError> {
         let (resp_tx, resp_rx) = oneshot::channel();
         self.op_tx
             .send(ChainOperation::ProcessRefund {
@@ -196,3 +237,44 @@ pub(crate) fn find_burn_events(
     }
     result
 }
+
+#[cfg(test)]
+mod tests {
+    use anyhow::anyhow;
+
+    use super::ProcessError;
+
+    #[test]
+    fn classify_direct_already_processed_is_already_processed() {
+        let err = anyhow!("deposit already processed");
+        assert!(matches!(
+            ProcessError::classify(err),
+            ProcessError::AlreadyProcessed
+        ));
+    }
+
+    /// The bridge panic string sits underneath a wrapper that comes from the
+    /// chain-client execution path. Classification must walk the error chain,
+    /// not only inspect the top-level `Display`.
+    #[test]
+    fn classify_wrapped_already_processed_is_already_processed() {
+        let err = anyhow!(
+            "worker operation failed: Execution error: Failed to execute Wasm module: \
+             RuntimeError: unreachable during Operation(0)"
+        )
+        .context("refund already processed");
+        assert!(matches!(
+            ProcessError::classify(err),
+            ProcessError::AlreadyProcessed
+        ));
+    }
+
+    #[test]
+    fn classify_unrelated_error_is_other() {
+        let err = anyhow!("something else");
+        assert!(matches!(
+            ProcessError::classify(err),
+            ProcessError::Other(_)
+        ));
+    }
+}

linera-bridge/src/relay/mod.rs

@@ -516,7 +516,11 @@ async fn serve_loop<E: linera_core::environment::Environment + 'static>(
                         }
                     }
                     linera::ChainOperation::ProcessDeposit { proof, response } => {
-                        let result = async {
+                        // Inner block returns anyhow::Error; the outer match
+                        // classifies it into ProcessError so the caller can
+                        // skip retrying when the chain already recorded this
+                        // deposit key.
+                        let inner: Result<()> = async {
                             let operations: Vec<_> = proof.log_indices.iter().map(|&log_index| {
                                 let op = crate::abi::BridgeOperation::ProcessDeposit {
                                     block_header_rlp: proof.block_header_rlp.clone(),
@@ -557,13 +561,14 @@ async fn serve_loop<E: linera_core::environment::Environment + 'static>(
                             };
                             Ok(())
                         }.await;
+                        let result = inner.map_err(linera::ProcessError::classify);
                         if response.send(result).is_err() {
                             tracing::debug!("ProcessDeposit response receiver dropped");
                         }
                         update_balance_metrics(&evm_client, &linera_client).await;
                     }
                     linera::ChainOperation::ProcessRefund { proof, response } => {
-                        let result = async {
+                        let inner: Result<()> = async {
                             let op = crate::abi::BridgeOperation::RefundBurn {
                                 block_header_rlp: proof.block_header_rlp,
                                 receipt_rlp: proof.receipt_rlp,
@@ -595,6 +600,7 @@ async fn serve_loop<E: linera_core::environment::Environment + 'static>(
                                 other => anyhow::bail!("RefundBurn not committed: {other:?}"),
                             }
                         }.await;
+                        let result = inner.map_err(linera::ProcessError::classify);
                         if response.send(result).is_err() {
                             tracing::debug!("ProcessRefund response receiver dropped");
                         }