espresso_node/

persistence.rs

//! Sequencer node persistence.
//!
//! This module implements the persistence required for a sequencer node to rejoin the network and
//! resume participating in consensus, in the event that its process crashes or is killed and loses
//! all in-memory state.
//!
//! This is distinct from the query service persistent storage found in the `api` module, which is
//! an extension that node operators can opt into. This module defines the minimum level of
//! persistence which is _required_ to run a node.

use std::collections::HashMap;

use alloy::primitives::{Address, U256};
use anyhow::Context;
use async_trait::async_trait;
use espresso_types::{
    PubKey,
    v0_3::{ChainConfig, RegisteredValidator},
};

pub mod fs;
pub mod migrations;
pub mod no_storage;
mod persistence_metrics;
pub mod sql;

/// RegisteredValidator without x25519_key/p2p_addr fields.
/// Used for migrating data written before x25519 support was added.
#[derive(serde::Serialize, serde::Deserialize)]
pub(crate) struct RegisteredValidatorNoX25519 {
    pub account: Address,
    pub stake_table_key: PubKey,
    pub state_ver_key: hotshot_types::light_client::StateVerKey,
    pub stake: U256,
    pub commission: u16,
    pub delegators: HashMap<Address, U256>,
    pub authenticated: bool,
}

impl RegisteredValidatorNoX25519 {
    pub fn migrate(self) -> RegisteredValidator<PubKey> {
        RegisteredValidator {
            account: self.account,
            stake_table_key: Some(self.stake_table_key),
            state_ver_key: Some(self.state_ver_key),
            stake: self.stake,
            commission: self.commission,
            delegators: self.delegators,
            authenticated: self.authenticated,
            x25519_key: None,
            p2p_addr: None,
        }
    }
}

// TODO: replace with a proper data migration if it is ever deemed worthwhile.
// Field order must match `RegisteredValidator<KEY>` in v0_3::stake_table exactly (bincode is order-sensitive).
#[derive(serde::Serialize, serde::Deserialize)]
pub(crate) struct RegisteredValidatorPreOption {
    pub account: Address,
    pub stake_table_key: PubKey,
    pub state_ver_key: hotshot_types::light_client::StateVerKey,
    pub stake: U256,
    pub commission: u16,
    pub delegators: HashMap<Address, U256>,
    pub authenticated: bool,
    pub x25519_key: Option<hotshot_types::x25519::PublicKey>,
    pub p2p_addr: Option<hotshot_types::addr::NetAddr>,
}

impl RegisteredValidatorPreOption {
    pub fn migrate(self) -> RegisteredValidator<PubKey> {
        RegisteredValidator {
            account: self.account,
            stake_table_key: Some(self.stake_table_key),
            state_ver_key: Some(self.state_ver_key),
            stake: self.stake,
            commission: self.commission,
            delegators: self.delegators,
            authenticated: self.authenticated,
            x25519_key: self.x25519_key,
            p2p_addr: self.p2p_addr,
        }
    }
}

// TODO: replace with a proper data migration if it is ever deemed worthwhile.
// Layout written after BLS fix (381b0f9207) but before Schnorr Option fix.
// stake_table_key is Option<KEY>, state_ver_key is raw StateVerKey.
// Field order must match exactly (bincode is order-sensitive).
#[derive(serde::Serialize, serde::Deserialize)]
pub(crate) struct RegisteredValidatorPreSchnorrOption {
    pub account: Address,
    pub stake_table_key: Option<PubKey>,
    pub state_ver_key: hotshot_types::light_client::StateVerKey,
    pub stake: U256,
    pub commission: u16,
    pub delegators: HashMap<Address, U256>,
    pub authenticated: bool,
    pub x25519_key: Option<hotshot_types::x25519::PublicKey>,
    pub p2p_addr: Option<hotshot_types::addr::NetAddr>,
}

impl RegisteredValidatorPreSchnorrOption {
    pub fn migrate(self) -> RegisteredValidator<PubKey> {
        RegisteredValidator {
            account: self.account,
            stake_table_key: self.stake_table_key,
            state_ver_key: Some(self.state_ver_key),
            stake: self.stake,
            commission: self.commission,
            delegators: self.delegators,
            authenticated: self.authenticated,
            x25519_key: self.x25519_key,
            p2p_addr: self.p2p_addr,
        }
    }
}

/// Update a `NetworkConfig` that may have originally been persisted with an old version.
fn migrate_network_config(
    mut network_config: serde_json::Value,
) -> anyhow::Result<serde_json::Value> {
    let config = network_config
        .get_mut("config")
        .context("missing field `config`")?
        .as_object_mut()
        .context("`config` must be an object")?;

    if !config.contains_key("builder_urls") {
        // When multi-builder support was added, the configuration field `builder_url: Url` was
        // replaced by an array `builder_urls: Vec<Url>`. If the saved config has no `builder_urls`
        // field, it is older than this change. Populate `builder_urls` with a singleton array
        // formed from the old value of `builder_url`, and delete the no longer used `builder_url`.
        let url = config
            .remove("builder_url")
            .context("missing field `builder_url`")?;
        config.insert("builder_urls".into(), vec![url].into());
    }

    // HotShotConfig was upgraded to include parameters for proposing and voting on upgrades.
    // Configs which were persisted before this upgrade may be missing these parameters. This
    // migration initializes them with a default. By default, we use JS MAX_SAFE_INTEGER for the
    // start parameters so that nodes will never do an upgrade, unless explicitly configured
    // otherwise.
    if !config.contains_key("start_proposing_view") {
        config.insert("start_proposing_view".into(), 9007199254740991u64.into());
    }
    if !config.contains_key("stop_proposing_view") {
        config.insert("stop_proposing_view".into(), 0.into());
    }
    if !config.contains_key("start_voting_view") {
        config.insert("start_voting_view".into(), 9007199254740991u64.into());
    }
    if !config.contains_key("stop_voting_view") {
        config.insert("stop_voting_view".into(), 0.into());
    }
    if !config.contains_key("start_proposing_time") {
        config.insert("start_proposing_time".into(), 9007199254740991u64.into());
    }
    if !config.contains_key("stop_proposing_time") {
        config.insert("stop_proposing_time".into(), 0.into());
    }
    if !config.contains_key("start_voting_time") {
        config.insert("start_voting_time".into(), 9007199254740991u64.into());
    }
    if !config.contains_key("stop_voting_time") {
        config.insert("stop_voting_time".into(), 0.into());
    }

    // HotShotConfig was upgraded to include an `epoch_height` parameter. Initialize with a default
    // if missing.
    if !config.contains_key("epoch_height") {
        config.insert("epoch_height".into(), 0.into());
    }

    // HotShotConfig was upgraded to include `drb_difficulty` and `drb_upgrade_difficulty` parameters. Initialize with a default
    // if missing.
    if !config.contains_key("drb_difficulty") {
        config.insert("drb_difficulty".into(), 0.into());
    }
    if !config.contains_key("drb_upgrade_difficulty") {
        config.insert("drb_upgrade_difficulty".into(), 0.into());
    }

    // HotShotConfig was upgraded to include `da_committeees`. Initialize with an empty `da_committees` if missing.
    if !config.contains_key("da_committees") {
        config.insert("da_committees".into(), serde_json::json!([]));
    }

    Ok(network_config)
}

#[async_trait]
pub trait ChainConfigPersistence: Sized + Send + Sync {
    async fn insert_chain_config(&mut self, chain_config: ChainConfig) -> anyhow::Result<()>;
}

#[cfg(test)]
mod tests {
    use std::{cmp::max, collections::BTreeMap, marker::PhantomData, sync::Arc, time::Duration};

    use alloy::{
        network::EthereumWallet,
        primitives::{Address, U256},
        providers::{Provider, ProviderBuilder, ext::AnvilApi},
    };
    use anyhow::bail;
    use async_lock::{Mutex, RwLock};
    use async_trait::async_trait;
    use committable::{Commitment, Committable};
    use espresso_contract_deployer::{
        Contract, Contracts, DEFAULT_EXIT_ESCROW_PERIOD_SECONDS, builder::DeployerArgsBuilder,
        network_config::light_client_genesis_from_stake_table,
    };
    use espresso_types::{
        Event, L1Client, L1ClientOptions, Leaf, Leaf2, NodeState, PubKey, SeqTypes, ValidatedState,
        traits::{
            EventConsumer, EventsPersistenceRead, MembershipPersistence, NullEventConsumer,
            PersistenceOptions, SequencerPersistence,
        },
        v0_3::{AuthenticatedValidator, EventKey, Fetcher, RegisteredValidator, StakeTableEvent},
    };
    use futures::{StreamExt, TryStreamExt, future::join_all};
    use hotshot::{
        InitializerEpochInfo,
        types::{BLSPubKey, SignatureKey},
    };
    use hotshot_contract_adapter::{
        sol_types::StakeTableV3::Delegated, stake_table::StakeTableContractVersion,
    };
    use hotshot_example_types::node_types::TEST_VERSIONS;
    use hotshot_new_protocol::message::Certificate2;
    use hotshot_query_service::{availability::BlockQueryData, testing::mocks::MOCK_UPGRADE};
    use hotshot_types::{
        data::{
            DaProposal2, EpochNumber, QuorumProposal2, QuorumProposalWrapper, VidCommitment,
            ViewNumber, ns_table::parse_ns_table, vid_commitment, vid_disperse::AvidMDisperseShare,
        },
        event::{EventType, HotShotAction, LeafInfo},
        light_client::StateKeyPair,
        message::{Proposal, UpgradeLock, convert_proposal},
        new_protocol::CoordinatorEvent,
        simple_certificate::{
            CertificatePair, NextEpochQuorumCertificate2, QuorumCertificate, QuorumCertificate2,
            UpgradeCertificate,
        },
        simple_vote::{
            NextEpochQuorumData2, QuorumData2, UpgradeProposalData, VersionedVoteData, Vote2Data,
        },
        traits::{EncodeBytes, block_contents::BlockHeader},
        utils::EpochTransitionIndicator,
        vid::avidm::{AvidMScheme, init_avidm_param},
        vote::HasViewNumber,
    };
    use indexmap::IndexMap;
    use staking_cli::demo::{DelegationConfig, StakingTransactions};
    use surf_disco::Client;
    use test_utils::reserve_tcp_port;
    use tide_disco::error::ServerError;
    use tokio::{spawn, time::sleep};
    use vbs::version::Version;
    use versions::{Upgrade, version};

    use crate::{
        RECENT_STAKE_TABLES_LIMIT, SequencerApiVersion,
        api::{
            Options,
            test_helpers::{STAKE_TABLE_CAPACITY_FOR_TEST, TestNetwork, TestNetworkConfigBuilder},
        },
        catchup::NullStateCatchup,
        testing::{TestConfigBuilder, staking_priv_keys},
    };

    #[async_trait]
    pub trait TestablePersistence: SequencerPersistence + MembershipPersistence {
        type Storage: Sync;

        async fn tmp_storage() -> Self::Storage;
        fn options(storage: &Self::Storage) -> impl PersistenceOptions<Persistence = Self>;

