Push tbv via block processor (#17)

* Best true value pusher generic.
Moved redis code to redis_backend

* Working!

* tests

Author: ZanCorDX

src/blocks_processor.rs

@@ -198,7 +198,6 @@ impl<HttpClientType: ClientT> BlocksProcessorClient<HttpClientType> {
         Ok(())
     }
 
-    /// T is parametric just because it too BIG to type
     fn handle_rpc_error(err: &jsonrpsee::core::Error, request: &ConsumeBuiltBlockRequest) {
         match err {
             jsonrpsee::core::Error::Call(error_object) => {

src/flashbots_config.rs

@@ -4,7 +4,6 @@
 use alloy_signer_local::PrivateKeySigner;
 use eyre::Context;
 use http::StatusCode;
-use jsonrpsee::http_client::HttpClientBuilder;
 use jsonrpsee::RpcModule;
 use rbuilder::building::builders::merging_builder::merging_build_backtest;
 use rbuilder::building::builders::UnfinishedBlockBuildingSinkFactory;
@@ -36,7 +35,6 @@ use reth_db::DatabaseEnv;
 use serde::Deserialize;
 use serde_with::serde_as;
 use tokio_util::sync::CancellationToken;
-use tower::ServiceBuilder;
 use tracing::{error, warn};
 use url::Url;
 
@@ -48,7 +46,6 @@ use crate::blocks_processor::{
     SIGNED_BLOCK_CONSUME_BUILT_BLOCK_METHOD,
 };
 use crate::build_info::rbuilder_version;
-use crate::flashbots_signer::FlashbotsSignerLayer;
 use crate::true_block_value_push::unfinished_block_building_sink_factory_wrapper::UnfinishedBlockBuildingSinkFactoryWrapper;
 
 use clickhouse::Client;
@@ -103,12 +100,16 @@ pub struct FlashbotsConfig {
     /// selected builder configurations
     pub builders: Vec<BuilderConfig>,
 
-    /// If this is Some then signed mode is used for blocks_processor and tbv_push is done via blocks_processor_url (signed block-processor also handles flashbots_reportBestTrueValue)
+    /// If this is Some then blocks_processor_url MUST be some and:
+    /// - signed mode is used for blocks_processor.
+    /// - tbv_push is done via blocks_processor_url (signed block-processor also handles flashbots_reportBestTrueValue).
     pub key_registration_url: Option<String>,
 
     pub blocks_processor_url: Option<String>,
 
-    /// For production: Some <=> key_registration_url is Some since it's used by smart-multiplexing.
+    /// Cfg to push tbv to redis.
+    /// For production we always need some tbv push (since it's used by smart-multiplexing.) so:
+    /// !Some(key_registration_url) => Some(tbv_push_redis)
     tbv_push_redis: Option<TBVPushRedisConfig>,
 }
 
@@ -251,30 +252,31 @@ impl FlashbotsConfig {
         &self,
         block_processor_key: Option<PrivateKeySigner>,
     ) -> eyre::Result<Box<dyn BidObserver + Send + Sync>> {
-        let bid_observer: Box<dyn BidObserver + Send + Sync> =
-            if let Some(url) = &self.blocks_processor_url {
-                if let Some(block_processor_key) = block_processor_key {
-                    let signing_middleware = FlashbotsSignerLayer::new(block_processor_key);
-                    let service_builder = ServiceBuilder::new()
-                        // map signer errors to http errors
-                        .map_err(jsonrpsee::http_client::transport::Error::Http)
-                        .layer(signing_middleware);
-                    let client = HttpClientBuilder::default()
-                        .set_middleware(service_builder)
-                        .build(url)?;
-                    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)?;
-                    Box::new(BlocksProcessorClientBidObserver::new(client))
-                }
+        let bid_observer: Box<dyn BidObserver + Send + Sync> = if let Some(url) =
+            &self.blocks_processor_url
+        {
+            if let Some(block_processor_key) = block_processor_key {
+                let client = crate::signed_http_client::create_client(url, block_processor_key)?;
+                let block_processor =
+                    BlocksProcessorClient::new(client, SIGNED_BLOCK_CONSUME_BUILT_BLOCK_METHOD);
+                Box::new(BlocksProcessorClientBidObserver::new(block_processor))
             } else {
-                Box::new(NullBidObserver {})
-            };
+                let client = BlocksProcessorClient::try_from(url)?;
+                Box::new(BlocksProcessorClientBidObserver::new(client))
+            }
+        } else {
+            if block_processor_key.is_some() {
+                return Self::bail_blocks_processor_url_not_set();
+            }
+            Box::new(NullBidObserver {})
+        };
         Ok(bid_observer)
     }
 
+    fn bail_blocks_processor_url_not_set<T>() -> Result<T, eyre::Report> {
+        eyre::bail!("blocks_processor_url should always be set if key_registration_url is set");
+    }
+
     /// Connects (UnfinishedBlockBuildingSinkFactoryWrapper->BlockSealingBidderFactory)->RelaySubmitSinkFactory
     /// RelaySubmitSinkFactory: submits final blocks to relays
     /// BlockSealingBidderFactory: performs sealing/bidding. Sends bids to the RelaySubmitSinkFactory
@@ -290,12 +292,15 @@ impl FlashbotsConfig {
         Arc<dyn BiddingServiceWinControl>,
     )> {
         let block_processor_key = if let Some(key_registration_url) = &self.key_registration_url {
+            if self.blocks_processor_url.is_none() {
+                return Self::bail_blocks_processor_url_not_set();
+            }
             Some(self.register_key(key_registration_url).await?)
         } else {
             None
         };
         // RelaySubmitSinkFactory
-        let bid_observer = self.create_block_processor_client(block_processor_key)?;
+        let bid_observer = self.create_block_processor_client(block_processor_key.clone())?;
         let (sink_sealed_factory, relays) = self
             .l1_config
             .create_relays_sealed_sink_factory(self.base_config.chain_spec()?, bid_observer)?;
@@ -321,32 +326,56 @@ impl FlashbotsConfig {
         ));
 
         // UnfinishedBlockBuildingSinkFactoryWrapper
-        let wrapped_sink_factory =
-            self.wrap_with_tbv_redis_pusher(&cancellation_token, &bid_value_source, sink_factory)?;
+        let wrapped_sink_factory = self.wrap_with_tbv_pusher(
+            &cancellation_token,
+            &bid_value_source,
+            block_processor_key,
+            sink_factory,
+        )?;
 
         Ok((wrapped_sink_factory, relays, bidding_service_win_control))
     }
 
-    fn wrap_with_tbv_redis_pusher(
+    /// Wraps the factory with one that sends a to TBV stream to our infra.
+    /// block_processor_key == Some -> We use signed block processor API.
+    /// block_processor_key == None -> If Some(tbv_push_redis) -> We send directly to a redis channel.
+    fn wrap_with_tbv_pusher(
         &self,
         cancellation_token: &CancellationToken,
         bid_value_source: &Arc<dyn BidValueSource + Send + Sync>,
+        block_processor_key: Option<PrivateKeySigner>,
         factory: Box<dyn UnfinishedBlockBuildingSinkFactory>,
     ) -> eyre::Result<Box<dyn UnfinishedBlockBuildingSinkFactory>> {
-        if let Some(cfg) = &self.tbv_push_redis {
+        if let Some(block_processor_key) = block_processor_key {
+            if let Some(blocks_processor_url) = &self.blocks_processor_url {
+                Ok(Box::new(
+                    UnfinishedBlockBuildingSinkFactoryWrapper::new_block_processor(
+                        factory,
+                        bid_value_source.clone(),
+                        blocks_processor_url.clone(),
+                        block_processor_key,
+                        cancellation_token.clone(),
+                    )?,
+                ))
+            } else {
+                Self::bail_blocks_processor_url_not_set()
+            }
+        } else if let Some(cfg) = &self.tbv_push_redis {
             let tbv_push_redis_url_value = cfg
                 .url
                 .as_ref()
                 .ok_or(eyre::Report::msg("Missing tbv_push_redis_url"))?
                 .value()
                 .context("tbv_push_redis_url")?;
-            Ok(Box::new(UnfinishedBlockBuildingSinkFactoryWrapper::new(
-                factory,
-                bid_value_source.clone(),
-                tbv_push_redis_url_value,
-                cfg.channel.clone(),
-                cancellation_token.clone(),
-            )?))
+            Ok(Box::new(
+                UnfinishedBlockBuildingSinkFactoryWrapper::new_redis(
+                    factory,
+                    bid_value_source.clone(),
+                    tbv_push_redis_url_value,
+                    cfg.channel.clone(),
+                    cancellation_token.clone(),
+                )?,
+            ))
         } else {
             Ok(factory)
         }

src/flashbots_signer.rs

@@ -109,27 +109,27 @@ where
                 "{:?}:0x{}",
                 signer.address(),
                 hex::encode(signature.as_bytes())
-            ))
-            .expect("Header contains invalid characters");
+            ))?;
             parts.headers.insert(FLASHBOTS_HEADER.clone(), header_val);
 
             let request = Request::from_parts(parts, Body::from(body_bytes.clone()));
             inner.call(request).await.map_err(Into::into)
         })
     }
 }
