solana-runtime 4.2.0-beta.0

use {
    crate::{
        bank::Bank,
        block_component_processor::vote_reward::{
            CalcVoteRewardUpdateVoteStatesError, calc_vote_rewards_update_vote_states,
        },
        leader_schedule_utils::leader_slot_index,
        validated_block_finalization::{
            BlockFinalizationCertError, ValidatedBlockFinalizationCert,
        },
        validated_reward_certificate::{Error as ValidatedRewardCertError, ValidatedRewardCert},
    },
    agave_votor_messages::{
        certificate::{Certificate, CertificateType},
        consensus_message::{Block, ConsensusMessage},
        migration::MigrationStatus,
        unverified_vote_message::UnverifiedCertificate,
    },
    crossbeam_channel::Sender,
    log::*,
    solana_clock::Slot,
    solana_entry::block_component::{
        BlockFooterV1, BlockMarkerV1, GenesisCertBlockMarker, VersionedBlockFooter,
        VersionedBlockHeader, VersionedBlockMarker, VersionedUpdateParent,
    },
    solana_hash::Hash,
    solana_pubkey::Pubkey,
    std::{collections::HashSet, sync::Arc},
    thiserror::Error,
};

pub(crate) mod vote_reward;

#[derive(Debug, Error)]
pub enum BankFooterError {
    #[error("calc vote rewards updating vote states failed with \"{0}\"")]
    CalcVoteRewardUpdateVoteStates(#[from] CalcVoteRewardUpdateVoteStatesError),
}

#[derive(Debug, Error)]
pub enum BlockComponentProcessorError {
    #[error("BlockComponent detected pre-migration")]
    BlockComponentPreMigration,
    #[error("GenesisCertificate marker detected when GenesisCertificate is already populated")]
    GenesisCertificateAlreadyPopulated,
    #[error("GenesisCertificate marker detected when the cluster has Alpenglow enabled at slot 0")]
    GenesisCertificateInAlpenglowCluster,
    #[error("GenesisCertificate marker detected on a block which is not a child of genesis")]
    GenesisCertificateOnNonChild,
    #[error("GenesisCertificate was invalid and failed to verify")]
    GenesisCertificateFailedVerification,
    #[error("GenesisCertificate marker must immediately follow the block header")]
    GenesisCertificateOutOfOrder,
    #[error("FinalizationCertificate was invalid or failed to verify {0}")]
    InvalidFinalizationCertificate(#[from] BlockFinalizationCertError),
    #[error("Missing block footer")]
    MissingBlockFooter,
    #[error("Missing parent marker (neither a header nor an update parent was present)")]
    MissingParentMarker,
    #[error("Multiple block footers detected")]
    MultipleBlockFooters,
    #[error("Multiple block headers detected")]
    MultipleBlockHeaders,
    #[error(
        "Block header parent slot mismatch: header={header_parent_slot}, bank={bank_parent_slot}"
    )]
    HeaderParentSlotMismatch {
        header_parent_slot: Slot,
        bank_parent_slot: Slot,
    },
    #[error("Multiple update parents detected")]
    MultipleUpdateParents,
    #[error("Nanosecond clock out of bounds")]
    NanosecondClockOutOfBounds,
    #[error("Spurious update parent")]
    SpuriousUpdateParent,
    #[error("UpdateParent marker is only valid in the first slot of a leader window: slot {0}")]
    UpdateParentNotFirstInLeaderWindow(Slot),
    #[error(
        "UpdateParent cannot be the initial parent marker unless replay starts at UpdateParent"
    )]
    UnexpectedInitialUpdateParent,
    #[error("Abandoned bank")]
    AbandonedBank(VersionedUpdateParent),
    #[error("invalid reward certs {0}")]
    InvalidRewardCerts(#[from] ValidatedRewardCertError),
    #[error("updating bank footer failed with \"{0}\"")]
    UpdateBankFooter(#[from] BankFooterError),
}

impl BlockComponentProcessorError {
    pub fn is_update_parent_recoverable_replay_error(&self) -> bool {
        match self {
            BlockComponentProcessorError::MissingParentMarker
            | BlockComponentProcessorError::MultipleBlockFooters
            | BlockComponentProcessorError::MultipleBlockHeaders
            | BlockComponentProcessorError::HeaderParentSlotMismatch { .. }
            | BlockComponentProcessorError::NanosecondClockOutOfBounds
            | BlockComponentProcessorError::UnexpectedInitialUpdateParent
            | BlockComponentProcessorError::GenesisCertificateOutOfOrder
            | BlockComponentProcessorError::AbandonedBank(_)
            | BlockComponentProcessorError::InvalidRewardCerts(_)
            | BlockComponentProcessorError::UpdateBankFooter(_)
            | BlockComponentProcessorError::InvalidFinalizationCertificate(_) => true,
            BlockComponentProcessorError::BlockComponentPreMigration
            | BlockComponentProcessorError::GenesisCertificateAlreadyPopulated
            | BlockComponentProcessorError::GenesisCertificateInAlpenglowCluster
            | BlockComponentProcessorError::GenesisCertificateOnNonChild
            | BlockComponentProcessorError::GenesisCertificateFailedVerification
            | BlockComponentProcessorError::MissingBlockFooter
            | BlockComponentProcessorError::MultipleUpdateParents
            | BlockComponentProcessorError::SpuriousUpdateParent
            | BlockComponentProcessorError::UpdateParentNotFirstInLeaderWindow(_) => false,
        }
    }
}

#[derive(Default)]
pub struct BlockComponentProcessor {
    has_header: bool,
    has_footer: bool,
    has_entry_batch: bool,
    update_parent: Option<VersionedUpdateParent>,
}

impl BlockComponentProcessor {
    pub fn on_final(
        &self,
        migration_status: &MigrationStatus,
        slot: Slot,
    ) -> Result<(), BlockComponentProcessorError> {
        // Only require block markers (header/footer) for slots where they should be present
        if !migration_status.should_allow_block_markers(slot) {
            return Ok(());
        }

        // Post-migration: both header and footer are required
        if !self.has_footer {
            return Err(BlockComponentProcessorError::MissingBlockFooter);
        }

        if !self.has_header && self.update_parent.is_none() {
            return Err(BlockComponentProcessorError::MissingParentMarker);
        }

        Ok(())
    }

    /// Process an entry batch.
    ///
    /// Validates that a parent marker (header or update parent) has been
    /// processed before any entry batches.
    pub fn on_entry_batch(
        &mut self,
        migration_status: &MigrationStatus,
        slot: Slot,
    ) -> Result<(), BlockComponentProcessorError> {
        if !migration_status.should_allow_block_markers(slot) {
            self.has_entry_batch = true;
            return Ok(());
        }

        // We must have either a header or an update parent prior to processing entry batches.
        if !self.has_header && self.update_parent.is_none() {
            return Err(BlockComponentProcessorError::MissingParentMarker);
        }

        self.has_entry_batch = true;
        Ok(())
    }

    /// Process a block marker:
    /// - Pre migration, no block markers are allowed
    /// - During the migration only header and genesis certificate are allowed:
    ///     - This is in case our node was slow in observing the completion of the migration
    ///     - By seeing the first alpenglow block, we can advance the migration phase
    /// - Once the migration is complete all markers are allowed
    pub fn on_marker(
        &mut self,
        bank: Arc<Bank>,
        parent_bank: Arc<Bank>,
        shred_version: u16,
        marker: VersionedBlockMarker,
        allow_initial_update_parent: bool,
        finalization_cert_sender: Option<&Sender<ConsensusMessage>>,
        migration_status: &MigrationStatus,
    ) -> Result<(), BlockComponentProcessorError> {
        let slot = bank.slot();
        let VersionedBlockMarker::V1(marker) = marker;

        let markers_fully_enabled = migration_status.should_allow_block_markers(slot);
        let in_migration = migration_status.is_in_migration();
        let fast_leader_handover_active =
            bank.feature_set.snapshot().alpenglow_fast_leader_handover;

        match marker {
            // Header and genesis cert can be processed either:
            // - once migration is fully enabled, or
            // - while we're still in the migration phase (to let us advance it)
            BlockMarkerV1::BlockHeader(header) if markers_fully_enabled || in_migration => {
                self.on_header(header.inner(), bank.parent_slot())
            }
            BlockMarkerV1::GenesisCertificate(genesis_cert_block_marker)
                if markers_fully_enabled || in_migration =>
            {
                self.on_genesis_cert_block_marker(
                    bank,
                    shred_version,
                    genesis_cert_block_marker.into_inner(),
                    migration_status,
                )
            }

            // Everything else is only valid once migration is complete
            BlockMarkerV1::BlockFooter(footer) if markers_fully_enabled => self.on_footer(
                bank,
                parent_bank,
                shred_version,
                footer.into_inner(),
                finalization_cert_sender,
            ),

            BlockMarkerV1::UpdateParent(update_parent) if markers_fully_enabled => {
                if fast_leader_handover_active {
                    self.on_update_parent(slot, update_parent.inner(), allow_initial_update_parent)
                } else {
                    Err(BlockComponentProcessorError::SpuriousUpdateParent)
                }
            }

            // Any other combination means we saw a marker too early
            _ => Err(BlockComponentProcessorError::BlockComponentPreMigration),
        }
    }

    /// Processes the genesis block marker with full verification
    pub fn on_genesis_cert_block_marker(
        &self,
        bank: Arc<Bank>,
        shred_version: u16,
        genesis_block_marker: GenesisCertBlockMarker,
        migration_status: &MigrationStatus,
    ) -> Result<(), BlockComponentProcessorError> {
        self.validate_genesis_cert_block_marker_position()?;
        self.process_genesis_cert_block_marker(
            bank,
            genesis_block_marker,
            migration_status,
            Some(shred_version),
        )
    }

    /// Processes a locally produced genesis certificate marker without
    /// re-verifying the certificate signature.
    pub fn on_genesis_cert_block_marker_leader(
        &self,
        bank: Arc<Bank>,
        genesis_block_marker: GenesisCertBlockMarker,
        migration_status: &MigrationStatus,
    ) -> Result<(), BlockComponentProcessorError> {
        self.process_genesis_cert_block_marker(bank, genesis_block_marker, migration_status, None)
    }

    fn validate_genesis_cert_block_marker_position(
        &self,
    ) -> Result<(), BlockComponentProcessorError> {
        if !self.has_header {
            return Err(BlockComponentProcessorError::MissingParentMarker);
        }

        if self.has_entry_batch || self.has_footer || self.update_parent.is_some() {
            return Err(BlockComponentProcessorError::GenesisCertificateOutOfOrder);
        }

        Ok(())
    }

    /// Performs verification if `shred_version` is specified
    fn process_genesis_cert_block_marker(
        &self,
        bank: Arc<Bank>,
        genesis_block_marker: GenesisCertBlockMarker,
        migration_status: &MigrationStatus,
        shred_version: Option<u16>,
    ) -> Result<(), BlockComponentProcessorError> {
        // Genesis Certificate is only allowed for direct child of genesis
        if bank.parent_slot() == 0 {
            return Err(BlockComponentProcessorError::GenesisCertificateInAlpenglowCluster);
        }

        let parent_block_id = bank
            .parent_block_id()
            .expect("Block id is populated for all slots > 0");
        if (bank.parent_slot(), parent_block_id)
            != (genesis_block_marker.slot, genesis_block_marker.block_id)
        {
            return Err(BlockComponentProcessorError::GenesisCertificateOnNonChild);
        }

        if bank.get_alpenglow_genesis_certificate().is_some() {
            return Err(BlockComponentProcessorError::GenesisCertificateAlreadyPopulated);
        }

        let genesis_cert_type = CertificateType::Genesis(Block {
            slot: genesis_block_marker.slot,
            block_id: genesis_block_marker.block_id,
        });
        let genesis_cert = match shred_version {
            Some(shred_version) => {
                let unverified_genesis_cert = UnverifiedCertificate {
                    cert_type: genesis_cert_type,
                    signature: genesis_block_marker.bls_signature,
                    bitmap: genesis_block_marker.bitmap,
                    shred_version,
                };
                Self::verify_genesis_certificate(&bank, unverified_genesis_cert)?
            }
            None => Certificate {
                cert_type: genesis_cert_type,
                signature: genesis_block_marker.bls_signature,
                bitmap: genesis_block_marker.bitmap,
            },
        };
        bank.set_alpenglow_genesis_certificate(&genesis_cert);
        bank.set_hashes_per_tick(None);

        if migration_status.is_alpenglow_enabled() {
            // We participated in the migration, nothing to do
            return Ok(());
        }

        // We missed the migration however we ingested the first alpenglow block.
        // This is either a result of startup replay, or in some weird cases steady state replay after a network partition.
        // Either way we ingest the genesis block details moving us to `ReadyToEnable`.
        // Since this is a direct child of genesis, and we are replaying, we know we have frozen the genesis block.
        // Then `load_frozen_forks` or `replay_stage` will take care of the rest.
        warn!(
            "{}: Alpenglow genesis marker processed during replay of {}. Transitioning Alpenglow \
             to ReadyToEnable",
            migration_status.my_pubkey(),
            bank.slot()
        );
        migration_status.set_genesis_block(
            genesis_cert
                .cert_type
                .to_block()
                .expect("Genesis cert must correspond to a block"),
        );
        migration_status.set_genesis_certificate(Arc::new(genesis_cert));
        assert!(migration_status.is_ready_to_enable());

        Ok(())
    }

    fn verify_genesis_certificate(
        bank: &Bank,
        cert: UnverifiedCertificate,
    ) -> Result<Certificate, BlockComponentProcessorError> {
        debug_assert!(cert.cert_type.is_genesis());

        let cert_slot = cert.cert_type.slot();
        let cert = bank.verify_certificate(cert).map_err(|_| {
            warn!(
                "Failed to verify genesis certificate for slot {cert_slot} in bank slot {}",
                bank.slot()
            );
            BlockComponentProcessorError::GenesisCertificateFailedVerification
        })?;

        Ok(cert)
    }

    fn on_footer(
        &mut self,
        bank: Arc<Bank>,
        parent_bank: Arc<Bank>,
        shred_version: u16,
        footer: VersionedBlockFooter,
        finalization_cert_sender: Option<&Sender<ConsensusMessage>>,
    ) -> Result<(), BlockComponentProcessorError> {
        if !self.has_header && self.update_parent.is_none() {
            return Err(BlockComponentProcessorError::MissingParentMarker);
        }

        if self.has_footer {
            return Err(BlockComponentProcessorError::MultipleBlockFooters);
        }

        let VersionedBlockFooter::V1(footer) = footer;

        Self::enforce_nanosecond_clock_bounds(&bank, &parent_bank, &footer)?;

        let BlockFooterV1 {
            bank_hash,
            block_producer_time_nanos,
            block_user_agent: _,
            block_final_cert,
            skip_reward_cert,
            notar_reward_cert,
        } = footer;

        let reward_cert = ValidatedRewardCert::try_new(
            &bank,
            shred_version,
            &skip_reward_cert,
            &notar_reward_cert,
        )?;
        let block_producer_time_nanos =
            Self::block_producer_time_nanos_as_i64(block_producer_time_nanos)?;
        let final_cert = block_final_cert
            .map(|final_cert| {
                ValidatedBlockFinalizationCert::try_from_footer(final_cert, &bank, shred_version)
                    .map_err(BlockComponentProcessorError::InvalidFinalizationCertificate)
            })
            .transpose()?;

        let (footer_input, pool_input) = match final_cert {
            None => (None, None),
            Some(cert) => {
                let (signers, finalize_cert, notarize_cert) = cert.into_parts();
                let final_slot = finalize_cert.cert_type.slot();
                (
                    Some((signers, final_slot)),
                    Some((finalize_cert, notarize_cert)),
                )
            }
        };

        Self::update_bank_with_footer_fields(
            &bank,
            block_producer_time_nanos,
            Some(bank_hash),
            reward_cert,
            footer_input
                .as_ref()
                .map(|(validators, slot)| (validators, *slot)),
        )?;

        // Send finalization cert(s) to consensus pool
        if let Some((finalize_cert, notarize_cert)) = pool_input
            && let Some(sender) = finalization_cert_sender
        {
            if let Some(notarize_cert) = notarize_cert {
                let cert = ConsensusMessage::Certificate(notarize_cert);
                // TODO blocking send.
                let _ = sender
                    .send(cert)
                    .inspect_err(|_| info!("ConsensusMessage sender disconnected"));
            }
            let cert = ConsensusMessage::Certificate(finalize_cert);
            // TODO blocking send.
            let _ = sender
                .send(cert)
                .inspect_err(|_| info!("ConsensusMessage sender disconnected"));
        }

        self.has_footer = true;
        Ok(())
    }

    fn on_header(
        &mut self,
        header: &VersionedBlockHeader,
        bank_parent_slot: Slot,
    ) -> Result<(), BlockComponentProcessorError> {
        if self.has_header {
            return Err(BlockComponentProcessorError::MultipleBlockHeaders);
        }

        if self.update_parent.is_some() {
            return Err(BlockComponentProcessorError::SpuriousUpdateParent);
        }

        let VersionedBlockHeader::V1(header) = header;
        if header.parent_slot != bank_parent_slot {
            return Err(BlockComponentProcessorError::HeaderParentSlotMismatch {
                header_parent_slot: header.parent_slot,
                bank_parent_slot,
            });
        }

        self.has_header = true;
        Ok(())
    }

    fn on_update_parent(
        &mut self,
        slot: Slot,
        update_parent: &VersionedUpdateParent,
        allow_initial_update_parent: bool,
    ) -> Result<(), BlockComponentProcessorError> {
        if self.update_parent.is_some() {
            return Err(BlockComponentProcessorError::MultipleUpdateParents);
        }

        if leader_slot_index(slot) != 0 {
            return Err(BlockComponentProcessorError::UpdateParentNotFirstInLeaderWindow(slot));
        }

        if !self.has_header && !allow_initial_update_parent {
            return Err(BlockComponentProcessorError::UnexpectedInitialUpdateParent);
        }

        self.update_parent = Some(update_parent.clone());

        if self.has_header {
            // Only an error in the sense that replay execution of this block
            // prefix is now over. Replay execution can continue after resetting
            // bank.
            Err(BlockComponentProcessorError::AbandonedBank(
                update_parent.clone(),
            ))
        } else {
            Ok(())
        }
    }

    fn enforce_nanosecond_clock_bounds(
        bank: &Bank,
        parent_bank: &Bank,
        footer: &BlockFooterV1,
    ) -> Result<(), BlockComponentProcessorError> {
        // Get parent time from the nanosecond clock account, or from the Tower-based
        // clock for the first Alpenglow block.
        let parent_time_nanos = parent_bank
            .get_nanosecond_clock()
            .unwrap_or_else(|| bank.clock().unix_timestamp.saturating_mul(1_000_000_000));

        let parent_slot = parent_bank.slot();
        let current_time_nanos =
            Self::block_producer_time_nanos_as_i64(footer.block_producer_time_nanos)?;
        let current_slot = bank.slot();
        let elapsed_slot_duration_nanos =
            bank.slot_range_duration_nanos(parent_slot.saturating_add(1), current_slot);

        let (lower_bound_nanos, upper_bound_nanos) =
            Self::nanosecond_time_bounds(parent_time_nanos, elapsed_slot_duration_nanos);

        let is_valid =
            lower_bound_nanos <= current_time_nanos && current_time_nanos <= upper_bound_nanos;

        match is_valid {
            true => Ok(()),
            false => Err(BlockComponentProcessorError::NanosecondClockOutOfBounds),
        }
    }

    /// Converts a footer timestamp to the signed nanosecond representation used
    /// by bank clock state.
    ///
    /// The `block_producer_time_nanos` parameter comes from wire-format footer
    /// data and is rejected if it cannot be represented as `i64`; wrapping it
    /// would make an extreme future timestamp look negative.
    fn block_producer_time_nanos_as_i64(
        block_producer_time_nanos: u64,
    ) -> Result<i64, BlockComponentProcessorError> {
        i64::try_from(block_producer_time_nanos)
            .map_err(|_| BlockComponentProcessorError::NanosecondClockOutOfBounds)
    }

    /// Given a parent time and elapsed slot duration, calculates inclusive
    /// block producer timestamp bounds.
    ///
    /// `parent_time_nanos` describes the parent bank's nanosecond clock.
    /// `elapsed_slot_duration_nanos` is the summed duration for all skipped
    /// and working slots after the parent. The returned `(lower_bound,
    /// upper_bound)` accepts timestamps where
    /// `lower_bound <= working_bank_time <= upper_bound`.
    ///
    /// Refer to
    /// https://github.com/solana-foundation/solana-improvement-documents/pull/363
    /// for details on the bounds calculation.
    pub fn nanosecond_time_bounds(
        parent_time_nanos: i64,
        elapsed_slot_duration_nanos: u128,
    ) -> (i64, i64) {
        let min_working_bank_time = parent_time_nanos.saturating_add(1);
        let max_working_bank_time_offset = elapsed_slot_duration_nanos
            .saturating_mul(2)
            .min(i64::MAX as u128) as i64;
        let max_working_bank_time = parent_time_nanos.saturating_add(max_working_bank_time_offset);

        (min_working_bank_time, max_working_bank_time)
    }

    pub fn update_bank_with_footer_fields(
        bank: &Bank,
        block_producer_time_nanos: i64,
        bank_hash: Option<Hash>,
        reward_cert: Option<ValidatedRewardCert>,
        final_cert_input: Option<(&HashSet<Pubkey>, Slot)>,
    ) -> Result<(), BankFooterError> {
        bank.update_clock_from_footer(block_producer_time_nanos);
        calc_vote_rewards_update_vote_states(
            bank,
            reward_cert,
            final_cert_input,
            block_producer_time_nanos,
        )?;

        if let Some(hash) = bank_hash {
            // Record expected bank hash from footer for later verification when the bank is frozen.
            bank.set_expected_bank_hash(hash);
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use {
        super::*,
        crate::{
            bank::{Bank, SlotLeader},
            bank_forks::BankForks,
            genesis_utils::{activate_all_features_alpenglow, create_genesis_config},
        },
        rand::Rng,
        solana_bls_signatures::BLS_SIGNATURE_AFFINE_SIZE,
        solana_clock::DEFAULT_MS_PER_SLOT,
        solana_entry::block_component::{
            BlockFooterV1, BlockHeaderV1, UpdateParentV1, VersionedUpdateParent,
        },
        solana_hash::Hash,
        std::sync::{Arc, RwLock},
    };

    const DEFAULT_NS_PER_SLOT: u64 = DEFAULT_MS_PER_SLOT * 1_000_000;

    fn create_test_bank() -> (Arc<Bank>, Arc<RwLock<BankForks>>) {
        let genesis_config_info = create_genesis_config(10_000);
        Bank::new_with_bank_forks_for_tests(&genesis_config_info.genesis_config)
    }

    fn create_test_bank_alpenglow() -> (Arc<Bank>, Arc<RwLock<BankForks>>) {
        let mut genesis_config_info = create_genesis_config(10_000);
        activate_all_features_alpenglow(&mut genesis_config_info.genesis_config);
        Bank::new_with_bank_forks_for_tests(&genesis_config_info.genesis_config)
    }

    fn create_child_bank(
        bank_forks: &RwLock<BankForks>,
        parent: &Arc<Bank>,
        slot: u64,
    ) -> Arc<Bank> {
        Bank::new_from_parent_with_bank_forks(
            bank_forks,
            parent.clone(),
            SlotLeader::new_unique(),
            slot,
        )
    }

    fn test_genesis_cert_marker() -> GenesisCertBlockMarker {
        GenesisCertBlockMarker {
            slot: 0,
            block_id: Hash::default(),
            bls_signature: solana_bls_signatures::Signature([0; BLS_SIGNATURE_AFFINE_SIZE]),
            bitmap: vec![],
        }
    }

    #[test]
    fn test_missing_header_error_on_entry_batch() {
        let migration_status = MigrationStatus::post_migration_status();
        let mut processor = BlockComponentProcessor::default();

        // Try to process entry batch without header - should fail
        let result = processor.on_entry_batch(&migration_status, 1);
        assert!(matches!(
            result,
            Err(BlockComponentProcessorError::MissingParentMarker)
        ));
    }

    #[test]
    fn test_missing_header_error_on_genesis_certificate() {
        let migration_status = MigrationStatus::post_migration_status();
        let mut processor = BlockComponentProcessor::default();
        let marker =
            VersionedBlockMarker::from_genesis_cert_block_marker(test_genesis_cert_marker());

        let (parent, bank_forks) = create_test_bank();
        let bank = create_child_bank(&bank_forks, &parent, 1);
        let shred_version = rand::rng().random();

        let result = processor.on_marker(
            bank,
            parent,
            shred_version,
            marker,
            false,
            None,
            &migration_status,
        );
        assert!(matches!(
            result,
            Err(BlockComponentProcessorError::MissingParentMarker)
        ));
    }

    #[test]
    fn test_genesis_certificate_after_entry_batch_errors() {
        let migration_status = MigrationStatus::post_migration_status();
        let mut processor = BlockComponentProcessor::default();
        let (parent, bank_forks) = create_test_bank();
        let bank = create_child_bank(&bank_forks, &parent, 1);
        let shred_version = rand::rng().random();

        let header = VersionedBlockMarker::from_block_header(BlockHeaderV1 {
            parent_slot: 0,
            parent_block_id: Hash::default(),
        });
        processor
            .on_marker(
                bank.clone(),
                parent.clone(),
                shred_version,
                header,
                false,
                None,
                &migration_status,
            )
            .unwrap();
        processor
            .on_entry_batch(&migration_status, bank.slot())
            .unwrap();

        let marker =
            VersionedBlockMarker::from_genesis_cert_block_marker(test_genesis_cert_marker());
        let result = processor.on_marker(
            bank,
            parent,
            shred_version,
            marker,
            false,
            None,
            &migration_status,
        );
        assert!(matches!(
            result,
            Err(BlockComponentProcessorError::GenesisCertificateOutOfOrder)
        ));
    }

    #[test]
    fn test_genesis_certificate_immediately_after_header_passes_order_check() {
        let migration_status = MigrationStatus::post_migration_status();
        let mut processor = BlockComponentProcessor::default();
        let (parent, bank_forks) = create_test_bank();
        let bank = create_child_bank(&bank_forks, &parent, 1);
        let shred_version = rand::rng().random();

        let header = VersionedBlockMarker::from_block_header(BlockHeaderV1 {
            parent_slot: 0,
            parent_block_id: Hash::default(),
        });
        processor
            .on_marker(
                bank.clone(),
                parent.clone(),
                shred_version,
                header,
                false,
                None,
                &migration_status,
            )
            .unwrap();

        let marker =
            VersionedBlockMarker::from_genesis_cert_block_marker(test_genesis_cert_marker());
        let result = processor.on_marker(
            bank,
            parent,
            shred_version,
            marker,
            false,
            None,
            &migration_status,
        );
        assert!(matches!(
            result,
            Err(BlockComponentProcessorError::GenesisCertificateInAlpenglowCluster)
        ));
    }

    #[test]
    fn test_missing_footer_error_on_slot_full() {
        let migration_status = MigrationStatus::post_migration_status();
        let processor = BlockComponentProcessor {
            has_header: true,
            ..BlockComponentProcessor::default()
        };

        // Try to mark slot as full without footer - should fail
        let result = processor.on_final(&migration_status, 1);
        assert!(matches!(
            result,
            Err(BlockComponentProcessorError::MissingBlockFooter)
        ));
    }

    #[test]
    fn test_multiple_headers_error() {
        let mut processor = BlockComponentProcessor::default();
        let header = VersionedBlockHeader::V1(BlockHeaderV1 {
            parent_slot: 0,
            parent_block_id: Hash::default(),
        });

        // First header should succeed
        assert!(processor.on_header(&header, 0).is_ok());

        // Second header should fail
        let result = processor.on_header(&header, 0);
        assert!(matches!(
            result,
            Err(BlockComponentProcessorError::MultipleBlockHeaders)
        ));
    }

    #[test]
    fn test_multiple_footers_error() {
        let mut processor = BlockComponentProcessor {
            has_header: true,
            ..Default::default()
        };

        let (parent, bank_forks) = create_test_bank();
        let bank = create_child_bank(&bank_forks, &parent, 1);
        let shred_version = rand::rng().random();

        // Calculate valid timestamp based on parent's time
        let parent_time_nanos = parent.clock().unix_timestamp.saturating_mul(1_000_000_000);
        let footer_time_nanos = parent_time_nanos + 400_000_000; // parent + 400ms

        let footer = VersionedBlockFooter::V1(BlockFooterV1 {
            bank_hash: Hash::new_unique(),
            block_producer_time_nanos: footer_time_nanos as u64,
            block_user_agent: vec![],
            block_final_cert: None,
            skip_reward_cert: None,
            notar_reward_cert: None,
        });

        // First footer should succeed
        processor
            .on_footer(
                bank.clone(),
                parent.clone(),
                shred_version,
                footer.clone(),
                None,
            )
            .unwrap();

        // Second footer should fail
        let err = processor
            .on_footer(bank, parent, shred_version, footer, None)
            .unwrap_err();
        assert!(matches!(
            err,
            BlockComponentProcessorError::MultipleBlockFooters
        ));
    }

    #[test]
    fn test_on_footer_sets_timestamp() {
        let mut processor = BlockComponentProcessor {
            has_header: true,
            ..Default::default()
        };

        let (parent, bank_forks) = create_test_bank();
        let bank = create_child_bank(&bank_forks, &parent, 1);
        let shred_version = rand::rng().random();

        // Calculate valid timestamp based on parent's time
        let parent_time_nanos = parent.clock().unix_timestamp.saturating_mul(1_000_000_000);
        let footer_time_nanos = parent_time_nanos + 200_000_000; // parent + 200ms
        let expected_time_secs = footer_time_nanos / 1_000_000_000;

        let footer = VersionedBlockFooter::V1(BlockFooterV1 {
            bank_hash: Hash::new_unique(),
            block_producer_time_nanos: footer_time_nanos as u64,
            block_user_agent: vec![],
            block_final_cert: None,
            skip_reward_cert: None,
            notar_reward_cert: None,
        });

        processor
            .on_footer(bank.clone(), parent, shred_version, footer, None)
            .unwrap();

        assert!(processor.has_footer);

        // Verify clock sysvar was updated with correct timestamp (nanos converted to seconds)
        assert_eq!(bank.clock().unix_timestamp, expected_time_secs);
    }

    #[test]
    fn test_on_header_sets_flag() {
        let mut processor = BlockComponentProcessor::default();
        let header = VersionedBlockHeader::V1(BlockHeaderV1 {
            parent_slot: 0,
            parent_block_id: Hash::default(),
        });

        processor.on_header(&header, 0).unwrap();
        assert!(processor.has_header);
    }

    #[test]
    fn test_on_header_parent_slot_mismatch_error() {
        let mut processor = BlockComponentProcessor::default();
        let header = VersionedBlockHeader::V1(BlockHeaderV1 {
            parent_slot: 2,
            parent_block_id: Hash::default(),
        });

        assert!(matches!(
            processor.on_header(&header, 0),
            Err(BlockComponentProcessorError::HeaderParentSlotMismatch {
                header_parent_slot: 2,
                bank_parent_slot: 0,
            })
        ));
    }

    #[test]
    fn test_on_marker_processes_header() {
        let migration_status = MigrationStatus::post_migration_status();
        let mut processor = BlockComponentProcessor::default();
        let marker = VersionedBlockMarker::from_block_header(BlockHeaderV1 {
            parent_slot: 0,
            parent_block_id: Hash::default(),
        });

        let (parent, bank_forks) = create_test_bank();
        let bank = create_child_bank(&bank_forks, &parent, 1);
        let shred_version = rand::rng().random();

        processor
            .on_marker(
                bank,
                parent,
                shred_version,
                marker,
                false,
                None,
                &migration_status,
            )
            .unwrap();
        assert!(processor.has_header);
    }

    #[test]
    fn test_on_marker_rejects_header_parent_slot_mismatch() {
        let migration_status = MigrationStatus::post_migration_status();
        let mut processor = BlockComponentProcessor::default();
        let marker = VersionedBlockMarker::from_block_header(BlockHeaderV1 {
            parent_slot: 7, // mismatches bank.parent_slot() below (0)
            parent_block_id: Hash::default(),
        });

        let (parent, bank_forks) = create_test_bank();
        let bank = create_child_bank(&bank_forks, &parent, 1);
        let shred_version = rand::rng().random();

        assert!(matches!(
            processor.on_marker(
                bank,
                parent,
                shred_version,
                marker,
                false,
                None,
                &migration_status
            ),
            Err(BlockComponentProcessorError::HeaderParentSlotMismatch {
                header_parent_slot: 7,
                bank_parent_slot: 0,
            })
        ));
    }

    #[test]
    fn test_on_marker_processes_footer() {
        let migration_status = MigrationStatus::post_migration_status();
        let mut processor = BlockComponentProcessor {
            has_header: true,
            ..Default::default()
        };

        let (parent, bank_forks) = create_test_bank();
        let bank = create_child_bank(&bank_forks, &parent, 1);
        let shred_version = rand::rng().random();

        // Calculate valid timestamp based on parent's time
        let parent_time_nanos = parent.clock().unix_timestamp.saturating_mul(1_000_000_000);
        let footer_time_nanos = parent_time_nanos + 300_000_000; // parent + 300ms
        let expected_time_secs = footer_time_nanos / 1_000_000_000;

        let marker = VersionedBlockMarker::from_block_footer(BlockFooterV1 {
            bank_hash: Hash::new_unique(),
            block_producer_time_nanos: footer_time_nanos as u64,
            block_user_agent: vec![],
            block_final_cert: None,
            skip_reward_cert: None,
            notar_reward_cert: None,
        });

        processor
            .on_marker(
                bank.clone(),
                parent,
                shred_version,
                marker,
                false,
                None,
                &migration_status,
            )
            .unwrap();
        assert!(processor.has_footer);

        // Verify clock sysvar was updated
        assert_eq!(bank.clock().unix_timestamp, expected_time_secs);
    }

    #[test]
    fn test_complete_workflow_success() {
        let migration_status = MigrationStatus::post_migration_status();
        let mut processor = BlockComponentProcessor::default();
        let (parent, bank_forks) = create_test_bank();
        let bank = create_child_bank(&bank_forks, &parent, 1);
        let shred_version = rand::rng().random();

        // Calculate valid timestamp based on parent's time
        let parent_time_nanos = parent.clock().unix_timestamp.saturating_mul(1_000_000_000);
        let footer_time_nanos = parent_time_nanos + 100_000_000; // parent + 100ms
        let expected_time_secs = footer_time_nanos / 1_000_000_000;

        // Process header
        let header = VersionedBlockHeader::V1(BlockHeaderV1 {
            parent_slot: 0,
            parent_block_id: Hash::default(),
        });
        processor.on_header(&header, bank.parent_slot()).unwrap();

        // Process some entry batches (not full yet)
        assert!(processor.on_entry_batch(&migration_status, 1).is_ok());

        // Process footer with valid timestamp
        let footer = VersionedBlockFooter::V1(BlockFooterV1 {
            bank_hash: Hash::new_unique(),
            block_producer_time_nanos: footer_time_nanos as u64,
            block_user_agent: vec![],
            block_final_cert: None,
            skip_reward_cert: None,
            notar_reward_cert: None,
        });
        processor
            .on_footer(bank.clone(), parent.clone(), shred_version, footer, None)
            .unwrap();

        // Verify clock sysvar was updated
        assert_eq!(bank.clock().unix_timestamp, expected_time_secs);

        // Entry batch after footer should still succeed
        let result = processor.on_entry_batch(&migration_status, 1);
        assert!(result.is_ok());
    }

    #[test]
    fn test_block_marker_detected_pre_migration() {
        let migration_status = MigrationStatus::default();
        let mut processor = BlockComponentProcessor::default();
        let (parent, bank_forks) = create_test_bank();
        let bank = create_child_bank(&bank_forks, &parent, 1);
        let shred_version = rand::rng().random();

        // Try to process a block header marker pre-migration - should fail
        let marker = VersionedBlockMarker::from_block_header(BlockHeaderV1 {
            parent_slot: 0,
            parent_block_id: Hash::default(),
        });

        let err = processor
            .on_marker(
                bank,
                parent,
                shred_version,
                marker,
                false,
                None,
                &migration_status,
            )
            .unwrap_err();
        assert!(matches!(
            err,
            BlockComponentProcessorError::BlockComponentPreMigration
        ));
    }

    #[test]
    fn test_footer_and_update_parent_rejected_pre_migration() {
        let migration_status = MigrationStatus::default();
        let (parent, bank_forks) = create_test_bank();
        let bank = create_child_bank(&bank_forks, &parent, 1);
        let shred_version = rand::rng().random();

        let parent_time_nanos = parent.clock().unix_timestamp.saturating_mul(1_000_000_000);
        let footer_marker = VersionedBlockMarker::from_block_footer(BlockFooterV1 {
            bank_hash: Hash::new_unique(),
            block_producer_time_nanos: (parent_time_nanos + 500_000_000) as u64,
            block_user_agent: vec![],
            block_final_cert: None,
            skip_reward_cert: None,
            notar_reward_cert: None,
        });

        let mut processor = BlockComponentProcessor::default();
        assert!(matches!(
            processor
                .on_marker(
                    bank.clone(),
                    parent.clone(),
                    shred_version,
                    footer_marker,
                    false,
                    None,
                    &migration_status
                )
                .unwrap_err(),
            BlockComponentProcessorError::BlockComponentPreMigration
        ));

        let update_parent_marker = VersionedBlockMarker::from_update_parent(UpdateParentV1 {
            new_parent_slot: 0,
            new_parent_block_id: Hash::default(),
        });

        let mut processor = BlockComponentProcessor::default();
        assert!(matches!(
            processor
                .on_marker(
                    bank,
                    parent,
                    shred_version,
                    update_parent_marker,
                    false,
                    None,
                    &migration_status
                )
                .unwrap_err(),
            BlockComponentProcessorError::BlockComponentPreMigration
        ));
    }

    #[test]
    fn test_entry_batch_pre_migration_succeeds() {
        let migration_status = MigrationStatus::default();
        let mut processor = BlockComponentProcessor::default();

        // Processing entry batches pre-migration (without markers) should succeed
        let result = processor.on_entry_batch(&migration_status, 1);
        assert!(result.is_ok());

        // Even with slot full
        let result = processor.on_entry_batch(&migration_status, 1);
        assert!(result.is_ok());
    }

    #[test]
    fn test_complete_workflow_post_migration() {
        let migration_status = MigrationStatus::post_migration_status();
        let mut processor = BlockComponentProcessor::default();
        let (parent, bank_forks) = create_test_bank();
        let bank = create_child_bank(&bank_forks, &parent, 1);
        let shred_version = rand::rng().random();

        // Process header marker
        let header_marker = VersionedBlockMarker::from_block_header(BlockHeaderV1 {
            parent_slot: 0,
            parent_block_id: Hash::default(),
        });
        processor
            .on_marker(
                bank.clone(),
                parent.clone(),
                shred_version,
                header_marker,
                false,
                None,
                &migration_status,
            )
            .unwrap();

        // Process entry batches
        assert!(processor.on_entry_batch(&migration_status, 1).is_ok());

        // Calculate valid timestamp based on parent's time
        let parent_time_nanos = parent.clock().unix_timestamp.saturating_mul(1_000_000_000);
        let footer_time_nanos = parent_time_nanos + 500_000_000; // parent + 500ms
        let expected_time_secs = footer_time_nanos / 1_000_000_000;

        // Process footer marker
        let footer_marker = VersionedBlockMarker::from_block_footer(BlockFooterV1 {
            bank_hash: Hash::new_unique(),
            block_producer_time_nanos: footer_time_nanos as u64,
            block_user_agent: vec![],
            block_final_cert: None,
            skip_reward_cert: None,
            notar_reward_cert: None,
        });
        processor
            .on_marker(
                bank.clone(),
                parent,
                shred_version,
                footer_marker,
                false,
                None,
                &migration_status,
            )
            .unwrap();

        // Verify clock sysvar was updated
        assert_eq!(bank.clock().unix_timestamp, expected_time_secs);

        // Entry batch after footer should still succeed
        let result = processor.on_entry_batch(&migration_status, 1);
        assert!(result.is_ok());
    }

    #[test]
    fn test_footer_without_header_errors() {
        let mut processor = BlockComponentProcessor::default();
        let (parent, bank_forks) = create_test_bank();
        let bank = create_child_bank(&bank_forks, &parent, 1);
        let shred_version = rand::rng().random();

        let footer = VersionedBlockFooter::V1(BlockFooterV1 {
            bank_hash: Hash::new_unique(),
            block_producer_time_nanos: 1_000_000_000,
            block_user_agent: vec![],
            block_final_cert: None,
            skip_reward_cert: None,
            notar_reward_cert: None,
        });

        // Try to process footer without header - should fail
        let err = processor
            .on_footer(bank, parent, shred_version, footer, None)
            .unwrap_err();
        assert!(matches!(
            err,
            BlockComponentProcessorError::MissingParentMarker
        ));
    }

    #[test]
    fn test_marker_with_footer_at_slot_full() {
        let migration_status = MigrationStatus::post_migration_status();
        let mut processor = BlockComponentProcessor::default();
        let (parent, bank_forks) = create_test_bank();
        let bank = create_child_bank(&bank_forks, &parent, 1);
        let shred_version = rand::rng().random();

        // Process header first
        processor.has_header = true;

        // Calculate valid timestamp based on parent's time
        let parent_time_nanos = parent.clock().unix_timestamp.saturating_mul(1_000_000_000);
        let footer_time_nanos = parent_time_nanos + 600_000_000; // parent + 600ms
        let expected_time_secs = footer_time_nanos / 1_000_000_000;

        // Process footer marker
        let footer_marker = VersionedBlockMarker::from_block_footer(BlockFooterV1 {
            bank_hash: Hash::new_unique(),
            block_producer_time_nanos: footer_time_nanos as u64,
            block_user_agent: vec![],
            block_final_cert: None,
            skip_reward_cert: None,
            notar_reward_cert: None,
        });

        // Should succeed - footer is processed
        processor
            .on_marker(
                bank.clone(),
                parent,
                shred_version,
                footer_marker,
                false,
                None,
                &migration_status,
            )
            .unwrap();
        assert!(processor.has_footer);

        // Verify clock sysvar was updated
        assert_eq!(bank.clock().unix_timestamp, expected_time_secs);
    }

    #[test]
    fn test_entry_batch_with_header_not_full_succeeds() {
        let migration_status = MigrationStatus::post_migration_status();
        let mut processor = BlockComponentProcessor {
            has_header: true,
            ..Default::default()
        };

        // Process entry batch with header but not full - should succeed even without footer
        let result = processor.on_entry_batch(&migration_status, 1);
        assert!(result.is_ok());
    }

    #[test]
    fn test_footer_sets_epoch_start_timestamp_on_epoch_change() {
        let mut processor = BlockComponentProcessor {
            has_header: true,
            ..Default::default()
        };
        let shred_version = rand::rng().random();

        // Create genesis bank
        let genesis_config_info = create_genesis_config(10_000);
        let (genesis_bank, bank_forks) =
            Bank::new_with_bank_forks_for_tests(&genesis_config_info.genesis_config);

        // Get epoch schedule to find first slot of next epoch
        let epoch_schedule = genesis_bank.epoch_schedule();
        let first_slot_in_epoch_1 = epoch_schedule.get_first_slot_in_epoch(1);

        // Create parent bank at last slot of epoch 0
        let mut parent = genesis_bank.clone();
        for slot in 1..first_slot_in_epoch_1 {
            parent = create_child_bank(&bank_forks, &parent, slot);
        }

        // Create bank at first slot of epoch 1
        let bank = create_child_bank(&bank_forks, &parent, first_slot_in_epoch_1);

        // Verify we're in epoch 1
        assert_eq!(bank.epoch(), 1);

        // Calculate valid timestamp based on parent's time
        let parent_slot = parent.slot();
        let parent_time_nanos = parent.clock().unix_timestamp.saturating_mul(1_000_000_000);
        let current_slot = bank.slot();
        let elapsed_slot_duration_nanos =
            bank.slot_range_duration_nanos(parent_slot.saturating_add(1), current_slot);

        // Use a timestamp in the middle of the valid range
        let (lower_bound, upper_bound) = BlockComponentProcessor::nanosecond_time_bounds(
            parent_time_nanos,
            elapsed_slot_duration_nanos,
        );
        let footer_time_nanos = (lower_bound + upper_bound) / 2;
        let expected_time_secs = footer_time_nanos / 1_000_000_000;

        let footer = VersionedBlockFooter::V1(BlockFooterV1 {
            bank_hash: Hash::new_unique(),
            block_producer_time_nanos: footer_time_nanos as u64,
            block_user_agent: vec![],
            block_final_cert: None,
            skip_reward_cert: None,
            notar_reward_cert: None,
        });

        processor
            .on_footer(bank.clone(), parent, shred_version, footer, None)
            .unwrap();

        // Verify clock sysvar was updated
        assert_eq!(bank.clock().unix_timestamp, expected_time_secs);

        // Verify epoch_start_timestamp was set correctly for the new epoch
        assert_eq!(bank.clock().epoch_start_timestamp, expected_time_secs);
    }

    // Helper function to test clock bounds enforcement
    fn test_clock_bounds_helper(
        slot_gap: u64,
        timestamp_fn: impl FnOnce(i64, i64, i64) -> i64,
        should_pass: bool,
    ) {
        let mut processor = BlockComponentProcessor {
            has_header: true,
            ..Default::default()
        };
        let shred_version = rand::rng().random();

        let (parent, bank_forks) = create_test_bank_alpenglow();
        let parent_time_nanos = parent.clock().unix_timestamp.saturating_mul(1_000_000_000);

        // Set up clock on parent so validation doesn't skip bounds checking
        parent.update_clock_from_footer(parent_time_nanos);

        let bank: Arc<Bank> = create_child_bank(&bank_forks, &parent, slot_gap);
        let elapsed_slot_duration_nanos = bank.slot_range_duration_nanos(1, slot_gap);

        let (lower_bound, upper_bound) = BlockComponentProcessor::nanosecond_time_bounds(
            parent_time_nanos,
            elapsed_slot_duration_nanos,
        );

        let footer_time_nanos = timestamp_fn(parent_time_nanos, lower_bound, upper_bound);

        let footer = VersionedBlockFooter::V1(BlockFooterV1 {
            bank_hash: Hash::new_unique(),
            block_producer_time_nanos: footer_time_nanos as u64,
            block_user_agent: vec![],
            block_final_cert: None,
            skip_reward_cert: None,
            notar_reward_cert: None,
        });

        let result = processor.on_footer(bank, parent, shred_version, footer, None);
        if should_pass {
            result.unwrap();
        } else {
            assert!(matches!(
                result.unwrap_err(),
                BlockComponentProcessorError::NanosecondClockOutOfBounds
            ));
        }
    }

    #[test]
    fn test_clock_bounds_at_minimum() {
        test_clock_bounds_helper(1, |_, lower, _| lower, true);
    }

    #[test]
    fn test_clock_bounds_at_maximum() {
        test_clock_bounds_helper(1, |_, _, upper| upper, true);
    }

    #[test]
    fn test_clock_bounds_below_minimum() {
        test_clock_bounds_helper(1, |_, lower, _| lower - 1, false);
    }

    #[test]
    fn test_clock_bounds_above_maximum() {
        test_clock_bounds_helper(1, |_, _, upper| upper + 1, false);
    }

    #[test]
    fn test_clock_bounds_multi_slot_gap() {
        // For 5 slots: upper_bound = parent_time + 2 * 5 * 400ms = parent_time + 4000ms
        // Use 2 seconds which is within bounds
        test_clock_bounds_helper(5, |_, lower, _| lower + 2_000_000_000, true);
    }

    #[test]
    fn test_clock_bounds_multi_slot_gap_exceeds() {
        // Exceed by 1 second beyond the upper bound
        test_clock_bounds_helper(5, |_, _, upper| upper + 1_000_000_000, false);
    }

    #[test]
    fn test_clock_bounds_timestamp_equals_parent() {
        // Timestamp equal to parent time (should fail, must be strictly greater)
        test_clock_bounds_helper(1, |parent_time, _, _| parent_time, false);
    }

    #[test]
    fn test_clock_bounds_without_parent_nanosecond_clock_rejects_out_of_bounds() {
        let mut processor = BlockComponentProcessor {
            has_header: true,
            ..Default::default()
        };
        let shred_version = rand::rng().random();

        let (parent, bank_forks) = create_test_bank_alpenglow();
        assert_eq!(parent.get_nanosecond_clock(), None);

        let bank = create_child_bank(&bank_forks, &parent, 1);
        let parent_time_nanos = bank.clock().unix_timestamp.saturating_mul(1_000_000_000);
        let elapsed_slot_duration_nanos =
            bank.slot_range_duration_nanos(parent.slot().saturating_add(1), bank.slot());
        let (_, upper_bound) = BlockComponentProcessor::nanosecond_time_bounds(
            parent_time_nanos,
            elapsed_slot_duration_nanos,
        );

        let footer = VersionedBlockFooter::V1(BlockFooterV1 {
            bank_hash: Hash::new_unique(),
            block_producer_time_nanos: u64::try_from(upper_bound.saturating_add(1)).unwrap(),
            block_user_agent: vec![],
            block_final_cert: None,
            skip_reward_cert: None,
            notar_reward_cert: None,
        });

        assert!(matches!(
            processor
                .on_footer(bank, parent, shred_version, footer, None)
                .unwrap_err(),
            BlockComponentProcessorError::NanosecondClockOutOfBounds
        ));
    }

    #[test]
    fn test_clock_bounds_rejects_timestamp_above_i64() {
        let mut processor = BlockComponentProcessor {
            has_header: true,
            ..Default::default()
        };
        let shred_version = rand::rng().random();

        let (parent, bank_forks) = create_test_bank_alpenglow();
        let parent_time_nanos = parent.clock().unix_timestamp.saturating_mul(1_000_000_000);
        parent.update_clock_from_footer(parent_time_nanos);
        let bank = create_child_bank(&bank_forks, &parent, 1);

        let footer = VersionedBlockFooter::V1(BlockFooterV1 {
            bank_hash: Hash::new_unique(),
            block_producer_time_nanos: u64::MAX,
            block_user_agent: vec![],
            block_final_cert: None,
            skip_reward_cert: None,
            notar_reward_cert: None,
        });

        assert!(matches!(
            processor
                .on_footer(bank, parent, shred_version, footer, None)
                .unwrap_err(),
            BlockComponentProcessorError::NanosecondClockOutOfBounds
        ));
    }

    // Helper function to test nanosecond_time_bounds calculation
    fn test_nanosecond_time_bounds_helper(
        parent_time_nanos: i64,
        elapsed_slot_duration_nanos: u128,
        expected_lower: i64,
        expected_upper: i64,
    ) {
        let (lower, upper) = BlockComponentProcessor::nanosecond_time_bounds(
            parent_time_nanos,
            elapsed_slot_duration_nanos,
        );

        assert_eq!(lower, expected_lower);
        assert_eq!(upper, expected_upper);
    }

    #[test]
    fn test_nanosecond_time_bounds_calculation() {
        // Test the nanosecond_time_bounds function directly
        // diff_slots = 15 - 10 = 5
        // lower = parent_time + 1
        // upper = parent_time + 2 * 5 * 400_000_000 = parent_time + 4_000_000_000
        let parent_slot = 10;
        let parent_time = 1_000_000_000_000; // 1000 seconds in nanos
        let working_slot = 15;
        let slot_delta = working_slot - parent_slot;
        test_nanosecond_time_bounds_helper(
            parent_time,
            u128::from(slot_delta).saturating_mul(u128::from(DEFAULT_NS_PER_SLOT)),
            parent_time + 1,
            parent_time + (2 * DEFAULT_NS_PER_SLOT * slot_delta) as i64,
        );
    }

    #[test]
    fn test_nanosecond_time_bounds_same_slot() {
        // Test with same slot (diff = 0)
        // diff_slots = 0
        // lower = parent_time + 1
        // upper = parent_time + 2 * 0 * 400_000_000 = parent_time
        // Note: In this case, lower > upper, so no timestamp would be valid
        // This is expected since we shouldn't have the same slot for parent and working bank
        let parent_time = 1_000_000_000_000;
        test_nanosecond_time_bounds_helper(parent_time, 0, parent_time + 1, parent_time);
    }

    #[test]
    fn test_nanosecond_time_bounds_saturates_upper_bound() {
        let parent_time = i64::MAX - 5;
        let (lower, upper) =
            BlockComponentProcessor::nanosecond_time_bounds(parent_time, u128::MAX);

        assert_eq!(lower, parent_time + 1);
        assert_eq!(upper, i64::MAX);
    }

    #[test]
    fn test_initial_up_reject() {
        let mut processor = BlockComponentProcessor::default();
        let update_parent = VersionedUpdateParent::V1(UpdateParentV1 {
            new_parent_slot: 0,
            new_parent_block_id: Hash::default(),
        });

        assert!(matches!(
            processor.on_update_parent(4, &update_parent, false),
            Err(BlockComponentProcessorError::UnexpectedInitialUpdateParent)
        ));
        assert!(processor.update_parent.is_none());
    }

    #[test]
    fn test_update_parent_rejects_non_first_leader_window_slot() {
        let mut processor = BlockComponentProcessor::default();
        let update_parent = VersionedUpdateParent::V1(UpdateParentV1 {
            new_parent_slot: 0,
            new_parent_block_id: Hash::default(),
        });

        assert!(matches!(
            processor.on_update_parent(5, &update_parent, true),
            Err(BlockComponentProcessorError::UpdateParentNotFirstInLeaderWindow(5))
        ));
        assert!(processor.update_parent.is_none());
    }

    #[test]
    fn test_initial_up_ok() {
        let mut processor = BlockComponentProcessor::default();
        let update_parent = VersionedUpdateParent::V1(UpdateParentV1 {
            new_parent_slot: 0,
            new_parent_block_id: Hash::default(),
        });

        processor.on_update_parent(4, &update_parent, true).unwrap();
        assert!(processor.update_parent.is_some());
    }

    #[test]
    fn test_update_parent_after_header_abandoned_bank() {
        let mut processor = BlockComponentProcessor::default();
        processor
            .on_header(
                &VersionedBlockHeader::V1(BlockHeaderV1 {
                    parent_slot: 0,
                    parent_block_id: Hash::default(),
                }),
                0,
            )
            .unwrap();

        let update_parent = VersionedUpdateParent::V1(UpdateParentV1 {
            new_parent_slot: 0,
            new_parent_block_id: Hash::default(),
        });

        assert!(matches!(
            processor.on_update_parent(4, &update_parent, false),
            Err(BlockComponentProcessorError::AbandonedBank(_))
        ));
    }

    #[test]
    fn test_multiple_update_parents_error() {
        let mut processor = BlockComponentProcessor::default();
        let update_parent = VersionedUpdateParent::V1(UpdateParentV1 {
            new_parent_slot: 0,
            new_parent_block_id: Hash::default(),
        });

        // First should succeed
        processor.on_update_parent(4, &update_parent, true).unwrap();

        // Second should fail
        assert!(matches!(
            processor.on_update_parent(4, &update_parent, true),
            Err(BlockComponentProcessorError::MultipleUpdateParents)
        ));
    }

    #[test]
    fn test_header_after_update_parent_error() {
        let mut processor = BlockComponentProcessor::default();
        processor
            .on_update_parent(
                4,
                &VersionedUpdateParent::V1(UpdateParentV1 {
                    new_parent_slot: 0,
                    new_parent_block_id: Hash::default(),
                }),
                true,
            )
            .unwrap();

        let header = VersionedBlockHeader::V1(BlockHeaderV1 {
            parent_slot: 0,
            parent_block_id: Hash::default(),
        });

        assert!(matches!(
            processor.on_header(&header, 0),
            Err(BlockComponentProcessorError::SpuriousUpdateParent)
        ));
    }

    #[test]
    fn test_workflow_with_update_parent() {
        let migration_status = MigrationStatus::post_migration_status();
        let mut processor = BlockComponentProcessor::default();
        let (parent, bank_forks) = create_test_bank();
        let bank = create_child_bank(&bank_forks, &parent, 4);
        let shred_version = rand::rng().random();

        processor
            .on_update_parent(
                bank.slot(),
                &VersionedUpdateParent::V1(UpdateParentV1 {
                    new_parent_slot: 0,
                    new_parent_block_id: Hash::default(),
                }),
                true,
            )
            .unwrap();

        assert!(processor.on_entry_batch(&migration_status, 1).is_ok());

        let parent_time_nanos = parent.clock().unix_timestamp.saturating_mul(1_000_000_000);
        let footer = VersionedBlockFooter::V1(BlockFooterV1 {
            bank_hash: Hash::new_unique(),
            block_producer_time_nanos: (parent_time_nanos + 100_000_000) as u64,
            block_user_agent: vec![],
            block_final_cert: None,
            skip_reward_cert: None,
            notar_reward_cert: None,
        });
        processor
            .on_footer(bank, parent, shred_version, footer, None)
            .unwrap();

        assert!(processor.on_final(&migration_status, 1).is_ok());
    }
}