        async fn connect(storage: &Self::Storage) -> Self {
            Self::options(storage).create().await.unwrap()
        }
    }

    #[rstest_reuse::template]
    #[rstest::rstest]
    #[case(PhantomData::<crate::persistence::sql::Persistence>)]
    #[case(PhantomData::<crate::persistence::fs::Persistence>)]
    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    pub fn persistence_types<P: TestablePersistence>(#[case] _p: PhantomData<P>) {}

    #[derive(Clone, Debug, Default)]
    struct EventCollector {
        events: Arc<RwLock<Vec<Event>>>,
    }

    impl EventCollector {
        async fn leaf_chain(&self) -> Vec<LeafInfo<SeqTypes>> {
            self.events
                .read()
                .await
                .iter()
                .flat_map(|event| {
                    let EventType::Decide { leaf_chain, .. } = &event.event else {
                        panic!("expected decide event, got {event:?}");
                    };
                    leaf_chain.iter().cloned().rev()
                })
                .collect::<Vec<_>>()
        }
    }

    #[async_trait]
    impl EventConsumer for EventCollector {
        async fn handle_event(&self, event: &CoordinatorEvent<SeqTypes>) -> anyhow::Result<()> {
            if let CoordinatorEvent::LegacyEvent(event) = event {
                self.events.write().await.push(event.clone());
            }
            Ok(())
        }
    }

    #[derive(Clone, Copy, Debug)]
    struct FailConsumer;

    #[async_trait]
    impl EventConsumer for FailConsumer {
        async fn handle_event(&self, _: &CoordinatorEvent<SeqTypes>) -> anyhow::Result<()> {
            bail!("mock error injection");
        }
    }

    #[rstest_reuse::apply(persistence_types)]
    pub async fn test_voted_view<P: TestablePersistence>(_p: PhantomData<P>) {
        let tmp = P::tmp_storage().await;
        let storage = P::connect(&tmp).await;

        // Initially, there is no saved view.
        assert_eq!(storage.load_latest_acted_view().await.unwrap(), None);

        // Store a view.
        let view1 = ViewNumber::genesis();
        storage
            .record_action(view1, None, HotShotAction::Vote)
            .await
            .unwrap();
        assert_eq!(
            storage.load_latest_acted_view().await.unwrap().unwrap(),
            view1
        );

        // Store a newer view, make sure storage gets updated.
        let view2 = view1 + 1;
        storage
            .record_action(view2, None, HotShotAction::Vote)
            .await
            .unwrap();
        assert_eq!(
            storage.load_latest_acted_view().await.unwrap().unwrap(),
            view2
        );

        // Store an old view, make sure storage is unchanged.
        storage
            .record_action(view1, None, HotShotAction::Vote)
            .await
            .unwrap();
        assert_eq!(
            storage.load_latest_acted_view().await.unwrap().unwrap(),
            view2
        );
    }

    #[rstest_reuse::apply(persistence_types)]
    pub async fn test_high_qc2_monotonic<P: TestablePersistence>(_p: PhantomData<P>) {
        let tmp = P::tmp_storage().await;
        let storage = P::connect(&tmp).await;

        // Initially there is no persisted lock.
        assert_eq!(storage.load_high_qc2().await.unwrap(), None);

        let mut qc = QuorumCertificate2::genesis(
            &ValidatedState::default(),
            &NodeState::mock(),
            TEST_VERSIONS.test,
        )
        .await;

        // Persist a lock at view 5.
        qc.view_number = ViewNumber::new(5);
        storage.append_high_qc2(qc.clone()).await.unwrap();
        assert_eq!(
            storage.load_high_qc2().await.unwrap().unwrap().view_number,
            ViewNumber::new(5)
        );

        // A newer lock advances the stored view.
        qc.view_number = ViewNumber::new(6);
        storage.append_high_qc2(qc.clone()).await.unwrap();
        assert_eq!(
            storage.load_high_qc2().await.unwrap().unwrap().view_number,
            ViewNumber::new(6)
        );

        // A stale (older) lock is a no-op: the compare-and-set never regresses
        // the persisted view, even though the file/row write is unconditional.
        qc.view_number = ViewNumber::new(4);
        storage.append_high_qc2(qc.clone()).await.unwrap();
        assert_eq!(
            storage.load_high_qc2().await.unwrap().unwrap().view_number,
            ViewNumber::new(6)
        );
    }

    #[rstest_reuse::apply(persistence_types)]
    pub async fn test_restart_view<P: TestablePersistence>(_p: PhantomData<P>) {
        let tmp = P::tmp_storage().await;
        let storage = P::connect(&tmp).await;

        // Initially, there is no saved view.
        assert_eq!(storage.load_restart_view().await.unwrap(), None);

        // Store a view.
        let view1 = ViewNumber::genesis();
        storage
            .record_action(view1, None, HotShotAction::Vote)
            .await
            .unwrap();
        assert_eq!(
            storage.load_restart_view().await.unwrap().unwrap(),
            view1 + 1
        );

        // Store a newer view, make sure storage gets updated.
        let view2 = view1 + 1;
        storage
            .record_action(view2, None, HotShotAction::Vote)
            .await
            .unwrap();
        assert_eq!(
            storage.load_restart_view().await.unwrap().unwrap(),
            view2 + 1
        );

        // Store an old view, make sure storage is unchanged.
        storage
            .record_action(view1, None, HotShotAction::Vote)
            .await
            .unwrap();
        assert_eq!(
            storage.load_restart_view().await.unwrap().unwrap(),
            view2 + 1
        );

        // store a higher proposed view, make sure storage is unchanged.
        storage
            .record_action(view2 + 1, None, HotShotAction::Propose)
            .await
            .unwrap();
        assert_eq!(
            storage.load_restart_view().await.unwrap().unwrap(),
            view2 + 1
        );

        // store a higher timeout vote view, make sure storage is unchanged.
        storage
            .record_action(view2 + 1, None, HotShotAction::TimeoutVote)
            .await
            .unwrap();
        assert_eq!(
            storage.load_restart_view().await.unwrap().unwrap(),
            view2 + 1
        );
    }

    #[rstest_reuse::apply(persistence_types)]
    pub async fn test_store_drb_input<P: TestablePersistence>(_p: PhantomData<P>) {
        use hotshot_types::drb::DrbInput;

        let tmp = P::tmp_storage().await;
        let storage = P::connect(&tmp).await;
        let difficulty_level = 10;

        // Initially, there is no saved info.
        if storage.load_drb_input(10).await.is_ok() {
            panic!("unexpected nonempty drb_input");
        }

        let drb_input_1 = DrbInput {
            epoch: 10,
            iteration: 10,
            value: [0u8; 32],
            difficulty_level,
        };

        let drb_input_2 = DrbInput {
            epoch: 10,
            iteration: 20,
            value: [0u8; 32],
            difficulty_level,
        };

        let drb_input_3 = DrbInput {
            epoch: 10,
            iteration: 30,
            value: [0u8; 32],
            difficulty_level,
        };

        let _ = storage.store_drb_input(drb_input_1.clone()).await;

        assert_eq!(storage.load_drb_input(10).await.unwrap(), drb_input_1);

        let _ = storage.store_drb_input(drb_input_3.clone()).await;

        // check that the drb input is overwritten
        assert_eq!(storage.load_drb_input(10).await.unwrap(), drb_input_3);

        let _ = storage.store_drb_input(drb_input_2.clone()).await;

        // check that the drb input is not overwritten by the older value
        assert_eq!(storage.load_drb_input(10).await.unwrap(), drb_input_3);
    }

    #[rstest_reuse::apply(persistence_types)]
    pub async fn test_epoch_info<P: TestablePersistence>(_p: PhantomData<P>) {
        let tmp = P::tmp_storage().await;
        let storage = P::connect(&tmp).await;

        // Initially, there is no saved info.
        assert_eq!(storage.load_start_epoch_info().await.unwrap(), Vec::new());

        // Store a drb result.
        storage
            .store_drb_result(EpochNumber::new(1), [1; 32])
            .await
            .unwrap();
        assert_eq!(
            storage.load_start_epoch_info().await.unwrap(),
            vec![InitializerEpochInfo::<SeqTypes> {
                epoch: EpochNumber::new(1),
                drb_result: [1; 32],
                block_header: None,
            }]
        );

        // Store a second DRB result
        storage
            .store_drb_result(EpochNumber::new(2), [3; 32])
            .await
            .unwrap();
        assert_eq!(
            storage.load_start_epoch_info().await.unwrap(),
            vec![
                InitializerEpochInfo::<SeqTypes> {
                    epoch: EpochNumber::new(1),
                    drb_result: [1; 32],
                    block_header: None,
                },
                InitializerEpochInfo::<SeqTypes> {
                    epoch: EpochNumber::new(2),
                    drb_result: [3; 32],
                    block_header: None,
                }
            ]
        );

        // Make a header
        let instance_state = NodeState::mock();
        let validated_state = hotshot_types::traits::ValidatedState::genesis(&instance_state).0;
        let leaf: Leaf2 = Leaf::genesis(&validated_state, &instance_state, MOCK_UPGRADE.base)
            .await
            .into();
        let header = leaf.block_header().clone();

        // Test storing the header
        storage
            .store_epoch_root(EpochNumber::new(1), header.clone())
            .await
            .unwrap();
        assert_eq!(
            storage.load_start_epoch_info().await.unwrap(),
            vec![
                InitializerEpochInfo::<SeqTypes> {
                    epoch: EpochNumber::new(1),
                    drb_result: [1; 32],
                    block_header: Some(header.clone()),
                },
                InitializerEpochInfo::<SeqTypes> {
                    epoch: EpochNumber::new(2),
                    drb_result: [3; 32],
                    block_header: None,
                }
            ]
        );

        // Store more than the limit
        let total_epochs = RECENT_STAKE_TABLES_LIMIT + 10;
        for i in 0..total_epochs {
            let epoch = EpochNumber::new(i);
            let drb = [i as u8; 32];
            storage
                .store_drb_result(epoch, drb)
                .await
                .unwrap_or_else(|_| panic!("Failed to store DRB result for epoch {i}"));
        }

        let results = storage.load_start_epoch_info().await.unwrap();

        // Check that only the most recent RECENT_STAKE_TABLES_LIMIT epochs are returned
        assert_eq!(
            results.len(),
            RECENT_STAKE_TABLES_LIMIT as usize,
            "Should return only the most recent {RECENT_STAKE_TABLES_LIMIT} epochs",
        );

        for (i, info) in results.iter().enumerate() {
            let expected_epoch =
                EpochNumber::new(total_epochs - RECENT_STAKE_TABLES_LIMIT + i as u64);
            let expected_drb = [(total_epochs - RECENT_STAKE_TABLES_LIMIT + i as u64) as u8; 32];
            assert_eq!(info.epoch, expected_epoch, "invalid epoch at index {i}",);
            assert_eq!(info.drb_result, expected_drb, "invalid DRB at index {i}",);
            assert!(info.block_header.is_none(), "Expected no block header");
        }
    }

    fn leaf_info(leaf: Leaf2) -> LeafInfo<SeqTypes> {
        LeafInfo {
            leaf,
            vid_share: None,
            state: Default::default(),
            delta: None,
            state_cert: None,
        }
    }

    #[rstest_reuse::apply(persistence_types)]
    pub async fn test_append_and_decide<P: TestablePersistence>(_p: PhantomData<P>) {
        let tmp = P::tmp_storage().await;
        let storage = P::connect(&tmp).await;

        // Test append VID
        assert_eq!(
            storage.load_vid_share(ViewNumber::new(0)).await.unwrap(),
            None
        );

        let leaf: Leaf2 = Leaf2::genesis(
            &ValidatedState::default(),
            &NodeState::mock(),
            TEST_VERSIONS.test.base,
        )
        .await;
        let leaf_payload = leaf.block_payload().unwrap();
        let leaf_payload_bytes_arc = leaf_payload.encode();

        let avidm_param = init_avidm_param(2).unwrap();
        let weights = vec![1u32; 2];

        let ns_table = parse_ns_table(
            leaf_payload.byte_len().as_usize(),
            &leaf_payload.ns_table().encode(),
        );
        let (payload_commitment, shares) =
            AvidMScheme::ns_disperse(&avidm_param, &weights, &leaf_payload_bytes_arc, ns_table)
                .unwrap();

        let (pubkey, privkey) = BLSPubKey::generated_from_seed_indexed([0; 32], 1);
        let signature = PubKey::sign(&privkey, &[]).unwrap();
        let mut vid = AvidMDisperseShare::<SeqTypes> {
            view_number: ViewNumber::new(0),
            payload_commitment,
            share: shares[0].clone(),
            recipient_key: pubkey,
            epoch: Some(EpochNumber::new(0)),
            target_epoch: Some(EpochNumber::new(0)),
            common: avidm_param,
        };
        let mut quorum_proposal = Proposal {
            data: QuorumProposalWrapper::<SeqTypes> {
                proposal: QuorumProposal2::<SeqTypes> {
                    epoch: None,
                    block_header: leaf.block_header().clone(),
                    view_number: ViewNumber::genesis(),
                    justify_qc: QuorumCertificate2::genesis(
                        &ValidatedState::default(),
                        &NodeState::mock(),
                        TEST_VERSIONS.test,
                    )
                    .await,
                    upgrade_certificate: None,
                    view_change_evidence: None,
                    next_drb_result: None,
                    next_epoch_justify_qc: None,
                    state_cert: None,
                },
            },
            signature,
            _pd: Default::default(),
        };

        let vid_share0 = convert_proposal(vid.clone().to_proposal(&privkey).unwrap().clone());

        storage.append_vid(&vid_share0).await.unwrap();

        assert_eq!(
            storage.load_vid_share(ViewNumber::new(0)).await.unwrap(),
            Some(vid_share0.clone())
        );

        vid.view_number = ViewNumber::new(1);

        let vid_share1 = convert_proposal(vid.clone().to_proposal(&privkey).unwrap().clone());
        storage.append_vid(&vid_share1).await.unwrap();

        assert_eq!(
            storage.load_vid_share(vid.view_number()).await.unwrap(),
            Some(vid_share1.clone())
        );

        vid.view_number = ViewNumber::new(2);

        let vid_share2 = convert_proposal(vid.clone().to_proposal(&privkey).unwrap().clone());
        storage.append_vid(&vid_share2).await.unwrap();

        assert_eq!(
            storage.load_vid_share(vid.view_number()).await.unwrap(),
            Some(vid_share2.clone())
        );

        vid.view_number = ViewNumber::new(3);

        let vid_share3 = convert_proposal(vid.clone().to_proposal(&privkey).unwrap().clone());
        storage.append_vid(&vid_share3).await.unwrap();

        assert_eq!(
            storage.load_vid_share(vid.view_number()).await.unwrap(),
            Some(vid_share3.clone())
        );

        let block_payload_signature = BLSPubKey::sign(&privkey, &leaf_payload_bytes_arc)
            .expect("Failed to sign block payload");

        let da_proposal_inner = DaProposal2::<SeqTypes> {
            encoded_transactions: leaf_payload_bytes_arc.clone(),
            metadata: leaf_payload.ns_table().clone(),
            view_number: ViewNumber::new(0),
            epoch: None,
            epoch_transition_indicator: EpochTransitionIndicator::NotInTransition,
        };

        let da_proposal = Proposal {
            data: da_proposal_inner,
            signature: block_payload_signature,
            _pd: Default::default(),
        };

        let vid_commitment = vid_commitment(
            &leaf_payload_bytes_arc,
            &leaf.block_header().metadata().encode(),
            2,
            TEST_VERSIONS.test.base,
        );

        storage
            .append_da2(&da_proposal, vid_commitment)
            .await
            .unwrap();

        assert_eq!(
            storage.load_da_proposal(ViewNumber::new(0)).await.unwrap(),
            Some(da_proposal.clone())
        );

        let mut da_proposal1 = da_proposal.clone();
        da_proposal1.data.view_number = ViewNumber::new(1);
        storage
            .append_da2(&da_proposal1.clone(), vid_commitment)
            .await
            .unwrap();

        assert_eq!(
            storage
                .load_da_proposal(da_proposal1.data.view_number)
                .await
                .unwrap(),
            Some(da_proposal1.clone())
        );

        let mut da_proposal2 = da_proposal1.clone();
        da_proposal2.data.view_number = ViewNumber::new(2);
        storage
            .append_da2(&da_proposal2.clone(), vid_commitment)
            .await
            .unwrap();

        assert_eq!(
            storage
                .load_da_proposal(da_proposal2.data.view_number)
                .await
                .unwrap(),
            Some(da_proposal2.clone())
        );

        let mut da_proposal3 = da_proposal2.clone();
        da_proposal3.data.view_number = ViewNumber::new(3);
        storage
            .append_da2(&da_proposal3.clone(), vid_commitment)
            .await
            .unwrap();

        assert_eq!(
            storage
                .load_da_proposal(da_proposal3.data.view_number)
                .await
                .unwrap(),
            Some(da_proposal3.clone())
        );

        let quorum_proposal1 = quorum_proposal.clone();

        storage
            .append_quorum_proposal2(&quorum_proposal1)
            .await
            .unwrap();

        assert_eq!(
            storage.load_quorum_proposals().await.unwrap(),
            BTreeMap::from_iter([(ViewNumber::genesis(), quorum_proposal1.clone())])
        );

        quorum_proposal.data.proposal.view_number = ViewNumber::new(1);
        let quorum_proposal2 = quorum_proposal.clone();
        storage
            .append_quorum_proposal2(&quorum_proposal2)
            .await
            .unwrap();

        assert_eq!(
            storage.load_quorum_proposals().await.unwrap(),
            BTreeMap::from_iter([
                (ViewNumber::genesis(), quorum_proposal1.clone()),
                (ViewNumber::new(1), quorum_proposal2.clone())
            ])
        );

        quorum_proposal.data.proposal.view_number = ViewNumber::new(2);
        quorum_proposal.data.proposal.justify_qc.view_number = ViewNumber::new(1);
        let quorum_proposal3 = quorum_proposal.clone();
        storage
            .append_quorum_proposal2(&quorum_proposal3)
            .await
            .unwrap();

        assert_eq!(
            storage.load_quorum_proposals().await.unwrap(),
            BTreeMap::from_iter([
                (ViewNumber::genesis(), quorum_proposal1.clone()),
                (ViewNumber::new(1), quorum_proposal2.clone()),
                (ViewNumber::new(2), quorum_proposal3.clone())
            ])
        );

        quorum_proposal.data.proposal.view_number = ViewNumber::new(3);
        quorum_proposal.data.proposal.justify_qc.view_number = ViewNumber::new(2);

        // This one should stick around after GC runs.
        let quorum_proposal4 = quorum_proposal.clone();
        storage
            .append_quorum_proposal2(&quorum_proposal4)
            .await
            .unwrap();

        assert_eq!(
            storage.load_quorum_proposals().await.unwrap(),
            BTreeMap::from_iter([
                (ViewNumber::genesis(), quorum_proposal1.clone()),
                (ViewNumber::new(1), quorum_proposal2.clone()),
                (ViewNumber::new(2), quorum_proposal3.clone()),
                (ViewNumber::new(3), quorum_proposal4.clone())
            ])
        );

        // Test decide and garbage collection. Pass in a leaf chain with no VID shares or payloads,
        // so we have to fetch the missing data from storage.
        let leaves = [
            Leaf2::from_quorum_proposal(&quorum_proposal1.data),
            Leaf2::from_quorum_proposal(&quorum_proposal2.data),
            Leaf2::from_quorum_proposal(&quorum_proposal3.data),
            Leaf2::from_quorum_proposal(&quorum_proposal4.data),
        ];
        let mut final_qc = leaves[3].justify_qc();
        final_qc.view_number += 1;
        final_qc.data.leaf_commit = Committable::commit(&leaf);
        let qcs = [
            CertificatePair::non_epoch_change(leaves[1].justify_qc()),
            CertificatePair::non_epoch_change(leaves[2].justify_qc()),
            CertificatePair::non_epoch_change(leaves[3].justify_qc()),
            CertificatePair::non_epoch_change(final_qc),
        ];

        assert_eq!(
            storage.load_anchor_view().await.unwrap(),
            ViewNumber::genesis()
        );

        let consumer = EventCollector::default();
        let leaf_chain = leaves
            .iter()
            .take(3)
            .map(|leaf| leaf_info(leaf.clone()))
            .zip(&qcs)
            .collect::<Vec<_>>();
        tracing::info!(?leaf_chain, "decide view 2");
        storage
            .append_decided_leaves(
                ViewNumber::new(2),
                leaf_chain.iter().map(|(leaf, qc)| (leaf, (*qc).clone())),
                None,
                &consumer,
            )
            .await
            .unwrap();
        assert_eq!(
            storage.load_anchor_view().await.unwrap(),
            ViewNumber::new(2)
        );

        for i in 0..=2 {
            assert_eq!(
                storage.load_da_proposal(ViewNumber::new(i)).await.unwrap(),
                None
            );

            assert_eq!(
                storage.load_vid_share(ViewNumber::new(i)).await.unwrap(),
                None
            );
        }

        assert_eq!(
            storage.load_da_proposal(ViewNumber::new(3)).await.unwrap(),
            Some(da_proposal3)
        );

        assert_eq!(
            storage.load_vid_share(ViewNumber::new(3)).await.unwrap(),
            Some(convert_proposal(vid_share3.clone()))
        );

        let proposals = storage.load_quorum_proposals().await.unwrap();
        assert_eq!(
            proposals,
            BTreeMap::from_iter([(ViewNumber::new(3), quorum_proposal4)])
        );

        // A decide event should have been processed.
        for (leaf, info) in leaves.iter().zip(consumer.leaf_chain().await.iter()) {
            assert_eq!(info.leaf, *leaf);
            let decided_vid_share = info.vid_share.as_ref().unwrap();
            assert_eq!(decided_vid_share.view_number(), leaf.view_number());
        }

        // The decided leaf should not have been garbage collected.
        assert_eq!(
            storage.load_anchor_leaf().await.unwrap(),
            Some((leaves[2].clone(), qcs[2].clone()))
        );
        assert_eq!(
            storage.load_anchor_view().await.unwrap(),
            leaves[2].view_number()
        );

        // Process a second decide event.
        let consumer = EventCollector::default();
        tracing::info!(leaf = ?leaves[3], qc = ?qcs[3], "decide view 3");
        storage
            .append_decided_leaves(
                ViewNumber::new(3),
                vec![(&leaf_info(leaves[3].clone()), qcs[3].clone())],
                None,
                &consumer,
            )
            .await
            .unwrap();
        assert_eq!(
            storage.load_anchor_view().await.unwrap(),
            ViewNumber::new(3)
        );

        // A decide event should have been processed.
        let events = consumer.events.read().await;
        assert_eq!(events.len(), 1);
        assert_eq!(events[0].view_number, ViewNumber::new(3));
        let EventType::Decide {
            committing_qc,
            leaf_chain,
            ..
        } = &events[0].event
        else {
            panic!("expected decide event, got {:?}", events[0]);
        };
        assert_eq!(**committing_qc, qcs[3]);
        assert_eq!(leaf_chain.len(), 1);
        let info = &leaf_chain[0];
        assert_eq!(info.leaf, leaves[3]);

        // The remaining data should have been GCed.
        assert_eq!(
            storage.load_da_proposal(ViewNumber::new(3)).await.unwrap(),
            None
        );

        assert_eq!(
            storage.load_vid_share(ViewNumber::new(3)).await.unwrap(),
            None
        );
        assert_eq!(
            storage.load_quorum_proposals().await.unwrap(),
            BTreeMap::new()
        );
    }

    #[rstest_reuse::apply(persistence_types)]
    pub async fn test_upgrade_certificate<P: TestablePersistence>(_p: PhantomData<P>) {
        let tmp = P::tmp_storage().await;
        let storage = P::connect(&tmp).await;

        // Test get upgrade certificate
        assert_eq!(storage.load_upgrade_certificate().await.unwrap(), None);

        let upgrade_data = UpgradeProposalData {
            old_version: Version { major: 0, minor: 1 },
            new_version: Version { major: 1, minor: 0 },
            decide_by: ViewNumber::genesis(),
            new_version_hash: Default::default(),
            old_version_last_view: ViewNumber::genesis(),
            new_version_first_view: ViewNumber::genesis(),
        };

        let decide_upgrade_certificate = UpgradeCertificate::<SeqTypes>::new(
            upgrade_data.clone(),
            upgrade_data.commit(),
            ViewNumber::genesis(),
            Default::default(),
            Default::default(),
        );
        let res = storage
            .store_upgrade_certificate(Some(decide_upgrade_certificate.clone()))
            .await;
        assert!(res.is_ok());

        let res = storage.load_upgrade_certificate().await.unwrap();
        let view_number = res.unwrap().view_number;
        assert_eq!(view_number, ViewNumber::genesis());

        let new_view_number_for_certificate = ViewNumber::new(50);
        let mut new_upgrade_certificate = decide_upgrade_certificate.clone();
        new_upgrade_certificate.view_number = new_view_number_for_certificate;

        let res = storage
            .store_upgrade_certificate(Some(new_upgrade_certificate.clone()))
            .await;
        assert!(res.is_ok());

        let res = storage.load_upgrade_certificate().await.unwrap();
        let view_number = res.unwrap().view_number;
        assert_eq!(view_number, new_view_number_for_certificate);
    }

    #[rstest_reuse::apply(persistence_types)]
    pub async fn test_next_epoch_quorum_certificate<P: TestablePersistence>(_p: PhantomData<P>) {
        let tmp = P::tmp_storage().await;
        let storage = P::connect(&tmp).await;

        //  test that next epoch qc2 does not exist
        assert_eq!(
            storage.load_next_epoch_quorum_certificate().await.unwrap(),
            None
        );

        let upgrade_lock = UpgradeLock::<SeqTypes>::new(TEST_VERSIONS.test);

        let genesis_view = ViewNumber::genesis();

        let leaf = Leaf2::genesis(
            &ValidatedState::default(),
            &NodeState::default(),
            TEST_VERSIONS.test.base,
        )
        .await;
        let data: NextEpochQuorumData2<SeqTypes> = QuorumData2 {
            leaf_commit: leaf.commit(),
            epoch: Some(EpochNumber::new(1)),
            block_number: Some(leaf.height()),
        }
        .into();

        let versioned_data =
            VersionedVoteData::new_infallible(data.clone(), genesis_view, &upgrade_lock);

        let bytes: [u8; 32] = versioned_data.commit().into();

        let next_epoch_qc = NextEpochQuorumCertificate2::new(
            data,
            Commitment::from_raw(bytes),
            genesis_view,
            None,
            PhantomData,
        );

        let res = storage
            .store_next_epoch_quorum_certificate(next_epoch_qc.clone())
            .await;
        assert!(res.is_ok());

        let res = storage.load_next_epoch_quorum_certificate().await.unwrap();
        let view_number = res.unwrap().view_number;
        assert_eq!(view_number, ViewNumber::genesis());

        let new_view_number_for_qc = ViewNumber::new(50);
        let mut new_qc = next_epoch_qc.clone();
        new_qc.view_number = new_view_number_for_qc;

        let res = storage
            .store_next_epoch_quorum_certificate(new_qc.clone())
            .await;
        assert!(res.is_ok());

        let res = storage.load_next_epoch_quorum_certificate().await.unwrap();
        let view_number = res.unwrap().view_number;
        assert_eq!(view_number, new_view_number_for_qc);
    }

    #[rstest_reuse::apply(persistence_types)]
    pub async fn test_decide_with_failing_event_consumer<P: TestablePersistence>(
        _p: PhantomData<P>,
    ) {
        let tmp = P::tmp_storage().await;
        let storage = P::connect(&tmp).await;

        // Create a short blockchain.
        let mut chain = vec![];

        let leaf: Leaf2 = Leaf::genesis(
            &ValidatedState::default(),
            &NodeState::mock(),
            MOCK_UPGRADE.base,
        )
        .await
        .into();
        let leaf_payload = leaf.block_payload().unwrap();
        let leaf_payload_bytes_arc = leaf_payload.encode();
        let avidm_param = init_avidm_param(2).unwrap();
        let weights = vec![1u32; 2];
        let ns_table = parse_ns_table(
            leaf_payload.byte_len().as_usize(),
            &leaf_payload.ns_table().encode(),
        );
        let (payload_commitment, shares) =
            AvidMScheme::ns_disperse(&avidm_param, &weights, &leaf_payload_bytes_arc, ns_table)
                .unwrap();

        let (pubkey, privkey) = BLSPubKey::generated_from_seed_indexed([0; 32], 1);
        let mut vid = AvidMDisperseShare::<SeqTypes> {
            view_number: ViewNumber::new(0),
            payload_commitment,
            share: shares[0].clone(),
            recipient_key: pubkey,
            epoch: Some(EpochNumber::new(0)),
            target_epoch: Some(EpochNumber::new(0)),
            common: avidm_param,
        }
        .to_proposal(&privkey)
        .unwrap()
        .clone();
        let mut quorum_proposal = QuorumProposalWrapper::<SeqTypes> {
            proposal: QuorumProposal2::<SeqTypes> {
                block_header: leaf.block_header().clone(),
                view_number: ViewNumber::genesis(),
                justify_qc: QuorumCertificate::genesis(
                    &ValidatedState::default(),
                    &NodeState::mock(),
                    TEST_VERSIONS.test,
                )
                .await
                .to_qc2(),
                upgrade_certificate: None,
                view_change_evidence: None,
                next_drb_result: None,
                next_epoch_justify_qc: None,
                epoch: None,
                state_cert: None,
            },
        };
        let mut qc = QuorumCertificate2::genesis(
            &ValidatedState::default(),
            &NodeState::mock(),
            TEST_VERSIONS.test,
        )
        .await;

        let block_payload_signature = BLSPubKey::sign(&privkey, &leaf_payload_bytes_arc)
            .expect("Failed to sign block payload");
        let mut da_proposal = Proposal {
            data: DaProposal2::<SeqTypes> {
                encoded_transactions: leaf_payload_bytes_arc.clone(),
                metadata: leaf_payload.ns_table().clone(),
                view_number: ViewNumber::new(0),
                epoch: Some(EpochNumber::new(0)),
                epoch_transition_indicator: EpochTransitionIndicator::NotInTransition,
            },
            signature: block_payload_signature,
            _pd: Default::default(),
        };

        let vid_commitment = vid_commitment(
            &leaf_payload_bytes_arc,
            &leaf.block_header().metadata().encode(),
            2,
            TEST_VERSIONS.test.base,
        );

        for i in 0..4 {
            quorum_proposal.proposal.view_number = ViewNumber::new(i);
            let leaf = Leaf2::from_quorum_proposal(&quorum_proposal);
            qc.view_number = leaf.view_number();
            qc.data.leaf_commit = Committable::commit(&leaf);
            vid.data.view_number = leaf.view_number();
            da_proposal.data.view_number = leaf.view_number();
            chain.push((leaf.clone(), qc.clone(), vid.clone(), da_proposal.clone()));
        }

        // Add proposals.
        for (_, _, vid, da) in &chain {
            tracing::info!(?da, ?vid, "insert proposal");
            storage.append_da2(da, vid_commitment).await.unwrap();
            storage
                .append_vid(&convert_proposal(vid.clone()))
                .await
                .unwrap();
        }

        // Decide 2 leaves, but fail in event processing.
        let leaf_chain = chain
            .iter()
            .take(2)
            .map(|(leaf, qc, ..)| (leaf_info(leaf.clone()), qc.clone()))
            .collect::<Vec<_>>();
        tracing::info!("decide with event handling failure");
        storage
            .append_decided_leaves(
                ViewNumber::new(1),
                leaf_chain
                    .iter()
                    .map(|(leaf, qc)| (leaf, CertificatePair::non_epoch_change(qc.clone()))),
                None,
                &FailConsumer,
            )
            .await
            .unwrap();
        // No garbage collection should have run.
        for i in 0..4 {
            tracing::info!(i, "check proposal availability");
            assert!(
                storage
                    .load_vid_share(ViewNumber::new(i))
                    .await
                    .unwrap()
                    .is_some()
            );
            assert!(
                storage
                    .load_da_proposal(ViewNumber::new(i))
                    .await
                    .unwrap()
                    .is_some()
            );
        }
        tracing::info!("check anchor leaf updated");
        assert_eq!(
            storage
                .load_anchor_leaf()
                .await
                .unwrap()
                .unwrap()
                .0
                .view_number(),
            ViewNumber::new(1)
        );
        assert_eq!(
            storage.load_anchor_view().await.unwrap(),
            ViewNumber::new(1)
        );

        // Now decide remaining leaves successfully. We should now garbage collect and process a
        // decide event for all the leaves.
        let consumer = EventCollector::default();
        let leaf_chain = chain
            .iter()
            .skip(2)
            .map(|(leaf, qc, ..)| (leaf_info(leaf.clone()), qc.clone()))
            .collect::<Vec<_>>();
        tracing::info!("decide successfully");
        storage
            .append_decided_leaves(
                ViewNumber::new(3),
                leaf_chain
                    .iter()
                    .map(|(leaf, qc)| (leaf, CertificatePair::non_epoch_change(qc.clone()))),
                None,
                &consumer,
            )
            .await
            .unwrap();
        // Garbage collection should have run.
        for i in 0..4 {
            tracing::info!(i, "check proposal garbage collected");
            assert!(
                storage
                    .load_vid_share(ViewNumber::new(i))
                    .await
                    .unwrap()
                    .is_none()
            );
            assert!(
                storage
                    .load_da_proposal(ViewNumber::new(i))
                    .await
                    .unwrap()
                    .is_none()
            );
        }
        tracing::info!("check anchor leaf updated");
        assert_eq!(
            storage
                .load_anchor_leaf()
                .await
                .unwrap()
                .unwrap()
                .0
                .view_number(),
            ViewNumber::new(3)
        );
        assert_eq!(
            storage.load_anchor_view().await.unwrap(),
            ViewNumber::new(3)
        );

        // Check decide event.
        tracing::info!("check decide event");
        let leaf_chain = consumer.leaf_chain().await;
        assert_eq!(leaf_chain.len(), 4, "{leaf_chain:#?}");
        for ((leaf, ..), info) in chain.iter().zip(leaf_chain.iter()) {
            assert_eq!(info.leaf, *leaf);
            let decided_vid_share = info.vid_share.as_ref().unwrap();
            assert_eq!(decided_vid_share.view_number(), leaf.view_number());
            assert!(info.leaf.block_payload().is_some());
        }
    }

    /// Splitting a decide into `persist_decided_leaves` (persist only, no events/GC) and a later
    /// `process_decided_events` loses no data: multiple persists can accumulate, and one process
    /// pass at the latest view drains the whole backlog and runs GC.
    #[rstest_reuse::apply(persistence_types)]
    pub async fn test_deferred_decide_processing<P: TestablePersistence>(_p: PhantomData<P>) {
        let tmp = P::tmp_storage().await;
        let storage = P::connect(&tmp).await;

        // Create a short blockchain (same setup as `test_decide_with_failing_event_consumer`).
        let mut chain = vec![];

        let leaf: Leaf2 = Leaf::genesis(
            &ValidatedState::default(),
            &NodeState::mock(),
            MOCK_UPGRADE.base,
        )
        .await
        .into();
        let leaf_payload = leaf.block_payload().unwrap();
        let leaf_payload_bytes_arc = leaf_payload.encode();
        let avidm_param = init_avidm_param(2).unwrap();
        let weights = vec![1u32; 2];
        let ns_table = parse_ns_table(
            leaf_payload.byte_len().as_usize(),
            &leaf_payload.ns_table().encode(),
        );
        let (payload_commitment, shares) =
            AvidMScheme::ns_disperse(&avidm_param, &weights, &leaf_payload_bytes_arc, ns_table)
                .unwrap();

        let (pubkey, privkey) = BLSPubKey::generated_from_seed_indexed([0; 32], 1);
        let mut vid = AvidMDisperseShare::<SeqTypes> {
            view_number: ViewNumber::new(0),
            payload_commitment,
            share: shares[0].clone(),
            recipient_key: pubkey,
            epoch: Some(EpochNumber::new(0)),
            target_epoch: Some(EpochNumber::new(0)),
            common: avidm_param,
        }
        .to_proposal(&privkey)
        .unwrap()
        .clone();
        let mut quorum_proposal = QuorumProposalWrapper::<SeqTypes> {
            proposal: QuorumProposal2::<SeqTypes> {
                block_header: leaf.block_header().clone(),
                view_number: ViewNumber::genesis(),
                justify_qc: QuorumCertificate::genesis(
                    &ValidatedState::default(),
                    &NodeState::mock(),
                    TEST_VERSIONS.test,
                )
                .await
                .to_qc2(),
                upgrade_certificate: None,
                view_change_evidence: None,
                next_drb_result: None,
                next_epoch_justify_qc: None,
                epoch: None,
                state_cert: None,
            },
        };
        let mut qc = QuorumCertificate2::genesis(
            &ValidatedState::default(),
            &NodeState::mock(),
            TEST_VERSIONS.test,
        )
        .await;

        let block_payload_signature = BLSPubKey::sign(&privkey, &leaf_payload_bytes_arc)
            .expect("Failed to sign block payload");
        let mut da_proposal = Proposal {
            data: DaProposal2::<SeqTypes> {
                encoded_transactions: leaf_payload_bytes_arc.clone(),
                metadata: leaf_payload.ns_table().clone(),
                view_number: ViewNumber::new(0),
                epoch: Some(EpochNumber::new(0)),
                epoch_transition_indicator: EpochTransitionIndicator::NotInTransition,
            },
            signature: block_payload_signature,
            _pd: Default::default(),
        };

        let vid_commitment = vid_commitment(
            &leaf_payload_bytes_arc,
            &leaf.block_header().metadata().encode(),
            2,
            TEST_VERSIONS.test.base,
        );

        for i in 0..4 {
            quorum_proposal.proposal.view_number = ViewNumber::new(i);
            let leaf = Leaf2::from_quorum_proposal(&quorum_proposal);
            qc.view_number = leaf.view_number();
            qc.data.leaf_commit = Committable::commit(&leaf);
            vid.data.view_number = leaf.view_number();
            da_proposal.data.view_number = leaf.view_number();
            chain.push((leaf.clone(), qc.clone(), vid.clone(), da_proposal.clone()));
        }

        // Per-view artifacts, as persisted at receive time (before any decide).
        for (_, _, vid, da) in &chain {
            storage.append_da2(da, vid_commitment).await.unwrap();
            storage
                .append_vid(&convert_proposal(vid.clone()))
                .await
                .unwrap();
        }

        let consumer = EventCollector::default();

        // Two decides with no processing in between (a lagging/coalescing processor): {0,1}, {2,3}.
        for (from, to, view) in [(0usize, 2usize, 1u64), (2, 4, 3)] {
            let leaf_chain = chain
                .iter()
                .take(to)
                .skip(from)
                .map(|(leaf, qc, ..)| (leaf_info(leaf.clone()), qc.clone()))
                .collect::<Vec<_>>();
            storage
                .persist_decided_leaves(
                    ViewNumber::new(view),
                    leaf_chain
                        .iter()
                        .map(|(leaf, qc)| (leaf, CertificatePair::non_epoch_change(qc.clone()))),
                    None,
                    &consumer,
                )
                .await
                .unwrap();
        }

        // Persist-only: leaves saved (anchor advanced), but no events emitted and no GC.
        assert_eq!(
            storage.load_anchor_view().await.unwrap(),
            ViewNumber::new(3),
            "anchor view should reflect the latest persisted leaf"
        );
        assert!(
            consumer.leaf_chain().await.is_empty(),
            "persist_decided_leaves must not emit decide events"
        );
        for i in 0..4 {
            assert!(
                storage
                    .load_vid_share(ViewNumber::new(i))
                    .await
                    .unwrap()
                    .is_some(),
                "persist_decided_leaves must not garbage collect VID shares"
            );
            assert!(
                storage
                    .load_da_proposal(ViewNumber::new(i))
                    .await
                    .unwrap()
                    .is_some(),
                "persist_decided_leaves must not garbage collect DA proposals"
            );
        }

        // A failing consumer propagates the error and leaves the cursor un-advanced: nothing is
        // GC'd and the range is retried below.
        storage
            .process_decided_events(ViewNumber::new(3), None, &FailConsumer)
            .await
            .unwrap_err();
        for i in 0..4 {
            assert!(
                storage
                    .load_da_proposal(ViewNumber::new(i))
                    .await
                    .unwrap()
                    .is_some(),
                "a failed process pass must not garbage collect anything"
            );
        }

        // One process pass at the latest view drains the whole backlog, runs GC, and reports the
        // cursor it advanced to.
        let processed = storage
            .process_decided_events(ViewNumber::new(3), None, &consumer)
            .await
            .unwrap();
        assert_eq!(
            processed,
            Some(ViewNumber::new(3)),
            "process_decided_events should report the highest processed view"
        );

        // All four leaves delivered, with payloads and VID shares reconstructed from storage.
        let leaf_chain = consumer.leaf_chain().await;
        assert_eq!(leaf_chain.len(), 4, "{leaf_chain:#?}");
        for ((leaf, ..), info) in chain.iter().zip(leaf_chain.iter()) {
            assert_eq!(info.leaf, *leaf);
            let decided_vid_share = info.vid_share.as_ref().unwrap();
            assert_eq!(decided_vid_share.view_number(), leaf.view_number());
            assert!(info.leaf.block_payload().is_some());
        }

        // GC ran for the processed range.
        for i in 0..4 {
            assert!(
                storage
                    .load_vid_share(ViewNumber::new(i))
                    .await
                    .unwrap()
                    .is_none(),
                "process_decided_events should have garbage collected VID shares"
            );
            assert!(
                storage
                    .load_da_proposal(ViewNumber::new(i))
                    .await
                    .unwrap()
                    .is_none(),
                "process_decided_events should have garbage collected DA proposals"
            );
        }

        // Re-processing with nothing new is a no-op.
        let consumer2 = EventCollector::default();
        storage
            .process_decided_events(ViewNumber::new(3), None, &consumer2)
            .await
            .unwrap();
        assert!(
            consumer2.leaf_chain().await.is_empty(),
            "re-processing an already-drained backlog must emit no events"
        );
        assert_eq!(
            storage.load_anchor_view().await.unwrap(),
            ViewNumber::new(3)
        );
    }

    /// A chain of `n` V6 mock leaves at views `0..n` with real, consecutive block heights (the
    /// other mock chains in this module reuse the genesis header, so every leaf has height 0).
    async fn consecutive_height_chain(n: u64) -> Vec<(Leaf2, QuorumCertificate2<SeqTypes>)> {
        let node_state = NodeState::mock().with_genesis_version(versions::NEW_PROTOCOL_VERSION);
        let genesis_leaf = Leaf2::genesis(
            &ValidatedState::default(),
            &node_state,
            TEST_VERSIONS.test.base,
        )
        .await;
        let mut quorum_proposal = QuorumProposalWrapper::<SeqTypes> {
            proposal: QuorumProposal2::<SeqTypes> {
                block_header: genesis_leaf.block_header().clone(),
                view_number: ViewNumber::genesis(),
                justify_qc: QuorumCertificate2::genesis(
                    &ValidatedState::default(),
                    &node_state,
                    TEST_VERSIONS.test,
                )
                .await,
                upgrade_certificate: None,
                view_change_evidence: None,
                next_drb_result: None,
                next_epoch_justify_qc: None,
                epoch: None,
                state_cert: None,
            },
        };
        let mut qc = QuorumCertificate2::genesis(
            &ValidatedState::default(),
            &node_state,
            TEST_VERSIONS.test,
        )
        .await;

        let mut chain = vec![];
        for i in 0..n {
            quorum_proposal.proposal.view_number = ViewNumber::new(i);
            *quorum_proposal.proposal.block_header.height_mut() = i;
            let leaf = Leaf2::from_quorum_proposal(&quorum_proposal);
            qc.view_number = leaf.view_number();
            qc.data.leaf_commit = Committable::commit(&leaf);
            chain.push((leaf, qc.clone()));
        }
        chain
    }

    /// Decide the leaves of `chain` at indices (== views) `range`.
    async fn decide_range<P: TestablePersistence>(
        storage: &P,
        chain: &[(Leaf2, QuorumCertificate2<SeqTypes>)],
        range: std::ops::Range<usize>,
        consumer: &(impl EventConsumer + 'static),
    ) {
        let decided_view = ViewNumber::new(range.end as u64 - 1);
        let leaf_chain = chain[range]
            .iter()
            .map(|(leaf, qc)| (leaf_info(leaf.clone()), qc.clone()))
            .collect::<Vec<_>>();
        storage
            .append_decided_leaves(
                decided_view,
                leaf_chain
                    .iter()
                    .map(|(leaf, qc)| (leaf, CertificatePair::non_epoch_change(qc.clone()))),
                None,
                consumer,
            )
            .await
            .unwrap();
    }

    /// Records the views delivered in decide events, in delivery order.
    #[derive(Clone, Debug, Default)]
    struct DecideViewCollector {
        views: Arc<RwLock<Vec<u64>>>,
    }

    #[async_trait]
    impl EventConsumer for DecideViewCollector {
        async fn handle_event(&self, event: &CoordinatorEvent<SeqTypes>) -> anyhow::Result<()> {
            let mut views = self.views.write().await;
            match event {
                // Decide events carry their leaves newest first.
                CoordinatorEvent::NewDecide { leaf_infos, .. } => views.extend(
                    leaf_infos
                        .iter()
                        .rev()
                        .map(|info| info.leaf.view_number().u64()),
                ),
                CoordinatorEvent::LegacyEvent(Event {
                    event: EventType::Decide { leaf_chain, .. },
                    ..
                }) => views.extend(
                    leaf_chain
                        .iter()
                        .rev()
                        .map(|info| info.leaf.view_number().u64()),
                ),
                _ => {},
            }
            Ok(())
        }
    }

    async fn received_views(consumer: &DecideViewCollector) -> Vec<u64> {
        consumer.views.read().await.clone()
    }

    /// Until a gap-fill decide arrives, event processing holds its cursor at the gap: nothing
    /// past it is delivered, and the pending leaves are neither skipped nor dropped.
    ///
    /// sql-only: the fs backend keeps the pop-oldest cursor and skips gaps (the consumer
    /// re-fetches missing blocks), trading gap-fill delivery for cheap decide passes.
    #[rstest::rstest]
    #[case(PhantomData::<crate::persistence::sql::Persistence>)]
    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    pub async fn test_decide_gap_holds_events<P: TestablePersistence>(#[case] _p: PhantomData<P>) {
        let tmp = P::tmp_storage().await;
        let storage = P::connect(&tmp).await;
        let consumer = DecideViewCollector::default();
        let chain = consecutive_height_chain(5).await;

        decide_range(&storage, &chain, 0..2, &consumer).await;
        assert_eq!(received_views(&consumer).await, vec![0, 1]);

        // Views 3..=4 decide while view 2 is still pending.
        decide_range(&storage, &chain, 3..5, &consumer).await;
        assert_eq!(
            received_views(&consumer).await,
            vec![0, 1],
            "leaves past a height gap must not be delivered before the gap is filled"
        );

        // Nothing was lost: the pending leaves are still in storage...
        assert_eq!(
            storage
                .load_anchor_leaf()
                .await
                .unwrap()
                .unwrap()
                .0
                .view_number(),
            ViewNumber::new(4)
        );

        // ...and re-processing neither skips the gap nor emits duplicates.
        storage
            .process_decided_events(ViewNumber::new(4), None, &consumer)
            .await
            .unwrap();
        assert_eq!(received_views(&consumer).await, vec![0, 1]);
    }

    /// Once the gap-fill decide arrives, the consumer receives the gap leaf and everything held
    /// up behind it, in order, exactly once each.
    ///
    /// sql-only: see [`test_decide_gap_holds_events`].
    #[rstest::rstest]
    #[case(PhantomData::<crate::persistence::sql::Persistence>)]
    #[test_log::test(tokio::test(flavor = "multi_thread"))]
    pub async fn test_decide_gap_fill_delivers_all<P: TestablePersistence>(
        #[case] _p: PhantomData<P>,
    ) {
        let tmp = P::tmp_storage().await;
        let storage = P::connect(&tmp).await;
        let consumer = DecideViewCollector::default();
        let chain = consecutive_height_chain(5).await;

        decide_range(&storage, &chain, 0..2, &consumer).await;
        decide_range(&storage, &chain, 3..5, &consumer).await;
        assert_eq!(received_views(&consumer).await, vec![0, 1]);

        // The gap-fill decide of view 2 arrives.
        decide_range(&storage, &chain, 2..3, &consumer).await;
        // A re-process pass at the watermark must not skip or duplicate anything.
        storage
            .process_decided_events(ViewNumber::new(4), None, &consumer)
            .await
            .unwrap();

        assert_eq!(
            received_views(&consumer).await,
            vec![0, 1, 2, 3, 4],
            "after the gap fills, every leaf is delivered in order, exactly once"
        );
    }

    #[rstest_reuse::apply(persistence_types)]
    pub async fn test_pruning<P: TestablePersistence>(_p: PhantomData<P>) {
        let tmp = P::tmp_storage().await;

        let mut options = P::options(&tmp);
        options.set_view_retention(1);
        let storage = options.create().await.unwrap();

        // Add some "old" data, from view 0.
        let leaf = Leaf::genesis(
            &ValidatedState::default(),
            &NodeState::mock(),
            MOCK_UPGRADE.base,
        )
        .await;
        let leaf_payload = leaf.block_payload().unwrap();
        let leaf_payload_bytes_arc = leaf_payload.encode();
        let avidm_param = init_avidm_param(2).unwrap();
        let weights = vec![1u32; 2];

        let ns_table = parse_ns_table(
            leaf_payload.byte_len().as_usize(),
            &leaf_payload.ns_table().encode(),
        );
        let (payload_commitment, shares) =
            AvidMScheme::ns_disperse(&avidm_param, &weights, &leaf_payload_bytes_arc, ns_table)
                .unwrap();

        let (pubkey, privkey) = BLSPubKey::generated_from_seed_indexed([0; 32], 1);
        let vid_share = convert_proposal(
            AvidMDisperseShare::<SeqTypes> {
                view_number: ViewNumber::new(0),
                payload_commitment,
                share: shares[0].clone(),
                recipient_key: pubkey,
                epoch: None,
                target_epoch: None,
                common: avidm_param,
            }
            .to_proposal(&privkey)
            .unwrap()
            .clone(),
        );

        let quorum_proposal = QuorumProposalWrapper::<SeqTypes> {
            proposal: QuorumProposal2::<SeqTypes> {
                block_header: leaf.block_header().clone(),
                view_number: ViewNumber::genesis(),
                justify_qc: QuorumCertificate::genesis(
                    &ValidatedState::default(),
                    &NodeState::mock(),
                    TEST_VERSIONS.test,
                )
                .await
                .to_qc2(),
                upgrade_certificate: None,
                view_change_evidence: None,
                next_drb_result: None,
                next_epoch_justify_qc: None,
                epoch: None,
                state_cert: None,
            },
        };
        let quorum_proposal_signature =
            BLSPubKey::sign(&privkey, &bincode::serialize(&quorum_proposal).unwrap())
                .expect("Failed to sign quorum proposal");
        let quorum_proposal = Proposal {
            data: quorum_proposal,
            signature: quorum_proposal_signature,
            _pd: Default::default(),
        };

        let block_payload_signature = BLSPubKey::sign(&privkey, &leaf_payload_bytes_arc)
            .expect("Failed to sign block payload");
        let da_proposal = Proposal {
            data: DaProposal2::<SeqTypes> {
                encoded_transactions: leaf_payload_bytes_arc,
                metadata: leaf_payload.ns_table().clone(),
                view_number: ViewNumber::new(0),
                epoch: None,
                epoch_transition_indicator: EpochTransitionIndicator::NotInTransition,
            },
            signature: block_payload_signature,
            _pd: Default::default(),
        };

        let leaf2: Leaf2 = leaf.clone().into();
        let vote_data = Vote2Data {
            leaf_commit: leaf2.commit(),
            epoch: EpochNumber::new(0),
            block_number: leaf2.height(),
        };
        let cert2 = Certificate2::new(
            vote_data.clone(),
            vote_data.commit(),
            ViewNumber::new(0),
            None,
            PhantomData,
        );

        storage
            .append_da2(&da_proposal, VidCommitment::V1(payload_commitment))
            .await
            .unwrap();
        storage.append_vid(&vid_share).await.unwrap();
        storage
            .append_quorum_proposal2(&quorum_proposal)
            .await
            .unwrap();
        storage
            .append_cert2(ViewNumber::new(0), cert2)
            .await
            .unwrap();

        // Decide a newer view, view 1.
        storage
            .append_decided_leaves(ViewNumber::new(1), [], None, &NullEventConsumer)
            .await
            .unwrap();

        // The old data is not more than the retention period (1 view) old, so it should not be
        // GCed.
        assert_eq!(
            storage
                .load_da_proposal(ViewNumber::new(0))
                .await
                .unwrap()
                .unwrap(),
            da_proposal
        );
        assert_eq!(
            storage
                .load_vid_share(ViewNumber::new(0))
                .await
                .unwrap()
                .unwrap(),
            vid_share
        );
        assert_eq!(
            storage
                .load_quorum_proposal(ViewNumber::new(0))
                .await
                .unwrap(),
            quorum_proposal
        );
        assert!(
            storage
                .load_cert2(ViewNumber::new(0))
                .await
                .unwrap()
                .is_some()
        );

        // Decide an even newer view, triggering GC of the old data.
        storage
            .append_decided_leaves(ViewNumber::new(2), [], None, &NullEventConsumer)
            .await
            .unwrap();
        assert!(
            storage
                .load_da_proposal(ViewNumber::new(0))
                .await
                .unwrap()
                .is_none()
        );
        assert!(
            storage
                .load_vid_share(ViewNumber::new(0))
                .await
                .unwrap()
                .is_none()
        );
        assert!(
            storage
                .load_quorum_proposal(ViewNumber::new(0))
                .await
                .is_err()
        );
        assert!(
            storage
                .load_cert2(ViewNumber::new(0))
                .await
                .unwrap()
                .is_none()
        );
    }

    async fn assert_events_eq<P: TestablePersistence>(
        persistence: &P,
        block: u64,
        stake_table_fetcher: &Fetcher,
        l1_client: &L1Client,
        stake_table_contract: Address,
    ) -> anyhow::Result<()> {
        // Load persisted events
        let (stored_l1, events) = persistence.load_events(0, block).await?;
        assert!(!events.is_empty());
        assert!(stored_l1.is_some());
        assert!(events.iter().all(|((l1_block, _), _)| *l1_block <= block));
        // Fetch events directly from the contract and compare with persisted data
        let contract_events = Fetcher::fetch_events_from_contract(
            l1_client.clone(),
            stake_table_contract,
            None,
            block,
        )
        .await?;
        assert_eq!(
            contract_events, events,
            "Events from contract and persistence do not match"
        );

        // Fetch events from stake table fetcher and compare with persisted data
        let fetched_events = stake_table_fetcher
            .fetch_and_store_stake_table_events(stake_table_contract, block)
            .await?;
        assert_eq!(fetched_events, events);

        Ok(())
    }

    // test for validating stake table event fetching from persistence,
    // ensuring that persisted data matches the on-chain events and that event fetcher work correctly.
    #[rstest_reuse::apply(persistence_types)]
    pub async fn test_stake_table_fetching_from_persistence<P: TestablePersistence>(
        #[values(
            StakeTableContractVersion::V1,
            StakeTableContractVersion::V2,
            StakeTableContractVersion::V3
        )]
        stake_table_version: StakeTableContractVersion,
        _p: PhantomData<P>,
    ) -> anyhow::Result<()> {
        let epoch_height = 20;

        let network_config = TestConfigBuilder::default()
            .epoch_height(epoch_height)
            .build();

        let anvil_provider = network_config.anvil().unwrap();

        let query_service_port =
            reserve_tcp_port().expect("OS should have ephemeral ports available");
        let query_api_options = Options::with_port(query_service_port);

        const NODE_COUNT: usize = 2;

        let storage = join_all((0..NODE_COUNT).map(|_| P::tmp_storage())).await;
        let persistence_options: [_; NODE_COUNT] = storage
            .iter()
            .map(P::options)
            .collect::<Vec<_>>()
            .try_into()
            .unwrap();

        let persistence = persistence_options[0].clone().create().await.unwrap();

        // Build the config with PoS hook
        let l1_url = network_config.l1_url();

        let upgrade = Upgrade::trivial(version(0, 3));

        let testnet_config = TestNetworkConfigBuilder::with_num_nodes()
            .api_config(query_api_options)
            .network_config(network_config.clone())
            .persistences(persistence_options.clone())
            .pos_hook(
                DelegationConfig::MultipleDelegators,
                stake_table_version,
                upgrade,
            )
            .await
            .expect("Pos deployment failed")
            .build();

        //start the network
        let test_network = TestNetwork::new(testnet_config, upgrade).await;

        let client: Client<ServerError, SequencerApiVersion> = Client::new(
            format!("http://localhost:{query_service_port}")
                .parse()
                .unwrap(),
        );
        client.connect(None).await;
        tracing::info!(query_service_port, "server running");

        // wait until we enter in epoch 3
        let _initial_blocks = client
            .socket("availability/stream/blocks/0")
            .subscribe::<BlockQueryData<SeqTypes>>()
            .await
            .unwrap()
            .take(40)
            .try_collect::<Vec<_>>()
            .await
            .unwrap();
        // Load initial persisted events and validate they exist.
        let membership_coordinator = test_network
            .server
            .consensus_handle()
            .membership_coordinator()
            .await;

        let l1_client = L1Client::new(vec![l1_url]).unwrap();
        let node_state = test_network.server.node_state();
        let chain_config = node_state.chain_config;
        let stake_table_contract = chain_config.stake_table_contract.unwrap();

        let current_membership = membership_coordinator.membership();
        {
            let membership_state = current_membership;
            let stake_table_fetcher = membership_state.fetcher();

            let block1 = anvil_provider
                .get_block_number()
                .await
                .expect("latest l1 block");

            assert_events_eq(
                &persistence,
                block1,
                stake_table_fetcher,
                &l1_client,
                stake_table_contract,
            )
            .await?;
        }
        let _epoch_4_blocks = client
            .socket("availability/stream/blocks/0")
            .subscribe::<BlockQueryData<SeqTypes>>()
            .await
            .unwrap()
            .take(65)
            .try_collect::<Vec<_>>()
            .await
            .unwrap();
        let block2 = anvil_provider
            .get_block_number()
            .await
            .expect("latest l1 block");

        {
            let membership_state = current_membership;
            let stake_table_fetcher = membership_state.fetcher();

            assert_events_eq(
                &persistence,
                block2,
                stake_table_fetcher,
                &l1_client,
                stake_table_contract,
            )
            .await?;
        }
        Ok(())
    }

