jetstreamer_firehose/

firehose.rs

use crossbeam_channel::{Receiver, Sender, unbounded};
use dashmap::{DashMap, DashSet};
use futures_util::future::BoxFuture;
use reqwest::{Client, Url};
use solana_address::Address;
use solana_geyser_plugin_manager::{
    block_metadata_notifier_interface::BlockMetadataNotifier,
    geyser_plugin_service::GeyserPluginServiceError,
};
use solana_hash::Hash;
use solana_ledger::entry_notifier_interface::EntryNotifier;
use solana_reward_info::RewardInfo;
use solana_rpc::{
    optimistically_confirmed_bank_tracker::SlotNotification,
    transaction_notifier_interface::TransactionNotifier,
};
use solana_runtime::bank::{KeyedRewardsAndNumPartitions, RewardType};
use solana_sdk_ids::vote::id as vote_program_id;
use solana_transaction::versioned::VersionedTransaction;
use std::{
    fmt::Display,
    future::Future,
    io,
    ops::Range,
    path::PathBuf,
    sync::{
        Arc,
        atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering},
    },
};
use thiserror::Error;
use tokio::{
    sync::broadcast::{self, error::TryRecvError},
    time::timeout,
};

use crate::{
    LOG_MODULE, SharedError,
    epochs::{epoch_to_slot_range, fetch_epoch_stream, slot_to_epoch},
    index::{SLOT_OFFSET_INDEX, SlotOffsetIndexError},
    node_reader::NodeReader,
    utils,
};

// Timeout applied to each asynchronous firehose operation (fetching epoch stream, reading
// header, seeking, reading next block). Adjust here to tune stall detection/restart
// aggressiveness.
const OP_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
// Epochs earlier than this were bincode-encoded in Old Faithful.
const BINCODE_EPOCH_CUTOFF: u64 = 157;

fn poll_shutdown(
    flag: &Arc<std::sync::atomic::AtomicBool>,
    receiver: &mut Option<broadcast::Receiver<()>>,
) -> bool {
    if let Some(rx) = receiver {
        match rx.try_recv() {
            Ok(_) | Err(TryRecvError::Lagged(_)) => {
                flag.store(true, Ordering::SeqCst);
            }
            Err(TryRecvError::Closed) => {
                flag.store(true, Ordering::SeqCst);
            }
            Err(TryRecvError::Empty) => {}
        }
    }
    flag.load(Ordering::SeqCst)
}

fn is_shutdown_error(err: &FirehoseError) -> bool {
    fn is_interrupted(inner: &(dyn std::error::Error + 'static)) -> bool {
        inner
            .downcast_ref::<io::Error>()
            .map(|io_err| io_err.kind() == io::ErrorKind::Interrupted)
            .unwrap_or(false)
    }

    match err {
        FirehoseError::BlockHandlerError(inner)
        | FirehoseError::TransactionHandlerError(inner)
        | FirehoseError::EntryHandlerError(inner)
        | FirehoseError::RewardHandlerError(inner)
        | FirehoseError::OnStatsHandlerError(inner) => is_interrupted(inner.as_ref()),
        _ => false,
    }
}

/// Errors that can occur while streaming the firehose. Errors that can occur while streaming
/// the firehose.
#[derive(Debug, Error)]
pub enum FirehoseError {
    /// HTTP client error surfaced from `reqwest`.
    Reqwest(reqwest::Error),
    /// Failure while reading the Old Faithful CAR header.
    ReadHeader(SharedError),
    /// Error emitted by the Solana Geyser plugin service.
    GeyserPluginService(GeyserPluginServiceError),
    /// Transaction notifier could not be acquired from the Geyser service.
    FailedToGetTransactionNotifier,
    /// Failure while reading data until the next block boundary.
    ReadUntilBlockError(SharedError),
    /// Failure while fetching an individual block.
    GetBlockError(SharedError),
    /// Failed to decode a node at the given index.
    NodeDecodingError(usize, SharedError),
    /// Error surfaced when querying the slot offset index.
    SlotOffsetIndexError(SlotOffsetIndexError),
    /// Failure while seeking to a slot within the Old Faithful CAR stream.
    SeekToSlotError(SharedError),
    /// Error surfaced during the plugin `on_load` stage.
    OnLoadError(SharedError),
    /// Error emitted while invoking the stats handler.
    OnStatsHandlerError(SharedError),
    /// Timeout reached while waiting for a firehose operation.
    OperationTimeout(&'static str),
    /// Transaction handler returned an error.
    TransactionHandlerError(SharedError),
    /// Entry handler returned an error.
    EntryHandlerError(SharedError),
    /// Reward handler returned an error.
    RewardHandlerError(SharedError),
    /// Block handler returned an error.
    BlockHandlerError(SharedError),
}

unsafe impl Send for FirehoseError {}
unsafe impl Sync for FirehoseError {}

impl Display for FirehoseError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            FirehoseError::Reqwest(e) => write!(f, "Reqwest error: {}", e),
            FirehoseError::ReadHeader(error) => {
                write!(f, "Error reading header: {}", error)
            }
            FirehoseError::GeyserPluginService(geyser_plugin_service_error) => write!(
                f,
                "Error initializing geyser plugin service: {}",
                geyser_plugin_service_error
            ),
            FirehoseError::FailedToGetTransactionNotifier => write!(
                f,
                "Failed to get transaction notifier from GeyserPluginService"
            ),
            FirehoseError::ReadUntilBlockError(error) => {
                write!(f, "Error reading until block: {}", error)
            }
            FirehoseError::GetBlockError(error) => write!(f, "Error getting block: {}", error),
            FirehoseError::NodeDecodingError(item_index, error) => {
                write!(
                    f,
                    "Error seeking, reading data from, or decoding data for data node {}: {}",
                    item_index, error
                )
            }
            FirehoseError::SlotOffsetIndexError(slot_offset_index_error) => write!(
                f,
                "Error getting info from slot offset index: {}",
                slot_offset_index_error
            ),
            FirehoseError::SeekToSlotError(error) => {
                write!(f, "Error seeking to slot: {}", error)
            }
            FirehoseError::OnLoadError(error) => write!(f, "Error on load: {}", error),
            FirehoseError::OnStatsHandlerError(error) => {
                write!(f, "Stats handler error: {}", error)
            }
            FirehoseError::OperationTimeout(op) => {
                write!(f, "Timeout while waiting for operation: {}", op)
            }
            FirehoseError::TransactionHandlerError(error) => {
                write!(f, "Transaction handler error: {}", error)
            }
            FirehoseError::EntryHandlerError(error) => {
                write!(f, "Entry handler error: {}", error)
            }
            FirehoseError::RewardHandlerError(error) => {
                write!(f, "Reward handler error: {}", error)
            }
            FirehoseError::BlockHandlerError(error) => {
                write!(f, "Block handler error: {}", error)
            }
        }
    }
}

impl From<reqwest::Error> for FirehoseError {
    fn from(e: reqwest::Error) -> Self {
        FirehoseError::Reqwest(e)
    }
}

impl From<GeyserPluginServiceError> for FirehoseError {
    fn from(e: GeyserPluginServiceError) -> Self {
        FirehoseError::GeyserPluginService(e)
    }
}

impl From<SlotOffsetIndexError> for FirehoseError {
    fn from(e: SlotOffsetIndexError) -> Self {
        FirehoseError::SlotOffsetIndexError(e)
    }
}

/// Per-thread progress information emitted by the firehose runner.
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct ThreadStats {
    /// Identifier of the worker thread reporting the stats.
    pub thread_id: usize,
    /// Timestamp captured when the thread began processing.
    pub start_time: std::time::Instant,
    /// Timestamp captured when the thread finished, if finished.
    pub finish_time: Option<std::time::Instant>,
    /// Slot range currently assigned to the thread (half-open, may shrink on restart).
    pub slot_range: Range<u64>,
    /// Original slot range assigned to the thread (half-open, never modified).
    pub initial_slot_range: Range<u64>,
    /// Latest slot processed by the thread.
    pub current_slot: u64,
    /// Total slots processed by the thread.
    pub slots_processed: u64,
    /// Number of blocks successfully processed.
    pub blocks_processed: u64,
    /// Number of slots skipped by the cluster leader.
    pub leader_skipped_slots: u64,
    /// Total transactions processed.
    pub transactions_processed: u64,
    /// Total entries processed.
    pub entries_processed: u64,
    /// Total rewards processed.
    pub rewards_processed: u64,
}

/// Aggregated firehose statistics covering all worker threads.
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct Stats {
    /// Per-thread statistics for the current update.
    pub thread_stats: ThreadStats,
    /// Timestamp captured when processing began.
    pub start_time: std::time::Instant,
    /// Timestamp captured when all processing finished, if finished.
    pub finish_time: Option<std::time::Instant>,
    /// Slot range currently being processed (half-open [start, end)).
    pub slot_range: Range<u64>,
    /// Aggregate slots processed across all threads.
    pub slots_processed: u64,
    /// Aggregate blocks processed across all threads.
    pub blocks_processed: u64,
    /// Aggregate skipped slots across all threads.
    pub leader_skipped_slots: u64,
    /// Aggregate transactions processed across all threads.
    pub transactions_processed: u64,
    /// Aggregate entries processed across all threads.
    pub entries_processed: u64,
    /// Aggregate rewards processed across all threads.
    pub rewards_processed: u64,
    /// Transactions processed since the previous stats pulse.
    pub transactions_since_last_pulse: u64,
    /// Blocks processed since the previous stats pulse.
    pub blocks_since_last_pulse: u64,
    /// Slots processed since the previous stats pulse.
    pub slots_since_last_pulse: u64,
    /// Elapsed time since the previous stats pulse.
    pub time_since_last_pulse: std::time::Duration,
}

/// Configuration for periodic stats emission via a [`Handler`] callback.
#[derive(Clone, PartialEq, Eq, Hash, Debug)]
pub struct StatsTracking<OnStats: Handler<Stats>> {
    /// Callback invoked whenever new stats are available.
    pub on_stats: OnStats,
    /// Emits a stats callback when the current slot is a multiple of this interval.
    pub tracking_interval_slots: u64,
}

#[inline(always)]
#[allow(clippy::too_many_arguments)]
async fn maybe_emit_stats<OnStats: Handler<Stats>>(
    stats_tracking: Option<&StatsTracking<OnStats>>,
    thread_index: usize,
    thread_stats: &ThreadStats,
    overall_slots_processed: &AtomicU64,
    overall_blocks_processed: &AtomicU64,
    overall_transactions_processed: &AtomicU64,
    overall_entries_processed: &AtomicU64,
    transactions_since_stats: &AtomicU64,
    blocks_since_stats: &AtomicU64,
    slots_since_stats: &AtomicU64,
    last_pulse: &Arc<AtomicU64>,
    base_instant: std::time::Instant,
) -> Result<(), (FirehoseError, u64)> {
    if let Some(stats_tracker) = stats_tracking {
        let total_slots = overall_slots_processed.load(Ordering::Relaxed);
        let total_blocks = overall_blocks_processed.load(Ordering::Relaxed);
        let total_transactions = overall_transactions_processed.load(Ordering::Relaxed);
        let total_entries = overall_entries_processed.load(Ordering::Relaxed);
        let now_nanos = base_instant.elapsed().as_nanos() as u64;
        let previous = last_pulse.swap(now_nanos, Ordering::Relaxed);
        let delta_nanos = now_nanos.saturating_sub(previous);
        let time_since_last_pulse = std::time::Duration::from_nanos(delta_nanos.max(1));
        let processed_transactions = transactions_since_stats.swap(0, Ordering::Relaxed);
        let processed_blocks = blocks_since_stats.swap(0, Ordering::Relaxed);
        let processed_slots = slots_since_stats.swap(0, Ordering::Relaxed);

        let stats = Stats {
            thread_stats: thread_stats.clone(),
            start_time: thread_stats.start_time,
            finish_time: thread_stats.finish_time,
            slot_range: thread_stats.slot_range.clone(),
            slots_processed: total_slots,
            blocks_processed: total_blocks,
            leader_skipped_slots: total_slots.saturating_sub(total_blocks),
            transactions_processed: total_transactions,
            entries_processed: total_entries,
            rewards_processed: thread_stats.rewards_processed,
            transactions_since_last_pulse: processed_transactions,
            blocks_since_last_pulse: processed_blocks,
            slots_since_last_pulse: processed_slots,
            time_since_last_pulse,
        };

        if let Err(e) = (stats_tracker.on_stats)(thread_index, stats).await {
            last_pulse.store(previous, Ordering::Relaxed);
            transactions_since_stats.fetch_add(processed_transactions, Ordering::Relaxed);
            blocks_since_stats.fetch_add(processed_blocks, Ordering::Relaxed);
            slots_since_stats.fetch_add(processed_slots, Ordering::Relaxed);
            return Err((
                FirehoseError::OnStatsHandlerError(e),
                thread_stats.current_slot,
            ));
        }
    }
    Ok(())
}

