alloy-flashblocks 0.1.0

//! FlashblockPoller for streaming pending block updates via RPC polling.

use alloy::{
    eips::BlockNumberOrTag,
    network::{BlockResponse, Network, primitives::HeaderResponse},
    providers::Provider,
};
use futures_util::Stream;
use std::{pin::Pin, task::Poll, time::Duration};
use tokio::time::{Interval, interval};

/// Flashblock emission interval (~200ms on Base).
pub const FLASHBLOCK_INTERVAL: Duration = Duration::from_millis(200);

/// Poll interval (0.6 × flashblock interval, following Alloy convention).
pub const POLL_INTERVAL: Duration = Duration::from_millis(
    (FLASHBLOCK_INTERVAL.as_millis() as f64 * 0.6) as u64
);

/// Polls for pending block updates via JSON-RPC.
///
/// Calls `eth_getBlockByNumber("pending")` every 120ms and yields
/// new blocks when the block hash changes.
pub struct FlashblockPoller<N: Network, P: Provider<N>> {
    provider: P,
    interval: Interval,
    last_hash: Option<alloy::primitives::B256>,
    pending_request:
        Option<Pin<Box<dyn std::future::Future<Output = Option<N::BlockResponse>> + Send>>>,
}

// Safety: All fields are Unpin (Box<dyn Future> is Unpin, Interval is Unpin, etc.)
impl<N: Network, P: Provider<N>> Unpin for FlashblockPoller<N, P> {}

impl<N: Network, P: Provider<N>> std::fmt::Debug for FlashblockPoller<N, P> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("FlashblockPoller")
            .field("last_hash", &self.last_hash)
            .field("has_pending_request", &self.pending_request.is_some())
            .finish_non_exhaustive()
    }
}

impl<N: Network, P: Provider<N> + Clone + Send + Sync + 'static> FlashblockPoller<N, P> {
    /// Create a new poller.
    pub fn new(provider: P) -> Self {
        Self {
            provider,
            interval: interval(POLL_INTERVAL),
            last_hash: None,
            pending_request: None,
        }
    }

    /// Convert to a boxed stream for easier usage.
    pub fn into_stream(self) -> Pin<Box<dyn Stream<Item = N::BlockResponse> + Send>>
    where
        Self: Send + 'static,
    {
        Box::pin(self)
    }
}

impl<N: Network, P: Provider<N> + Clone + Send + Sync + 'static> Stream for FlashblockPoller<N, P> {
    type Item = N::BlockResponse;

    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> Poll<Option<Self::Item>> {
        let this = self.get_mut();

        loop {
            // If we have a pending request, poll it
            if let Some(ref mut fut) = this.pending_request {
                match fut.as_mut().poll(cx) {
                    Poll::Ready(result) => {
                        this.pending_request = None;

                        if let Some(block) = result {
                            let block_hash = block.header().hash();

                            // Only yield if block changed
                            if this.last_hash != Some(block_hash) {
                                this.last_hash = Some(block_hash);
                                return Poll::Ready(Some(block));
                            }
                        }
                        // Block didn't change or was None, continue to next tick
                    }
                    Poll::Pending => return Poll::Pending,
                }
            }

            // Wait for next interval tick
            match this.interval.poll_tick(cx) {
                Poll::Ready(_) => {
                    // Start a new request
                    let provider = this.provider.clone();
                    this.pending_request = Some(Box::pin(async move {
                        provider
                            .get_block_by_number(BlockNumberOrTag::Pending)
                            .await
                            .ok()
                            .flatten()
                    }));
                }
                Poll::Pending => return Poll::Pending,
            }
        }
    }
}