-/*
+
 #[cfg(test)]
 mod tests {
     use super::*;
+    use alloy_signer_local::PrivateKeySigner;
     use http::Response;
     use hyper::Body;
     use std::convert::Infallible;
     use tower::{service_fn, ServiceExt};
 
     #[tokio::test]
     async fn test_signature() {
-        let fb_signer = LocalWallet::new(&mut thread_rng());
+        let fb_signer = PrivateKeySigner::random();
 
         // mock service that returns the request headers
         let svc = FlashbotsSigner {
@@ -164,19 +164,18 @@ mod tests {
 
         let signer_address = format!("{:?}", fb_signer.address());
         let expected_signature = fb_signer
-            .sign_message(format!("0x{:x}", H256::from(keccak256(bytes.clone()))))
+            .sign_message(format!("{:?}", keccak256(bytes.clone())).as_bytes())
             .await
-            .unwrap()
-            .to_string();
-
+            .unwrap();
+        let expected_signature = hex::encode(expected_signature.as_bytes());
         // verify that the header contains expected address and signature
         assert_eq!(header_address, signer_address);
         assert_eq!(header_signature, expected_signature);
     }
 
     #[tokio::test]
     async fn test_skips_non_json() {
-        let fb_signer = LocalWallet::new(&mut thread_rng());
+        let fb_signer = PrivateKeySigner::random();
 
         // mock service that returns the request headers
         let svc = FlashbotsSigner {
@@ -210,7 +209,7 @@ mod tests {
 
     #[tokio::test]
     async fn test_returns_error_when_not_post() {
-        let fb_signer = LocalWallet::new(&mut thread_rng());
+        let fb_signer = PrivateKeySigner::random();
 
         // mock service that returns the request headers
         let svc = FlashbotsSigner {
@@ -240,5 +239,52 @@ mod tests {
         // should be an error
         assert!(res.is_err());
     }
+
+    /// Uses a static private key and compares the signature generated by this package to the signature
+    /// generated by the `cast` CLI.
+    /// Test copied from https://github.com/flashbots/go-utils/blob/main/signature/signature_test.go#L146 501d395be6a9802494ef1ef25a755acaa4448c17 (TestSignatureCreateCompareToCastAndEthers)
+    #[tokio::test]
+    async fn test_signature_cast() {
+        let fb_signer: PrivateKeySigner =
+            "fad9c8855b740a0b7ed4c221dbad0f33a83a49cad6b3fe8d5817ac83d38b6a19"
+                .parse()
+                .unwrap();
+        // mock service that returns the request headers
+        let svc = FlashbotsSigner {
+            signer: fb_signer.clone(),
+            inner: service_fn(|_req: Request<Body>| async {
+                let (parts, _) = _req.into_parts();
+
+                let mut res = Response::builder();
+                for (k, v) in parts.headers.iter() {
+                    res = res.header(k, v);
+                }
+                let res = res.body(Body::empty()).unwrap();
+                Ok::<_, Infallible>(res)
+            }),
+        };
+
+        // build request
+        let bytes = "Hello".as_bytes();
+        let req = Request::builder()
+            .method(http::Method::POST)
+            .header(http::header::CONTENT_TYPE, "application/json")
+            .body(Body::from(bytes))
+            .unwrap();
+
+        let res = svc.oneshot(req).await.unwrap();
+
+        let header = res.headers().get("x-flashbots-signature").unwrap();
+        let header = header.to_str().unwrap();
+        let header = header.split(":0x").collect::<Vec<_>>();
+        let header_address = header[0];
+        let header_signature = header[1];
+        // I generated the signature using the cast CLI:
+        // 	cast wallet sign --private-key fad9c8855b740a0b7ed4c221dbad0f33a83a49cad6b3fe8d5817ac83d38b6a19  $(cast from-utf8 $(cast keccak Hello))
+        let signer_address = "0x96216849c49358B10257cb55b28eA603c874b05E".to_lowercase();
+        let expected_signature = "1446053488f02d460c012c84c4091cd5054d98c6cfca01b65f6c1a72773e80e60b8a4931aeee7ed18ce3adb45b2107e8c59e25556c1f871a8334e30e5bddbed21c";
+        // verify that the header contains expected address and signature
+        assert_eq!(header_address, signer_address);
+        assert_eq!(header_signature, expected_signature);
+    }
 }
-*/

