feat(rpc): add newFlashblockTransactions subscription type (#287)

Adds `newFlashblockTransactions` to eth_subscribe for streaming
transactions from flashblocks as they're sequenced. These have higher
inclusion confidence than standard newPendingTransactions since they've
already been included by the sequencer.

Accepts optional boolean param:
- true: full transaction objects
- false (default): hashes only

Changes:
- NewFlashblockTransactions variant in BaseSubscriptionKind
- get_pending_transactions/get_pending_transaction_hashes on PendingBlocks
- Stream functions and subscription handler in pubsub
- Tests for both modes

Closes #280

Author: madisoncarter1234

crates/flashblocks/src/pending_blocks.rs

@@ -280,6 +280,16 @@ impl PendingBlocks {
 
         logs
     }
+
+    /// Returns all pending transactions from flashblocks.
+    pub fn get_pending_transactions(&self) -> Vec<Transaction> {
+        self.transactions.clone()
+    }
+
+    /// Returns the hashes of all pending transactions from flashblocks.
+    pub fn get_pending_transaction_hashes(&self) -> Vec<B256> {
+        self.transactions.iter().map(|tx| tx.tx_hash()).collect()
+    }
 }
 
 impl PendingBlocksAPI for Guard<Option<Arc<PendingBlocks>>> {

crates/rpc/src/base/pubsub.rs

@@ -6,6 +6,7 @@
 
 use std::sync::Arc;
 
+use alloy_primitives::B256;
 use alloy_rpc_types_eth::{Filter, Log, pubsub::Params};
 use base_reth_flashblocks::FlashblocksAPI;
 use jsonrpsee::{
@@ -15,6 +16,7 @@ use jsonrpsee::{
     server::SubscriptionMessage,
 };
 use op_alloy_network::Optimism;
+use op_alloy_rpc_types::Transaction;
 use reth_rpc::eth::EthPubSub as RethEthPubSub;
 use reth_rpc_eth_api::{
     EthApiTypes, RpcBlock, RpcNodeCore, RpcTransaction,
@@ -112,6 +114,52 @@ impl<Eth, FB> EthPubSub<Eth, FB> {
             },
         )
     }
+
+    /// Returns a stream that yields full transactions from pending flashblocks
+    fn new_flashblock_transactions_full_stream(
+        flashblocks_state: Arc<FB>,
+    ) -> impl Stream<Item = Vec<Transaction>>
+    where
+        FB: FlashblocksAPI + Send + Sync + 'static,
+    {
+        BroadcastStream::new(flashblocks_state.subscribe_to_flashblocks()).filter_map(|result| {
+            let pending_blocks = match result {
+                Ok(blocks) => blocks,
+                Err(err) => {
+                    error!(
+                        message = "Error in flashblocks stream for transactions",
+                        error = %err
+                    );
+                    return None;
+                }
+            };
+            let txs = pending_blocks.get_pending_transactions();
+            if txs.is_empty() { None } else { Some(txs) }
+        })
+    }
+
+    /// Returns a stream that yields transaction hashes from pending flashblocks
+    fn new_flashblock_transactions_hash_stream(
+        flashblocks_state: Arc<FB>,
+    ) -> impl Stream<Item = Vec<B256>>
+    where
+        FB: FlashblocksAPI + Send + Sync + 'static,
+    {
+        BroadcastStream::new(flashblocks_state.subscribe_to_flashblocks()).filter_map(|result| {
+            let pending_blocks = match result {
+                Ok(blocks) => blocks,
+                Err(err) => {
+                    error!(
+                        message = "Error in flashblocks stream for transaction hashes",
+                        error = %err
+                    );
+                    return None;
+                }
+            };
+            let hashes = pending_blocks.get_pending_transaction_hashes();
+            if hashes.is_empty() { None } else { Some(hashes) }
+        })
+    }
 }
 
 #[async_trait]
@@ -165,6 +213,29 @@ where
                     pipe_from_stream(sink, stream).await;
                 });
             }
+            BaseSubscriptionKind::NewFlashblockTransactions => {
+                // Extract full_transactions param, default to false (hash only)
+                let full = match params {
+                    Some(Params::Bool(full)) => full,
+                    _ => false,
+                };
+
+                if full {
+                    let stream = Self::new_flashblock_transactions_full_stream(Arc::clone(
+                        &self.flashblocks_state,
+                    ));
+                    tokio::spawn(async move {
+                        pipe_from_stream(sink, stream).await;
+                    });
+                } else {
+                    let stream = Self::new_flashblock_transactions_hash_stream(Arc::clone(
+                        &self.flashblocks_state,
+                    ));
+                    tokio::spawn(async move {
+                        pipe_from_stream(sink, stream).await;
+                    });
+                }
+            }
         }
 
         Ok(())

crates/rpc/src/base/types.rs