#[inline(always)]
fn fetch_add_if(tracking_enabled: bool, atomic: &AtomicU64, value: u64) {
    if tracking_enabled {
        atomic.fetch_add(value, Ordering::Relaxed);
    }
}

fn clear_pending_skip(map: &DashMap<usize, DashSet<u64>>, thread_id: usize, slot: u64) -> bool {
    map.get(&thread_id)
        .map(|set| set.remove(&slot).is_some())
        .unwrap_or(false)
}

fn decode_transaction_status_meta_from_frame(
    slot: u64,
    reassembled_metadata: Vec<u8>,
) -> Result<solana_transaction_status::TransactionStatusMeta, SharedError> {
    if reassembled_metadata.is_empty() {
        // Early epochs often omit metadata entirely.
        return Ok(solana_transaction_status::TransactionStatusMeta::default());
    }

    match utils::decompress_zstd(reassembled_metadata.clone()) {
        Ok(decompressed) => {
            decode_transaction_status_meta(slot, decompressed.as_slice()).map_err(|err| {
                Box::new(std::io::Error::other(format!(
                    "decode transaction metadata (slot {slot}): {err}"
                ))) as SharedError
            })
        }
        Err(decomp_err) => {
            // If the frame was not zstd-compressed (common for very early data), try to
            // decode the raw bytes directly before bailing.
            decode_transaction_status_meta(slot, reassembled_metadata.as_slice()).map_err(|err| {
                Box::new(std::io::Error::other(format!(
                    "transaction metadata not zstd-compressed for slot {slot}; raw decode failed (raw_err={err}, decompress_err={decomp_err})"
                ))) as SharedError
            })
        }
    }
}

fn decode_transaction_status_meta(
    slot: u64,
    metadata_bytes: &[u8],
) -> Result<solana_transaction_status::TransactionStatusMeta, SharedError> {
    let epoch = slot_to_epoch(slot);
    let mut bincode_err: Option<String> = None;
    if epoch < BINCODE_EPOCH_CUTOFF {
        match bincode::deserialize::<solana_storage_proto::StoredTransactionStatusMeta>(
            metadata_bytes,
        ) {
            Ok(stored) => return Ok(stored.into()),
            Err(err) => {
                bincode_err = Some(err.to_string());
            }
        }
    }

    let bin_err_for_proto = bincode_err.clone();
    let proto: solana_storage_proto::convert::generated::TransactionStatusMeta =
        prost_011::Message::decode(metadata_bytes).map_err(|err| {
            // If we already tried bincode, surface both failures for easier debugging.
            if let Some(ref bin_err) = bin_err_for_proto {
                Box::new(std::io::Error::other(format!(
                    "protobuf decode transaction metadata failed (epoch {epoch}); bincode failed earlier: {bin_err}; protobuf error: {err}"
                ))) as SharedError
            } else {
                Box::new(std::io::Error::other(format!(
                    "protobuf decode transaction metadata: {err}"
                ))) as SharedError
            }
        })?;

    proto.try_into().map_err(|err| {
        if let Some(ref bin_err) = bincode_err {
            Box::new(std::io::Error::other(format!(
                "convert transaction metadata proto failed (epoch {epoch}); bincode failed earlier: {bin_err}; conversion error: {err}"
            ))) as SharedError
        } else {
            Box::new(std::io::Error::other(format!(
                "convert transaction metadata proto: {err}"
            ))) as SharedError
        }
    })
}

#[cfg(test)]
mod metadata_decode_tests {
    use super::{decode_transaction_status_meta, decode_transaction_status_meta_from_frame};
    use solana_message::v0::LoadedAddresses;
    use solana_storage_proto::StoredTransactionStatusMeta;
    use solana_transaction_status::TransactionStatusMeta;

    fn sample_meta() -> TransactionStatusMeta {
        let mut meta = TransactionStatusMeta::default();
        meta.fee = 42;
        meta.pre_balances = vec![1, 2];
        meta.post_balances = vec![3, 4];
        meta.log_messages = Some(vec!["hello".into()]);
        meta.pre_token_balances = Some(Vec::new());
        meta.post_token_balances = Some(Vec::new());
        meta.rewards = Some(Vec::new());
        meta.compute_units_consumed = Some(7);
        meta.cost_units = Some(9);
        meta.loaded_addresses = LoadedAddresses::default();
        meta
    }

    #[test]
    fn decodes_bincode_metadata_for_early_epochs() {
        let stored = StoredTransactionStatusMeta {
            status: Ok(()),
            fee: 42,
            pre_balances: vec![1, 2],
            post_balances: vec![3, 4],
            inner_instructions: None,
            log_messages: Some(vec!["hello".into()]),
            pre_token_balances: Some(Vec::new()),
            post_token_balances: Some(Vec::new()),
            rewards: Some(Vec::new()),
            return_data: None,
            compute_units_consumed: Some(7),
            cost_units: Some(9),
        };
        let bytes = bincode::serialize(&stored).expect("bincode serialize");
        let decoded = decode_transaction_status_meta(0, &bytes).expect("decode");
        assert_eq!(decoded, TransactionStatusMeta::from(stored));
    }

    #[test]
    fn decodes_protobuf_metadata_for_later_epochs() {
        let meta = sample_meta();
        let generated: solana_storage_proto::convert::generated::TransactionStatusMeta =
            meta.clone().into();
        let bytes = prost_011::Message::encode_to_vec(&generated);
        let decoded = decode_transaction_status_meta(157 * 432000, &bytes).expect("decode");
        assert_eq!(decoded, meta);
    }

    #[test]
    fn falls_back_to_proto_when_early_epoch_bytes_are_proto() {
        let meta = sample_meta();
        let generated: solana_storage_proto::convert::generated::TransactionStatusMeta =
            meta.clone().into();
        let bytes = prost_011::Message::encode_to_vec(&generated);
        // Epoch 100 should try bincode first; if those bytes are proto, we must fall back.
        let decoded = decode_transaction_status_meta(100 * 432000, &bytes).expect("decode");
        assert_eq!(decoded, meta);
    }

    #[test]
    fn empty_frame_decodes_to_default() {
        let decoded = decode_transaction_status_meta_from_frame(0, Vec::new()).expect("decode");
        assert_eq!(decoded, TransactionStatusMeta::default());
    }

    #[test]
    fn raw_bincode_frame_without_zstd_still_decodes() {
        let stored = StoredTransactionStatusMeta {
            status: Ok(()),
            fee: 1,
            pre_balances: vec![],
            post_balances: vec![],
            inner_instructions: None,
            log_messages: None,
            pre_token_balances: Some(Vec::new()),
            post_token_balances: Some(Vec::new()),
            rewards: Some(Vec::new()),
            return_data: None,
            compute_units_consumed: None,
            cost_units: None,
        };
        let raw_bytes = bincode::serialize(&stored).expect("serialize");
        let decoded =
            decode_transaction_status_meta_from_frame(0, raw_bytes).expect("decode fallback");
        assert_eq!(decoded, TransactionStatusMeta::from(stored));
    }
}

/// Firehose transaction payload passed to [`Handler`] callbacks.
#[derive(Debug, Clone)]
pub struct TransactionData {
    /// Slot that contains the transaction.
    pub slot: u64,
    /// Index of the transaction within the slot.
    pub transaction_slot_index: usize,
    /// Transaction signature.
    pub signature: solana_signature::Signature,
    /// Hash of the transaction message.
    pub message_hash: Hash,
    /// Indicates whether the transaction is a vote.
    pub is_vote: bool,
    /// Status metadata returned by the Solana runtime.
    pub transaction_status_meta: solana_transaction_status::TransactionStatusMeta,
    /// Fully decoded transaction.
    pub transaction: VersionedTransaction,
}

/// Block entry metadata passed to [`Handler`] callbacks.
#[derive(Debug, Clone)]
pub struct EntryData {
    /// Slot that generated the entry.
    pub slot: u64,
    /// Index of the entry within the slot.
    pub entry_index: usize,
    /// Range of transaction indexes covered by the entry.
    pub transaction_indexes: Range<usize>,
    /// Number of hashes associated with the entry.
    pub num_hashes: u64,
    /// Entry hash.
    pub hash: Hash,
}

/// Reward data conveyed to reward [`Handler`] callbacks.
#[derive(Debug, Clone)]
pub struct RewardsData {
    /// Slot the rewards correspond to.
    pub slot: u64,
    /// Reward recipients and their associated reward information.
    pub rewards: Vec<(Address, RewardInfo)>,
}

/// Block-level data streamed to block handlers.
#[derive(Debug)]
pub enum BlockData {
    /// Fully populated block payload with ledger metadata.
    Block {
        /// Parent slot number.
        parent_slot: u64,
        /// Parent block hash.
        parent_blockhash: Hash,
        /// Current block slot.
        slot: u64,
        /// Current block hash.
        blockhash: Hash,
        /// Rewards keyed by account and partition information.
        rewards: KeyedRewardsAndNumPartitions,
        /// Optional Unix timestamp for the block.
        block_time: Option<i64>,
        /// Optional ledger block height.
        block_height: Option<u64>,
        /// Number of executed transactions in the block.
        executed_transaction_count: u64,
        /// Number of entries contained in the block.
        entry_count: u64,
    },
    /// Marker indicating the slot appears skipped (either truly skipped or it is late and will
    /// arrive out of order).
    PossibleLeaderSkipped {
        /// Slot number that either lacked a block or may still arrive later.
        slot: u64,
    },
}

impl BlockData {
    /// Returns the slot associated with this block or skipped slot.
    #[inline(always)]
    pub const fn slot(&self) -> u64 {
        match self {
            BlockData::Block { slot, .. } => *slot,
            BlockData::PossibleLeaderSkipped { slot } => *slot,
        }
    }

    /// Returns `true` if this record currently represents a missing/possibly skipped slot.
    #[inline(always)]
    pub const fn was_skipped(&self) -> bool {
        matches!(self, BlockData::PossibleLeaderSkipped { .. })
    }

    /// Returns the optional block time when available.
    #[inline(always)]
    pub const fn block_time(&self) -> Option<i64> {
        match self {
            BlockData::Block { block_time, .. } => *block_time,
            BlockData::PossibleLeaderSkipped { .. } => None,
        }
    }
}

type HandlerResult = Result<(), SharedError>;
type HandlerFuture = BoxFuture<'static, HandlerResult>;

/// Asynchronous callback invoked for each firehose event of type `Data`.
pub trait Handler<Data>: Fn(usize, Data) -> HandlerFuture + Send + Sync + Clone + 'static {}

impl<Data, F> Handler<Data> for F where
    F: Fn(usize, Data) -> HandlerFuture + Send + Sync + Clone + 'static
{
}

/// Function pointer alias for [`Handler`] callbacks.
pub type HandlerFn<Data> = fn(usize, Data) -> HandlerFuture;
/// Convenience alias for block handlers accepted by [`firehose`].
pub type OnBlockFn = HandlerFn<BlockData>;
/// Convenience alias for transaction handlers accepted by [`firehose`].
pub type OnTxFn = HandlerFn<TransactionData>;
/// Convenience alias for entry handlers accepted by [`firehose`].
pub type OnEntryFn = HandlerFn<EntryData>;
/// Convenience alias for reward handlers accepted by [`firehose`].
pub type OnRewardFn = HandlerFn<RewardsData>;
/// Type alias for [`StatsTracking`] using simple function pointers.
pub type StatsTracker = StatsTracking<HandlerFn<Stats>>;
/// Convenience alias for firehose error handlers.
pub type OnErrorFn = HandlerFn<FirehoseErrorContext>;
/// Convenience alias for stats tracking handlers accepted by [`firehose`].
pub type OnStatsTrackingFn = StatsTracking<HandlerFn<Stats>>;

