feat(builder): add support for block access list (EIP-7928) for flashblocks builder

bin/node/src/args.rs

@@ -259,6 +259,14 @@ pub struct FlashblocksRpcArgs {
     )]
     pub flashblocks_subscription_max_addresses: usize,
 
+    /// Disable EIP-7928 block access list usage on the flashblocks RPC node.
+    #[arg(
+        long = "xlayer.flashblocks-disable-access-list",
+        help = "Disable EIP-7928 access list usage on the flashblocks RPC node",
+        default_value = "false"
+    )]
+    pub flashblocks_disable_access_list: bool,
+
     /// Enable flashblocks RPC state comparison debug mode
     #[arg(
         long = "xlayer.flashblocks-debug-state-comparison",

bin/node/src/main.rs

@@ -208,6 +208,10 @@ fn main() {
                                     .xlayer_args
                                     .flashblocks_rpc
                                     .flashblocks_debug_state_comparison,
+                                disable_access_list: args
+                                    .xlayer_args
+                                    .flashblocks_rpc
+                                    .flashblocks_disable_access_list,
                             },
                             FlashblocksPersistCtx {
                                 datadir,

crates/builder/Cargo.toml

@@ -54,16 +54,18 @@ reth-tasks.workspace = true
 reth-ipc = { workspace = true, optional = true }
 
 # alloy
+alloy-eip7928 = { workspace = true, features = ["serde", "rlp"] }
 alloy-primitives.workspace = true
-alloy-consensus.workspace = true
+alloy-consensus = { workspace = true, features = ["std"] }
 alloy-contract.workspace = true
 alloy-eips.workspace = true
 alloy-evm.workspace = true
+alloy-rlp = { workspace = true, features = ["derive", "std"] }
 alloy-rpc-types-engine.workspace = true
 alloy-rpc-types-eth.workspace = true
 alloy-network.workspace = true
 alloy-provider.workspace = true
-alloy-sol-types = { workspace = true, features = ["json"] }
+alloy-sol-types = { workspace = true, features = ["std", "json"] }
 
 # op
 alloy-op-evm.workspace = true
@@ -100,6 +102,7 @@ multiaddr = { workspace = true }
 
 # misc
 anyhow = "1"
+brotli = { workspace = true, features = ["std"] }
 clap.workspace = true
 dashmap.workspace = true
 derive_more.workspace = true

crates/builder/src/broadcast/compress.rs

@@ -0,0 +1,108 @@
+//! Shared compression helpers for the flashblocks wire protocol.
+//!
+//! The wire format is auto-detecting: a payload is treated as legacy uncompressed
+//! JSON if its first non-whitespace byte is `{`, otherwise it is decoded as a
+//! brotli-compressed frame. This lets us upgrade the producer to emit compressed
+//! frames without breaking older subscribers, and lets us replay legacy
+//! persistence files unchanged.
+
+use std::{borrow::Cow, io};
+
+use alloy_primitives::Bytes;
+use brotli::enc::BrotliEncoderParams;
+
+/// Brotli quality level. Q5 is the empirical sweet spot for JSON-shaped
+/// flashblock payloads — sub-10ms encode at ~5x size reduction on 1.2MB inputs.
+pub const BROTLI_QUALITY: u32 = 5;
+
+/// Brotli sliding-window size in log2 bytes. 22 is the brotli default.
+pub const BROTLI_LGWIN: u32 = 22;
+
+/// Compress `json` bytes with brotli at [`BROTLI_QUALITY`].
+pub fn brotli_encode(json: &[u8]) -> io::Result<Bytes> {
+    let mut compressed = Vec::with_capacity(json.len() / 4);
+    let params = BrotliEncoderParams {
+        quality: BROTLI_QUALITY as i32,
+        lgwin: BROTLI_LGWIN as i32,
+        ..Default::default()
+    };
+    let mut input = json;
+    brotli::BrotliCompress(&mut input, &mut compressed, &params)?;
+    Ok(Bytes::from(compressed))
+}
+
+/// Decode a wire frame, auto-detecting between legacy uncompressed JSON and
+/// brotli-compressed bytes.
+///
+/// Returns the input borrowed if it is already JSON (leading `{` after
+/// optional whitespace), otherwise allocates a decompressed buffer.
+pub fn try_decompress(bytes: &[u8]) -> io::Result<Cow<'_, [u8]>> {
+    if bytes.trim_ascii_start().starts_with(b"{") {
+        return Ok(Cow::Borrowed(bytes));
+    }
+
+    let mut decompressor = brotli::Decompressor::new(bytes, 4096);
+    let mut decompressed = Vec::new();
+    io::copy(&mut decompressor, &mut decompressed)?;
+    Ok(Cow::Owned(decompressed))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn brotli_round_trip() {
+        let original = br#"{"hello":"world","numbers":[1,2,3],"nested":{"a":"b"}}"#;
+        let compressed = brotli_encode(original).expect("encode succeeds");
+        assert!(!compressed.is_empty(), "compressed output is non-empty");
+        let decoded = try_decompress(&compressed).expect("decompress succeeds");
+        assert_eq!(decoded.as_ref(), original);
+    }
+
+    #[test]
+    fn brotli_compresses_repetitive_input() {
+        // EIP-7928 access list JSON is highly repetitive; expect significant compression.
+        let original: Vec<u8> =
+            br#"{"address":"0x0000000000000000000000000000000000000042","balance_changes":[{"block_access_index":0,"post_balance":"0x100"}]}"#
+                .repeat(100);
+        let compressed = brotli_encode(&original).expect("encode succeeds");
+        assert!(compressed.len() < original.len() / 5, "expected >5x compression ratio");
+    }
+
+    #[test]
+    fn auto_detect_legacy_json_passthrough() {
+        let json = br#"{"key":"value"}"#;
+        let result = try_decompress(json).expect("passthrough succeeds");
+        assert_eq!(result.as_ref(), json);
+        assert!(matches!(result, Cow::Borrowed(_)), "legacy JSON must be borrowed, not copied");
+    }
+
+    #[test]
+    fn auto_detect_legacy_json_with_leading_whitespace() {
+        let json = b"   \n\t  {\"key\":\"value\"}";
+        let result = try_decompress(json).expect("passthrough succeeds");
+        assert_eq!(result.as_ref(), json);
+    }
+
+    #[test]
+    fn auto_detect_brotli_decompresses() {
+        let original = br#"{"key":"value","data":"some larger payload for brotli to compress"}"#;
+        let compressed = brotli_encode(original).expect("encode");
+        assert!(
+            !compressed.starts_with(b"{"),
+            "compressed output must not start with `{{` (would defeat auto-detect)"
+        );
+        let result = try_decompress(&compressed).expect("decompress succeeds");
+        assert_eq!(result.as_ref(), original);
+        assert!(matches!(result, Cow::Owned(_)), "compressed input must allocate decoded buffer");
+    }
+
+    #[test]
+    fn malformed_brotli_returns_err() {
+        // Bytes that don't start with `{` and aren't valid brotli.
+        let garbage = &[0xff_u8, 0xff, 0xff, 0xff, 0xff];
+        let result = try_decompress(garbage);
+        assert!(result.is_err(), "malformed brotli must error, not silently pass through");
+    }
+}

