Update to use op-alloy flashblock types

Author: 0x00101010

Justfile

@@ -4,4 +4,13 @@ build:
 
 # Build docker image (debug)
 build-debug:
-    docker buildx build --build-arg RELEASE=false -t flashbots/rollup-boost:develop .
+    docker buildx build --build-arg RELEASE=false -t flashbots/rollup-boost:develop .
+
+clippy:
+    cargo clippy --workspace -- -D warnings
+
+fmt:
+    cargo fmt --all
+
+test:
+    cargo nextest run --workspace

crates/rollup-boost-types/src/flashblocks.rs

@@ -7,6 +7,7 @@ use futures::{SinkExt, StreamExt};
 use lru::LruCache;
 use rollup_boost_types::flashblocks::FlashblocksPayloadV1;
 use std::io::ErrorKind::TimedOut;
+use op_alloy_rpc_types_engine::OpFlashblockPayload;
 use std::num::NonZeroUsize;
 use std::sync::Arc;
 use std::sync::Mutex;
@@ -44,23 +45,23 @@ enum FlashblocksReceiverError {
     TaskPanic(String),
 
     #[error("Failed to send message to sender: {0}")]
-    SendError(#[from] Box<tokio::sync::mpsc::error::SendError<FlashblocksPayloadV1>>),
+    SendError(#[from] Box<tokio::sync::mpsc::error::SendError<OpFlashblockPayload>>),
 
     #[error("Ping mutex poisoned")]
     MutexPoisoned,
 }
 
 pub struct FlashblocksReceiverService {
     url: Url,
-    sender: mpsc::Sender<FlashblocksPayloadV1>,
+    sender: mpsc::Sender<OpFlashblockPayload>,
     websocket_config: FlashblocksWebsocketConfig,
     metrics: FlashblocksWsInboundMetrics,
 }
 
 impl FlashblocksReceiverService {
     pub fn new(
         url: Url,
-        sender: mpsc::Sender<FlashblocksPayloadV1>,
+        sender: mpsc::Sender<OpFlashblockPayload>,
         websocket_config: FlashblocksWebsocketConfig,
     ) -> Self {
         Self {
@@ -167,7 +168,7 @@ impl FlashblocksReceiverService {
                             Some(Ok(msg)) => match msg {
                                 Message::Text(text) => {
                                     metrics.messages_received.increment(1);
-                                    match serde_json::from_str::<FlashblocksPayloadV1>(&text) {
+                                    match serde_json::from_str::<OpFlashblockPayload>(&text) {
                                         Ok(flashblocks_msg) => sender.send(flashblocks_msg).await.map_err(|e| {
                                                 FlashblocksReceiverError::SendError(Box::new(e))
                                             })?,
@@ -254,12 +255,12 @@ mod tests {
         addr: SocketAddr,
     ) -> eyre::Result<(
         watch::Sender<bool>,
-        mpsc::Sender<FlashblocksPayloadV1>,
+        mpsc::Sender<OpFlashblockPayload>,
         mpsc::Receiver<()>,
         url::Url,
     )> {
         let (term_tx, mut term_rx) = watch::channel(false);
-        let (send_tx, mut send_rx) = mpsc::channel::<FlashblocksPayloadV1>(100);
+        let (send_tx, mut send_rx) = mpsc::channel::<OpFlashblockPayload>(100);
         let (send_ping_tx, send_ping_rx) = mpsc::channel::<()>(100);
 
         let listener = TcpListener::bind(addr)?;
@@ -423,12 +424,12 @@ mod tests {
 
         // Send a message to the websocket server
         send_msg
-            .send(FlashblocksPayloadV1::default())
+            .send(OpFlashblockPayload::default())
             .await
             .expect("message sent to websocket server");
 
         let msg = rx.recv().await.expect("message received from websocket");
-        assert_eq!(msg, FlashblocksPayloadV1::default());
+        assert_eq!(msg, OpFlashblockPayload::default());
 
         // Drop the websocket server and start another one with the same address
         // The FlashblocksReceiverService should reconnect to the new server
@@ -440,12 +441,12 @@ mod tests {
         // start a new server with the same address
         let (term, send_msg, _, _url) = start(addr).await?;
         send_msg
-            .send(FlashblocksPayloadV1::default())
+            .send(OpFlashblockPayload::default())
             .await
             .expect("message sent to websocket server");
 
         let msg = rx.recv().await.expect("message received from websocket");
-        assert_eq!(msg, FlashblocksPayloadV1::default());
+        assert_eq!(msg, OpFlashblockPayload::default());
         term.send(true).expect("termination signal sent");
 
         Ok(())

crates/rollup-boost/src/flashblocks/inbound.rs

@@ -6,7 +6,7 @@ use core::{
     task::{Context, Poll},
 };
 use futures::{Sink, SinkExt, StreamExt};
-use rollup_boost_types::flashblocks::FlashblocksPayloadV1;
+use op_alloy_rpc_types_engine::OpFlashblockPayload;
 use std::{io, net::TcpListener, sync::Arc};
 use tokio::{
     net::TcpStream,
@@ -22,7 +22,7 @@ use tokio_tungstenite::{accept_async, tungstenite::Message};
 /// A WebSockets publisher that accepts connections from client websockets and broadcasts to them
 /// updates about new flashblocks. It maintains a count of sent messages and active subscriptions.
 ///
-/// This is modelled as a `futures::Sink` that can be used to send `FlashblocksPayloadV1` messages.
+/// This is modelled as a `futures::Sink` that can be used to send `OpFlashblockPayload` messages.
 pub struct WebSocketPublisher {
     sent: Arc<AtomicUsize>,
     subs: Arc<AtomicUsize>,
@@ -55,7 +55,7 @@ impl WebSocketPublisher {
         })
     }
 
-    pub fn publish(&self, payload: &FlashblocksPayloadV1) -> io::Result<()> {
+    pub fn publish(&self, payload: &OpFlashblockPayload) -> io::Result<()> {
         // Serialize the payload to a UTF-8 string
         // serialize only once, then just copy around only a pointer
         // to the serialized data for each subscription.
@@ -223,14 +223,14 @@ impl Debug for WebSocketPublisher {
     }
 }
 
-impl Sink<&FlashblocksPayloadV1> for WebSocketPublisher {
+impl Sink<&OpFlashblockPayload> for WebSocketPublisher {
     type Error = eyre::Report;
 
     fn poll_ready(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
         Poll::Ready(Ok(()))
     }
 
-    fn start_send(self: Pin<&mut Self>, item: &FlashblocksPayloadV1) -> Result<(), Self::Error> {
+    fn start_send(self: Pin<&mut Self>, item: &OpFlashblockPayload) -> Result<(), Self::Error> {
         self.publish(item)?;
         Ok(())
     }

crates/rollup-boost/src/flashblocks/outbound.rs

@@ -13,10 +13,10 @@ use op_alloy_rpc_types_engine::{
     OpExecutionPayloadEnvelopeV3, OpExecutionPayloadEnvelopeV4, OpExecutionPayloadV4,
     OpPayloadAttributes,
 };
-use rollup_boost_types::flashblocks::{
-    ExecutionPayloadBaseV1, ExecutionPayloadFlashblockDeltaV1, FlashblocksPayloadV1,
+use op_alloy_rpc_types_engine::{
+    OpFlashblockPayload, OpFlashblockPayloadBase, OpFlashblockPayloadDelta,
 };
-use rollup_boost_types::payload::{NewPayload, OpExecutionPayloadEnvelope, PayloadVersion};
+use reth_optimism_payload_builder::payload_id_optimism;
 use std::io;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicU64, Ordering};
@@ -42,21 +42,21 @@ pub enum FlashblocksError {
 // Simplify actor messages to just handle shutdown
 #[derive(Debug)]
 enum FlashblocksEngineMessage {
-    FlashblocksPayloadV1(FlashblocksPayloadV1),
+    OpFlashblockPayload(OpFlashblockPayload),
 }
 
 #[derive(Clone, Debug, Default)]
 pub struct FlashblockBuilder {
-    base: Option<ExecutionPayloadBaseV1>,
-    flashblocks: Vec<ExecutionPayloadFlashblockDeltaV1>,
+    base: Option<OpFlashblockPayloadBase>,
+    flashblocks: Vec<OpFlashblockPayloadDelta>,
 }
 
 impl FlashblockBuilder {
     pub fn new() -> Self {
         Self::default()
     }
 
-    pub fn extend(&mut self, payload: FlashblocksPayloadV1) -> Result<(), FlashblocksError> {
+    pub fn extend(&mut self, payload: OpFlashblockPayload) -> Result<(), FlashblocksError> {
         tracing::debug!(message = "Extending payload", payload_id = %payload.payload_id, index = payload.index, has_base=payload.base.is_some());
 
         // Validate the index is contiguous
@@ -239,7 +239,7 @@ impl FlashblocksService {
 
     async fn on_event(&mut self, event: FlashblocksEngineMessage) {
         match event {
-            FlashblocksEngineMessage::FlashblocksPayloadV1(payload) => {
+            FlashblocksEngineMessage::OpFlashblockPayload(payload) => {
                 self.metrics.messages_processed.increment(1);
 
                 tracing::debug!(
@@ -297,9 +297,9 @@ impl FlashblocksService {
         }
     }
 
-    pub async fn run(&mut self, mut stream: mpsc::Receiver<FlashblocksPayloadV1>) {
+    pub async fn run(&mut self, mut stream: mpsc::Receiver<OpFlashblockPayload>) {
         while let Some(event) = stream.recv().await {
-            self.on_event(FlashblocksEngineMessage::FlashblocksPayloadV1(event))
+            self.on_event(FlashblocksEngineMessage::OpFlashblockPayload(event))
                 .await;
         }
     }
@@ -461,7 +461,7 @@ mod tests {
         let mut builder = FlashblockBuilder::new();
 
         // Error: First payload must have a base
-        let result = builder.extend(FlashblocksPayloadV1 {
+        let result = builder.extend(OpFlashblockPayload {
             payload_id: PayloadId::default(),
             index: 0,
             ..Default::default()
@@ -470,21 +470,21 @@ mod tests {
         assert_eq!(result.unwrap_err(), FlashblocksError::MissingBasePayload);
 
         // Ok: First payload is correct if it has base and index 0
-        let result = builder.extend(FlashblocksPayloadV1 {
+        let result = builder.extend(OpFlashblockPayload {
             payload_id: PayloadId::default(),
             index: 0,
-            base: Some(ExecutionPayloadBaseV1 {
+            base: Some(OpFlashblockPayloadBase {
                 ..Default::default()
             }),
             ..Default::default()
         });
         assert!(result.is_ok());
 
         // Error: First payload must have index 0
-        let result = builder.extend(FlashblocksPayloadV1 {
+        let result = builder.extend(OpFlashblockPayload {
             payload_id: PayloadId::default(),
             index: 1,
-            base: Some(ExecutionPayloadBaseV1 {
+            base: Some(OpFlashblockPayloadBase {
                 ..Default::default()
             }),
             ..Default::default()
@@ -493,7 +493,7 @@ mod tests {
         assert_eq!(result.unwrap_err(), FlashblocksError::UnexpectedBasePayload);
 
         // Error: Second payload must have a follow-up index
-        let result = builder.extend(FlashblocksPayloadV1 {
+        let result = builder.extend(OpFlashblockPayload {
             payload_id: PayloadId::default(),
             index: 2,
             base: None,
@@ -503,7 +503,7 @@ mod tests {
         assert_eq!(result.unwrap_err(), FlashblocksError::InvalidIndex);
 
         // Ok: Second payload has the correct index
-        let result = builder.extend(FlashblocksPayloadV1 {
+        let result = builder.extend(OpFlashblockPayload {
             payload_id: PayloadId::default(),
             index: 1,
             base: None,