Submit delayed payments to block processor (#742)

Author: dvush

Cargo.lock

@@ -178,6 +178,7 @@ tracing = "0.1.37"
 metrics = { version = "0.24.1" }
 ahash = "0.8.6"
 time = { version = "0.3.36", features = ["macros", "formatting", "parsing"] }
+uuid = { version = "1.6.1", features = ["serde", "v5", "v4"] }
 
 eth-sparse-mpt = { path = "crates/eth-sparse-mpt" }
 bid-scraper = { path = "crates/bid-scraper" }
@@ -186,3 +187,4 @@ rbuilder-primitives = { path = "crates/rbuilder-primitives" }
 rbuilder-config = { path = "crates/rbuilder-config" }
 sysperf = { path = "crates/sysperf" }
 metrics_macros = { path = "crates/rbuilder/src/telemetry/metrics_macros" }
+

Cargo.toml

@@ -57,6 +57,7 @@ clickhouse = { version = "0.12.2", features = ["time", "uuid", "native-tls"] }
 futures-util.workspace = true
 parking_lot.workspace = true
 lazy_static.workspace = true
+uuid.workspace = true
 
 [build-dependencies]
 built = { version = "0.7.1", features = ["git2", "chrono"] }

crates/rbuilder-operator/Cargo.toml

@@ -1,9 +1,6 @@
-use alloy_primitives::{BlockHash, U256};
+use alloy_primitives::{Address, BlockHash, U256};
 use exponential_backoff::Backoff;
-use jsonrpsee::{
-    core::{client::ClientT, traits::ToRpcParams},
-    http_client::{HttpClient, HttpClientBuilder},
-};
+use jsonrpsee::core::{client::ClientT, traits::ToRpcParams};
 use rbuilder::{
     building::BuiltBlockTrace,
     live_builder::{
@@ -14,7 +11,7 @@ use rbuilder::{
 use rbuilder_primitives::{
     mev_boost::SubmitBlockRequest,
     serialize::{RawBundle, RawShareBundle},
-    Bundle, Order,
+    Bundle, Order, OrderId,
 };
 use reth_primitives::SealedBlock;
 use serde::{Deserialize, Serialize};
@@ -23,12 +20,12 @@ use serde_with::{serde_as, DisplayFromStr};
 use std::{sync::Arc, time::Duration};
 use time::format_description::well_known;
 use tracing::{error, warn, Span};
+use uuid::Uuid;
 
 use crate::metrics::inc_submit_block_errors;
 
 const BLOCK_PROCESSOR_ERROR_CATEGORY: &str = "block_processor";
-const DEFAULT_BLOCK_CONSUME_BUILT_BLOCK_METHOD: &str = "block_consumeBuiltBlockV2";
-pub const SIGNED_BLOCK_CONSUME_BUILT_BLOCK_METHOD: &str = "flashbots_consumeBuiltBlockV2";
+pub const SIGNED_BLOCK_CONSUME_BUILT_BLOCK_METHOD: &str = "flashbots_consumeBuiltBlockV3";
 
 #[derive(Debug, Serialize, Deserialize)]
 #[serde(rename_all = "camelCase")]
@@ -67,6 +64,14 @@ struct BlocksProcessorHeader {
     pub number: Option<U256>,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Hash)]
+#[serde(rename_all = "camelCase")]
+pub struct BlockProcessorDelayedPayments {
+    pub source: Uuid,
+    pub value: U256,
+    pub address: Address,
+}
+
 type ConsumeBuiltBlockRequest = (
     BlocksProcessorHeader,
     String,
@@ -78,6 +83,7 @@ type ConsumeBuiltBlockRequest = (
     String,
     U256,
     U256,
+    Vec<BlockProcessorDelayedPayments>,
 );
 
 /// Struct to avoid copying ConsumeBuiltBlockRequest since HttpClient::request eats the parameter.
@@ -113,22 +119,6 @@ pub struct BlocksProcessorClient<HttpClientType> {
     consume_built_block_method: &'static str,
 }
 
-impl BlocksProcessorClient<HttpClient> {
-    pub fn try_from(
-        url: &str,
-        max_request_size: u32,
-        max_concurrent_requests: usize,
-    ) -> eyre::Result<Self> {
-        Ok(Self {
-            client: HttpClientBuilder::default()
-                .max_request_size(max_request_size)
-                .max_concurrent_requests(max_concurrent_requests)
-                .build(url)?,
-            consume_built_block_method: DEFAULT_BLOCK_CONSUME_BUILT_BLOCK_METHOD,
-        })
-    }
-}
-
 /// RawBundle::encode_no_blobs but more compatible.
 fn encode_bundle_for_blocks_processor(mut bundle: Bundle) -> RawBundle {
     // set to 0 when none
@@ -187,6 +177,36 @@ impl<HttpClientType: ClientT> BlocksProcessorClient<HttpClientType> {
 
         let used_share_bundles = Self::get_used_sbundles(built_block_trace);
 
+        let delayed_payments = built_block_trace
+            .included_orders
+            .iter()
+            .filter_map(|res| {
+                let delayed_kickback = if let Some(k) = &res.delayed_kickback {
+                    k
+                } else {
+                    return None;
+                };
+
+                if delayed_kickback.should_pay_in_block {
+                    return None;
+                }
+
+                let bundle_uuid = match res.order.id() {
+                    OrderId::Bundle(uuid) => uuid,
+                    _ => {
+                        error!(order = ?res.order.id(), "Delayed kickback is found for non-bundle");
+                        return None;
+                    }
+                };
+
+                Some(BlockProcessorDelayedPayments {
+                    source: bundle_uuid,
+                    value: delayed_kickback.payout_value,
+                    address: delayed_kickback.recipient,
+                })
+            })
+            .collect();
+
         let params: ConsumeBuiltBlockRequest = (
             header,
             closed_at,
@@ -198,6 +218,7 @@ impl<HttpClientType: ClientT> BlocksProcessorClient<HttpClientType> {
             builder_name,
             built_block_trace.true_bid_value,
             best_bid_value,
+            delayed_payments,
         );
         let request = ConsumeBuiltBlockRequestArc::new(params);
         let backoff = backoff();
@@ -375,4 +396,19 @@ mod tests {
         dbg!(total_sleep_time);
         assert!(total_sleep_time > 40 && total_sleep_time < 90);
     }
+
+    #[test]
+    fn test_delayed_payment_serialize() {
+        let value = BlockProcessorDelayedPayments {
+            address: alloy_primitives::address!("93Ea7cB31f76B982601321b2A0d93Ec9A948236D"),
+            value: U256::from(16),
+            source: Uuid::try_parse("ff7b2232-b30d-4889-9258-c3632ba4bfc0").unwrap(),
+        };
+
+        let value_str = serde_json::to_string(&value).unwrap();
+
+        let expected_str = r#"{"source":"ff7b2232-b30d-4889-9258-c3632ba4bfc0","value":"0x10","address":"0x93ea7cb31f76b982601321b2a0d93ec9a948236d"}"#;
+
+        assert_eq!(value_str, expected_str);
+    }
 }

crates/rbuilder-operator/src/blocks_processor.rs

@@ -274,25 +274,22 @@ impl FlashbotsConfig {
         block_processor_key: Option<PrivateKeySigner>,
     ) -> eyre::Result<Option<Box<dyn BidObserver + Send + Sync>>> {
         if let Some(url) = &self.blocks_processor_url {
-            let bid_observer: Box<dyn BidObserver + Send + Sync> =
-                if let Some(block_processor_key) = block_processor_key {
-                    let client = crate::signed_http_client::create_client(
-                        url,
-                        block_processor_key,
-                        self.blocks_processor_max_request_size_bytes,
-                        self.blocks_processor_max_concurrent_requests,
-                    )?;
-                    let block_processor =
-                        BlocksProcessorClient::new(client, SIGNED_BLOCK_CONSUME_BUILT_BLOCK_METHOD);
-                    Box::new(BlocksProcessorClientBidObserver::new(block_processor))
-                } else {
-                    let client = BlocksProcessorClient::try_from(
-                        url,
-                        self.blocks_processor_max_request_size_bytes,
-                        self.blocks_processor_max_concurrent_requests,
-                    )?;
-                    Box::new(BlocksProcessorClientBidObserver::new(client))
-                };
+            let bid_observer: Box<dyn BidObserver + Send + Sync> = if let Some(
+                block_processor_key,
+            ) = block_processor_key
+            {
+                let client = crate::signed_http_client::create_client(
+                    url,
+                    block_processor_key,
+                    self.blocks_processor_max_request_size_bytes,
+                    self.blocks_processor_max_concurrent_requests,
+                )?;
+                let block_processor =
+                    BlocksProcessorClient::new(client, SIGNED_BLOCK_CONSUME_BUILT_BLOCK_METHOD);
+                Box::new(BlocksProcessorClientBidObserver::new(block_processor))
+            } else {
+                eyre::bail!("Unsigned block processing is not supported: if blocks_processor_url is set, a block_processor_key must also be provided");
+            };
             Ok(Some(bid_observer))
         } else {
             if block_processor_key.is_some() {

crates/rbuilder-operator/src/flashbots_config.rs

@@ -101,7 +101,7 @@ mev-share-sse.workspace = true
 beacon-api-client.workspace = true
 ethereum-consensus.workspace = true
 jsonrpsee = { version = "0.20.3", features = ["full"] }
-uuid = { version = "1.6.1", features = ["serde", "v5", "v4"] }
+uuid.workspace = true
 prometheus.workspace = true
 warp.workspace = true
 lazy_static.workspace = true