crates/builder/src/broadcast/frame.rs

@@ -0,0 +1,118 @@
+//! Encode-once carrier types for the flashblocks broadcast pipeline.
+//!
+//! When the sequencer produces a flashblock, we serialize+compress it exactly
+//! once and pass the resulting bytes through both fan-out paths (P2P + WS).
+//! When a relay node receives a flashblock, it decodes once for internal
+//! consumers (validator, persistence) but forwards the original bytes to its
+//! downstream subscribers without re-encoding.
+
+use std::{io, sync::Arc};
+
+use alloy_primitives::Bytes;
+use serde::{de::DeserializeOwned, Serialize};
+
+use crate::broadcast::{compress, types::Message};
+
+/// Wire-ready bytes paired with their decoded form.
+#[derive(Clone, Debug)]
+pub struct BroadcastFrame {
+    /// Brotli-compressed JSON, ready to send on the WS pipe or libp2p stream.
+    pub bytes: Bytes,
+    /// Already-decoded structured form, for in-process consumers.
+    pub decoded: Arc<Message>,
+}
+
+impl BroadcastFrame {
+    /// Decode wire bytes into a complete frame.
+    ///
+    /// Auto-detects between brotli-compressed and legacy uncompressed JSON via
+    /// [`compress::try_decompress`]. The original `bytes` are retained on the
+    /// frame so relay paths can forward them downstream without re-encoding.
+    pub fn from_bytes(bytes: Bytes) -> io::Result<Self> {
+        let decoded: Message = decode(&bytes)?;
+        Ok(Self { bytes, decoded: Arc::new(decoded) })
+    }
+}
+
+/// Serialize `value` to JSON and brotli-compress the result.
+///
+/// Single encode path for any wire envelope on the broadcast layer (P2P
+/// [`Message`], WS `XLayerFlashblockMessage`, etc).
+pub fn encode<T: Serialize>(value: &T) -> io::Result<Bytes> {
+    let json = serde_json::to_vec(value)?;
+    compress::brotli_encode(&json)
+}
+
+/// Decode wire bytes into `T`, auto-detecting compressed vs. legacy JSON.
+///
+/// Single decode path for any wire envelope. The auto-detection (leading-`{`
+/// heuristic) lives in [`compress::try_decompress`], so a producer that emits
+/// compressed bytes and a producer that emits legacy uncompressed JSON can
+/// both be consumed without coordination.
+pub fn decode<T: DeserializeOwned>(bytes: &[u8]) -> io::Result<T> {
+    let json = compress::try_decompress(bytes)?;
+    serde_json::from_slice::<T>(&json)
+        .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, format!("decode: {e}")))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::broadcast::{XLayerFlashblockMessage, XLayerFlashblockPayload};
+
+    fn sample_message() -> Message {
+        Message::from_flashblock_payload(XLayerFlashblockMessage::from_flashblock_payload(
+            XLayerFlashblockPayload::default(),
+        ))
+    }
+
+    #[test]
+    fn encode_decode_round_trip() {
+        let original = sample_message();
+        let bytes = encode(&original).expect("encode");
+        let decoded: Message = decode(&bytes).expect("decode");
+        assert_eq!(decoded, original);
+    }
+
+    #[test]
+    fn encode_decode_round_trip_xlayer_flashblock_message() {
+        // The same encode/decode path works for any Serialize/DeserializeOwned type —
+        // the WS-side stream uses this with XLayerFlashblockMessage.
+        let original =
+            XLayerFlashblockMessage::from_flashblock_payload(XLayerFlashblockPayload::default());
+        let bytes = encode(&original).expect("encode");
+        let decoded: XLayerFlashblockMessage = decode(&bytes).expect("decode");
+        assert_eq!(decoded, original);
+    }
+
+    #[test]
+    fn from_bytes_round_trip() {
+        let original = sample_message();
+        let bytes = encode(&original).expect("encode");
+        let frame = BroadcastFrame::from_bytes(bytes.clone()).expect("from_bytes");
+        assert_eq!(*frame.decoded, original);
+        // Frame retains the original bytes byte-for-byte for relay.
+        assert_eq!(frame.bytes, bytes);
+    }
+
+    #[test]
+    fn from_bytes_accepts_legacy_uncompressed_json() {
+        let original = sample_message();
+        let json = Bytes::from(serde_json::to_vec(&original).expect("json"));
+        // Legacy callers might send uncompressed JSON — must still decode.
+        let frame = BroadcastFrame::from_bytes(json).expect("decode legacy");
+        assert_eq!(*frame.decoded, original);
+    }
+
+    #[test]
+    fn encoded_bytes_are_not_legacy_json() {
+        // The encoded bytes must not start with `{` so that the auto-detect heuristic
+        // routes them through brotli decompression on the receive side.
+        let original = sample_message();
+        let bytes = encode(&original).expect("encode");
+        assert!(
+            !bytes.starts_with(b"{"),
+            "compressed frame must not look like legacy JSON to the auto-detect heuristic"
+        );
+    }
+}