BM-1933: Additional Order Fulfillment Checks

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,18 @@ fn extract_tx_log<E: SolEvent + Debug + Clone>(
     }
 }
 
+fn validate_fulfill_receipt(receipt: TransactionReceipt) -> Result<(), MarketError> {
+    if let Some(log) = receipt.decoded_log::<IBoundlessMarket::PaymentRequirementsFailed>() {
+        let raw_error = Bytes::copy_from_slice(&log.error);
+        match IBoundlessMarketErrors::abi_decode(&raw_error) {
+            Ok(err) => Err(MarketError::PaymentRequirementsFailed(err)),
+            Err(_) => Err(MarketError::PaymentRequirementsFailedUnknownError(raw_error)),
+        }
+    } else {
+        Ok(())
+    }
+}
+
 /// Data returned when querying for a RequestSubmitted event
 #[derive(Debug, Clone)]
 pub struct RequestSubmittedEventData {
@@ -729,7 +750,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 +772,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 +805,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 +834,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 +877,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 +920,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 +959,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 +998,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

@@ -21,8 +21,9 @@ use alloy::{
 use alloy_primitives::Bytes;
 use boundless_market::{
     contracts::{
-        boundless_market::{FulfillmentTx, UnlockedRequest},
+        boundless_market::{FulfillmentTx, MarketError, UnlockedRequest},
         hit_points::default_allowance,
+        IBoundlessMarket::IBoundlessMarketErrors,
         AssessorReceipt, FulfillmentData, FulfillmentDataType, Offer, Predicate, ProofRequest,
         RequestId, RequestStatus, Requirements,
     },
@@ -434,11 +435,17 @@ async fn test_e2e_no_payment() {
         };
 
         let balance_before = ctx.prover_market.balance_of(some_other_address).await.unwrap();
-        // fulfill the request.
-        ctx.prover_market
+        // fulfill the request. This call should fail payment requirements since the lock belongs
+        // to a different prover, but the request itself still becomes fulfilled on-chain.
+        let err = ctx
+            .prover_market
             .fulfill(FulfillmentTx::new(vec![fulfillment.clone()], assessor_fill.clone()))
             .await
-            .unwrap();
+            .expect_err("expected payment requirement failure for mismatched locker");
+        match err {
+            MarketError::PaymentRequirementsFailed(IBoundlessMarketErrors::RequestIsLocked(_)) => {}
+            other => panic!("unexpected error when fulfilling with mismatched prover: {other:?}"),
+        }
         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};
@@ -890,16 +890,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(
@@ -932,6 +935,7 @@ where
                 config.clone(),
                 order_state_tx.clone(),
                 self.priority_requestors.clone(),
+                market.clone(),
             )
             .await
             .context("Failed to initialize proving service")?,
@@ -947,9 +951,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);
+                }
             }
         }