import flashblocks types through rblib (#48)

Author: akundaz

Cargo.toml

@@ -17,8 +17,7 @@ authors = ["Flashbots <[email protected]>"]
 exclude = [".github/"]
 
 [workspace.dependencies]
-rblib = { git = "https://github.com/flashbots/rblib", rev = "9eccd22047c3f8978ae022d575dca749a81684a8" }
-rollup-boost-types = { git = "https://github.com/flashbots/rollup-boost", rev = "b282527e5f46d0915d1e5bdf44338593bc8beb2d" }
+rblib = { git = "https://github.com/flashbots/rblib", rev = "e9f6a539a5d4bb8b3a2cae5f02d8adc05d126c37" }
 
 futures = "0.3"
 tokio = "1.46"
@@ -67,7 +66,6 @@ debug = ["tokio/full", "tokio/tracing", "dep:console-subscriber"]
 
 [dependencies]
 rblib = { workspace = true }
-rollup-boost-types = { workspace = true }
 
 futures = { workspace = true }
 tokio = { workspace = true }

src/publish.rs

@@ -16,16 +16,27 @@ use {
 	futures::{SinkExt, StreamExt},
 	parking_lot::RwLock,
 	rblib::{
-		alloy::{consensus::BlockHeader, eips::Encodable2718, primitives::U256},
+		alloy::{
+			consensus::BlockHeader,
+			eips::Encodable2718,
+			optimism::rpc_types_engine::{
+				OpFlashblockPayload,
+				OpFlashblockPayloadBase,
+				OpFlashblockPayloadDelta,
+				OpFlashblockPayloadMetadata,
+			},
+			primitives::U256,
+		},
 		prelude::{ext::CheckpointOpExt, *},
 		reth::node::builder::PayloadBuilderAttributes,
 	},
-	rollup_boost_types::flashblocks::{
-		ExecutionPayloadBaseV1,
-		ExecutionPayloadFlashblockDeltaV1,
-		FlashblocksPayloadV1,
+	std::{
+		collections::BTreeMap,
+		io,
+		net::TcpListener,
+		sync::Arc,
+		time::Instant,
 	},
-	std::{io, net::TcpListener, sync::Arc, time::Instant},
 	tokio::{
 		net::TcpStream,
 		sync::{
@@ -58,7 +69,7 @@ pub struct PublishFlashblock {
 	/// Set once at the begining of the payload job, captures immutable
 	/// information about the payload that is being built. This info is derived
 	/// from the payload attributes parameter on the FCU from the EL node.
-	block_base: RwLock<Option<ExecutionPayloadBaseV1>>,
+	block_base: RwLock<Option<OpFlashblockPayloadBase>>,
 
 	/// Metrics for monitoring flashblock publishing.
 	metrics: Metrics,
@@ -107,7 +118,7 @@ impl Step<Flashblocks> for PublishFlashblock {
 		// TODO: Consider moving this into its own step
 		let base = self.block_base.write().take();
 		let (_excess_blob_gas, blob_gas_used) = payload.blob_fields();
-		let diff = ExecutionPayloadFlashblockDeltaV1 {
+		let diff = OpFlashblockPayloadDelta {
 			state_root: sealed_block.block().state_root,
 			receipts_root: sealed_block.block().receipts_root,
 			logs_bloom: sealed_block.block().logs_bloom,
@@ -128,12 +139,17 @@ impl Step<Flashblocks> for PublishFlashblock {
 		let index = flashblock_number.index();
 
 		// Push the contents of the payload
-		if let Err(e) = self.sink.publish(&FlashblocksPayloadV1 {
+		if let Err(e) = self.sink.publish(&OpFlashblockPayload {
 			base,
 			diff,
 			payload_id: ctx.block().payload_id(),
 			index,
-			metadata: serde_json::Value::Null,
+			metadata: OpFlashblockPayloadMetadata {
+				block_number: ctx.block().number(),
+				// TODO: Fill in the following fields
+				new_account_balances: BTreeMap::new(),
+				receipts: BTreeMap::new(),
+			},
 		}) {
 			self.metrics.websocket_publish_errors_total.increment(1);
 			tracing::error!("Failed to publish flashblock to websocket: {e}");
@@ -174,7 +190,7 @@ impl Step<Flashblocks> for PublishFlashblock {
 		self.times.on_job_started(&self.metrics);
 
 		// this remains constant for the entire payload job.
-		self.block_base.write().replace(ExecutionPayloadBaseV1 {
+		self.block_base.write().replace(OpFlashblockPayloadBase {
 			parent_beacon_block_root: ctx
 				.block()
 				.attributes()
@@ -406,7 +422,7 @@ impl WebSocketSink {
 	/// Called once by the `PublishFlashblock` pipeline step every time there is a
 	/// non-empty flashblock that needs to be broadcasted to all external
 	/// subscribers.
-	pub fn publish(&self, payload: &FlashblocksPayloadV1) -> io::Result<usize> {
+	pub fn publish(&self, payload: &OpFlashblockPayload) -> io::Result<usize> {
 		// 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.

src/tests/utils/ws.rs

@@ -7,7 +7,7 @@ use {
 	futures::{FutureExt, StreamExt},
 	itertools::Itertools,
 	orx_concurrent_vec::ConcurrentVec,
-	rollup_boost_types::flashblocks::FlashblocksPayloadV1,
+	rblib::alloy::optimism::rpc_types_engine::OpFlashblockPayload,
 	std::{sync::Arc, time::Instant},
 	tokio::{net::TcpStream, sync::oneshot, task::JoinHandle},
 	tokio_tungstenite::{
@@ -88,7 +88,7 @@ impl Deref for WebSocketObserver {
 #[derive(Clone, Debug, Deref)]
 pub struct ObservedFlashblock {
 	#[deref]
-	pub block: FlashblocksPayloadV1,
+	pub block: OpFlashblockPayload,
 	pub at: Instant,
 }
 
@@ -152,7 +152,7 @@ impl Observations {
 
 	fn record_message(&self, msg: Message) {
 		if let Message::Text(string) = &msg {
-			match serde_json::from_str::<FlashblocksPayloadV1>(string) {
+			match serde_json::from_str::<OpFlashblockPayload>(string) {
 				Ok(fb_payload) => self.record_flashblock(fb_payload),
 				Err(error) => {
 					self.errors.push(error.into());
@@ -163,7 +163,7 @@ impl Observations {
 		self.messages.push(msg);
 	}
 
-	fn record_flashblock(&self, payload: FlashblocksPayloadV1) {
+	fn record_flashblock(&self, payload: OpFlashblockPayload) {
 		debug!(
 			index = payload.index,
 			payload_id = ?payload.payload_id,

tools/fbutil/Cargo.toml

@@ -11,7 +11,7 @@ authors.workspace = true
 exclude.workspace = true
 
 [dependencies]
-rollup-boost-types = { workspace = true }
+rblib = { workspace = true }
 
 futures = { workspace = true }
 tokio = { workspace = true }

tools/fbutil/src/watch.rs

@@ -5,7 +5,7 @@ use {
 	colored::Colorize,
 	futures::StreamExt,
 	human_format::Formatter,
-	rollup_boost_types::flashblocks::FlashblocksPayloadV1,
+	rblib::alloy::optimism::rpc_types_engine::OpFlashblockPayload,
 	std::time::Instant,
 	tokio_tungstenite::{connect_async, tungstenite::Message},
 };
@@ -18,7 +18,7 @@ pub async fn run(cli: &Cli, _: &WatchArgs) -> eyre::Result<()> {
 	println!("Connected to {}", cli.ws);
 
 	let mut prev_at: Option<Instant> = None;
-	let mut prev: Option<FlashblocksPayloadV1> = None;
+	let mut prev: Option<OpFlashblockPayload> = None;
 
 	while let Some(msg) = stream.next().await {
 		let now = Instant::now();
@@ -31,7 +31,7 @@ pub async fn run(cli: &Cli, _: &WatchArgs) -> eyre::Result<()> {
 			eyre::bail!("Received non-text message: {msg:?}");
 		};
 
-		let block: FlashblocksPayloadV1 = serde_json::from_str(&msg)?;
+		let block: OpFlashblockPayload = serde_json::from_str(&msg)?;
 
 		if block.index == 0 {
 			println!(); // start of new block