/// Metadata describing a firehose worker failure.
#[derive(Clone, Debug)]
pub struct FirehoseErrorContext {
    /// Thread index that encountered the error.
    pub thread_id: usize,
    /// Slot the worker was processing when the error surfaced.
    pub slot: u64,
    /// Epoch derived from the failing slot.
    pub epoch: u64,
    /// Stringified error payload for display/logging.
    pub error_message: String,
}

/// Streams blocks, transactions, entries, rewards, and stats to user-provided handlers.
///
/// The requested `slot_range` is half-open `[start, end)`; on recoverable errors the
/// runner restarts from the last processed slot to maintain coverage.
#[inline]
#[allow(clippy::too_many_arguments)]
pub async fn firehose<OnBlock, OnTransaction, OnEntry, OnRewards, OnStats, OnError>(
    threads: u64,
    slot_range: Range<u64>,
    on_block: Option<OnBlock>,
    on_tx: Option<OnTransaction>,
    on_entry: Option<OnEntry>,
    on_rewards: Option<OnRewards>,
    on_error: Option<OnError>,
    stats_tracking: Option<StatsTracking<OnStats>>,
    shutdown_signal: Option<broadcast::Receiver<()>>,
) -> Result<(), (FirehoseError, u64)>
where
    OnBlock: Handler<BlockData>,
    OnTransaction: Handler<TransactionData>,
    OnEntry: Handler<EntryData>,
    OnRewards: Handler<RewardsData>,
    OnStats: Handler<Stats>,
    OnError: Handler<FirehoseErrorContext>,
{
    if threads == 0 {
        return Err((
            FirehoseError::OnLoadError("Number of threads must be greater than 0".into()),
            slot_range.start,
        ));
    }
    let client = Client::new();
    log::info!(target: LOG_MODULE, "starting firehose...");
    log::info!(target: LOG_MODULE, "index base url: {}", SLOT_OFFSET_INDEX.base_url());

    let slot_range = Arc::new(slot_range);

    // divide slot_range into n subranges
    let subranges = generate_subranges(&slot_range, threads);
    if threads > 1 {
        log::debug!(target: LOG_MODULE, "⚡ thread sub-ranges: {:?}", subranges);
    }

    let firehose_start = std::time::Instant::now();
    let shutdown_flag = Arc::new(AtomicBool::new(false));
    if let Some(ref rx) = shutdown_signal {
        let mut rx = rx.resubscribe();
        let flag = shutdown_flag.clone();
        tokio::spawn(async move {
            if rx.recv().await.is_ok() {
                log::info!(target: LOG_MODULE, "shutdown signal received; notifying firehose threads");
                flag.store(true, Ordering::SeqCst);
            }
        });
    }
    let mut handles = Vec::new();
    // Shared per-thread error counters
    let error_counts: Arc<Vec<AtomicU32>> =
        Arc::new((0..subranges.len()).map(|_| AtomicU32::new(0)).collect());

    let overall_slots_processed: Arc<AtomicU64> = Arc::new(AtomicU64::new(0));
    let overall_blocks_processed: Arc<AtomicU64> = Arc::new(AtomicU64::new(0));
    let overall_transactions_processed: Arc<AtomicU64> = Arc::new(AtomicU64::new(0));
    let overall_entries_processed: Arc<AtomicU64> = Arc::new(AtomicU64::new(0));
    let pending_skipped_slots: Arc<DashMap<usize, DashSet<u64>>> = Arc::new(DashMap::new());

    for (thread_index, mut slot_range) in subranges.into_iter().enumerate() {
        let error_counts = error_counts.clone();
        let client = client.clone();
        let on_block = on_block.clone();
        let on_tx = on_tx.clone();
        let on_entry = on_entry.clone();
        let on_reward = on_rewards.clone();
        let on_error = on_error.clone();
        let overall_slots_processed = overall_slots_processed.clone();
        let overall_blocks_processed = overall_blocks_processed.clone();
        let overall_transactions_processed = overall_transactions_processed.clone();
        let overall_entries_processed = overall_entries_processed.clone();
        let stats_tracking = stats_tracking.clone();
        let transactions_since_stats = Arc::new(AtomicU64::new(0));
        let blocks_since_stats = Arc::new(AtomicU64::new(0));
        let slots_since_stats = Arc::new(AtomicU64::new(0));
        let last_pulse = Arc::new(AtomicU64::new(0));
        let transactions_since_stats_cloned = transactions_since_stats.clone();
        let blocks_since_stats_cloned = blocks_since_stats.clone();
        let slots_since_stats_cloned = slots_since_stats.clone();
        let last_pulse_cloned = last_pulse.clone();
        let shutdown_flag = shutdown_flag.clone();
        let pending_skipped_slots = pending_skipped_slots.clone();
        let thread_shutdown_rx = shutdown_signal.as_ref().map(|rx| rx.resubscribe());

        let handle = tokio::spawn(async move {
            let transactions_since_stats = transactions_since_stats_cloned;
            let blocks_since_stats = blocks_since_stats_cloned;
            let slots_since_stats = slots_since_stats_cloned;
            let last_pulse = last_pulse_cloned;
            let mut shutdown_rx = thread_shutdown_rx;
            let start_time = firehose_start;
            last_pulse.store(
                firehose_start.elapsed().as_nanos() as u64,
                Ordering::Relaxed,
            );
            let log_target = format!("{}::T{:03}", LOG_MODULE, thread_index);
            let mut skip_until_index = None;
            let last_emitted_slot = slot_range.start.saturating_sub(1);
            let block_enabled = on_block.is_some();
            let tx_enabled = on_tx.is_some();
            let entry_enabled = on_entry.is_some();
            let reward_enabled = on_reward.is_some();
            let tracking_enabled = stats_tracking.is_some();
            if block_enabled {
                pending_skipped_slots.entry(thread_index).or_default();
            }
            let mut last_counted_slot = slot_range.start.saturating_sub(1);
            let mut last_emitted_slot_global = slot_range.start.saturating_sub(1);
            let mut thread_stats = if tracking_enabled {
                Some(ThreadStats {
                    thread_id: thread_index,
                    start_time,
                    finish_time: None,
                    slot_range: slot_range.clone(),
                    initial_slot_range: slot_range.clone(),
                    current_slot: slot_range.start,
                    slots_processed: 0,
                    blocks_processed: 0,
                    leader_skipped_slots: 0,
                    transactions_processed: 0,
                    entries_processed: 0,
                    rewards_processed: 0,
                })
            } else {
                None
            };

            // let mut triggered = false;
            while let Err((err, slot)) = async {
                let mut last_emitted_slot = last_emitted_slot_global;
                if poll_shutdown(&shutdown_flag, &mut shutdown_rx) {
                    log::info!(
                        target: &log_target,
                        "shutdown requested; terminating firehose thread {}",
                        thread_index
                    );
                    return Ok(());
                }
                let epoch_range = slot_to_epoch(slot_range.start)..=slot_to_epoch(slot_range.end - 1);
                log::info!(
                    target: &log_target,
                    "slot range: {} (epoch {}) ... {} (epoch {})",
                    slot_range.start,
                    slot_to_epoch(slot_range.start),
                    slot_range.end,
                    slot_to_epoch(slot_range.end)
                );

                log::info!(target: &log_target, "🚒 starting firehose...");

                // for each epoch
                let mut current_slot: Option<u64> = None;
                for epoch_num in epoch_range.clone() {
                    if poll_shutdown(&shutdown_flag, &mut shutdown_rx) {
                        log::info!(
                            target: &log_target,
                            "shutdown requested; terminating firehose thread {}",
                            thread_index
                        );
                        return Ok(());
                    }
                    log::info!(target: &log_target, "entering epoch {}", epoch_num);
                    let stream = match timeout(OP_TIMEOUT, fetch_epoch_stream(epoch_num, &client)).await {
                        Ok(stream) => stream,
                        Err(_) => {
                            return Err((FirehoseError::OperationTimeout("fetch_epoch_stream"), current_slot.unwrap_or(slot_range.start)));
                        }
                    };
                    let mut reader = NodeReader::new(stream);

                    let header_fut = reader.read_raw_header();
                    let header = match timeout(OP_TIMEOUT, header_fut).await {
                        Ok(res) => res
                            .map_err(FirehoseError::ReadHeader)
                            .map_err(|e| (e, current_slot.unwrap_or(slot_range.start)))?,
                        Err(_) => {
                            return Err((FirehoseError::OperationTimeout("read_raw_header"), current_slot.unwrap_or(slot_range.start)));
                        }
                    };
                    log::debug!(target: &log_target, "read epoch {} header: {:?}", epoch_num, header);

                    let (epoch_start, epoch_end_inclusive) = epoch_to_slot_range(epoch_num);
                    let local_start = std::cmp::max(slot_range.start, epoch_start);
                    let local_end_inclusive =
                        std::cmp::min(slot_range.end.saturating_sub(1), epoch_end_inclusive);
                    if local_start > local_end_inclusive {
                        log::debug!(
                            target: &log_target,
                            "epoch {} has no overlap with thread range ({}..{}), skipping",
                            epoch_num,
                            slot_range.start,
                            slot_range.end
                        );
                        continue;
                    }

                    let mut previous_blockhash = Hash::default();
                    let mut latest_entry_blockhash = Hash::default();
                    // Reset counters to align to the local epoch slice; prevents boundary slots
                    // from being treated as already-counted after a restart.
                    last_counted_slot = local_start.saturating_sub(1);
                    current_slot = None;
                    if tracking_enabled
                        && let Some(ref mut stats) = thread_stats {
                            stats.current_slot = local_start;
                            stats.slot_range.start = local_start;
                        }

                    if local_start > epoch_start {
                        // Seek to the previous slot so the stream includes all nodes
                        // (transactions, entries, rewards) that precede the block payload for
                        // `local_start`.
                        let seek_slot = local_start.saturating_sub(1);
                        let seek_fut = reader.seek_to_slot(seek_slot);
                        match timeout(OP_TIMEOUT, seek_fut).await {
                            Ok(res) => res.map_err(|e| (e, current_slot.unwrap_or(slot_range.start)))?,
                            Err(_) => {
                                return Err((FirehoseError::OperationTimeout("seek_to_slot"), current_slot.unwrap_or(slot_range.start)));
                            }
                        }
                    }

                    // for each item in each block
                    let mut item_index = 0;
                    let mut displayed_skip_message = false;
                    loop {
                        if poll_shutdown(&shutdown_flag, &mut shutdown_rx) {
                            log::info!(
                                target: &log_target,
                                "shutdown requested; terminating firehose thread {}",
                                thread_index
                            );
                            return Ok(());
                        }
                        let read_fut = reader.read_until_block();
                        let nodes = match timeout(OP_TIMEOUT, read_fut).await {
                            Ok(result) => result
                                .map_err(FirehoseError::ReadUntilBlockError)
                                .map_err(|e| (e, current_slot.unwrap_or(slot_range.start)))?,
                            Err(_) => {
                                log::warn!(target: &log_target, "timeout reading next block, retrying (will restart)...");
                                return Err((FirehoseError::OperationTimeout("read_until_block"), current_slot.map(|s| s + 1).unwrap_or(slot_range.start)));
                            }
                        };
                        if nodes.is_empty() {
                            log::info!(
                                target: &log_target,
                                "reached end of epoch {}",
                                epoch_num
                            );
                            break;
                        }
                        if let Some(last_node) = nodes.0.last()
                            && !last_node.get_node().is_block()
                        {
                            log::info!(target: &log_target, "reached end of epoch {}", epoch_num);
                            break;
                        }
                        let block = nodes
                            .get_block()
                            .map_err(FirehoseError::GetBlockError)
                            .map_err(|e| (e, current_slot.unwrap_or(slot_range.start)))?;
                        log::debug!(
                            target: &log_target,
                            "read {} items from epoch {}, now at slot {}",
                            item_index,
                            epoch_num,
                            block.slot
                        );
                        let slot = block.slot;
                        if slot > local_end_inclusive {
                            log::debug!(
                                target: &log_target,
                                "reached end of local slice at slot {} (epoch {}), stopping",
                                slot,
                                epoch_num
                            );
                            break;
                        }
                        if slot >= slot_range.end {
                            log::info!(target: &log_target, "reached end of slot range at slot {}", slot);
                            // Return early to terminate the firehose thread cleanly. We use >=
                            // because slot_range is half-open [start, end), so any slot equal
                            // to end is out-of-range and must not be processed. Do not emit
                            // synthetic skipped slots here; another thread may own the boundary.
                            if block_enabled {
                                pending_skipped_slots.remove(&thread_index);
                            }
                            return Ok(());
                        }
                        debug_assert!(slot < slot_range.end, "processing out-of-range slot {} (end {})", slot, slot_range.end);
                        if slot < slot_range.start {
                            if slot.saturating_add(1) == slot_range.start {
                                log::debug!(
                                    target: &log_target,
                                    "priming reader with preceding slot {}, skipping",
                                    slot
                                );
                            } else {
                                log::warn!(
                                    target: &log_target,
                                    "encountered slot {} before start of range {}, skipping",
                                    slot,
                                    slot_range.start
                                );
                            }
                            continue;
                        }
                        current_slot = Some(slot);
                        let mut entry_index: usize = 0;
                        let mut this_block_executed_transaction_count: u64 = 0;
                        let mut this_block_entry_count: u64 = 0;
                        let mut this_block_rewards: Vec<(Address, RewardInfo)> = Vec::new();

                        for node_with_cid in &nodes.0 {
                            item_index += 1;
                            if let Some(skip) = skip_until_index {
                                if item_index < skip {
                                    if !displayed_skip_message {
                                        log::info!(
                                            target: &log_target,
                                            "skipping until index {} (at {})",
                                            skip,
                                            item_index
                                        );
                                        displayed_skip_message = true;
                                    }
                                    continue;
                                } else {
                                    log::info!(
                                        target: &log_target,
                                        "reached target index {}, resuming...",
                                        skip
                                    );
                                    skip_until_index = None;
                                }
                            }
                            let node = node_with_cid.get_node();

                            if let Some(ref mut stats) = thread_stats {
                                stats.current_slot = slot;
                            }

                            let error_slot = current_slot.unwrap_or(slot_range.start);

                            use crate::node::Node::*;
                            match node {
                                Transaction(tx) => {
                                    if tx_enabled
                                        && let Some(on_tx_cb) = on_tx.as_ref()
                                    {
                                        let error_slot = current_slot.unwrap_or(slot_range.start);
                                        let versioned_tx = tx.as_parsed().map_err(|err| {
                                            (
                                                FirehoseError::NodeDecodingError(item_index, err),
                                                error_slot,
                                            )
                                        })?;
                                        let reassembled_metadata = nodes
                                            .reassemble_dataframes(tx.metadata.clone())
                                            .map_err(|err| {
                                                (
                                                    FirehoseError::NodeDecodingError(item_index, err),
                                                    error_slot,
                                                )
                                            })?;

                                        let as_native_metadata = decode_transaction_status_meta_from_frame(
                                            block.slot,
                                            reassembled_metadata,
                                        )
                                        .map_err(|err| {
                                            (
                                                FirehoseError::NodeDecodingError(item_index, err),
                                                error_slot,
                                            )
                                        })?;

                                        let message_hash = {
                                            #[cfg(feature = "verify-transaction-signatures")]
                                            {
                                                versioned_tx.verify_and_hash_message().map_err(|err| {
                                                    (
                                                        FirehoseError::TransactionHandlerError(Box::new(err)),
                                                        error_slot,
                                                    )
                                                })?
                                            }
                                            #[cfg(not(feature = "verify-transaction-signatures"))]
                                            {
                                                versioned_tx.message.hash()
                                            }
                                        };
                                        let signature = versioned_tx
                                            .signatures
                                            .first()
                                            .ok_or_else(|| {
                                                Box::new(std::io::Error::new(
                                                    std::io::ErrorKind::InvalidData,
                                                    "transaction missing signature",
                                                )) as SharedError
                                            })
                                            .map_err(|err| {
                                                (
                                                    FirehoseError::NodeDecodingError(
                                                        item_index,
                                                        err,
                                                    ),
                                                    error_slot,
                                                )
                                            })?;
                                        let is_vote = is_simple_vote_transaction(&versioned_tx);

                                        on_tx_cb(
                                            thread_index,
                                            TransactionData {
                                                slot: block.slot,
                                                transaction_slot_index: tx.index.unwrap() as usize,
                                                signature: *signature,
                                                message_hash,
                                                is_vote,
                                                transaction_status_meta: as_native_metadata.clone(),
                                                transaction: versioned_tx.clone(),
                                            },
                                        )
                                        .await
                                        .map_err(|e| {
                                            (
                                                FirehoseError::TransactionHandlerError(e),
                                                error_slot,
                                            )
                                        })?;
                                    }
                                    fetch_add_if(
                                        tracking_enabled,
                                        &overall_transactions_processed,
                                        1,
                                    );
                                    if let Some(ref mut stats) = thread_stats {
                                        stats.transactions_processed += 1;
                                    }
                                    transactions_since_stats.fetch_add(1, Ordering::Relaxed);
                                }
                                Entry(entry) => {
                                    let entry_hash = Hash::from(entry.hash.to_bytes());
                                    let entry_transaction_count = entry.transactions.len();
                                    let entry_transaction_count_u64 = entry_transaction_count as u64;
                                    let starting_transaction_index_u64 =
                                        this_block_executed_transaction_count;
                                    latest_entry_blockhash = entry_hash;
                                    this_block_executed_transaction_count += entry_transaction_count_u64;
                                    this_block_entry_count += 1;

                                    if entry_enabled && let Some(on_entry_cb) = on_entry.as_ref() {
                                        let starting_transaction_index = usize::try_from(
                                            starting_transaction_index_u64,
                                        )
                                        .map_err(|err| {
                                            (
                                                FirehoseError::EntryHandlerError(Box::new(err)),
                                                error_slot,
                                            )
                                        })?;
                                        let transaction_indexes_end =
                                            starting_transaction_index + entry_transaction_count;
                                        on_entry_cb(
                                            thread_index,
                                            EntryData {
                                                slot: block.slot,
                                                entry_index,
                                                transaction_indexes: starting_transaction_index
                                                    ..transaction_indexes_end,
                                                num_hashes: entry.num_hashes,
                                                hash: entry_hash,
                                            },
                                        )
                                        .await
                                        .map_err(|e| {
                                            (
                                                FirehoseError::EntryHandlerError(e),
                                                error_slot,
                                            )
                                        })?;
                                    }
                                    entry_index += 1;
                                    fetch_add_if(
                                        tracking_enabled,
                                        &overall_entries_processed,
                                        1,
                                    );
                                    if let Some(ref mut stats) = thread_stats {
                                        stats.entries_processed += 1;
                                    }
                                }
                                Block(block) => {
                                    let prev_last_counted_slot = last_counted_slot;
                                    let thread_stats_snapshot = thread_stats.as_ref().map(|stats| {
                                        (
                                            stats.slots_processed,
                                            stats.blocks_processed,
                                            stats.leader_skipped_slots,
                                            stats.current_slot,
                                        )
                                    });

                                    let next_expected_slot = prev_last_counted_slot.saturating_add(1);
                                    let skip_start_from_previous = last_counted_slot.saturating_add(1);
                                    let skip_start = skip_start_from_previous.max(next_expected_slot);

                                    let skipped_epoch = slot_to_epoch(last_counted_slot);
                                    for skipped_slot in skip_start..slot {
                                        if slot_to_epoch(skipped_slot) != skipped_epoch {
                                            break;
                                        }
                                        log::debug!(
                                            target: &log_target,
                                            "leader skipped slot {} (prev_counted {}, current slot {})",
                                            skipped_slot,
                                            prev_last_counted_slot,
                                            slot,
                                        );
                                        if block_enabled {
                                            pending_skipped_slots
                                                .entry(thread_index)
                                                .or_default()
                                                .insert(skipped_slot);
                                        }
                                        if block_enabled
                                            && let Some(on_block_cb) = on_block.as_ref()
                                            && skipped_slot > last_emitted_slot {
                                                last_emitted_slot = skipped_slot;
                                                on_block_cb(
                                                    thread_index,
                                                    BlockData::PossibleLeaderSkipped {
                                                        slot: skipped_slot,
                                                    },
                                                )
                                                .await
                                                .map_err(|e| {
                                                    (
                                                        FirehoseError::BlockHandlerError(e),
                                                        error_slot,
                                                    )
                                                })?;
                                            }
                                        if tracking_enabled {
                                            overall_slots_processed.fetch_add(1, Ordering::Relaxed);
                                            slots_since_stats.fetch_add(1, Ordering::Relaxed);
                                            if let Some(ref mut stats) = thread_stats {
                                                stats.leader_skipped_slots += 1;
                                                stats.slots_processed += 1;
                                                stats.current_slot = skipped_slot;
                                            }
                                        }
                                        last_counted_slot = skipped_slot;
                                    }

                                    let cleared_pending_skip = if block_enabled {
                                        clear_pending_skip(
                                            &pending_skipped_slots,
                                            thread_index,
                                            slot,
                                        )
                                    } else {
                                        false
                                    };

                                    if slot <= last_counted_slot && !cleared_pending_skip {
                                        log::debug!(
                                            target: &log_target,
                                            "duplicate block {}, already counted (last_counted={})",
                                            slot,
                                            last_counted_slot,
                                        );
                                        this_block_rewards.clear();
                                        continue;
                                    }

                                    if block_enabled {
                                        if let Some(on_block_cb) = on_block.as_ref() {
                                            let keyed_rewards = std::mem::take(&mut this_block_rewards);
                                            if slot > last_emitted_slot {
                                                last_emitted_slot = slot;
                                                on_block_cb(
                                                    thread_index,
                                                    BlockData::Block {
                                                        parent_slot: block.meta.parent_slot,
                                                        parent_blockhash: previous_blockhash,
                                                        slot: block.slot,
                                                        blockhash: latest_entry_blockhash,
                                                        rewards: KeyedRewardsAndNumPartitions {
                                                            keyed_rewards,
                                                            num_partitions: None,
                                                        },
                                                        block_time: Some(block.meta.blocktime as i64),
                                                        block_height: block.meta.block_height,
                                                        executed_transaction_count:
                                                            this_block_executed_transaction_count,
                                                        entry_count: this_block_entry_count,
                                                    },
                                                )
                                                .await
                                                .map_err(|e| {
                                                    (
                                                        FirehoseError::BlockHandlerError(e),
                                                        error_slot,
                                                    )
                                                })?;
                                            }
                                        }
                                    } else {
                                        this_block_rewards.clear();
                                    }
                                    previous_blockhash = latest_entry_blockhash;

                                    if tracking_enabled {
                                        overall_slots_processed.fetch_add(1, Ordering::Relaxed);
                                        overall_blocks_processed.fetch_add(1, Ordering::Relaxed);
                                        slots_since_stats.fetch_add(1, Ordering::Relaxed);
                                        blocks_since_stats.fetch_add(1, Ordering::Relaxed);
                                        if let Some(ref mut stats) = thread_stats {
                                            stats.blocks_processed += 1;
                                            stats.slots_processed += 1;
                                            stats.current_slot = slot;
                                        }

                                        if let (Some(stats_tracking_cfg), Some(thread_stats_ref)) =
                                            (&stats_tracking, thread_stats.as_mut())
                                            && slot % stats_tracking_cfg.tracking_interval_slots == 0
                                                && let Err(err) = maybe_emit_stats(
                                                    stats_tracking.as_ref(),
                                                    thread_index,
                                                    thread_stats_ref,
                                                    &overall_slots_processed,
                                                    &overall_blocks_processed,
                                                    &overall_transactions_processed,
                                                    &overall_entries_processed,
                                                &transactions_since_stats,
                                                &blocks_since_stats,
                                                &slots_since_stats,
                                                &last_pulse,
                                                start_time,
                                            )
                                            .await
                                            {
                                                blocks_since_stats.fetch_sub(1, Ordering::Relaxed);
                                                    slots_since_stats.fetch_sub(1, Ordering::Relaxed);
                                                    overall_blocks_processed
                                                        .fetch_sub(1, Ordering::Relaxed);
                                                    overall_slots_processed
                                                        .fetch_sub(1, Ordering::Relaxed);
                                                    if let Some((
                                                        prev_slots_processed,
                                                        prev_blocks_processed,
                                                        prev_leader_skipped,
                                                        prev_current_slot,
                                                    )) = thread_stats_snapshot
                                                    {
                                                        thread_stats_ref.slots_processed =
                                                            prev_slots_processed;
                                                        thread_stats_ref.blocks_processed =
                                                            prev_blocks_processed;
                                                        thread_stats_ref.leader_skipped_slots =
                                                            prev_leader_skipped;
                                                        thread_stats_ref.current_slot =
                                                            prev_current_slot;
                                                    }
                                                    last_counted_slot = prev_last_counted_slot;
                                                    return Err(err);
                                                }
                                    }

                                    if slot > last_counted_slot {
                                        last_counted_slot = slot;
                                    }
                                }
                                Subset(_subset) => (),
                                Epoch(_epoch) => (),
                                Rewards(rewards) => {
                                    if reward_enabled || block_enabled {
                                        let reassembled = nodes
                                            .reassemble_dataframes(rewards.data.clone())
                                            .map_err(|err| {
                                                (
                                                    FirehoseError::NodeDecodingError(item_index, err),
                                                    current_slot.unwrap_or(slot_range.start),
                                                )
                                            })?;
                                        if reassembled.is_empty() {
                                            this_block_rewards.clear();
                                            if reward_enabled
                                                && let Some(on_reward_cb) = on_reward.as_ref()
                                            {
                                                on_reward_cb(
                                                    thread_index,
                                                    RewardsData {
                                                        slot: block.slot,
                                                        rewards: Vec::new(),
                                                    },
                                                )
                                                .await
                                                .map_err(|e| {
                                                    (
                                                        FirehoseError::RewardHandlerError(e),
                                                        error_slot,
                                                    )
                                                })?;
                                            }
                                            continue;
                                        }

                                        let decompressed = utils::decompress_zstd(reassembled)
                                            .map_err(|err| {
                                                (
                                                    FirehoseError::NodeDecodingError(
                                                        item_index,
                                                        err,
                                                    ),
                                                    error_slot,
                                                )
                                            })?;

                                        let decoded =
                                            prost_011::Message::decode(decompressed.as_slice())
                                                .map_err(|err| {
                                                    (
                                                        FirehoseError::NodeDecodingError(
                                                            item_index,
                                                            Box::new(err),
                                                        ),
                                                        error_slot,
                                                    )
                                                })?;
                                        let keyed_rewards = convert_proto_rewards(&decoded)
                                            .map_err(|err| {
                                                (
                                                    FirehoseError::NodeDecodingError(item_index, err),
                                                    error_slot,
                                                )
                                            })?;
                                        if reward_enabled
                                            && let Some(on_reward_cb) = on_reward.as_ref()
                                        {
                                            on_reward_cb(
                                                thread_index,
                                                RewardsData {
                                                    slot: block.slot,
                                                    rewards: keyed_rewards.clone(),
                                                },
                                            )
                                            .await
                                            .map_err(|e| {
                                                (
                                                    FirehoseError::RewardHandlerError(e),
                                                    error_slot,
                                                )
                                            })?;
                                        }
                                        this_block_rewards = keyed_rewards;
                                        if let Some(ref mut stats) = thread_stats {
                                            stats.rewards_processed +=
                                                this_block_rewards.len() as u64;
                                        }
                                    }
                                }
                                DataFrame(_data_frame) => (),
                            }
                        }
                        if block.slot == slot_range.end - 1 {
                            let finish_time = std::time::Instant::now();
                            let elapsed = finish_time.duration_since(start_time);
                            log::info!(target: &log_target, "processed slot {}", block.slot);
                            let elapsed_pretty = human_readable_duration(elapsed);
                            log::info!(
                                target: &log_target,
                                "processed {} slots across {} epochs in {}.",
                                slot_range.end - slot_range.start,
                                slot_to_epoch(slot_range.end) + 1 - slot_to_epoch(slot_range.start),
                                elapsed_pretty
                            );
                            log::info!(target: &log_target, "a 🚒 firehose thread completed its work.");
                            // On completion, report threads with non-zero error counts for
                            // visibility.
                            let summary: String = error_counts
                                .iter()
                                .enumerate()
                                .filter_map(|(i, c)| {
                                    let v = c.load(Ordering::Relaxed);
                                    if v > 0 {
                                        Some(format!("{:03}({})", i, v))
                                    } else {
                                        None
                                    }
                                })
                                .collect::<Vec<_>>()
                                .join(", ");
                            if !summary.is_empty() {
                                log::debug!(target: &log_target, "threads with errors: {}", summary);
                            }
                            return Ok(());
                        }
                    }
                    if let Some(expected_last_slot) = slot_range.end.checked_sub(1)
                        && last_counted_slot < expected_last_slot
                    {
                        // Do not synthesize skipped slots during final flush; another thread may
                        // cover the remaining range (especially across epoch boundaries).
                    }
                    if let Some(ref mut stats) = thread_stats {
                        stats.finish_time = Some(std::time::Instant::now());
                        maybe_emit_stats(
                            stats_tracking.as_ref(),
                            thread_index,
                            stats,
                            &overall_slots_processed,
                            &overall_blocks_processed,
                            &overall_transactions_processed,
                            &overall_entries_processed,
                            &transactions_since_stats,
                            &blocks_since_stats,
                            &slots_since_stats,
                            &last_pulse,
                            start_time,
                        )
                        .await?;
                    }
                    if block_enabled {
                        pending_skipped_slots.remove(&thread_index);
                    }
                    log::info!(target: &log_target, "thread {} has finished its work", thread_index);
                    }
                    Ok(())
            }
            .await
            {
                if is_shutdown_error(&err) {
                    log::info!(
                        target: &log_target,
                        "shutdown requested; terminating firehose thread {}",
                        thread_index
                    );
                    break;
                }
                let epoch = slot_to_epoch(slot);
                let item_index = match &err {
                    FirehoseError::NodeDecodingError(item_index, _) => *item_index,
                    _ => 0,
                };
                let error_message = err.to_string();
                log::error!(
                    target: &log_target,
                    "🧯💦🔥 firehose encountered an error at slot {} in epoch {} and will roll back one slot and retry:",
                    slot,
                    epoch
                );
                log::error!(target: &log_target, "{}", error_message);
                if matches!(err, FirehoseError::SlotOffsetIndexError(_)) {
                    // Clear cached index data for this epoch to avoid retrying with a bad/partial index.
                    SLOT_OFFSET_INDEX.invalidate_epoch(epoch);
                }
                if let Some(on_error_cb) = on_error.clone() {
                    let context = FirehoseErrorContext {
                        thread_id: thread_index,
                        slot,
                        epoch,
                        error_message: error_message.clone(),
                    };
                    if let Err(handler_err) = on_error_cb(thread_index, context).await {
                        log::error!(
                            target: &log_target,
                            "on_error handler failed: {}",
                            handler_err
                        );
                    }
                }
                // Increment this thread's error counter
                error_counts[thread_index].fetch_add(1, Ordering::Relaxed);
                log::warn!(
                    target: &log_target,
                    "restarting from slot {} at index {}",
                    slot,
                    item_index,
                );
                // Update slot range to resume from the failed slot, not the original start.
                // Reset local tracking so we don't treat the resumed slot range as already counted.
                // If we've already counted this slot, resume from the next one to avoid duplicates.
                if slot <= last_counted_slot {
                    slot_range.start = last_counted_slot.saturating_add(1);
                } else {
                    slot_range.start = slot;
                }
                // Reset pulse timer to exclude downtime from next rate calc.
                last_pulse.store(start_time.elapsed().as_nanos() as u64, Ordering::Relaxed);
                if tracking_enabled
                    && let Some(ref mut stats_ref) = thread_stats {
                        stats_ref.slot_range.start = slot_range.start;
                        stats_ref.slot_range.end = slot_range.end;
                        // initial_slot_range remains unchanged for progress reporting.
                    }
                if block_enabled {
                    pending_skipped_slots.remove(&thread_index);
                }
                skip_until_index = Some(item_index);
                last_emitted_slot_global = last_emitted_slot;
            }
        });
        handles.push(handle);
    }

    // Wait for all threads to complete
    for handle in handles {
        handle.await.unwrap();
    }
    if stats_tracking.is_some() {
        let elapsed = firehose_start.elapsed();
        let elapsed_secs = elapsed.as_secs_f64();
        let total_slots = overall_slots_processed.load(Ordering::Relaxed);
        let total_blocks = overall_blocks_processed.load(Ordering::Relaxed);
        let total_transactions = overall_transactions_processed.load(Ordering::Relaxed);
        let total_leader_skipped = total_slots.saturating_sub(total_blocks);
        let total_errors: u64 = error_counts
            .iter()
            .map(|counter| counter.load(Ordering::Relaxed) as u64)
            .sum();
        let overall_tps = if elapsed_secs > 0.0 {
            total_transactions as f64 / elapsed_secs
        } else {
            0.0
        };
        log::info!(
            target: LOG_MODULE,
            "firehose summary: elapsed={:.2}s, slots={}, blocks={}, leader_skipped={}, transactions={}, overall_tps={:.2}, total_errors={}",
            elapsed_secs,
            total_slots,
            total_blocks,
            total_leader_skipped,
            total_transactions,
            overall_tps,
            total_errors
        );
    }
    if shutdown_flag.load(Ordering::SeqCst) {
        log::info!(target: LOG_MODULE, "firehose shutdown complete; all threads exited cleanly.");
    } else {
        log::info!(target: LOG_MODULE, "🚒 firehose finished successfully.");
    }
    Ok(())
}

