BM-1933: Additional Order Fulfillment Checks

crates/boundless-market/Cargo.toml

@@ -66,4 +66,4 @@ serde_json = { workspace = true }
 
 [dev-dependencies]
 boundless-test-utils = { workspace = true }
-tracing-test = { workspace = true }
+tracing-test = { workspace = true, features = ["no-env-filter"] }

crates/boundless-market/src/contracts/boundless_market.rs

@@ -28,21 +28,20 @@ use alloy::{
     signers::Signer,
 };
 
-use alloy_sol_types::{SolCall, SolEvent};
+use alloy_sol_types::{SolCall, SolEvent, SolInterface};
 use anyhow::{anyhow, Context, Result};
 use risc0_ethereum_contracts::event_query::EventQueryConfig;
 use thiserror::Error;
 
-use crate::{
-    contracts::token::{IERC20Permit, IHitPoints::IHitPointsErrors, Permit, IERC20},
-    deployments::collateral_token_supports_permit,
-};
-
 use super::{
     eip712_domain, AssessorReceipt, EIP712DomainSaltless, Fulfillment,
-    IBoundlessMarket::{self, IBoundlessMarketInstance, ProofDelivered},
+    IBoundlessMarket::{self, IBoundlessMarketErrors, IBoundlessMarketInstance, ProofDelivered},
     Offer, ProofRequest, RequestError, RequestId, RequestStatus, TxnErr, TXN_CONFIRM_TIMEOUT,
 };
+use crate::{
+    contracts::token::{IERC20Permit, IHitPoints::IHitPointsErrors, Permit, IERC20},
+    deployments::collateral_token_supports_permit,
+};
 
 /// Fraction of collateral the protocol gives to the prover who fills an order that was locked by another prover but expired
 /// This is determined by the constant SLASHING_BURN_BPS defined in the BoundlessMarket contract.
@@ -113,6 +112,16 @@ pub enum MarketError {
     /// Timeout reached.
     #[error("Timeout: 0x{0:x}")]
     TimeoutReached(U256),
+
+    /// Payment requirements failed
+    #[error("Payment requirements failed during order fulfillment: {0:?}")]
+    PaymentRequirementsFailed(IBoundlessMarketErrors),
+
+    /// Payment requirements failed, unable to decode error
+    #[error(
+        "Payment requirements failed during order fulfillment: unrecognized error payload {0:?}"
+    )]
+    PaymentRequirementsFailedUnknownError(Bytes),
 }
 
 impl From<alloy::contract::Error> for MarketError {
@@ -209,6 +218,39 @@ fn extract_tx_log<E: SolEvent + Debug + Clone>(
     }
 }
 