@@ -57,6 +57,16 @@ pub enum BaseSubscriptionKind {
     /// Unlike standard `logs` subscription which only includes logs from confirmed blocks,
     /// this includes logs from the current pending flashblock state.
     PendingLogs,
+    /// New flashblock transactions subscription.
+    ///
+    /// Returns transactions from flashblocks as they are sequenced, providing higher inclusion
+    /// confidence than standard `newPendingTransactions` which returns mempool transactions.
+    /// Flashblock transactions have been included by the sequencer and are effectively preconfirmed.
+    ///
+    /// Accepts an optional boolean parameter:
+    /// - `true`: Returns full transaction objects
+    /// - `false` (default): Returns only transaction hashes
+    NewFlashblockTransactions,
 }
 
 impl ExtendedSubscriptionKind {

crates/rpc/tests/flashblocks_rpc.rs

@@ -985,3 +985,109 @@ async fn test_eth_subscribe_new_heads() -> eyre::Result<()> {
 
     Ok(())
 }
+
+#[tokio::test]
+async fn test_eth_subscribe_new_flashblock_transactions_hashes() -> eyre::Result<()> {
+    let setup = TestSetup::new().await?;
+    let _provider = setup.harness.provider();
+    let ws_url = setup.harness.ws_url();
+    let (mut ws_stream, _) = connect_async(&ws_url).await?;
+
+    // Subscribe to newFlashblockTransactions with default (hash only) mode
+    ws_stream
+        .send(Message::Text(
+            json!({
+                "jsonrpc": "2.0",
+                "id": 1,
+                "method": "eth_subscribe",
+                "params": ["newFlashblockTransactions"]
+            })
+            .to_string()
+            .into(),
+        ))
+        .await?;
+
+    let response = ws_stream.next().await.unwrap()?;
+    let sub: serde_json::Value = serde_json::from_str(response.to_text()?)?;
+    assert_eq!(sub["jsonrpc"], "2.0");
+    assert_eq!(sub["id"], 1);
+    let subscription_id = sub["result"].as_str().expect("subscription id expected");
+
+    // Send first flashblock with L1 deposit tx
+    setup.send_flashblock(setup.create_first_payload()).await?;
+
+    let notification = ws_stream.next().await.unwrap()?;
+    let notif: serde_json::Value = serde_json::from_str(notification.to_text()?)?;
+    assert_eq!(notif["method"], "eth_subscription");
+    assert_eq!(notif["params"]["subscription"], subscription_id);
+
+    // Result should be an array of transaction hashes (strings)
+    let txs = notif["params"]["result"].as_array().expect("expected array of tx hashes");
+    assert_eq!(txs.len(), 1);
+    assert!(txs[0].is_string(), "Expected hash string, got: {:?}", txs[0]);
+
+    // Send second flashblock with more transactions
+    setup.send_flashblock(setup.create_second_payload()).await?;
+
+    let notification2 = ws_stream.next().await.unwrap()?;
+    let notif2: serde_json::Value = serde_json::from_str(notification2.to_text()?)?;
+    let txs2 = notif2["params"]["result"].as_array().expect("expected array of tx hashes");
+    assert_eq!(txs2.len(), 6);
+    assert!(txs2.iter().all(|tx| tx.is_string()));
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn test_eth_subscribe_new_flashblock_transactions_full() -> eyre::Result<()> {
+    let setup = TestSetup::new().await?;
+    let _provider = setup.harness.provider();
+    let ws_url = setup.harness.ws_url();
+    let (mut ws_stream, _) = connect_async(&ws_url).await?;
+
+    // Subscribe to newFlashblockTransactions with full transaction objects (true)
+    ws_stream
+        .send(Message::Text(
+            json!({
+                "jsonrpc": "2.0",
+                "id": 1,
+                "method": "eth_subscribe",
+                "params": ["newFlashblockTransactions", true]
+            })
+            .to_string()
+            .into(),
+        ))
+        .await?;
+
+    let response = ws_stream.next().await.unwrap()?;
+    let sub: serde_json::Value = serde_json::from_str(response.to_text()?)?;
+    assert_eq!(sub["jsonrpc"], "2.0");
+    assert_eq!(sub["id"], 1);
+    let subscription_id = sub["result"].as_str().expect("subscription id expected");
+
+    // Send flashblocks
+    setup.send_flashblock(setup.create_first_payload()).await?;
+
+    let notification = ws_stream.next().await.unwrap()?;
+    let notif: serde_json::Value = serde_json::from_str(notification.to_text()?)?;
+    assert_eq!(notif["method"], "eth_subscription");
+    assert_eq!(notif["params"]["subscription"], subscription_id);
+
+    // Result should be an array of full transaction objects
+    let txs = notif["params"]["result"].as_array().expect("expected array of transactions");
+    assert_eq!(txs.len(), 1);
+    // Full transaction objects have fields like "hash", "from", "to", etc.
+    assert!(txs[0]["hash"].is_string(), "Expected full tx with hash field");
+    assert!(txs[0]["blockNumber"].is_string(), "Expected full tx with blockNumber field");
+
+    // Send second flashblock with more transactions
+    setup.send_flashblock(setup.create_second_payload()).await?;
+
+    let notification2 = ws_stream.next().await.unwrap()?;
+    let notif2: serde_json::Value = serde_json::from_str(notification2.to_text()?)?;
+    let txs2 = notif2["params"]["result"].as_array().expect("expected array of transactions");
+    assert_eq!(txs2.len(), 6);
+    assert!(txs2.iter().all(|tx| tx["hash"].is_string() && tx["blockNumber"].is_string()));
+
+    Ok(())
+}