#[allow(clippy::result_large_err)]
/// Builds a Geyser-backed firehose and returns a slot notification stream.
///
/// This helper is used by [`firehose`] when Geyser plugins need to be stood up in-process
/// rather than relying solely on remote streams. The provided `slot_range` is treated as a
/// half-open interval `[start, end)`, and the thread will restart from the last processed
/// slot on recoverable errors to maintain coverage.
pub fn firehose_geyser(
    rt: Arc<tokio::runtime::Runtime>,
    slot_range: Range<u64>,
    geyser_config_files: Option<&[PathBuf]>,
    index_base_url: &Url,
    client: &Client,
    on_load: impl Future<Output = Result<(), SharedError>> + Send + 'static,
    threads: u64,
) -> Result<Receiver<SlotNotification>, (FirehoseError, u64)> {
    if threads == 0 {
        return Err((
            FirehoseError::OnLoadError("Number of threads must be greater than 0".into()),
            slot_range.start,
        ));
    }
    log::info!(target: LOG_MODULE, "starting firehose...");
    log::info!(target: LOG_MODULE, "index base url: {}", index_base_url);
    let (confirmed_bank_sender, confirmed_bank_receiver) = unbounded();
    let mut entry_notifier_maybe = None;
    let mut block_meta_notifier_maybe = None;
    let mut transaction_notifier_maybe = None;
    if let Some(geyser_config_files) = geyser_config_files {
        log::debug!(target: LOG_MODULE, "geyser config files: {:?}", geyser_config_files);

        let service =
            solana_geyser_plugin_manager::geyser_plugin_service::GeyserPluginService::new(
                confirmed_bank_receiver.clone(),
                true,
                geyser_config_files,
            )
            .map_err(|e| (e.into(), slot_range.start))?;

        transaction_notifier_maybe = Some(
            service
                .get_transaction_notifier()
                .ok_or(FirehoseError::FailedToGetTransactionNotifier)
                .map_err(|e| (e, slot_range.start))?,
        );

        entry_notifier_maybe = service.get_entry_notifier();
        block_meta_notifier_maybe = service.get_block_metadata_notifier();

        log::debug!(target: LOG_MODULE, "geyser plugin service initialized.");
    }

    if entry_notifier_maybe.is_some() {
        log::debug!(target: LOG_MODULE, "entry notifications enabled")
    } else {
        log::debug!(target: LOG_MODULE, "none of the plugins have enabled entry notifications")
    }
    log::info!(target: LOG_MODULE, "running on_load...");
    rt.spawn(on_load);

    let slot_range = Arc::new(slot_range);
    let transaction_notifier_maybe = Arc::new(transaction_notifier_maybe);
    let entry_notifier_maybe = Arc::new(entry_notifier_maybe);
    let block_meta_notifier_maybe = Arc::new(block_meta_notifier_maybe);
    let confirmed_bank_sender = Arc::new(confirmed_bank_sender);

    // divide slot_range into n subranges
    let subranges = generate_subranges(&slot_range, threads);
    if threads > 1 {
        log::info!(target: LOG_MODULE, "⚡ thread sub-ranges: {:?}", subranges);
    }

    let mut handles = Vec::new();
    // Shared per-thread error counters
    let error_counts: Arc<Vec<AtomicU32>> =
        Arc::new((0..subranges.len()).map(|_| AtomicU32::new(0)).collect());

    for (i, slot_range) in subranges.into_iter().enumerate() {
        let transaction_notifier_maybe = (*transaction_notifier_maybe).clone();
        let entry_notifier_maybe = (*entry_notifier_maybe).clone();
        let block_meta_notifier_maybe = (*block_meta_notifier_maybe).clone();
        let confirmed_bank_sender = (*confirmed_bank_sender).clone();
        let client = client.clone();
        let error_counts = error_counts.clone();

        let rt_clone = rt.clone();

        let handle = std::thread::spawn(move || {
            rt_clone.block_on(async {
                firehose_geyser_thread(
                    slot_range,
                    transaction_notifier_maybe,
                    entry_notifier_maybe,
                    block_meta_notifier_maybe,
                    confirmed_bank_sender,
                    &client,
                    if threads > 1 { Some(i) } else { None },
                    error_counts,
                )
                .await
                .unwrap();
            });
        });
        handles.push(handle);
    }

    // Wait for all threads to complete
    for handle in handles {
        handle.join().unwrap();
    }
    log::info!(target: LOG_MODULE, "🚒 firehose finished successfully.");
    if let Some(block_meta_notifier) = block_meta_notifier_maybe.as_ref() {
        block_meta_notifier.notify_block_metadata(
            u64::MAX,
            "unload",
            u64::MAX,
            "unload",
            &KeyedRewardsAndNumPartitions {
                keyed_rewards: vec![],
                num_partitions: None,
            },
            None,
            None,
            0,
            0,
        );
    }
    Ok(confirmed_bank_receiver)
}

