fix: pr feedback

Author: troykessler

rust/main/agents/relayer/src/relay_api/handlers.rs

@@ -17,6 +17,7 @@ use tracing::{debug, error, info, warn};
 
 use crate::msg::pending_message::{MessageContext, PendingMessage};
 use crate::relay_api::metrics::RelayApiMetrics;
+use crate::relay_api::ExtractedMessage;
 
 /// Bounded cache for tracking recently submitted tx hashes to prevent replay attacks
 pub struct TxHashCache {
@@ -414,18 +415,19 @@ async fn create_relay(
         .as_ref()
         .ok_or_else(|| ServerError::InternalError("Send channels not configured".to_string()))?;
 
-    // 3. Process each message
-    let mut processed_messages = Vec::new();
+    // 3. First pass: Validate ALL messages before processing any
+    // This prevents partial success where some messages are sent to channels
+    // and others fail validation, which would prevent retries (tx_hash already cached)
+    struct ValidatedMessage {
+        extracted: ExtractedMessage,
+        msg_ctx: Arc<MessageContext>,
+        app_context: Option<String>,
+        send_channel: UnboundedSender<QueueOperation>,
+    }
 
-    for extracted in extracted_messages {
-        info!(
-            message_id = ?extracted.message_id,
-            origin = extracted.origin_domain,
-            destination = extracted.destination_domain,
-            nonce = extracted.message.nonce,
-            "Processing message"
-        );
+    let mut validated_messages = Vec::new();
 
+    for extracted in &extracted_messages {
         // Get message context for (origin, destination)
         let msg_ctx = msg_ctxs
             .get(&(extracted.origin_domain, extracted.destination_domain))
@@ -436,8 +438,59 @@ async fn create_relay(
                 ))
             })?;
 
-        // Classify app_context for metrics
-        // Use short timeouts to avoid blocking the response
+        // Apply message filtering (whitelist, blacklist, address blacklist)
+        if let Some(whitelist) = &state.message_whitelist {
+            if !whitelist.msg_matches(&extracted.message, true) {
+                if let Some(ref metrics) = state.metrics {
+                    metrics.inc_failure("message_not_whitelisted");
+                }
+                return Err(ServerError::InvalidRequest(
+                    "Message not whitelisted".to_string(),
+                ));
+            }
+        }
+
+        if let Some(blacklist) = &state.message_blacklist {
+            if blacklist.msg_matches(&extracted.message, false) {
+                if let Some(ref metrics) = state.metrics {
+                    metrics.inc_failure("message_blacklisted");
+                }
+                return Err(ServerError::InvalidRequest(
+                    "Message blacklisted".to_string(),
+                ));
+            }
+        }
+
+        if let Some(blacklist) = &state.address_blacklist {
+            if let Some(blacklisted_address) =
+                blacklist.find_blacklisted_address(&extracted.message)
+            {
+                debug!(
+                    blacklisted_address = hex::encode(blacklisted_address),
+                    message_id = ?extracted.message_id,
+                    "Rejecting message involving blacklisted address"
+                );
+                if let Some(ref metrics) = state.metrics {
+                    metrics.inc_failure("message_blacklisted_address");
+                }
+                return Err(ServerError::InvalidRequest(
+                    "Message involves a blacklisted address".to_string(),
+                ));
+            }
+        }
+
+        // Verify send channel exists for destination
+        let send_channel = send_channels
+            .get(&extracted.destination_domain)
+            .ok_or_else(|| {
+                ServerError::InvalidRequest(format!(
+                    "No send channel for destination domain {}",
+                    extracted.destination_domain
+                ))
+            })?
+            .clone();
+
+        // Classify app_context for metrics (with short timeouts)
         let recipient_ism = tokio::time::timeout(
             Duration::from_millis(500),
             msg_ctx
@@ -501,76 +554,44 @@ async fn create_relay(
         debug!(
             message_id = ?extracted.message_id,
             app_context = ?app_context,
-            "Classified message app context"
+            "Validated and classified message"
         );
 
-        // Apply message filtering (whitelist, blacklist, address blacklist)
-        // Skip if not whitelisted
-        if let Some(whitelist) = &state.message_whitelist {
-            if !whitelist.msg_matches(&extracted.message, true) {
-                if let Some(ref metrics) = state.metrics {
-                    metrics.inc_failure("message_not_whitelisted");
-                }
-                return Err(ServerError::InvalidRequest(
-                    "Message not whitelisted".to_string(),
-                ));
-            }
-        }
-
-        // Skip if message is blacklisted
-        if let Some(blacklist) = &state.message_blacklist {
-            if blacklist.msg_matches(&extracted.message, false) {
-                if let Some(ref metrics) = state.metrics {
-                    metrics.inc_failure("message_blacklisted");
-                }
-                return Err(ServerError::InvalidRequest(
-                    "Message blacklisted".to_string(),
-                ));
-            }
-        }
+        validated_messages.push(ValidatedMessage {
+            extracted: extracted.clone(),
+            msg_ctx: msg_ctx.clone(),
+            app_context,
+            send_channel,
+        });
+    }
 
-        // Skip if message involves a blacklisted address
-        if let Some(blacklist) = &state.address_blacklist {
-            if let Some(blacklisted_address) =
-                blacklist.find_blacklisted_address(&extracted.message)
-            {
-                // Log internally but don't expose the address in the API response
-                debug!(
-                    blacklisted_address = hex::encode(blacklisted_address),
-                    message_id = ?extracted.message_id,
-                    "Rejecting message involving blacklisted address"
-                );
-                if let Some(ref metrics) = state.metrics {
-                    metrics.inc_failure("message_blacklisted_address");
-                }
-                return Err(ServerError::InvalidRequest(
-                    "Message involves a blacklisted address".to_string(),
-                ));
-            }
-        }
+    // 4. Second pass: Process all validated messages
+    // All messages passed validation, safe to send to channels and persist to DB
+    let mut processed_messages = Vec::new();
 
-        // Get send channel for destination
-        let send_channel = send_channels
-            .get(&extracted.destination_domain)
-            .ok_or_else(|| {
-                ServerError::InvalidRequest(format!(
-                    "No send channel for destination domain {}",
-                    extracted.destination_domain
-                ))
-            })?;
+    for validated in validated_messages {
+        let extracted = &validated.extracted;
+        info!(
+            message_id = ?extracted.message_id,
+            origin = extracted.origin_domain,
+            destination = extracted.destination_domain,
+            nonce = extracted.message.nonce,
+            "Processing validated message"
+        );
 
         // Create PendingMessage with classified app_context
         let pending_msg = PendingMessage::new(
             extracted.message.clone(),
-            msg_ctx.clone(),
+            validated.msg_ctx.clone(),
             PendingOperationStatus::FirstPrepareAttempt,
-            app_context.clone(),
+            validated.app_context.clone(),
             0, // Max retries - relay API messages fail fast, no retries
         );
 
         // CRITICAL: Send to channel FIRST, before persisting to DB
         // This ensures we only persist messages that were successfully queued for processing
-        send_channel
+        validated
+            .send_channel
             .send(Box::new(pending_msg) as QueueOperation)
             .map_err(|e| {
                 error!(
@@ -584,7 +605,7 @@ async fn create_relay(
         info!(
             message_id = ?extracted.message_id,
             destination = extracted.destination_domain,
-            app_context = ?app_context,
+            app_context = ?validated.app_context,
             "Successfully sent message to processor channel"
         );