src/lib.rs

@@ -6,4 +6,5 @@ pub mod build_info;
 pub mod flashbots_config;
 pub mod flashbots_signer;
 pub mod metrics;
+pub mod signed_http_client;
 mod true_block_value_push;

src/signed_http_client.rs

@@ -0,0 +1,29 @@
+use crate::flashbots_signer::{FlashbotsSigner, FlashbotsSignerLayer};
+use alloy_signer_local::PrivateKeySigner;
+use jsonrpsee::http_client::transport::Error as JsonError;
+use jsonrpsee::http_client::HttpClientBuilder;
+use jsonrpsee::http_client::{transport::HttpBackend, HttpClient};
+use tower::ServiceBuilder;
+type MapErrorFn = fn(Box<dyn std::error::Error + Send + Sync>) -> JsonError;
+
+const fn map_error(err: Box<dyn std::error::Error + Send + Sync>) -> JsonError {
+    JsonError::Http(err)
+}
+
+pub type SignedHttpClient =
+    HttpClient<tower::util::MapErr<FlashbotsSigner<PrivateKeySigner, HttpBackend>, MapErrorFn>>;
+
+pub fn create_client(
+    url: &str,
+    signer: PrivateKeySigner,
+) -> Result<SignedHttpClient, jsonrpsee::core::Error> {
+    let signing_middleware = FlashbotsSignerLayer::new(signer);
+    let service_builder = ServiceBuilder::new()
+        // Coerce to function pointer and remove the + 'static added to the closure
+        .map_err(map_error as MapErrorFn)
+        .layer(signing_middleware);
+    let client = HttpClientBuilder::default()
+        .set_middleware(service_builder)
+        .build(url)?;
+    Ok(client)
+}