#[allow(clippy::too_many_arguments)]
#[allow(clippy::result_large_err)]
async fn firehose_geyser_thread(
    mut slot_range: Range<u64>,
    transaction_notifier_maybe: Option<Arc<dyn TransactionNotifier + Send + Sync + 'static>>,
    entry_notifier_maybe: Option<Arc<dyn EntryNotifier + Send + Sync + 'static>>,
    block_meta_notifier_maybe: Option<Arc<dyn BlockMetadataNotifier + Send + Sync + 'static>>,
    confirmed_bank_sender: Sender<SlotNotification>,
    client: &Client,
    thread_index: Option<usize>,
    error_counts: Arc<Vec<AtomicU32>>,
) -> Result<(), (FirehoseError, u64)> {
    let start_time = std::time::Instant::now();
    let log_target = if let Some(thread_index) = thread_index {
        format!("{}::T{:03}", LOG_MODULE, thread_index)
    } else {
        LOG_MODULE.to_string()
    };
    let initial_slot_range = slot_range.clone();
    let mut skip_until_index = None;
    let mut last_counted_slot = slot_range.start.saturating_sub(1);
    // let mut triggered = false;
    while let Err((err, slot)) = async {
            let epoch_range = slot_to_epoch(slot_range.start)..=slot_to_epoch(slot_range.end - 1);
            log::info!(
                target: &log_target,
                "slot range: {} (epoch {}) ... {} (epoch {})",
                slot_range.start,
                slot_to_epoch(slot_range.start),
                slot_range.end,
                slot_to_epoch(slot_range.end)
            );

            log::info!(target: &log_target, "🚒 starting firehose...");

            // for each epoch
            let mut current_slot: Option<u64> = None;
            for epoch_num in epoch_range.clone() {
                log::info!(target: &log_target, "entering epoch {}", epoch_num);
                let stream = match timeout(OP_TIMEOUT, fetch_epoch_stream(epoch_num, client)).await {
                    Ok(stream) => stream,
                    Err(_) => {
                        return Err((FirehoseError::OperationTimeout("fetch_epoch_stream"), current_slot.unwrap_or(slot_range.start)));
                    }
                };
                let mut reader = NodeReader::new(stream);

                let header_fut = reader.read_raw_header();
                let header = match timeout(OP_TIMEOUT, header_fut).await {
                    Ok(res) => res
                        .map_err(FirehoseError::ReadHeader)
                        .map_err(|e| (e, current_slot.unwrap_or(slot_range.start)))?,
                    Err(_) => {
                        return Err((FirehoseError::OperationTimeout("read_raw_header"), current_slot.unwrap_or(slot_range.start)));
                    }
                };
                log::debug!(target: &log_target, "read epoch {} header: {:?}", epoch_num, header);

                let (epoch_start, epoch_end_inclusive) = epoch_to_slot_range(epoch_num);
                let local_start = std::cmp::max(slot_range.start, epoch_start);
                let local_end_inclusive =
                    std::cmp::min(slot_range.end.saturating_sub(1), epoch_end_inclusive);
                if local_start > local_end_inclusive {
                    log::debug!(
                        target: &log_target,
                        "epoch {} has no overlap with thread range ({}..{}), skipping",
                        epoch_num,
                        slot_range.start,
                        slot_range.end
                    );
                    continue;
                }

                let mut todo_previous_blockhash = Hash::default();
                let mut todo_latest_entry_blockhash = Hash::default();
                // Reset counters to align to the local epoch slice; prevents boundary slots
                // from being treated as already-counted after a restart.
                last_counted_slot = local_start.saturating_sub(1);
                current_slot = None;

                if local_start > epoch_start {
                    // Seek to the slot immediately preceding the requested range so the reader
                    // captures the full node set (transactions, entries, rewards) for the
                    // target block on the next iteration.
                    let seek_slot = local_start.saturating_sub(1);
                    let seek_fut = reader.seek_to_slot(seek_slot);
                    match timeout(OP_TIMEOUT, seek_fut).await {
                        Ok(res) => res.map_err(|e| (e, current_slot.unwrap_or(slot_range.start)))?,
                        Err(_) => {
                            return Err((FirehoseError::OperationTimeout("seek_to_slot"), current_slot.unwrap_or(slot_range.start)));
                        }
                    }
                }

                // for each item in each block
                let mut item_index = 0;
                let mut displayed_skip_message = false;
                loop {
                    let read_fut = reader.read_until_block();
                    let nodes = match timeout(OP_TIMEOUT, read_fut).await {
                        Ok(result) => result
                            .map_err(FirehoseError::ReadUntilBlockError)
                            .map_err(|e| (e, current_slot.unwrap_or(slot_range.start)))?,
                        Err(_) => {
                            log::warn!(target: &log_target, "timeout reading next block, retrying (will restart)...");
                            let restart_slot =
                                current_slot.map(|s| s + 1).unwrap_or(slot_range.start);
                            return Err((
                                FirehoseError::OperationTimeout("read_until_block"),
                                restart_slot,
                            ));
                        }
                    };
                    if nodes.is_empty() {
                        log::info!(
                            target: &log_target,
                            "reached end of epoch {}",
                            epoch_num
                        );
                        break;
                    }
                    // ignore epoch and subset nodes at end of car file loop { if
                    // nodes.0.is_empty() { break; } if let Some(node) = nodes.0.last() { if
                    //     node.get_node().is_epoch() { log::debug!(target: &log_target,
                    //         "skipping epoch node for epoch {}", epoch_num); nodes.0.pop(); }
                    //     else if node.get_node().is_subset() { nodes.0.pop(); } else if
                    //     node.get_node().is_block() { break; } } } if nodes.0.is_empty() {
                    //         log::info!(target: &log_target, "reached end of epoch {}",
                    //             epoch_num); break; }
                    if let Some(last_node) = nodes.0.last()
                        && !last_node.get_node().is_block() {
                            log::info!(target: &log_target, "reached end of epoch {}", epoch_num);
                            break;
                        }
                    let block = nodes
                        .get_block()
                        .map_err(FirehoseError::GetBlockError)
                        .map_err(|e| (e, current_slot.unwrap_or(slot_range.start)))?;
                    log::debug!(
                        target: &log_target,
                        "read {} items from epoch {}, now at slot {}",
                        item_index,
                        epoch_num,
                        block.slot
                    );
                    let slot = block.slot;
                    if slot > local_end_inclusive {
                        log::debug!(
                            target: &log_target,
                            "reached end of local slice at slot {} (epoch {}), stopping",
                            slot,
                            epoch_num
                        );
                        break;
                    }
                    if slot >= slot_range.end {
                        log::info!(target: &log_target, "reached end of slot range at slot {}", slot);
                        // Return early to terminate the firehose thread cleanly. We use >=
                        // because slot_range is half-open [start, end), so any slot equal to
                        // end is out-of-range and must not be processed.
                        return Ok(());
                    }
                    debug_assert!(slot < slot_range.end, "processing out-of-range slot {} (end {})", slot, slot_range.end);
                    if slot < local_start {
                        if slot.saturating_add(1) == local_start {
                            log::debug!(
                                target: &log_target,
                                "priming reader with preceding slot {}, skipping",
                                slot
                            );
                        } else {
                            log::warn!(
                                target: &log_target,
                                "encountered slot {} before start of range {}, skipping",
                                slot,
                                local_start
                            );
                        }
                        continue;
                    }
                    current_slot = Some(slot);
                    let mut entry_index: usize = 0;
                    let mut this_block_executed_transaction_count: u64 = 0;
                    let mut this_block_entry_count: u64 = 0;
                    let mut this_block_rewards: Vec<(Address, RewardInfo)> = Vec::new();

                    if slot <= last_counted_slot {
                        log::debug!(
                            target: &log_target,
                            "duplicate block {}, already counted (last_counted={})",
                            slot,
                            last_counted_slot,
                        );
                        this_block_rewards.clear();
                        continue;
                    }

                    nodes.each(|node_with_cid| -> Result<(), SharedError> {
                        item_index += 1;
                        // if item_index == 100000 && !triggered { log::info!("simulating
                        //     error"); triggered = true; return
                        //     Err(Box::new(GeyserReplayError::NodeDecodingError(item_index,
                        //     Box::new(std::io::Error::new( std::io::ErrorKind::Other,
                        //         "simulated error", )), ))); }
                        if let Some(skip) = skip_until_index {
                            if item_index < skip {
                                if !displayed_skip_message {
                                    log::info!(
                                        target: &log_target,
                                        "skipping until index {} (at {})",
                                        skip,
                                        item_index
                                    );
                                    displayed_skip_message = true;
                                }
                                return Ok(());
                            } else {
                                log::info!(
                                    target: &log_target,
                                    "reached target index {}, resuming...",
                                    skip
                                );
                                skip_until_index = None;
                            }
                        }
                        let node = node_with_cid.get_node();

                        use crate::node::Node::*;
                        match node {
                            Transaction(tx) => {
                                let versioned_tx = tx.as_parsed()?;
                                let reassembled_metadata = nodes.reassemble_dataframes(tx.metadata.clone())?;

                                let as_native_metadata = decode_transaction_status_meta_from_frame(
                                    block.slot,
                                    reassembled_metadata,
                                )?;

                                let message_hash = {
                                    #[cfg(feature = "verify-transaction-signatures")]
                                    {
                                        versioned_tx.verify_and_hash_message()?
                                    }
                                    #[cfg(not(feature = "verify-transaction-signatures"))]
                                    {
                                        // Signature verification is optional because it is
                                        // extremely expensive at replay scale.
                                        versioned_tx.message.hash()
                                    }
                                };
                                let signature = versioned_tx
                                    .signatures
                                    .first()
                                    .ok_or_else(|| {
                                        Box::new(std::io::Error::new(
                                            std::io::ErrorKind::InvalidData,
                                            "transaction missing signature",
                                        )) as SharedError
                                    })?;
                                let is_vote = is_simple_vote_transaction(&versioned_tx);

                                if let Some(transaction_notifier) = transaction_notifier_maybe.as_ref() {
                                    transaction_notifier.notify_transaction(
                                        block.slot,
                                        tx.index.unwrap() as usize,
                                        signature,
                                        &message_hash,
                                        is_vote,
                                        &as_native_metadata,
                                        &versioned_tx,
                                    );
                                }

                            }
                            Entry(entry) => {
                                let entry_hash = Hash::from(entry.hash.to_bytes());
                                let entry_transaction_count = entry.transactions.len();
                                let entry_transaction_count_u64 = entry_transaction_count as u64;
                                let starting_transaction_index =
                                    usize::try_from(this_block_executed_transaction_count).map_err(|_| {
                                        Box::new(std::io::Error::other(
                                            "transaction index exceeds usize range",
                                        )) as SharedError
                                    })?;
                                todo_latest_entry_blockhash = entry_hash;
                                this_block_executed_transaction_count += entry_transaction_count_u64;
                                this_block_entry_count += 1;
                                if entry_notifier_maybe.is_none() {
                                    return Ok(());
                                }
                                let entry_notifier = entry_notifier_maybe.as_ref().unwrap();
                                let entry_summary = solana_entry::entry::EntrySummary {
                                    num_hashes: entry.num_hashes,
                                    hash: Hash::from(entry.hash.to_bytes()),
                                    num_transactions: entry_transaction_count_u64,
                                };
                                entry_notifier.notify_entry(
                                    block.slot,
                                    entry_index,
                                    &entry_summary,
                                    starting_transaction_index,
                                );
                                entry_index += 1;
                            }
                            Block(block) => {
                                let notification = SlotNotification::Root((block.slot, block.meta.parent_slot));
                                confirmed_bank_sender.send(notification).unwrap();

                                if block_meta_notifier_maybe.is_none() {
                                    last_counted_slot = block.slot;
                                    return Ok(());
                                }
                                let keyed_rewards = std::mem::take(&mut this_block_rewards);
                                let block_meta_notifier = block_meta_notifier_maybe.as_ref().unwrap();
                                block_meta_notifier.notify_block_metadata(
                                    block.meta.parent_slot,
                                    todo_previous_blockhash.to_string().as_str(),
                                    block.slot,
                                    todo_latest_entry_blockhash.to_string().as_str(),
                                    &KeyedRewardsAndNumPartitions {
                                        keyed_rewards,
                                        num_partitions: None,
                                    },
                                    Some(block.meta.blocktime as i64),
                                    block.meta.block_height,
                                    this_block_executed_transaction_count,
                                    this_block_entry_count,
                                );
                                todo_previous_blockhash = todo_latest_entry_blockhash;
                                last_counted_slot = block.slot;
                                std::thread::yield_now();
                            }
                            Subset(_subset) => (),
                            Epoch(_epoch) => (),
                            Rewards(rewards) => {
                                if !rewards.is_complete() {
                                    let reassembled = nodes.reassemble_dataframes(rewards.data.clone())?;
                                    let decompressed = utils::decompress_zstd(reassembled)?;
                                    let decoded = prost_011::Message::decode(decompressed.as_slice()).map_err(|err| {
                                        Box::new(std::io::Error::other(
                                            std::format!("Error decoding rewards: {:?}", err),
                                        ))
                                    })?;
                                    this_block_rewards = convert_proto_rewards(&decoded)?;
                                }
                            }
                            DataFrame(_data_frame) => (),
                        }
                        Ok(())
                    })
                .map_err(|e| FirehoseError::NodeDecodingError(item_index, e)).map_err(|e| (e, current_slot.unwrap_or(slot_range.start)))?;
                    if block.slot == slot_range.end - 1 {
                        let finish_time = std::time::Instant::now();
                        let elapsed = finish_time.duration_since(start_time);
                        log::info!(target: &log_target, "processed slot {}", block.slot);
                        let elapsed_pretty = human_readable_duration(elapsed);
                        log::info!(
                            target: &log_target,
                            "processed {} slots across {} epochs in {}.",
                            initial_slot_range.end - initial_slot_range.start,
                            slot_to_epoch(initial_slot_range.end)
                                + 1
                                - slot_to_epoch(initial_slot_range.start),
                            elapsed_pretty
                        );
                        log::info!(target: &log_target, "a 🚒 firehose thread finished completed its work.");
                        // On completion, report threads with non-zero error counts for
                        // visibility.
                        let summary: String = error_counts
                            .iter()
                            .enumerate()
                            .filter_map(|(i, c)| {
                                let v = c.load(Ordering::Relaxed);
                                if v > 0 { Some(format!("{:03}({})", i, v)) } else { None }
                            })
                            .collect::<Vec<_>>()
                            .join(", ");
                        if !summary.is_empty() {
                            log::debug!(target: &log_target, "threads with errors: {}", summary);
                        }
                        return Ok(());
                    }
                }
            }
            Ok(())
}
.await
{
        if is_shutdown_error(&err) {
            log::info!(
                target: &log_target,
                "shutdown requested; terminating firehose thread {:?}",
                thread_index
            );
            return Ok(());
        }
        log::error!(
            target: &log_target,
            "🧯💦🔥 firehose encountered an error at slot {} in epoch {} and will roll back one slot and retry:",
            slot,
            slot_to_epoch(slot)
            );
            log::error!(target: &log_target, "{}", err);
            if matches!(err, FirehoseError::SlotOffsetIndexError(_)) {
                // Clear cached index data for this epoch to avoid retrying with a bad/partial index.
                SLOT_OFFSET_INDEX.invalidate_epoch(slot_to_epoch(slot));
            }
            let item_index = match err {
                FirehoseError::NodeDecodingError(item_index, _) => item_index,
                _ => 0,
            };
            // Increment this thread's error counter
            let idx = thread_index.unwrap_or(0);
            error_counts[idx].fetch_add(1, Ordering::Relaxed);
            log::warn!(
                target: &log_target,
                "restarting from slot {} at index {}",
                slot,
                item_index,
            );
            // Update slot range to resume from the failed slot, not the original start.
            // If the failing slot was already fully processed, resume from the next slot.
            if slot <= last_counted_slot {
                slot_range.start = last_counted_slot.saturating_add(1);
            } else {
                slot_range.start = slot;
            }
            skip_until_index = Some(item_index);
}
    Ok(())
}