+fn validate_fulfill_receipt(receipt: TransactionReceipt) -> Result<(), MarketError> {
+    for (idx, log) in receipt.inner.logs().iter().enumerate() {
+        if log.topic0().is_some_and(|topic| {
+            *topic == IBoundlessMarket::PaymentRequirementsFailed::SIGNATURE_HASH
+        }) {
+            match log.log_decode::<IBoundlessMarket::PaymentRequirementsFailed>() {
+                Ok(decoded) => {
+                    let raw_error = Bytes::copy_from_slice(decoded.inner.data.error.as_ref());
+                    match IBoundlessMarketErrors::abi_decode(&raw_error) {
+                        Ok(err) => tracing::warn!(
+                            tx_hash = ?receipt.transaction_hash,
+                            log_index = idx,
+                            "Payment requirements failed for at least one fulfillment: {err:?}"
+                        ),
+                        Err(_) => tracing::warn!(
+                            tx_hash = ?receipt.transaction_hash,
+                            log_index = idx,
+                            raw = ?raw_error,
+                            "Payment requirements failed for at least one fulfillment, but error payload was unrecognized"
+                        ),
+                    }
+                }
+                Err(err) => tracing::warn!(
+                    tx_hash = ?receipt.transaction_hash,
+                    log_index = idx,
+                    "Failed to decode PaymentRequirementsFailed event: {err:?}"
+                ),
+            }
+        }
+    }
+    Ok(())
+}
+
 /// Data returned when querying for a RequestSubmitted event
 #[derive(Debug, Clone)]
 pub struct RequestSubmittedEventData {
@@ -729,7 +771,7 @@ impl<P: Provider> BoundlessMarketService<P> {
 
         tracing::info!("Submitted proof for batch {:?}: {}", fill_ids, receipt.transaction_hash);
 
-        Ok(())
+        validate_fulfill_receipt(receipt)
     }
 
     /// Fulfill a batch of requests by delivering the proof for each application and withdraw from the prover balance.
@@ -751,7 +793,7 @@ impl<P: Provider> BoundlessMarketService<P> {
 
         tracing::info!("Submitted proof for batch {:?}: {}", fill_ids, receipt.transaction_hash);
 
-        Ok(())
+        validate_fulfill_receipt(receipt)
     }
 
     /// Combined function to submit a new merkle root to the set-verifier and call `fulfill`.
@@ -784,7 +826,7 @@ impl<P: Provider> BoundlessMarketService<P> {
 
         tracing::info!("Submitted merkle root and proof for batch {}", tx_receipt.transaction_hash);
 
-        Ok(())
+        validate_fulfill_receipt(tx_receipt)
     }
 
     /// Combined function to submit a new merkle root to the set-verifier and call `fulfillAndWithdraw`.
@@ -813,7 +855,7 @@ impl<P: Provider> BoundlessMarketService<P> {
 
         tracing::info!("Submitted merkle root and proof for batch {}", tx_receipt.transaction_hash);
 
-        Ok(())
+        validate_fulfill_receipt(tx_receipt)
     }
 
     /// A combined call to `IBoundlessMarket.priceRequest` and `IBoundlessMarket.fulfill`.
@@ -856,7 +898,7 @@ impl<P: Provider> BoundlessMarketService<P> {
 
         tracing::info!("Fulfilled proof for batch {}", tx_receipt.transaction_hash);
 
-        Ok(())
+        validate_fulfill_receipt(tx_receipt)
     }
 
     /// A combined call to `IBoundlessMarket.priceRequest` and `IBoundlessMarket.fulfillAndWithdraw`.
@@ -899,7 +941,7 @@ impl<P: Provider> BoundlessMarketService<P> {
 
         tracing::info!("Fulfilled proof for batch {}", tx_receipt.transaction_hash);
 
-        Ok(())
+        validate_fulfill_receipt(tx_receipt)
     }
 
     /// Combined function to submit a new merkle root to the set-verifier and call `priceAndfulfill`.
@@ -938,7 +980,7 @@ impl<P: Provider> BoundlessMarketService<P> {
 
         tracing::info!("Submitted merkle root and proof for batch {}", tx_receipt.transaction_hash);
 
-        Ok(())
+        validate_fulfill_receipt(tx_receipt)
     }
 
     /// Combined function to submit a new merkle root to the set-verifier and call `priceAndFulfillAndWithdraw`.
@@ -977,7 +1019,7 @@ impl<P: Provider> BoundlessMarketService<P> {
 
         tracing::info!("Submitted merkle root and proof for batch {}", tx_receipt.transaction_hash);
 
-        Ok(())
+        validate_fulfill_receipt(tx_receipt)
     }
 
     /// Checks if a request is locked in.

crates/boundless-market/tests/e2e.rs

@@ -374,6 +374,7 @@ async fn test_e2e_price_and_fulfill_batch() {
 }
 
 #[tokio::test]
+#[traced_test]
 async fn test_e2e_no_payment() {
     // Setup anvil
     let anvil = Anvil::new().spawn();
@@ -434,11 +435,13 @@ async fn test_e2e_no_payment() {
         };
 
         let balance_before = ctx.prover_market.balance_of(some_other_address).await.unwrap();
-        // fulfill the request.
+        // fulfill the request. This call emits a PaymentRequirementsFailed log since the lock
+        // belongs to a different prover, but the request itself still becomes fulfilled on-chain.
         ctx.prover_market
             .fulfill(FulfillmentTx::new(vec![fulfillment.clone()], assessor_fill.clone()))
             .await
-            .unwrap();
+            .expect("fulfillment should succeed even if payment requirements fail");
+        assert!(logs_contain("Payment requirements failed for at least one fulfillment"));
         assert!(ctx.customer_market.is_fulfilled(request_id).await.unwrap());
         let balance_after = ctx.prover_market.balance_of(some_other_address).await.unwrap();
         assert!(balance_before == balance_after);

crates/broker/src/lib.rs

@@ -22,7 +22,7 @@ use crate::storage::create_uri_handler;
 use alloy::{
     network::Ethereum,
     primitives::{Address, Bytes, FixedBytes, U256},
-    providers::{Provider, WalletProvider},
+    providers::{DynProvider, Provider, WalletProvider},
     signers::local::PrivateKeySigner,
 };
 use anyhow::{Context, Result};
@@ -897,16 +897,19 @@ where
             Arc::new(provers::DefaultProver::new())
         };
 
+        let prover_addr = self.provider.default_signer_address();
+
         let (pricing_tx, pricing_rx) = mpsc::channel(PRICING_CHANNEL_CAPACITY);
 
-        let collateral_token_decimals = BoundlessMarketService::new(
+        let market = Arc::new(BoundlessMarketService::new(
             self.deployment().boundless_market_address,
-            self.provider.clone(),
-            Address::ZERO,
-        )
-        .collateral_token_decimals()
-        .await
-        .context("Failed to get stake token decimals. Possible RPC error.")?;
+            DynProvider::new(self.provider.clone()),
+            prover_addr,
+        ));
+        let collateral_token_decimals = market
+            .collateral_token_decimals()
+            .await
+            .context("Failed to get stake token decimals. Possible RPC error.")?;
 
         // Spin up the order picker to pre-flight and find orders to lock
         let order_picker = Arc::new(order_picker::OrderPicker::new(
@@ -939,6 +942,7 @@ where
                 config.clone(),
                 order_state_tx.clone(),
                 self.priority_requestors.clone(),
+                market.clone(),
             )
             .await
             .context("Failed to initialize proving service")?,
@@ -954,9 +958,6 @@ where
             Ok(())
         });
 
-        let prover_addr =
-            self.args.private_key.as_ref().expect("Private key must be set").address();
-
         let order_monitor = Arc::new(order_monitor::OrderMonitor::new(
             self.db.clone(),
             self.provider.clone(),

crates/broker/src/order_monitor.rs

@@ -469,8 +469,16 @@ where
             } else if !is_within_deadline(&order, current_block_timestamp, min_deadline) {
                 self.skip_order(&order, "expired").await;
             } else if is_target_time_reached(&order, current_block_timestamp) {
-                tracing::info!("Request 0x{:x} was locked by another prover but expired unfulfilled, setting status to pending proving", order.request.id);
-                candidate_orders.push(order);
+                if self.market.is_fulfilled(order.request.id).await? {
+                    tracing::debug!(
+                    "Lock expiry timeout occurred, but 0x{:x} was already fulfilled by another prover. Skipping.",
+                    order.request.id
+                );
+                    self.skip_order(&order, "was fulfilled by other").await;
+                } else {
+                    tracing::info!("Request 0x{:x} was locked by another prover but expired unfulfilled, setting status to pending proving", order.request.id);
+                    candidate_orders.push(order);
+                }
             }
         }