    #[rstest_reuse::apply(persistence_types)]
    pub async fn test_stake_table_background_fetching<P: TestablePersistence>(
        #[values(
            StakeTableContractVersion::V1,
            StakeTableContractVersion::V2,
            StakeTableContractVersion::V3
        )]
        stake_table_version: StakeTableContractVersion,
        _p: PhantomData<P>,
    ) -> anyhow::Result<()> {
        use espresso_types::v0_3::ChainConfig;
        use hotshot_contract_adapter::stake_table::StakeTableContractVersion;

        let blocks_per_epoch = 10;

        let network_config = TestConfigBuilder::<1>::default()
            .epoch_height(blocks_per_epoch)
            .build();

        let anvil_provider = network_config.anvil().unwrap();

        let (genesis_state, genesis_stake) = light_client_genesis_from_stake_table(
            &network_config.hotshot_config().hotshot_stake_table(),
            STAKE_TABLE_CAPACITY_FOR_TEST,
        )
        .unwrap();

        let (_, priv_keys): (Vec<_>, Vec<_>) = (0..20)
            .map(|i| <PubKey as SignatureKey>::generated_from_seed_indexed([1; 32], i as u64))
            .unzip();
        let state_key_pairs = (0..20)
            .map(|i| StateKeyPair::generate_from_seed_indexed([2; 32], i as u64))
            .collect::<Vec<_>>();

        let validators = staking_priv_keys(&priv_keys, &state_key_pairs, &[], 20);

        let deployer = ProviderBuilder::new()
            .wallet(EthereumWallet::from(network_config.signer().clone()))
            .connect_http(network_config.l1_url().clone());

        let mut contracts = Contracts::new();
        let args = DeployerArgsBuilder::default()
            .deployer(deployer.clone())
            .rpc_url(network_config.l1_url().clone())
            .mock_light_client(true)
            .genesis_lc_state(genesis_state)
            .genesis_st_state(genesis_stake)
            .blocks_per_epoch(blocks_per_epoch)
            .epoch_start_block(1)
            .exit_escrow_period(U256::from(max(
                blocks_per_epoch * 15 + 100,
                DEFAULT_EXIT_ESCROW_PERIOD_SECONDS,
            )))
            .multisig_pauser(network_config.signer().address())
            .token_name("Espresso".to_string())
            .token_symbol("ESP".to_string())
            .initial_token_supply(U256::from(3590000000u64))
            .ops_timelock_delay(U256::from(0))
            .ops_timelock_admin(network_config.signer().address())
            .ops_timelock_proposers(vec![network_config.signer().address()])
            .ops_timelock_executors(vec![network_config.signer().address()])
            .safe_exit_timelock_delay(U256::from(10))
            .safe_exit_timelock_admin(network_config.signer().address())
            .safe_exit_timelock_proposers(vec![network_config.signer().address()])
            .safe_exit_timelock_executors(vec![network_config.signer().address()])
            .build()
            .unwrap();

        match stake_table_version {
            StakeTableContractVersion::V1 => args.deploy_to_stake_table_v1(&mut contracts).await,
            StakeTableContractVersion::V2 => args.deploy_to_stake_table_v2(&mut contracts).await,
            StakeTableContractVersion::V3 => args.deploy_to_stake_table_v3(&mut contracts).await,
        }
        .expect("contracts deployed");

        let st_addr = contracts
            .address(Contract::StakeTableProxy)
            .expect("StakeTableProxy deployed");
        let l1_url = network_config.l1_url().clone();

        let mut planned_txns = StakingTransactions::create(
            l1_url.clone(),
            &deployer,
            st_addr,
            validators,
            None,
            DelegationConfig::MultipleDelegators,
        )
        .await
        .expect("stake table setup failed");

        planned_txns
            .apply_prerequisites()
            .await
            .expect("prerequisites failed");

        // Ensure we have at least one stake table affecting transaction
        planned_txns.apply_one().await.expect("send tx failed");

        // new block every 1s
        anvil_provider
            .anvil_set_interval_mining(1)
            .await
            .expect("interval mining");

        // spawn a separate task
        // this is going to keep registering validators and multiple delegators
        // the interval mining is set to 1s so each transaction finalization would take atleast 1s
        spawn({
            async move {
                {
                    while let Some(receipt) =
                        planned_txns.apply_one().await.expect("send tx failed")
                    {
                        tracing::debug!(?receipt, "transaction finalized");
                    }
                }
            }
        });

        let storage = P::tmp_storage().await;
        let persistence = P::options(&storage).create().await.unwrap();

        let l1_client = L1ClientOptions {
            stake_table_update_interval: Duration::from_secs(7),
            l1_retry_delay: Duration::from_millis(10),
            l1_events_max_block_range: 10000,
            ..Default::default()
        }
        .connect(vec![l1_url])
        .unwrap();
        l1_client.spawn_tasks().await;

        let fetcher = Fetcher::new(
            Arc::new(NullStateCatchup::default()),
            Arc::new(Mutex::new(persistence.clone())),
            l1_client.clone(),
            ChainConfig {
                stake_table_contract: Some(st_addr),
                base_fee: 0.into(),
                ..Default::default()
            },
        );

        // sleep so that we have enough events
        sleep(Duration::from_secs(20)).await;

        fetcher.spawn_update_loop().await;
        let mut prev_l1_block = 0;
        let mut prev_events_len = 0;
        for _i in 0..10 {
            // Wait for more than update interval to assert that persistence was updated
            // L1 update interval is 7s in this test

            tokio::time::sleep(std::time::Duration::from_secs(8)).await;

            let block = anvil_provider
                .get_block_number()
                .await
                .expect("latest l1 block");

            let (read_offset, persisted_events) = persistence.load_events(0, block).await?;
            let read_offset = read_offset.unwrap();
            let l1_block = match read_offset {
                EventsPersistenceRead::Complete => block,
                EventsPersistenceRead::UntilL1Block(block) => block,
            };

            tracing::info!("{l1_block:?}, persistence events = {persisted_events:?}.");
            assert!(persisted_events.len() > prev_events_len);

            assert!(l1_block > prev_l1_block, "events not updated");

            let contract_events =
                Fetcher::fetch_events_from_contract(l1_client.clone(), st_addr, None, l1_block)
                    .await?;
            assert_eq!(persisted_events, contract_events);

            prev_l1_block = l1_block;
            prev_events_len = persisted_events.len();
        }

        Ok(())
    }

    #[rstest_reuse::apply(persistence_types)]
    pub async fn test_membership_persistence<P: TestablePersistence>(
        _p: PhantomData<P>,
    ) -> anyhow::Result<()> {
        let tmp = P::tmp_storage().await;
        let mut opt = P::options(&tmp);

        let storage = opt.create().await.unwrap();

        let validator = AuthenticatedValidator::mock();
        let mut st = IndexMap::new();
        st.insert(validator.account, validator);

        storage
            .store_stake(EpochNumber::new(10), st.clone(), None, None)
            .await?;

        let (table, ..) = storage.load_stake(EpochNumber::new(10)).await?.unwrap();
        assert_eq!(st, table);

        let val2 = AuthenticatedValidator::mock();
        let mut st2 = IndexMap::new();
        st2.insert(val2.account, val2);
        storage
            .store_stake(EpochNumber::new(11), st2.clone(), None, None)
            .await?;

        let tables = storage.load_latest_stake(4).await?.unwrap();
        let mut iter = tables.iter();
        assert_eq!(
            Some(&(EpochNumber::new(11), (st2.clone(), None), None)),
            iter.next()
        );
        assert_eq!(Some(&(EpochNumber::new(10), (st, None), None)), iter.next());
        assert_eq!(None, iter.next());

        for i in 0..=20 {
            storage
                .store_stake(EpochNumber::new(i), st2.clone(), None, None)
                .await?;
        }

        let tables = storage.load_latest_stake(5).await?.unwrap();
        let mut iter = tables.iter();
        assert_eq!(
            Some(&(EpochNumber::new(20), (st2.clone(), None), None)),
            iter.next()
        );
        assert_eq!(
            Some(&(EpochNumber::new(19), (st2.clone(), None), None)),
            iter.next()
        );
        assert_eq!(
            Some(&(EpochNumber::new(18), (st2.clone(), None), None)),
            iter.next()
        );
        assert_eq!(
            Some(&(EpochNumber::new(17), (st2.clone(), None), None)),
            iter.next()
        );
        assert_eq!(
            Some(&(EpochNumber::new(16), (st2, None), None)),
            iter.next()
        );
        assert_eq!(None, iter.next());

        Ok(())
    }

    #[rstest_reuse::apply(persistence_types)]
    pub async fn test_delete_stake_tables<P: TestablePersistence>(
        _p: PhantomData<P>,
    ) -> anyhow::Result<()> {
        let tmp = P::tmp_storage().await;
        let mut opt = P::options(&tmp);
        let storage = opt.create().await.unwrap();

        let event1 = StakeTableEvent::Delegate(Delegated {
            delegator: Address::ZERO,
            validator: Address::ZERO,
            amount: U256::from(100),
        });
        let event2 = StakeTableEvent::Delegate(Delegated {
            delegator: Address::ZERO,
            validator: Address::ZERO,
            amount: U256::from(200),
        });

        let l1_block = 42u64;
        let events: Vec<(EventKey, StakeTableEvent)> =
            vec![((l1_block, 0), event1), ((l1_block, 1), event2)];

        storage.store_events(l1_block, events.clone()).await?;

        let (read_offset, loaded_events) = storage.load_events(0_u64, l1_block).await?;
        assert!(read_offset.is_some());
        assert_eq!(loaded_events.len(), 2);
        assert_eq!(loaded_events, events);

        // Store some validators
        let v = RegisteredValidator::mock();
        let mut vmap = IndexMap::new();
        vmap.insert(v.account, v);
        storage
            .store_all_validators(EpochNumber::new(1), vmap.clone())
            .await?;

        let loaded = storage
            .load_all_validators(EpochNumber::new(1), 0, 10)
            .await?;
        assert_eq!(loaded.len(), 1);

        storage.delete_stake_tables().await?;

        // Events cleared
        let (read_offset, loaded_events) = storage.load_events(0_u64, l1_block).await?;
        assert!(read_offset.is_none());
        assert!(loaded_events.is_empty());

        // Validators cleared
        let loaded = storage
            .load_all_validators(EpochNumber::new(1), 0, 10)
            .await
            .unwrap_or_default();
        assert!(loaded.is_empty());

        let event3 = StakeTableEvent::Delegate(Delegated {
            delegator: Address::ZERO,
            validator: Address::ZERO,
            amount: U256::from(300),
        });
        let new_events: Vec<(EventKey, StakeTableEvent)> = vec![((l1_block, 0), event3)];
        storage.store_events(l1_block, new_events.clone()).await?;

        let (read_offset, loaded_events) = storage.load_events(0_u64, l1_block).await?;
        assert!(read_offset.is_some());
        assert_eq!(loaded_events.len(), 1);
        assert_eq!(loaded_events, new_events);

        Ok(())
    }

    #[rstest_reuse::apply(persistence_types)]
    pub async fn test_store_and_load_all_validators<P: TestablePersistence>(
        _p: PhantomData<P>,
    ) -> anyhow::Result<()> {
        let tmp = P::tmp_storage().await;
        let mut opt = P::options(&tmp);
        let storage = opt.create().await.unwrap();

        let mut vmap1 = IndexMap::new();
        for _i in 0..25 {
            let v = RegisteredValidator::mock();
            vmap1.insert(v.account, v);
        }
        storage
            .store_all_validators(EpochNumber::new(10), vmap1.clone())
            .await?;

        let mut expected_all: Vec<_> = vmap1.clone().into_values().collect();
        expected_all.sort_by_key(|v| v.account);

        // Load all
        let loaded_all = storage
            .load_all_validators(EpochNumber::new(10), 0, 100)
            .await?;
        // SQLite returns a different ordered list even though there is an `ORDER BY address ASC` clause
        assert_eq!(expected_all, loaded_all);

        // Load first 10
        let loaded_first_10 = storage
            .load_all_validators(EpochNumber::new(10), 0, 10)
            .await?;

        assert_eq!(expected_all[..10], loaded_first_10);

        // Load next 10
        let loaded_next_10 = storage
            .load_all_validators(EpochNumber::new(10), 10, 10)
            .await?;

        assert_eq!(expected_all[10..20], loaded_next_10);

        // Load remaining 5
        let loaded_last_5 = storage
            .load_all_validators(EpochNumber::new(10), 20, 10)
            .await?;

        assert_eq!(expected_all[20..], loaded_last_5);

        // offset beyond size should return empty
        let loaded_empty = storage
            .load_all_validators(EpochNumber::new(10), 100, 10)
            .await?;
        assert!(loaded_empty.is_empty());

        // epoch 11
        let validator2 = RegisteredValidator::mock();
        let mut vmap2 = IndexMap::new();
        vmap2.insert(validator2.account, validator2.clone());

        storage
            .store_all_validators(EpochNumber::new(11), vmap2.clone())
            .await?;

        let mut expected_epoch11: Vec<_> = vmap2.clone().into_values().collect();
        expected_epoch11.sort_by_key(|v| v.account);

        let loaded2 = storage
            .load_all_validators(EpochNumber::new(11), 0, 100)
            .await?;

        assert_eq!(expected_epoch11, loaded2);

        // Epoch 10 still there
        let loaded1_again = storage
            .load_all_validators(EpochNumber::new(10), 0, 100)
            .await?;

        assert_eq!(expected_all, loaded1_again);

        Ok(())
    }

    #[rstest_reuse::apply(persistence_types)]
    pub async fn test_non_consecutive_decide<P: TestablePersistence>(_p: PhantomData<P>) {
        let tmp = P::tmp_storage().await;
        let storage = P::connect(&tmp).await;

        let genesis_leaf: Leaf2 = Leaf2::genesis(
            &ValidatedState::default(),
            &NodeState::mock(),
            TEST_VERSIONS.test.base,
        )
        .await;
        let mut quorum_proposal = QuorumProposalWrapper::<SeqTypes> {
            proposal: QuorumProposal2::<SeqTypes> {
                epoch: None,
                block_header: genesis_leaf.block_header().clone(),
                view_number: genesis_leaf.view_number(),
                justify_qc: QuorumCertificate2::genesis(
                    &ValidatedState::default(),
                    &NodeState::mock(),
                    TEST_VERSIONS.test,
                )
                .await,
                upgrade_certificate: None,
                view_change_evidence: None,
                next_drb_result: None,
                next_epoch_justify_qc: None,
                state_cert: None,
            },
        };

        let leaf0 = Leaf2::from_quorum_proposal(&quorum_proposal);

        quorum_proposal.proposal.view_number = ViewNumber::new(2);
        *quorum_proposal.proposal.block_header.height_mut() = 2;
        quorum_proposal.proposal.justify_qc.view_number = ViewNumber::new(1);
        let leaf2 = Leaf2::from_quorum_proposal(&quorum_proposal);

        let mut qc0 = leaf0.justify_qc();
        qc0.data.leaf_commit = Committable::commit(&leaf0);

        let mut qc2 = leaf2.justify_qc();
        qc2.view_number += 1;
        qc2.data.leaf_commit = Committable::commit(&leaf2);

        let mut deciding_qc = qc2.clone();
        deciding_qc.view_number += 1;

        // Decide the first leaf, but fail to generate a decide event.
        storage
            .append_decided_leaves(
                ViewNumber::new(0),
                [(
                    &leaf_info(leaf0.clone()),
                    CertificatePair::non_epoch_change(qc0),
                )],
                None,
                &FailConsumer,
            )
            .await
            .unwrap();

        // Later, decide a new leaf, but skipping some leaf in the middle. This should generate
        // decide events for both the leaves, correctly separating into two events since the leaves
        // are non-consecutive, and correctly applying `deciding_qc` only to the last event.
        let consumer = EventCollector::default();
        storage
            .append_decided_leaves(
                ViewNumber::new(2),
                [(
                    &leaf_info(leaf2.clone()),
                    CertificatePair::non_epoch_change(qc2),
                )],
                Some(Arc::new(CertificatePair::non_epoch_change(
                    deciding_qc.clone(),
                ))),
                &consumer,
            )
            .await
            .unwrap();

        let events = consumer.events.read().await;
        assert_eq!(events.len(), 2);

        let EventType::Decide {
            leaf_chain: leaf_chain0,
            deciding_qc: deciding_qc0,
            ..
        } = &events[0].event
        else {
            panic!("expected decide event, got {:?}", events[0].event);
        };
        assert_eq!(leaf_chain0.len(), 1);
        assert_eq!(leaf_chain0[0].leaf, leaf0);
        assert_eq!(*deciding_qc0, None);

        let EventType::Decide {
            leaf_chain: leaf_chain2,
            deciding_qc: deciding_qc2,
            ..
        } = &events[1].event
        else {
            panic!("expected decide event, got {:?}", events[1].event);
        };
        assert_eq!(leaf_chain2.len(), 1);
        assert_eq!(leaf_chain2[0].leaf, leaf2);
        assert_eq!(deciding_qc2.as_ref().unwrap().qc(), &deciding_qc);
    }
}