#[inline]
fn is_simple_vote_transaction(versioned_tx: &VersionedTransaction) -> bool {
    if !(1..=2).contains(&versioned_tx.signatures.len()) {
        return false;
    }

    if !matches!(
        versioned_tx.version(),
        solana_transaction::versioned::TransactionVersion::Legacy(_)
    ) {
        return false;
    }

    let instructions = versioned_tx.message.instructions();
    if instructions.len() != 1 {
        return false;
    }

    let program_index = instructions[0].program_id_index as usize;
    versioned_tx
        .message
        .static_account_keys()
        .get(program_index)
        .map(|program_id| program_id == &vote_program_id())
        .unwrap_or(false)
}

#[inline(always)]
fn convert_proto_rewards(
    proto_rewards: &solana_storage_proto::convert::generated::Rewards,
) -> Result<Vec<(Address, RewardInfo)>, SharedError> {
    let mut keyed_rewards = Vec::with_capacity(proto_rewards.rewards.len());
    for proto_reward in proto_rewards.rewards.iter() {
        let reward = RewardInfo {
            reward_type: match proto_reward.reward_type - 1 {
                0 => RewardType::Fee,
                1 => RewardType::Rent,
                2 => RewardType::Staking,
                3 => RewardType::Voting,
                typ => {
                    return Err(Box::new(std::io::Error::other(format!(
                        "unsupported reward type {}",
                        typ
                    ))));
                }
            },
            lamports: proto_reward.lamports,
            post_balance: proto_reward.post_balance,
            commission: proto_reward.commission.parse::<u8>().ok(),
        };
        let pubkey = proto_reward
            .pubkey
            .parse::<Address>()
            .map_err(|err| Box::new(err) as SharedError)?;
        keyed_rewards.push((pubkey, reward));
    }
    Ok(keyed_rewards)
}

