Fix an unbound loop when using a misbehaving relay (#423)

Author: jclapis

crates/pbs/src/mev_boost/submit_block.rs

@@ -98,6 +98,7 @@ async fn submit_block_with_timeout(
     let mut remaining_timeout_ms = timeout_ms;
     let mut retry = 0;
     let mut backoff = Duration::from_millis(250);
+    let mut request_api_version = api_version;
 
     loop {
         let start_request = Instant::now();
@@ -108,12 +109,21 @@ async fn submit_block_with_timeout(
             headers.clone(),
             remaining_timeout_ms,
             retry,
-            &api_version,
+            &request_api_version,
             fork_name,
         )
         .await
         {
-            Ok(response) => return Ok(response),
+            Ok(response) => {
+                // If the original request was for v2 but we had to fall back to v1, return a v2
+                // response
+                if request_api_version == BuilderApiVersion::V1 &&
+                    api_version != request_api_version
+                {
+                    return Ok(None);
+                }
+                return Ok(response);
+            }
 
             Err(err) if err.should_retry() => {
                 tokio::time::sleep(backoff).await;
@@ -127,12 +137,15 @@ async fn submit_block_with_timeout(
                 }
             }
 
-            Err(err) if err.is_not_found() && matches!(api_version, BuilderApiVersion::V2) => {
+            Err(err)
+                if err.is_not_found() && matches!(request_api_version, BuilderApiVersion::V2) =>
+            {
                 warn!(
                     relay_id = relay.id.as_ref(),
                     "relay does not support v2 endpoint, retrying with v1"
                 );
                 url = relay.submit_block_url(BuilderApiVersion::V1)?;
+                request_api_version = BuilderApiVersion::V1;
             }
 
             Err(err) => return Err(err),

tests/src/mock_relay.rs

@@ -46,6 +46,8 @@ pub struct MockRelayState {
     pub chain: Chain,
     pub signer: BlsSecretKey,
     large_body: bool,
+    supports_submit_block_v2: bool,
+    use_not_found_for_submit_block: bool,
     received_get_header: Arc<AtomicU64>,
     received_get_status: Arc<AtomicU64>,
     received_register_validator: Arc<AtomicU64>,
@@ -69,6 +71,12 @@ impl MockRelayState {
     pub fn large_body(&self) -> bool {
         self.large_body
     }
+    pub fn supports_submit_block_v2(&self) -> bool {
+        self.supports_submit_block_v2
+    }
+    pub fn use_not_found_for_submit_block(&self) -> bool {
+        self.use_not_found_for_submit_block
+    }
     pub fn set_response_override(&self, status: StatusCode) {
         *self.response_override.write().unwrap() = Some(status);
     }
@@ -80,6 +88,8 @@ impl MockRelayState {
             chain,
             signer,
             large_body: false,
+            supports_submit_block_v2: true,
+            use_not_found_for_submit_block: false,
             received_get_header: Default::default(),
             received_get_status: Default::default(),
             received_register_validator: Default::default(),
@@ -91,6 +101,14 @@ impl MockRelayState {
     pub fn with_large_body(self) -> Self {
         Self { large_body: true, ..self }
     }
+
+    pub fn with_no_submit_block_v2(self) -> Self {
+        Self { supports_submit_block_v2: false, ..self }
+    }
+
+    pub fn with_not_found_for_submit_block(self) -> Self {
+        Self { use_not_found_for_submit_block: true, ..self }
+    }
 }
 
 pub fn mock_relay_app_router(state: Arc<MockRelayState>) -> Router {
@@ -100,7 +118,11 @@ pub fn mock_relay_app_router(state: Arc<MockRelayState>) -> Router {
         .route(REGISTER_VALIDATOR_PATH, post(handle_register_validator))
         .route(SUBMIT_BLOCK_PATH, post(handle_submit_block_v1));
 
-    let v2_builder_routes = Router::new().route(SUBMIT_BLOCK_PATH, post(handle_submit_block_v2));
+    let v2_builder_routes = if state.supports_submit_block_v2 {
+        Router::new().route(SUBMIT_BLOCK_PATH, post(handle_submit_block_v2))
+    } else {
+        Router::new()
+    };
 
     let builder_router_v1 = Router::new().nest(BUILDER_V1_API_PATH, v1_builder_routes);
     let builder_router_v2 = Router::new().nest(BUILDER_V2_API_PATH, v2_builder_routes);
@@ -165,6 +187,9 @@ async fn handle_submit_block_v1(
     State(state): State<Arc<MockRelayState>>,
     Json(submit_block): Json<SignedBlindedBeaconBlock>,
 ) -> Response {
+    if state.use_not_found_for_submit_block() {
+        return StatusCode::NOT_FOUND.into_response();
+    }
     state.received_submit_block.fetch_add(1, Ordering::Relaxed);
     if state.large_body() {
         (StatusCode::OK, Json(vec![1u8; 1 + MAX_SIZE_SUBMIT_BLOCK_RESPONSE])).into_response()
@@ -192,6 +217,9 @@ async fn handle_submit_block_v1(
     }
 }
 async fn handle_submit_block_v2(State(state): State<Arc<MockRelayState>>) -> Response {
+    if state.use_not_found_for_submit_block() {
+        return StatusCode::NOT_FOUND.into_response();
+    }
     state.received_submit_block.fetch_add(1, Ordering::Relaxed);
     (StatusCode::ACCEPTED, "").into_response()
 }

tests/tests/pbs_post_blinded_blocks.rs

@@ -17,7 +17,7 @@ use tracing::info;
 
 #[tokio::test]
 async fn test_submit_block_v1() -> Result<()> {
-    let res = submit_block_impl(3800, &BuilderApiVersion::V1).await?;
+    let res = submit_block_impl(3800, &BuilderApiVersion::V1, false, false).await?;
     assert_eq!(res.status(), StatusCode::OK);
 
     let signed_blinded_block = load_test_signed_blinded_block();
@@ -32,12 +32,31 @@ async fn test_submit_block_v1() -> Result<()> {
 
 #[tokio::test]
 async fn test_submit_block_v2() -> Result<()> {
-    let res = submit_block_impl(3850, &BuilderApiVersion::V2).await?;
+    let res = submit_block_impl(3802, &BuilderApiVersion::V2, false, false).await?;
     assert_eq!(res.status(), StatusCode::ACCEPTED);
     assert_eq!(res.bytes().await?.len(), 0);
     Ok(())
 }
 
+// Test that when submitting a block using v2 to a relay that does not support
+// v2, PBS falls back to v1 and successfully submits the block.
+#[tokio::test]
+async fn test_submit_block_v2_without_relay_support() -> Result<()> {
+    let res = submit_block_impl(3804, &BuilderApiVersion::V2, true, false).await?;
+    assert_eq!(res.status(), StatusCode::ACCEPTED);
+    assert_eq!(res.bytes().await?.len(), 0);
+    Ok(())
+}
+
+// Test that when submitting a block using v2 to a relay that returns 404s
+// for both v1 and v2, PBS doesn't loop forever.
+#[tokio::test]
+async fn test_submit_block_on_broken_relay() -> Result<()> {
+    let res = submit_block_impl(3806, &BuilderApiVersion::V2, true, true).await?;
+    assert_eq!(res.status(), StatusCode::BAD_GATEWAY);
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_submit_block_too_large() -> Result<()> {
     setup_test_env();
@@ -68,7 +87,12 @@ async fn test_submit_block_too_large() -> Result<()> {
     Ok(())
 }
 
-async fn submit_block_impl(pbs_port: u16, api_version: &BuilderApiVersion) -> Result<Response> {
+async fn submit_block_impl(
+    pbs_port: u16,
+    api_version: &BuilderApiVersion,
+    remove_v2_support: bool,
+    force_404s: bool,
+) -> Result<Response> {
     setup_test_env();
     let signer = random_secret();
     let pubkey = signer.public_key();
@@ -77,7 +101,14 @@ async fn submit_block_impl(pbs_port: u16, api_version: &BuilderApiVersion) -> Re
 
     // Run a mock relay
     let relays = vec![generate_mock_relay(pbs_port + 1, pubkey)?];
-    let mock_state = Arc::new(MockRelayState::new(chain, signer));
+    let mut mock_relay_state = MockRelayState::new(chain, signer);
+    if remove_v2_support {
+        mock_relay_state = mock_relay_state.with_no_submit_block_v2();
+    }
+    if force_404s {
+        mock_relay_state = mock_relay_state.with_not_found_for_submit_block();
+    }
+    let mock_state = Arc::new(mock_relay_state);
     tokio::spawn(start_mock_relay_service(mock_state.clone(), pbs_port + 1));
 
     // Run the PBS service
@@ -99,6 +130,7 @@ async fn submit_block_impl(pbs_port: u16, api_version: &BuilderApiVersion) -> Re
             mock_validator.do_submit_block_v2(Some(signed_blinded_block)).await?
         }
     };
-    assert_eq!(mock_state.received_submit_block(), 1);
+    let expected_count = if force_404s { 0 } else { 1 };
+    assert_eq!(mock_state.received_submit_block(), expected_count);
     Ok(res)
 }