#[inline]
/// Splits `slot_range` into nearly-even sub-ranges for the given thread count.
pub fn generate_subranges(slot_range: &Range<u64>, threads: u64) -> Vec<Range<u64>> {
    let total = slot_range.end - slot_range.start;
    let slots_per_thread = total / threads;
    let remainder = total % threads;

    let ranges: Vec<Range<u64>> = (0..threads)
        .map(|i| {
            // Distribute remainder slots to the first `remainder` threads
            let extra_slot = if i < remainder { 1 } else { 0 };
            let start = slot_range.start + i * slots_per_thread + i.min(remainder);
            let end = start + slots_per_thread + extra_slot;
            start..end
        })
        .collect();

    // Verify that ranges cover all slots exactly
    let total_covered: u64 = ranges.iter().map(|r| r.end - r.start).sum();
    assert_eq!(
        total_covered, total,
        "Range generation failed: {} threads should cover {} slots but only cover {}",
        threads, total, total_covered
    );

    // Verify no gaps between ranges
    for i in 1..ranges.len() {
        assert_eq!(
            ranges[i - 1].end,
            ranges[i].start,
            "Gap found between thread {} (ends at {}) and thread {} (starts at {})",
            i - 1,
            ranges[i - 1].end,
            i,
            ranges[i].start
        );
    }

    log::info!(
        target: LOG_MODULE,
        "Generated {} thread ranges covering {} slots total",
        threads,
        total_covered
    );
    ranges
}

fn human_readable_duration(duration: std::time::Duration) -> String {
    if duration.is_zero() {
        return "0s".into();
    }
    let total_secs = duration.as_secs();
    if total_secs < 60 {
        let secs_f = duration.as_secs_f64();
        if total_secs == 0 {
            format!("{:.2}s", secs_f)
        } else if duration.subsec_millis() == 0 {
            format!("{}s", total_secs)
        } else {
            format!("{:.2}s", secs_f)
        }
    } else {
        let mut secs = total_secs;
        let days = secs / 86_400;
        secs %= 86_400;
        let hours = secs / 3_600;
        secs %= 3_600;
        let minutes = secs / 60;
        secs %= 60;
        if days > 0 {
            if hours > 0 {
                format!("{days}d{hours}h")
            } else {
                format!("{days}d")
            }
        } else if hours > 0 {
            if minutes > 0 {
                format!("{hours}h{minutes}m")
            } else {
                format!("{hours}h")
            }
        } else if minutes > 0 {
            if secs > 0 {
                format!("{minutes}m{secs}s")
            } else {
                format!("{minutes}m")
            }
        } else {
            format!("{secs}s")
        }
    }
}

#[cfg(test)]
fn log_stats_handler(thread_id: usize, stats: Stats) -> HandlerFuture {
    Box::pin(async move {
        let elapsed = stats.start_time.elapsed();
        let elapsed_secs = elapsed.as_secs_f64();
        let tps = if elapsed_secs > 0.0 {
            stats.transactions_processed as f64 / elapsed_secs
        } else {
            0.0
        };
        log::info!(
            target: LOG_MODULE,
            "thread {thread_id} stats: current_slot={}, slots_processed={}, blocks_processed={}, txs={}, entries={}, rewards={}, elapsed_s={:.2}, tps={:.2}",
            stats.thread_stats.current_slot,
            stats.slots_processed,
            stats.blocks_processed,
            stats.transactions_processed,
            stats.entries_processed,
            stats.rewards_processed,
            elapsed_secs,
            tps
        );
        Ok(())
    })
}

#[cfg(test)]
use futures_util::FutureExt;
#[cfg(test)]
use serial_test::serial;
#[cfg(test)]
use std::sync::{Mutex, OnceLock};

#[tokio::test(flavor = "multi_thread")]
async fn test_firehose_epoch_800() {
    use dashmap::DashSet;
    use std::sync::atomic::{AtomicU64, Ordering};
    solana_logger::setup_with_default("info");
    const THREADS: usize = 4;
    const NUM_SLOTS_TO_COVER: u64 = 50;
    static PREV_BLOCK: [AtomicU64; THREADS] = [const { AtomicU64::new(0) }; THREADS];
    static NUM_SKIPPED_BLOCKS: AtomicU64 = AtomicU64::new(0);
    static NUM_BLOCKS: AtomicU64 = AtomicU64::new(0);
    static SEEN_SKIPPED: OnceLock<DashSet<u64>> = OnceLock::new();
    static SEEN_SLOTS: OnceLock<DashSet<u64>> = OnceLock::new();
    static MIN_TRANSACTIONS: AtomicU64 = AtomicU64::new(u64::MAX);
    let stats_tracking = StatsTracking {
        on_stats: log_stats_handler,
        tracking_interval_slots: 10,
    };

    for prev in PREV_BLOCK.iter() {
        prev.store(0, Ordering::Relaxed);
    }
    NUM_SKIPPED_BLOCKS.store(0, Ordering::Relaxed);
    NUM_BLOCKS.store(0, Ordering::Relaxed);
    MIN_TRANSACTIONS.store(u64::MAX, Ordering::Relaxed);
    SEEN_SLOTS.get_or_init(DashSet::new).clear();
    SEEN_SKIPPED.get_or_init(DashSet::new).clear();

    firehose(
        THREADS.try_into().unwrap(),
        (345600000 - NUM_SLOTS_TO_COVER / 2)..(345600000 + NUM_SLOTS_TO_COVER / 2),
        Some(|thread_id: usize, block: BlockData| {
            async move {
                let _prev =
                    PREV_BLOCK[thread_id % PREV_BLOCK.len()].swap(block.slot(), Ordering::Relaxed);
                if block.was_skipped() {
                    log::info!(
                        target: LOG_MODULE,
                        "leader skipped block {} on thread {}",
                        block.slot(),
                        thread_id,
                    );
                } else {
                    /*log::info!(
                        target: LOG_MODULE,
                        "got block {} on thread {}",
                        block.slot(),
                        thread_id,
                    );*/
                }

                let first_time = SEEN_SLOTS.get_or_init(DashSet::new).insert(block.slot());
                if block.was_skipped() {
                    NUM_SKIPPED_BLOCKS.fetch_add(1, Ordering::Relaxed);
                    SEEN_SKIPPED.get_or_init(DashSet::new).insert(block.slot());
                } else {
                    if first_time {
                        NUM_BLOCKS.fetch_add(1, Ordering::Relaxed);
                        if let BlockData::Block {
                            executed_transaction_count,
                            ..
                        } = &block
                        {
                            let executed = *executed_transaction_count;
                            let _ = MIN_TRANSACTIONS.fetch_update(
                                Ordering::Relaxed,
                                Ordering::Relaxed,
                                |current| {
                                    if executed < current {
                                        Some(executed)
                                    } else {
                                        None
                                    }
                                },
                            );
                        }
                    }
                }
                Ok(())
            }
            .boxed()
        }),
        None::<OnTxFn>,
        None::<OnEntryFn>,
        None::<OnRewardFn>,
        None::<OnErrorFn>,
        Some(stats_tracking),
        None,
    )
    .await
    .unwrap();
    let seen = SEEN_SLOTS.get_or_init(DashSet::new).len() as u64;
    assert_eq!(
        seen, NUM_SLOTS_TO_COVER,
        "expected to see exactly {NUM_SLOTS_TO_COVER} unique slots, saw {seen}"
    );
    let mut skipped: Vec<u64> = SEEN_SKIPPED
        .get_or_init(DashSet::new)
        .iter()
        .map(|v| *v)
        .collect();
    skipped.sort_unstable();
    // 345600000 is present but empty; still emitted as a block. Skip set should not include it.
    const EXPECTED_SKIPPED: [u64; 6] = [
        345_600_004,
        345_600_005,
        345_600_008,
        345_600_009,
        345_600_010,
        345_600_011,
    ];
    assert_eq!(skipped, EXPECTED_SKIPPED, "unexpected skipped slots");
    assert!(NUM_BLOCKS.load(Ordering::Relaxed) > 0);
}

#[tokio::test(flavor = "multi_thread")]
async fn test_firehose_target_slot_transactions() {
    use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
    solana_logger::setup_with_default("info");
    const TARGET_SLOT: u64 = 376_273_722;
    const SLOT_RADIUS: u64 = 50;
    const EXPECTED_TRANSACTIONS: u64 = 1414;
    const EXPECTED_NON_VOTE_TRANSACTIONS: u64 = 511;
    static FOUND: AtomicBool = AtomicBool::new(false);
    static OBSERVED_TXS: AtomicU64 = AtomicU64::new(0);
    static OBSERVED_NON_VOTE: AtomicU64 = AtomicU64::new(0);

    FOUND.store(false, Ordering::Relaxed);
    OBSERVED_TXS.store(0, Ordering::Relaxed);
    OBSERVED_NON_VOTE.store(0, Ordering::Relaxed);

    firehose(
        4,
        (TARGET_SLOT - SLOT_RADIUS)..(TARGET_SLOT + SLOT_RADIUS),
        Some(|_thread_id: usize, block: BlockData| {
            async move {
                if block.slot() == TARGET_SLOT {
                    assert!(
                        !block.was_skipped(),
                        "target slot {TARGET_SLOT} was marked leader skipped",
                    );
                    if let BlockData::Block {
                        executed_transaction_count,
                        ..
                    } = block
                    {
                        OBSERVED_TXS.store(executed_transaction_count, Ordering::Relaxed);
                        FOUND.store(true, Ordering::Relaxed);
                        assert_eq!(
                            executed_transaction_count, EXPECTED_TRANSACTIONS,
                            "unexpected transaction count for slot {TARGET_SLOT}"
                        );
                        assert_eq!(
                            OBSERVED_NON_VOTE.load(Ordering::Relaxed),
                            EXPECTED_NON_VOTE_TRANSACTIONS,
                            "unexpected non-vote transaction count for slot {TARGET_SLOT}"
                        );
                    }
                }
                Ok(())
            }
            .boxed()
        }),
        Some(|_thread_id: usize, transaction: TransactionData| {
            async move {
                if transaction.slot == TARGET_SLOT && !transaction.is_vote {
                    OBSERVED_NON_VOTE.fetch_add(1, Ordering::Relaxed);
                }
                Ok(())
            }
            .boxed()
        }),
        None::<OnEntryFn>,
        None::<OnRewardFn>,
        None::<OnErrorFn>,
        None::<OnStatsTrackingFn>,
        None,
    )
    .await
    .unwrap();

    assert!(
        FOUND.load(Ordering::Relaxed),
        "target slot was not processed"
    );
    assert_eq!(
        OBSERVED_TXS.load(Ordering::Relaxed),
        EXPECTED_TRANSACTIONS,
        "recorded transaction count mismatch"
    );
}

#[tokio::test(flavor = "multi_thread")]
async fn test_firehose_epoch_850_votes_present() {
    use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
    solana_logger::setup_with_default("info");
    const TARGET_SLOT: u64 = 367_200_100; // epoch 850
    const SLOT_RADIUS: u64 = 10;
    static SEEN_BLOCK: AtomicBool = AtomicBool::new(false);
    static VOTE_TXS: AtomicU64 = AtomicU64::new(0);
    static TOTAL_TXS: AtomicU64 = AtomicU64::new(0);

    SEEN_BLOCK.store(false, Ordering::Relaxed);
    VOTE_TXS.store(0, Ordering::Relaxed);
    TOTAL_TXS.store(0, Ordering::Relaxed);

    firehose(
        2,
        (TARGET_SLOT - SLOT_RADIUS)..(TARGET_SLOT + SLOT_RADIUS),
        Some(|_thread_id: usize, block: BlockData| {
            async move {
                if block.slot() == TARGET_SLOT {
                    assert!(
                        !block.was_skipped(),
                        "target slot {TARGET_SLOT} was marked leader skipped",
                    );
                    SEEN_BLOCK.store(true, Ordering::Relaxed);
                }
                Ok(())
            }
            .boxed()
        }),
        Some(|_thread_id: usize, transaction: TransactionData| {
            async move {
                if transaction.slot == TARGET_SLOT {
                    TOTAL_TXS.fetch_add(1, Ordering::Relaxed);
                    if transaction.is_vote {
                        VOTE_TXS.fetch_add(1, Ordering::Relaxed);
                    }
                }
                Ok(())
            }
            .boxed()
        }),
        None::<OnEntryFn>,
        None::<OnRewardFn>,
        None::<OnErrorFn>,
        None::<OnStatsTrackingFn>,
        None,
    )
    .await
    .unwrap();

    assert!(
        SEEN_BLOCK.load(Ordering::Relaxed),
        "target slot was not processed"
    );
    assert!(
        TOTAL_TXS.load(Ordering::Relaxed) > 0,
        "no transactions counted in target slot"
    );
    assert_eq!(VOTE_TXS.load(Ordering::Relaxed), 991);
}

#[cfg(test)]
#[serial]
#[tokio::test(flavor = "multi_thread")]
async fn test_firehose_restart_loses_coverage_without_reset() {
    use std::collections::HashMap;
    solana_logger::setup_with_default("info");
    const THREADS: usize = 1;
    const START_SLOT: u64 = 345_600_000;
    const NUM_SLOTS: u64 = 8;

    static COVERAGE: OnceLock<Mutex<HashMap<u64, u32>>> = OnceLock::new();
    COVERAGE
        .get_or_init(|| Mutex::new(HashMap::new()))
        .lock()
        .unwrap()
        .clear();
    static FAIL_TRIGGERED: AtomicBool = AtomicBool::new(false);
    static SEEN_BLOCKS: AtomicU64 = AtomicU64::new(0);
    FAIL_TRIGGERED.store(false, Ordering::Relaxed);
    SEEN_BLOCKS.store(0, Ordering::Relaxed);

    firehose(
        THREADS.try_into().unwrap(),
        START_SLOT..(START_SLOT + NUM_SLOTS),
        Some(|_thread_id: usize, block: BlockData| {
            async move {
                // Force an error after at least one block has been seen so restart happens mid-range.
                if !block.was_skipped()
                    && SEEN_BLOCKS.load(Ordering::Relaxed) > 0
                    && !FAIL_TRIGGERED.swap(true, Ordering::SeqCst)
                {
                    return Err("synthetic handler failure to exercise restart".into());
                }
                let mut coverage = COVERAGE
                    .get_or_init(|| Mutex::new(HashMap::new()))
                    .lock()
                    .unwrap();
                *coverage.entry(block.slot()).or_insert(0) += 1;
                if !block.was_skipped() {
                    SEEN_BLOCKS.fetch_add(1, Ordering::Relaxed);
                }
                Ok(())
            }
            .boxed()
        }),
        None::<OnTxFn>,
        None::<OnEntryFn>,
        None::<OnRewardFn>,
        None::<OnErrorFn>,
        None::<OnStatsTrackingFn>,
        None,
    )
    .await
    .unwrap();

    let coverage = COVERAGE.get().unwrap().lock().unwrap();
    for slot in START_SLOT..(START_SLOT + NUM_SLOTS) {
        assert!(
            coverage.contains_key(&slot),
            "missing coverage for slot {slot} after restart"
        );
    }
}

#[cfg(test)]
#[serial]
#[tokio::test(flavor = "multi_thread")]
async fn test_firehose_gap_coverage_near_known_missing_range() {
    use std::collections::HashSet;
    solana_logger::setup_with_default("info");
    const GAP_START: u64 = 378864000;
    const START_SLOT: u64 = GAP_START - 1000;
    const END_SLOT: u64 = GAP_START + 1000;
    const THREADS: usize = 16;

    static COVERAGE: OnceLock<Mutex<HashSet<u64>>> = OnceLock::new();
    COVERAGE
        .get_or_init(|| Mutex::new(HashSet::new()))
        .lock()
        .unwrap()
        .clear();

    firehose(
        THREADS.try_into().unwrap(),
        START_SLOT..(END_SLOT + 1),
        Some(|_thread_id: usize, block: BlockData| {
            async move {
                if block.was_skipped() {
                    return Ok(());
                }
                let slot = block.slot();
                COVERAGE
                    .get_or_init(|| Mutex::new(HashSet::new()))
                    .lock()
                    .unwrap()
                    .insert(slot);
                Ok(())
            }
            .boxed()
        }),
        None::<OnTxFn>,
        None::<OnEntryFn>,
        None::<OnRewardFn>,
        None::<OnErrorFn>,
        None::<OnStatsTrackingFn>,
        None,
    )
    .await
    .unwrap();

    let mut coverage = COVERAGE
        .get_or_init(|| Mutex::new(HashSet::new()))
        .lock()
        .unwrap()
        .clone();

    // ignore a known 4-slot leader skipped gap
    coverage.insert(378864396);
    coverage.insert(378864397);
    coverage.insert(378864398);
    coverage.insert(378864399);

    let expected: Vec<u64> = (START_SLOT..=END_SLOT).collect();
    let missing: Vec<u64> = expected
        .iter()
        .copied()
        .filter(|slot| !coverage.contains(slot))
        .collect();
    assert!(
        missing.is_empty(),
        "missing slots in {START_SLOT}..={END_SLOT}; count={}, first few={:?}",
        missing.len(),
        &missing[..missing.len().min(10)]
    );
}