espresso_node/
persistence.rs

1//! Sequencer node persistence.
2//!
3//! This module implements the persistence required for a sequencer node to rejoin the network and
4//! resume participating in consensus, in the event that its process crashes or is killed and loses
5//! all in-memory state.
6//!
7//! This is distinct from the query service persistent storage found in the `api` module, which is
8//! an extension that node operators can opt into. This module defines the minimum level of
9//! persistence which is _required_ to run a node.
10
11use std::collections::HashMap;
12
13use alloy::primitives::{Address, U256};
14use anyhow::Context;
15use async_trait::async_trait;
16use espresso_types::{
17    PubKey,
18    v0_3::{ChainConfig, RegisteredValidator},
19};
20
21pub mod fs;
22pub mod no_storage;
23mod persistence_metrics;
24/// Persistence backed by a SQL database, with one module per backend.
25///
26/// [`postgres`](sql::postgres) and [`sqlite`](sql::sqlite) compile the same code against their own
27/// database. [`sql::Options::embedded_db`] picks one, and the options are shared by both.
28pub mod sql {
29    mod options;
30
31    pub use options::*;
32
33    pub mod postgres;
34    pub mod sqlite;
35}
36pub(crate) mod storage_probe;
37
38/// RegisteredValidator without x25519_key/p2p_addr fields.
39/// Used for migrating data written before x25519 support was added.
40#[derive(serde::Serialize, serde::Deserialize)]
41pub(crate) struct RegisteredValidatorNoX25519 {
42    pub account: Address,
43    pub stake_table_key: PubKey,
44    pub state_ver_key: hotshot_types::light_client::StateVerKey,
45    pub stake: U256,
46    pub commission: u16,
47    pub delegators: HashMap<Address, U256>,
48    pub authenticated: bool,
49}
50
51impl RegisteredValidatorNoX25519 {
52    pub fn migrate(self) -> RegisteredValidator<PubKey> {
53        RegisteredValidator {
54            account: self.account,
55            stake_table_key: Some(self.stake_table_key),
56            state_ver_key: Some(self.state_ver_key),
57            stake: self.stake,
58            commission: self.commission,
59            delegators: self.delegators,
60            authenticated: self.authenticated,
61            x25519_key: None,
62            p2p_addr: None,
63        }
64    }
65}
66
67// TODO: replace with a proper data migration if it is ever deemed worthwhile.
68// Field order must match `RegisteredValidator<KEY>` in v0_3::stake_table exactly (bincode is order-sensitive).
69#[derive(serde::Serialize, serde::Deserialize)]
70pub(crate) struct RegisteredValidatorPreOption {
71    pub account: Address,
72    pub stake_table_key: PubKey,
73    pub state_ver_key: hotshot_types::light_client::StateVerKey,
74    pub stake: U256,
75    pub commission: u16,
76    pub delegators: HashMap<Address, U256>,
77    pub authenticated: bool,
78    pub x25519_key: Option<hotshot_types::x25519::PublicKey>,
79    pub p2p_addr: Option<hotshot_types::addr::NetAddr>,
80}
81
82impl RegisteredValidatorPreOption {
83    pub fn migrate(self) -> RegisteredValidator<PubKey> {
84        RegisteredValidator {
85            account: self.account,
86            stake_table_key: Some(self.stake_table_key),
87            state_ver_key: Some(self.state_ver_key),
88            stake: self.stake,
89            commission: self.commission,
90            delegators: self.delegators,
91            authenticated: self.authenticated,
92            x25519_key: self.x25519_key,
93            p2p_addr: self.p2p_addr,
94        }
95    }
96}
97
98// TODO: replace with a proper data migration if it is ever deemed worthwhile.
99// Layout written after BLS fix (381b0f9207) but before Schnorr Option fix.
100// stake_table_key is Option<KEY>, state_ver_key is raw StateVerKey.
101// Field order must match exactly (bincode is order-sensitive).
102#[derive(serde::Serialize, serde::Deserialize)]
103pub(crate) struct RegisteredValidatorPreSchnorrOption {
104    pub account: Address,
105    pub stake_table_key: Option<PubKey>,
106    pub state_ver_key: hotshot_types::light_client::StateVerKey,
107    pub stake: U256,
108    pub commission: u16,
109    pub delegators: HashMap<Address, U256>,
110    pub authenticated: bool,
111    pub x25519_key: Option<hotshot_types::x25519::PublicKey>,
112    pub p2p_addr: Option<hotshot_types::addr::NetAddr>,
113}
114
115impl RegisteredValidatorPreSchnorrOption {
116    pub fn migrate(self) -> RegisteredValidator<PubKey> {
117        RegisteredValidator {
118            account: self.account,
119            stake_table_key: self.stake_table_key,
120            state_ver_key: Some(self.state_ver_key),
121            stake: self.stake,
122            commission: self.commission,
123            delegators: self.delegators,
124            authenticated: self.authenticated,
125            x25519_key: self.x25519_key,
126            p2p_addr: self.p2p_addr,
127        }
128    }
129}
130
131/// Update a `NetworkConfig` that may have originally been persisted with an old version.
132fn migrate_network_config(
133    mut network_config: serde_json::Value,
134) -> anyhow::Result<serde_json::Value> {
135    let config = network_config
136        .get_mut("config")
137        .context("missing field `config`")?
138        .as_object_mut()
139        .context("`config` must be an object")?;
140
141    if !config.contains_key("builder_urls") {
142        // When multi-builder support was added, the configuration field `builder_url: Url` was
143        // replaced by an array `builder_urls: Vec<Url>`. If the saved config has no `builder_urls`
144        // field, it is older than this change. Populate `builder_urls` with a singleton array
145        // formed from the old value of `builder_url`, and delete the no longer used `builder_url`.
146        let url = config
147            .remove("builder_url")
148            .context("missing field `builder_url`")?;
149        config.insert("builder_urls".into(), vec![url].into());
150    }
151
152    // HotShotConfig was upgraded to include parameters for proposing and voting on upgrades.
153    // Configs which were persisted before this upgrade may be missing these parameters. This
154    // migration initializes them with a default. By default, we use JS MAX_SAFE_INTEGER for the
155    // start parameters so that nodes will never do an upgrade, unless explicitly configured
156    // otherwise.
157    if !config.contains_key("start_proposing_view") {
158        config.insert("start_proposing_view".into(), 9007199254740991u64.into());
159    }
160    if !config.contains_key("stop_proposing_view") {
161        config.insert("stop_proposing_view".into(), 0.into());
162    }
163    if !config.contains_key("start_voting_view") {
164        config.insert("start_voting_view".into(), 9007199254740991u64.into());
165    }
166    if !config.contains_key("stop_voting_view") {
167        config.insert("stop_voting_view".into(), 0.into());
168    }
169    if !config.contains_key("start_proposing_time") {
170        config.insert("start_proposing_time".into(), 9007199254740991u64.into());
171    }
172    if !config.contains_key("stop_proposing_time") {
173        config.insert("stop_proposing_time".into(), 0.into());
174    }
175    if !config.contains_key("start_voting_time") {
176        config.insert("start_voting_time".into(), 9007199254740991u64.into());
177    }
178    if !config.contains_key("stop_voting_time") {
179        config.insert("stop_voting_time".into(), 0.into());
180    }
181
182    // HotShotConfig was upgraded to include an `epoch_height` parameter. Initialize with a default
183    // if missing.
184    if !config.contains_key("epoch_height") {
185        config.insert("epoch_height".into(), 0.into());
186    }
187
188    // HotShotConfig was upgraded to include `drb_difficulty` and `drb_upgrade_difficulty` parameters. Initialize with a default
189    // if missing.
190    if !config.contains_key("drb_difficulty") {
191        config.insert("drb_difficulty".into(), 0.into());
192    }
193    if !config.contains_key("drb_upgrade_difficulty") {
194        config.insert("drb_upgrade_difficulty".into(), 0.into());
195    }
196
197    // HotShotConfig was upgraded to include `da_committeees`. Initialize with an empty `da_committees` if missing.
198    if !config.contains_key("da_committees") {
199        config.insert("da_committees".into(), serde_json::json!([]));
200    }
201
202    Ok(network_config)
203}
204
205#[async_trait]
206pub trait ChainConfigPersistence: Sized + Send + Sync {
207    async fn insert_chain_config(&mut self, chain_config: ChainConfig) -> anyhow::Result<()>;
208}
209
210#[cfg(test)]
211mod tests {
212    use std::{cmp::max, collections::BTreeMap, marker::PhantomData, sync::Arc, time::Duration};
213
214    use alloy::{
215        network::EthereumWallet,
216        primitives::{Address, U256},
217        providers::{Provider, ProviderBuilder, ext::AnvilApi},
218    };
219    use anyhow::{bail, ensure};
220    use async_lock::{Mutex, RwLock};
221    use async_trait::async_trait;
222    use committable::{Commitment, Committable};
223    use espresso_contract_deployer::{
224        Contract, Contracts, DEFAULT_EXIT_ESCROW_PERIOD_SECONDS, builder::DeployerArgsBuilder,
225        network_config::light_client_genesis_from_stake_table,
226    };
227    use espresso_types::{
228        L1Client, L1ClientOptions, Leaf, Leaf2, NodeState, PubKey, SeqTypes, TEST_UPGRADE,
229        ValidatedState,
230        traits::{
231            EventConsumer, EventsPersistenceRead, MembershipPersistence, NullEventConsumer,
232            PersistenceOptions, SequencerPersistence,
233        },
234        v0_3::{AuthenticatedValidator, EventKey, Fetcher, RegisteredValidator, StakeTableEvent},
235    };
236    use futures::{StreamExt, TryStreamExt, future::join_all};
237    use hotshot_contract_adapter::{
238        sol_types::StakeTableV3::Delegated, stake_table::StakeTableContractVersion,
239    };
240    use hotshot_example_types::node_types::TEST_VERSIONS;
241    use hotshot_new_protocol::message::Certificate2;
242    use hotshot_query_service::{availability::BlockQueryData, testing::mocks::MOCK_UPGRADE};
243    use hotshot_types::{
244        data::{
245            DaProposal2, EpochNumber, QuorumProposal2, QuorumProposalWrapper, VidCommitment,
246            ViewNumber,
247            ns_table::parse_ns_table,
248            vid_commitment,
249            vid_disperse::{AvidMDisperseShare, AvidmGf2DisperseShare},
250        },
251        event::{HotShotAction, LeafInfo},
252        initializer::InitializerEpochInfo,
253        light_client::{LightClientState, StakeTableState, StateKeyPair},
254        message::{Proposal, UpgradeLock, convert_proposal},
255        new_protocol::CoordinatorEvent,
256        signature_key::BLSPubKey,
257        simple_certificate::{
258            CertificatePair, LightClientStateUpdateCertificateV2, NextEpochQuorumCertificate2,
259            QuorumCertificate, QuorumCertificate2, UpgradeCertificate,
260        },
261        simple_vote::{
262            NextEpochQuorumData2, QuorumData2, UpgradeProposalData, VersionedVoteData, Vote2Data,
263        },
264        traits::{EncodeBytes, block_contents::BlockHeader, signature_key::SignatureKey},
265        utils::EpochTransitionIndicator,
266        vid::{
267            avidm::{AvidMScheme, init_avidm_param},
268            avidm_gf2::{AvidmGf2Scheme, init_avidm_gf2_param},
269        },
270        vote::HasViewNumber,
271    };
272    use http_client::{Client, error::ClientErr};
273    use indexmap::IndexMap;
274    use staking_cli::demo::{DelegationConfig, StakingTransactions};
275    use test_utils::reserve_tcp_port;
276    use tokio::{spawn, time::sleep};
277    use vbs::version::Version;
278    use versions::Upgrade;
279
280    use crate::{
281        RECENT_STAKE_TABLES_LIMIT, SequencerApiVersion,
282        api::{
283            Options,
284            data_source::testing::TestableSqlDataSource,
285            test_helpers::{STAKE_TABLE_CAPACITY_FOR_TEST, TestNetwork, TestNetworkConfigBuilder},
286        },
287        catchup::NullStateCatchup,
288        legacy_chain::LegacyChain,
289        persistence::sql::DECIDE_GAP_FILL_HORIZON,
290        prefetch_stake_table_events,
291        testing::{TestConfigBuilder, staking_priv_keys},
292    };
293
294    #[async_trait]
295    pub trait TestablePersistence: SequencerPersistence + MembershipPersistence {
296        type Storage: Sync;
297
298        async fn tmp_storage() -> Self::Storage;
299        fn options(storage: &Self::Storage) -> impl PersistenceOptions<Persistence = Self>;
300
301        async fn connect(storage: &Self::Storage) -> Self {
302            Self::options(storage).create().await.unwrap()
303        }
304    }
305
306    #[rstest_reuse::template]
307    #[rstest::rstest]
308    #[case::postgres(PhantomData::<crate::persistence::sql::postgres::Persistence>)]
309    #[case::sqlite(PhantomData::<crate::persistence::sql::sqlite::Persistence>)]
310    #[case::fs(PhantomData::<crate::persistence::fs::Persistence>)]
311    #[test_log::test(tokio::test(flavor = "multi_thread"))]
312    pub fn persistence_types<P: TestablePersistence>(#[case] _p: PhantomData<P>) {}
313
314    #[derive(Clone, Debug, Default)]
315    struct EventCollector {
316        events: Arc<RwLock<Vec<CoordinatorEvent<SeqTypes>>>>,
317    }
318
319    impl EventCollector {
320        /// The leaves of every decide event received so far, oldest first.
321        async fn leaf_chain(&self) -> Vec<LeafInfo<SeqTypes>> {
322            self.events
323                .read()
324                .await
325                .iter()
326                .flat_map(|event| {
327                    let CoordinatorEvent::NewDecide { leaf_infos, .. } = event else {
328                        panic!("expected decide event, got {event:?}");
329                    };
330                    leaf_infos.iter().cloned().rev()
331                })
332                .collect::<Vec<_>>()
333        }
334    }
335
336    #[async_trait]
337    impl EventConsumer for EventCollector {
338        async fn handle_event(&self, event: &CoordinatorEvent<SeqTypes>) -> anyhow::Result<()> {
339            self.events.write().await.push(event.clone());
340            Ok(())
341        }
342    }
343
344    #[derive(Clone, Copy, Debug)]
345    struct FailConsumer;
346
347    #[async_trait]
348    impl EventConsumer for FailConsumer {
349        async fn handle_event(&self, _: &CoordinatorEvent<SeqTypes>) -> anyhow::Result<()> {
350            bail!("mock error injection");
351        }
352    }
353
354    #[rstest_reuse::apply(persistence_types)]
355    pub async fn test_voted_view<P: TestablePersistence>(_p: PhantomData<P>) {
356        let tmp = P::tmp_storage().await;
357        let storage = P::connect(&tmp).await;
358
359        // Initially, there is no saved view.
360        assert_eq!(storage.load_latest_acted_view().await.unwrap(), None);
361
362        // Store a view.
363        let view1 = ViewNumber::genesis();
364        storage
365            .record_action(view1, HotShotAction::Vote)
366            .await
367            .unwrap();
368        assert_eq!(
369            storage.load_latest_acted_view().await.unwrap().unwrap(),
370            view1
371        );
372
373        // Store a newer view, make sure storage gets updated.
374        let view2 = view1 + 1;
375        storage
376            .record_action(view2, HotShotAction::Vote)
377            .await
378            .unwrap();
379        assert_eq!(
380            storage.load_latest_acted_view().await.unwrap().unwrap(),
381            view2
382        );
383
384        // Store an old view, make sure storage is unchanged.
385        storage
386            .record_action(view1, HotShotAction::Vote)
387            .await
388            .unwrap();
389        assert_eq!(
390            storage.load_latest_acted_view().await.unwrap().unwrap(),
391            view2
392        );
393    }
394
395    #[rstest_reuse::apply(persistence_types)]
396    pub async fn test_high_qc2_monotonic<P: TestablePersistence>(_p: PhantomData<P>) {
397        let tmp = P::tmp_storage().await;
398        let storage = P::connect(&tmp).await;
399
400        // Initially there is no persisted lock.
401        assert_eq!(storage.load_high_qc2().await.unwrap(), None);
402
403        let mut qc = QuorumCertificate2::genesis(
404            &ValidatedState::default(),
405            &NodeState::mock(),
406            TEST_VERSIONS.test,
407        )
408        .await;
409
410        // Persist a lock at view 5.
411        qc.view_number = ViewNumber::new(5);
412        storage.append_high_qc2(qc.clone()).await.unwrap();
413        assert_eq!(
414            storage.load_high_qc2().await.unwrap().unwrap().view_number,
415            ViewNumber::new(5)
416        );
417
418        // A newer lock advances the stored view.
419        qc.view_number = ViewNumber::new(6);
420        storage.append_high_qc2(qc.clone()).await.unwrap();
421        assert_eq!(
422            storage.load_high_qc2().await.unwrap().unwrap().view_number,
423            ViewNumber::new(6)
424        );
425
426        // A stale (older) lock is a no-op: the compare-and-set never regresses
427        // the persisted view, even though the file/row write is unconditional.
428        qc.view_number = ViewNumber::new(4);
429        storage.append_high_qc2(qc.clone()).await.unwrap();
430        assert_eq!(
431            storage.load_high_qc2().await.unwrap().unwrap().view_number,
432            ViewNumber::new(6)
433        );
434
435        // Locks order by epoch first: a later epoch at an earlier view replaces
436        // the stored lock, and an earlier epoch at a later view does not.
437        let epoch = qc.data.epoch.unwrap_or_else(EpochNumber::genesis);
438        let at = |qc: &QuorumCertificate2<SeqTypes>| (qc.data.epoch, qc.view_number);
439        qc.data.epoch = Some(epoch + 1);
440        qc.view_number = ViewNumber::new(5);
441        storage.append_high_qc2(qc.clone()).await.unwrap();
442        assert_eq!(
443            at(&storage.load_high_qc2().await.unwrap().unwrap()),
444            (Some(epoch + 1), ViewNumber::new(5))
445        );
446        qc.data.epoch = Some(epoch);
447        qc.view_number = ViewNumber::new(9);
448        storage.append_high_qc2(qc.clone()).await.unwrap();
449        assert_eq!(
450            at(&storage.load_high_qc2().await.unwrap().unwrap()),
451            (Some(epoch + 1), ViewNumber::new(5))
452        );
453    }
454
455    #[rstest_reuse::apply(persistence_types)]
456    pub async fn test_boundary_qc2_monotonic<P: TestablePersistence>(_p: PhantomData<P>) {
457        let tmp = P::tmp_storage().await;
458        let storage = P::connect(&tmp).await;
459
460        assert_eq!(storage.load_boundary_qc2().await.unwrap(), None);
461
462        let mut qc = QuorumCertificate2::genesis(
463            &ValidatedState::default(),
464            &NodeState::mock(),
465            TEST_VERSIONS.test,
466        )
467        .await;
468        let at = |qc: &QuorumCertificate2<SeqTypes>| (qc.data.epoch, qc.view_number);
469
470        // Epoch 2, view 5.
471        qc.data.epoch = Some(EpochNumber::new(2));
472        qc.view_number = ViewNumber::new(5);
473        storage.append_boundary_qc2(qc.clone()).await.unwrap();
474        assert_eq!(
475            at(&storage.load_boundary_qc2().await.unwrap().unwrap()),
476            (Some(EpochNumber::new(2)), ViewNumber::new(5))
477        );
478
479        // An earlier epoch at a later view is older: locks order by epoch first.
480        qc.data.epoch = Some(EpochNumber::new(1));
481        qc.view_number = ViewNumber::new(9);
482        storage.append_boundary_qc2(qc.clone()).await.unwrap();
483        assert_eq!(
484            at(&storage.load_boundary_qc2().await.unwrap().unwrap()),
485            (Some(EpochNumber::new(2)), ViewNumber::new(5))
486        );
487
488        // A later epoch replaces it.
489        qc.data.epoch = Some(EpochNumber::new(3));
490        qc.view_number = ViewNumber::new(4);
491        storage.append_boundary_qc2(qc.clone()).await.unwrap();
492        assert_eq!(
493            at(&storage.load_boundary_qc2().await.unwrap().unwrap()),
494            (Some(EpochNumber::new(3)), ViewNumber::new(4))
495        );
496
497        // Within an epoch, a later view replaces it and an earlier one does not.
498        qc.view_number = ViewNumber::new(6);
499        storage.append_boundary_qc2(qc.clone()).await.unwrap();
500        assert_eq!(
501            at(&storage.load_boundary_qc2().await.unwrap().unwrap()),
502            (Some(EpochNumber::new(3)), ViewNumber::new(6))
503        );
504        qc.view_number = ViewNumber::new(5);
505        storage.append_boundary_qc2(qc.clone()).await.unwrap();
506        assert_eq!(
507            at(&storage.load_boundary_qc2().await.unwrap().unwrap()),
508            (Some(EpochNumber::new(3)), ViewNumber::new(6))
509        );
510    }
511
512    #[rstest_reuse::apply(persistence_types)]
513    pub async fn test_restart_view<P: TestablePersistence>(_p: PhantomData<P>) {
514        let tmp = P::tmp_storage().await;
515        let storage = P::connect(&tmp).await;
516
517        // Initially, there is no saved view.
518        assert_eq!(storage.load_restart_view().await.unwrap(), None);
519
520        // Store a view.
521        let view1 = ViewNumber::genesis();
522        storage
523            .record_action(view1, HotShotAction::Vote)
524            .await
525            .unwrap();
526        assert_eq!(
527            storage.load_restart_view().await.unwrap().unwrap(),
528            view1 + 1
529        );
530
531        // Store a newer view, make sure storage gets updated.
532        let view2 = view1 + 1;
533        storage
534            .record_action(view2, HotShotAction::Vote)
535            .await
536            .unwrap();
537        assert_eq!(
538            storage.load_restart_view().await.unwrap().unwrap(),
539            view2 + 1
540        );
541
542        // Store an old view, make sure storage is unchanged.
543        storage
544            .record_action(view1, HotShotAction::Vote)
545            .await
546            .unwrap();
547        assert_eq!(
548            storage.load_restart_view().await.unwrap().unwrap(),
549            view2 + 1
550        );
551
552        // store a higher proposed view, make sure storage is unchanged.
553        storage
554            .record_action(view2 + 1, HotShotAction::Propose)
555            .await
556            .unwrap();
557        assert_eq!(
558            storage.load_restart_view().await.unwrap().unwrap(),
559            view2 + 1
560        );
561
562        // store a higher timeout vote view, make sure storage is unchanged.
563        storage
564            .record_action(view2 + 1, HotShotAction::TimeoutVote)
565            .await
566            .unwrap();
567        assert_eq!(
568            storage.load_restart_view().await.unwrap().unwrap(),
569            view2 + 1
570        );
571    }
572
573    #[rstest_reuse::apply(persistence_types)]
574    pub async fn test_store_drb_input<P: TestablePersistence>(_p: PhantomData<P>) {
575        use hotshot_types::drb::DrbInput;
576
577        let tmp = P::tmp_storage().await;
578        let storage = P::connect(&tmp).await;
579        let difficulty_level = 10;
580
581        // Initially, there is no saved info.
582        if storage.load_drb_input(10).await.is_ok() {
583            panic!("unexpected nonempty drb_input");
584        }
585
586        let drb_input_1 = DrbInput {
587            epoch: 10,
588            iteration: 10,
589            value: [0u8; 32],
590            difficulty_level,
591        };
592
593        let drb_input_2 = DrbInput {
594            epoch: 10,
595            iteration: 20,
596            value: [0u8; 32],
597            difficulty_level,
598        };
599
600        let drb_input_3 = DrbInput {
601            epoch: 10,
602            iteration: 30,
603            value: [0u8; 32],
604            difficulty_level,
605        };
606
607        let _ = storage.store_drb_input(drb_input_1.clone()).await;
608
609        assert_eq!(storage.load_drb_input(10).await.unwrap(), drb_input_1);
610
611        let _ = storage.store_drb_input(drb_input_3.clone()).await;
612
613        // check that the drb input is overwritten
614        assert_eq!(storage.load_drb_input(10).await.unwrap(), drb_input_3);
615
616        let _ = storage.store_drb_input(drb_input_2.clone()).await;
617
618        // check that the drb input is not overwritten by the older value
619        assert_eq!(storage.load_drb_input(10).await.unwrap(), drb_input_3);
620    }
621
622    #[rstest_reuse::apply(persistence_types)]
623    pub async fn test_epoch_info<P: TestablePersistence>(_p: PhantomData<P>) {
624        let tmp = P::tmp_storage().await;
625        let storage = P::connect(&tmp).await;
626
627        // Initially, there is no saved info.
628        assert_eq!(storage.load_start_epoch_info().await.unwrap(), Vec::new());
629
630        // Store a drb result.
631        storage
632            .store_drb_result(EpochNumber::new(1), [1; 32])
633            .await
634            .unwrap();
635        assert_eq!(
636            storage.load_start_epoch_info().await.unwrap(),
637            vec![InitializerEpochInfo::<SeqTypes> {
638                epoch: EpochNumber::new(1),
639                drb_result: [1; 32],
640                block_header: None,
641            }]
642        );
643
644        // Store a second DRB result
645        storage
646            .store_drb_result(EpochNumber::new(2), [3; 32])
647            .await
648            .unwrap();
649        assert_eq!(
650            storage.load_start_epoch_info().await.unwrap(),
651            vec![
652                InitializerEpochInfo::<SeqTypes> {
653                    epoch: EpochNumber::new(1),
654                    drb_result: [1; 32],
655                    block_header: None,
656                },
657                InitializerEpochInfo::<SeqTypes> {
658                    epoch: EpochNumber::new(2),
659                    drb_result: [3; 32],
660                    block_header: None,
661                }
662            ]
663        );
664
665        // Make a header
666        let instance_state = NodeState::mock();
667        let validated_state = hotshot_types::traits::ValidatedState::genesis(&instance_state).0;
668        let leaf: Leaf2 = Leaf::genesis(&validated_state, &instance_state, MOCK_UPGRADE.base)
669            .await
670            .into();
671        let header = leaf.block_header().clone();
672
673        // Test storing the header
674        storage
675            .store_epoch_root(EpochNumber::new(1), header.clone())
676            .await
677            .unwrap();
678        assert_eq!(
679            storage.load_start_epoch_info().await.unwrap(),
680            vec![
681                InitializerEpochInfo::<SeqTypes> {
682                    epoch: EpochNumber::new(1),
683                    drb_result: [1; 32],
684                    block_header: Some(header.clone()),
685                },
686                InitializerEpochInfo::<SeqTypes> {
687                    epoch: EpochNumber::new(2),
688                    drb_result: [3; 32],
689                    block_header: None,
690                }
691            ]
692        );
693
694        // Store more than the limit
695        let total_epochs = RECENT_STAKE_TABLES_LIMIT + 10;
696        for i in 0..total_epochs {
697            let epoch = EpochNumber::new(i);
698            let drb = [i as u8; 32];
699            storage
700                .store_drb_result(epoch, drb)
701                .await
702                .unwrap_or_else(|_| panic!("Failed to store DRB result for epoch {i}"));
703        }
704
705        let results = storage.load_start_epoch_info().await.unwrap();
706
707        // Check that only the most recent RECENT_STAKE_TABLES_LIMIT epochs are returned
708        assert_eq!(
709            results.len(),
710            RECENT_STAKE_TABLES_LIMIT as usize,
711            "Should return only the most recent {RECENT_STAKE_TABLES_LIMIT} epochs",
712        );
713
714        for (i, info) in results.iter().enumerate() {
715            let expected_epoch =
716                EpochNumber::new(total_epochs - RECENT_STAKE_TABLES_LIMIT + i as u64);
717            let expected_drb = [(total_epochs - RECENT_STAKE_TABLES_LIMIT + i as u64) as u8; 32];
718            assert_eq!(info.epoch, expected_epoch, "invalid epoch at index {i}",);
719            assert_eq!(info.drb_result, expected_drb, "invalid DRB at index {i}",);
720            assert!(info.block_header.is_none(), "Expected no block header");
721        }
722    }
723
724    fn leaf_info(leaf: Leaf2) -> LeafInfo<SeqTypes> {
725        LeafInfo {
726            leaf,
727            vid_share: None,
728            state: Default::default(),
729            delta: None,
730            state_cert: None,
731        }
732    }
733
734    #[rstest_reuse::apply(persistence_types)]
735    pub async fn test_append_and_decide<P: TestablePersistence>(_p: PhantomData<P>) {
736        let tmp = P::tmp_storage().await;
737        let storage = P::connect(&tmp).await;
738
739        // Test append VID
740        assert_eq!(
741            storage.load_vid_share(ViewNumber::new(0)).await.unwrap(),
742            None
743        );
744
745        let node_state = NodeState::mock().with_genesis_version(versions::NEW_PROTOCOL_VERSION);
746        let leaf: Leaf2 =
747            Leaf2::genesis(&ValidatedState::default(), &node_state, TEST_UPGRADE.base).await;
748        let leaf_payload = leaf.block_payload().unwrap();
749        let leaf_payload_bytes_arc = leaf_payload.encode();
750
751        let avidm_param = init_avidm_gf2_param(2).unwrap();
752        let weights = vec![1u32; 2];
753
754        let ns_table = parse_ns_table(
755            leaf_payload.byte_len().as_usize(),
756            &leaf_payload.ns_table().encode(),
757        );
758        let (payload_commitment, common, shares) =
759            AvidmGf2Scheme::ns_disperse(&avidm_param, &weights, &leaf_payload_bytes_arc, ns_table)
760                .unwrap();
761
762        let (pubkey, privkey) = BLSPubKey::generated_from_seed_indexed([0; 32], 1);
763        let signature = PubKey::sign(&privkey, &[]).unwrap();
764        let mut vid = AvidmGf2DisperseShare::<SeqTypes> {
765            view_number: ViewNumber::new(0),
766            payload_commitment,
767            share: shares[0].clone(),
768            recipient_key: pubkey,
769            epoch: Some(EpochNumber::new(0)),
770            target_epoch: Some(EpochNumber::new(0)),
771            common,
772        };
773        let mut quorum_proposal = Proposal {
774            data: QuorumProposalWrapper::<SeqTypes> {
775                proposal: QuorumProposal2::<SeqTypes> {
776                    epoch: None,
777                    block_header: leaf.block_header().clone(),
778                    view_number: ViewNumber::genesis(),
779                    justify_qc: QuorumCertificate2::genesis(
780                        &ValidatedState::default(),
781                        &node_state,
782                        TEST_UPGRADE,
783                    )
784                    .await,
785                    upgrade_certificate: None,
786                    view_change_evidence: None,
787                    next_drb_result: None,
788                    next_epoch_justify_qc: None,
789                    state_cert: None,
790                },
791            },
792            signature,
793            _pd: Default::default(),
794        };
795
796        let vid_share0 = convert_proposal(vid.clone().to_proposal(&privkey).unwrap().clone());
797
798        storage.append_vid(&vid_share0).await.unwrap();
799
800        assert_eq!(
801            storage.load_vid_share(ViewNumber::new(0)).await.unwrap(),
802            Some(vid_share0.clone())
803        );
804
805        vid.view_number = ViewNumber::new(1);
806
807        let vid_share1 = convert_proposal(vid.clone().to_proposal(&privkey).unwrap().clone());
808        storage.append_vid(&vid_share1).await.unwrap();
809
810        assert_eq!(
811            storage.load_vid_share(vid.view_number()).await.unwrap(),
812            Some(vid_share1.clone())
813        );
814
815        vid.view_number = ViewNumber::new(2);
816
817        let vid_share2 = convert_proposal(vid.clone().to_proposal(&privkey).unwrap().clone());
818        storage.append_vid(&vid_share2).await.unwrap();
819
820        assert_eq!(
821            storage.load_vid_share(vid.view_number()).await.unwrap(),
822            Some(vid_share2.clone())
823        );
824
825        vid.view_number = ViewNumber::new(3);
826
827        let vid_share3 = convert_proposal(vid.clone().to_proposal(&privkey).unwrap().clone());
828        storage.append_vid(&vid_share3).await.unwrap();
829
830        assert_eq!(
831            storage.load_vid_share(vid.view_number()).await.unwrap(),
832            Some(vid_share3.clone())
833        );
834
835        let block_payload_signature = BLSPubKey::sign(&privkey, &leaf_payload_bytes_arc)
836            .expect("Failed to sign block payload");
837
838        let da_proposal_inner = DaProposal2::<SeqTypes> {
839            encoded_transactions: leaf_payload_bytes_arc.clone(),
840            metadata: leaf_payload.ns_table().clone(),
841            view_number: ViewNumber::new(0),
842            epoch: None,
843            epoch_transition_indicator: EpochTransitionIndicator::NotInTransition,
844        };
845
846        let da_proposal = Proposal {
847            data: da_proposal_inner,
848            signature: block_payload_signature,
849            _pd: Default::default(),
850        };
851
852        let vid_commitment = vid_commitment(
853            &leaf_payload_bytes_arc,
854            &leaf.block_header().metadata().encode(),
855            2,
856            TEST_UPGRADE.base,
857        );
858
859        storage
860            .append_da2(&da_proposal, vid_commitment)
861            .await
862            .unwrap();
863
864        assert_eq!(
865            storage.load_da_proposal(ViewNumber::new(0)).await.unwrap(),
866            Some(da_proposal.clone())
867        );
868
869        let mut da_proposal1 = da_proposal.clone();
870        da_proposal1.data.view_number = ViewNumber::new(1);
871        storage
872            .append_da2(&da_proposal1.clone(), vid_commitment)
873            .await
874            .unwrap();
875
876        assert_eq!(
877            storage
878                .load_da_proposal(da_proposal1.data.view_number)
879                .await
880                .unwrap(),
881            Some(da_proposal1.clone())
882        );
883
884        let mut da_proposal2 = da_proposal1.clone();
885        da_proposal2.data.view_number = ViewNumber::new(2);
886        storage
887            .append_da2(&da_proposal2.clone(), vid_commitment)
888            .await
889            .unwrap();
890
891        assert_eq!(
892            storage
893                .load_da_proposal(da_proposal2.data.view_number)
894                .await
895                .unwrap(),
896            Some(da_proposal2.clone())
897        );
898
899        let mut da_proposal3 = da_proposal2.clone();
900        da_proposal3.data.view_number = ViewNumber::new(3);
901        storage
902            .append_da2(&da_proposal3.clone(), vid_commitment)
903            .await
904            .unwrap();
905
906        assert_eq!(
907            storage
908                .load_da_proposal(da_proposal3.data.view_number)
909                .await
910                .unwrap(),
911            Some(da_proposal3.clone())
912        );
913
914        let quorum_proposal1 = quorum_proposal.clone();
915
916        storage
917            .append_quorum_proposal2(&quorum_proposal1)
918            .await
919            .unwrap();
920
921        assert_eq!(
922            storage.load_quorum_proposals().await.unwrap(),
923            BTreeMap::from_iter([(ViewNumber::genesis(), quorum_proposal1.clone())])
924        );
925
926        quorum_proposal.data.proposal.view_number = ViewNumber::new(1);
927        let quorum_proposal2 = quorum_proposal.clone();
928        storage
929            .append_quorum_proposal2(&quorum_proposal2)
930            .await
931            .unwrap();
932
933        assert_eq!(
934            storage.load_quorum_proposals().await.unwrap(),
935            BTreeMap::from_iter([
936                (ViewNumber::genesis(), quorum_proposal1.clone()),
937                (ViewNumber::new(1), quorum_proposal2.clone())
938            ])
939        );
940
941        quorum_proposal.data.proposal.view_number = ViewNumber::new(2);
942        quorum_proposal.data.proposal.justify_qc.view_number = ViewNumber::new(1);
943        let quorum_proposal3 = quorum_proposal.clone();
944        storage
945            .append_quorum_proposal2(&quorum_proposal3)
946            .await
947            .unwrap();
948
949        assert_eq!(
950            storage.load_quorum_proposals().await.unwrap(),
951            BTreeMap::from_iter([
952                (ViewNumber::genesis(), quorum_proposal1.clone()),
953                (ViewNumber::new(1), quorum_proposal2.clone()),
954                (ViewNumber::new(2), quorum_proposal3.clone())
955            ])
956        );
957
958        quorum_proposal.data.proposal.view_number = ViewNumber::new(3);
959        quorum_proposal.data.proposal.justify_qc.view_number = ViewNumber::new(2);
960
961        // This one should stick around after GC runs.
962        let quorum_proposal4 = quorum_proposal.clone();
963        storage
964            .append_quorum_proposal2(&quorum_proposal4)
965            .await
966            .unwrap();
967
968        assert_eq!(
969            storage.load_quorum_proposals().await.unwrap(),
970            BTreeMap::from_iter([
971                (ViewNumber::genesis(), quorum_proposal1.clone()),
972                (ViewNumber::new(1), quorum_proposal2.clone()),
973                (ViewNumber::new(2), quorum_proposal3.clone()),
974                (ViewNumber::new(3), quorum_proposal4.clone())
975            ])
976        );
977
978        // Test decide and garbage collection. Pass in a leaf chain with no VID shares or payloads,
979        // so we have to fetch the missing data from storage.
980        let leaves = [
981            Leaf2::from_quorum_proposal(&quorum_proposal1.data),
982            Leaf2::from_quorum_proposal(&quorum_proposal2.data),
983            Leaf2::from_quorum_proposal(&quorum_proposal3.data),
984            Leaf2::from_quorum_proposal(&quorum_proposal4.data),
985        ];
986        let mut final_qc = leaves[3].justify_qc().clone();
987        final_qc.view_number += 1;
988        final_qc.data.leaf_commit = Committable::commit(&leaf);
989        let qcs = [
990            CertificatePair::non_epoch_change(leaves[1].justify_qc().clone()),
991            CertificatePair::non_epoch_change(leaves[2].justify_qc().clone()),
992            CertificatePair::non_epoch_change(leaves[3].justify_qc().clone()),
993            CertificatePair::non_epoch_change(final_qc),
994        ];
995
996        assert_eq!(
997            storage.load_anchor_view().await.unwrap(),
998            ViewNumber::genesis()
999        );
1000
1001        let consumer = EventCollector::default();
1002        let leaf_chain = leaves
1003            .iter()
1004            .take(3)
1005            .map(|leaf| leaf_info(leaf.clone()))
1006            .zip(&qcs)
1007            .collect::<Vec<_>>();
1008        tracing::info!(?leaf_chain, "decide view 2");
1009        storage
1010            .append_decided_leaves(
1011                ViewNumber::new(2),
1012                leaf_chain.iter().map(|(leaf, qc)| (leaf, (*qc).clone())),
1013                &consumer,
1014            )
1015            .await
1016            .unwrap();
1017        assert_eq!(
1018            storage.load_anchor_view().await.unwrap(),
1019            ViewNumber::new(2)
1020        );
1021
1022        for i in 0..=2 {
1023            assert_eq!(
1024                storage.load_da_proposal(ViewNumber::new(i)).await.unwrap(),
1025                None
1026            );
1027
1028            assert_eq!(
1029                storage.load_vid_share(ViewNumber::new(i)).await.unwrap(),
1030                None
1031            );
1032        }
1033
1034        assert_eq!(
1035            storage.load_da_proposal(ViewNumber::new(3)).await.unwrap(),
1036            Some(da_proposal3)
1037        );
1038
1039        assert_eq!(
1040            storage.load_vid_share(ViewNumber::new(3)).await.unwrap(),
1041            Some(convert_proposal(vid_share3.clone()))
1042        );
1043
1044        let proposals = storage.load_quorum_proposals().await.unwrap();
1045        assert_eq!(
1046            proposals,
1047            BTreeMap::from_iter([(ViewNumber::new(3), quorum_proposal4)])
1048        );
1049
1050        // A decide event should have been processed.
1051        for (leaf, info) in leaves.iter().zip(consumer.leaf_chain().await.iter()) {
1052            assert_eq!(info.leaf, *leaf);
1053            let decided_vid_share = info.vid_share.as_ref().unwrap();
1054            assert_eq!(decided_vid_share.view_number(), leaf.view_number());
1055        }
1056
1057        // The decided leaf should not have been garbage collected.
1058        assert_eq!(
1059            storage.load_anchor_leaf().await.unwrap(),
1060            Some((leaves[2].clone(), qcs[2].clone()))
1061        );
1062        assert_eq!(
1063            storage.load_anchor_view().await.unwrap(),
1064            leaves[2].view_number()
1065        );
1066
1067        // Process a second decide event.
1068        let consumer = EventCollector::default();
1069        tracing::info!(leaf = ?leaves[3], qc = ?qcs[3], "decide view 3");
1070        storage
1071            .append_decided_leaves(
1072                ViewNumber::new(3),
1073                vec![(&leaf_info(leaves[3].clone()), qcs[3].clone())],
1074                &consumer,
1075            )
1076            .await
1077            .unwrap();
1078        assert_eq!(
1079            storage.load_anchor_view().await.unwrap(),
1080            ViewNumber::new(3)
1081        );
1082
1083        // A decide event should have been processed.
1084        let events = consumer.events.read().await;
1085        assert_eq!(events.len(), 1);
1086        let CoordinatorEvent::NewDecide {
1087            leaf_infos, cert1, ..
1088        } = &events[0]
1089        else {
1090            panic!("expected decide event, got {:?}", events[0]);
1091        };
1092        assert_eq!(cert1, qcs[3].qc());
1093        assert_eq!(leaf_infos.len(), 1);
1094        let info = &leaf_infos[0];
1095        assert_eq!(info.leaf, leaves[3]);
1096        assert_eq!(info.leaf.view_number(), ViewNumber::new(3));
1097
1098        // The remaining data should have been GCed.
1099        assert_eq!(
1100            storage.load_da_proposal(ViewNumber::new(3)).await.unwrap(),
1101            None
1102        );
1103
1104        assert_eq!(
1105            storage.load_vid_share(ViewNumber::new(3)).await.unwrap(),
1106            None
1107        );
1108        assert_eq!(
1109            storage.load_quorum_proposals().await.unwrap(),
1110            BTreeMap::new()
1111        );
1112    }
1113
1114    #[rstest_reuse::apply(persistence_types)]
1115    pub async fn test_upgrade_certificate<P: TestablePersistence>(_p: PhantomData<P>) {
1116        let tmp = P::tmp_storage().await;
1117        let storage = P::connect(&tmp).await;
1118
1119        // Test get upgrade certificate
1120        assert_eq!(storage.load_upgrade_certificate().await.unwrap(), None);
1121
1122        let upgrade_data = UpgradeProposalData {
1123            old_version: Version { major: 0, minor: 1 },
1124            new_version: Version { major: 1, minor: 0 },
1125            decide_by: ViewNumber::genesis(),
1126            new_version_hash: Default::default(),
1127            old_version_last_view: ViewNumber::genesis(),
1128            new_version_first_view: ViewNumber::genesis(),
1129        };
1130
1131        let decide_upgrade_certificate = UpgradeCertificate::<SeqTypes>::new(
1132            upgrade_data.clone(),
1133            upgrade_data.commit(),
1134            ViewNumber::genesis(),
1135            Default::default(),
1136            Default::default(),
1137        );
1138        let res = storage
1139            .store_upgrade_certificate(Some(decide_upgrade_certificate.clone()))
1140            .await;
1141        assert!(res.is_ok());
1142
1143        let res = storage.load_upgrade_certificate().await.unwrap();
1144        let view_number = res.unwrap().view_number;
1145        assert_eq!(view_number, ViewNumber::genesis());
1146
1147        let new_view_number_for_certificate = ViewNumber::new(50);
1148        let mut new_upgrade_certificate = decide_upgrade_certificate.clone();
1149        new_upgrade_certificate.view_number = new_view_number_for_certificate;
1150
1151        let res = storage
1152            .store_upgrade_certificate(Some(new_upgrade_certificate.clone()))
1153            .await;
1154        assert!(res.is_ok());
1155
1156        let res = storage.load_upgrade_certificate().await.unwrap();
1157        let view_number = res.unwrap().view_number;
1158        assert_eq!(view_number, new_view_number_for_certificate);
1159    }
1160
1161    /// The `NewProtocolStorage` impl on `Arc<P>` (the object the coordinator actually holds) is the
1162    /// persist-before-vote hook. It must durably route to the `high_qc2` table and stay monotonic.
1163    #[rstest_reuse::apply(persistence_types)]
1164    pub async fn test_arc_storage_append_high_qc2_persists<P: TestablePersistence>(
1165        _p: PhantomData<P>,
1166    ) {
1167        use hotshot_new_protocol::storage::NewProtocolStorage;
1168
1169        let tmp = P::tmp_storage().await;
1170        let storage = P::connect(&tmp).await;
1171        // Production wraps the persistence as `Arc<P>`; that is the impl under test.
1172        let arc = Arc::new(storage.clone());
1173
1174        assert_eq!(storage.load_high_qc2().await.unwrap(), None);
1175
1176        let mut high_qc = QuorumCertificate2::genesis(
1177            &ValidatedState::default(),
1178            &NodeState::mock(),
1179            TEST_VERSIONS.test,
1180        )
1181        .await;
1182        high_qc.view_number = ViewNumber::new(7);
1183        NewProtocolStorage::append_high_qc2(&arc, high_qc.clone())
1184            .await
1185            .unwrap();
1186        assert_eq!(
1187            storage.load_high_qc2().await.unwrap().unwrap().view_number,
1188            ViewNumber::new(7)
1189        );
1190
1191        // A stale write through the wrapper must not regress the persisted view.
1192        let mut stale = high_qc.clone();
1193        stale.view_number = ViewNumber::new(3);
1194        NewProtocolStorage::append_high_qc2(&arc, stale)
1195            .await
1196            .unwrap();
1197        assert_eq!(
1198            storage.load_high_qc2().await.unwrap().unwrap().view_number,
1199            ViewNumber::new(7)
1200        );
1201    }
1202
1203    /// `load_consensus_state` must fold the running high QC persisted via `append_high_qc2` back into
1204    /// the recovered `high_qc` (which `Consensus::new` then uses to restore `locked_view` past the
1205    /// decided anchor). With no decided anchor, recovery starts from the genesis QC at view 0 and
1206    /// must pick up a persisted high QC at a higher view.
1207    #[rstest_reuse::apply(persistence_types)]
1208    pub async fn test_load_consensus_state_recovers_high_qc<P: TestablePersistence>(
1209        _p: PhantomData<P>,
1210    ) {
1211        let tmp = P::tmp_storage().await;
1212        let storage = P::connect(&tmp).await;
1213
1214        let mut high_qc = QuorumCertificate2::genesis(
1215            &ValidatedState::default(),
1216            &NodeState::mock(),
1217            TEST_VERSIONS.test,
1218        )
1219        .await;
1220        high_qc.view_number = ViewNumber::new(9);
1221        storage.append_high_qc2(high_qc).await.unwrap();
1222
1223        let (initializer, _anchor_view) = storage
1224            .load_consensus_state(
1225                NodeState::mock(),
1226                Upgrade::trivial(versions::EPOCH_REWARD_VERSION),
1227            )
1228            .await
1229            .unwrap();
1230
1231        // Without the recovery fold this would be the genesis view (0); with it, the persisted lock.
1232        assert_eq!(initializer.high_qc().view_number, ViewNumber::new(9));
1233    }
1234
1235    /// `load_consensus_state` adopts the running high QC persisted via `update_high_qc2` over an
1236    /// older eQC checkpoint.
1237    #[rstest_reuse::apply(persistence_types)]
1238    pub async fn test_load_consensus_state_prefers_running_high_qc_over_eqc<
1239        P: TestablePersistence,
1240    >(
1241        _p: PhantomData<P>,
1242    ) {
1243        let tmp = P::tmp_storage().await;
1244        let storage = P::connect(&tmp).await;
1245
1246        // Build a next-epoch QC we can re-stamp at different views.
1247        let upgrade_lock = UpgradeLock::<SeqTypes>::new(TEST_VERSIONS.test);
1248        let leaf = Leaf2::genesis(
1249            &ValidatedState::default(),
1250            &NodeState::default(),
1251            TEST_VERSIONS.test.base,
1252        )
1253        .await;
1254        let data: NextEpochQuorumData2<SeqTypes> = QuorumData2 {
1255            leaf_commit: leaf.commit(),
1256            epoch: Some(EpochNumber::new(1)),
1257            block_number: Some(leaf.height()),
1258        }
1259        .into();
1260        let versioned_data =
1261            VersionedVoteData::new_infallible(data.clone(), ViewNumber::new(5), &upgrade_lock);
1262        let bytes: [u8; 32] = versioned_data.commit().into();
1263        let next_epoch_qc = NextEpochQuorumCertificate2::new(
1264            data,
1265            Commitment::from_raw(bytes),
1266            ViewNumber::new(5),
1267            None,
1268            PhantomData,
1269        );
1270
1271        // eQC checkpoint: high QC + corresponding next-epoch QC, both at view 5.
1272        let mut eqc_high_qc = QuorumCertificate2::genesis(
1273            &ValidatedState::default(),
1274            &NodeState::mock(),
1275            TEST_VERSIONS.test,
1276        )
1277        .await;
1278        eqc_high_qc.view_number = ViewNumber::new(5);
1279        storage
1280            .store_eqc(eqc_high_qc.clone(), next_epoch_qc.clone())
1281            .await
1282            .unwrap();
1283
1284        // A newer running high QC (view 9): recovery adopts it over the eQC checkpoint.
1285        let epoch_height = 10;
1286        let mut running_high_qc = eqc_high_qc.clone();
1287        running_high_qc.view_number = ViewNumber::new(9);
1288        running_high_qc.data.block_number = Some(epoch_height);
1289        storage
1290            .append_high_qc2(running_high_qc.clone())
1291            .await
1292            .unwrap();
1293
1294        let (initializer, _anchor_view) = storage
1295            .load_consensus_state(
1296                NodeState::mock().with_epoch_height(epoch_height),
1297                Upgrade::trivial(versions::EPOCH_REWARD_VERSION),
1298            )
1299            .await
1300            .unwrap();
1301
1302        assert_eq!(initializer.high_qc().view_number, ViewNumber::new(9));
1303    }
1304
1305    #[rstest_reuse::apply(persistence_types)]
1306    pub async fn test_decide_with_failing_event_consumer<P: TestablePersistence>(
1307        _p: PhantomData<P>,
1308    ) {
1309        let tmp = P::tmp_storage().await;
1310        let storage = P::connect(&tmp).await;
1311
1312        // Create a short blockchain.
1313        let mut chain = vec![];
1314
1315        let node_state = NodeState::mock().with_genesis_version(versions::NEW_PROTOCOL_VERSION);
1316        let leaf: Leaf2 =
1317            Leaf2::genesis(&ValidatedState::default(), &node_state, TEST_UPGRADE.base).await;
1318        let leaf_payload = leaf.block_payload().unwrap();
1319        let leaf_payload_bytes_arc = leaf_payload.encode();
1320        let avidm_param = init_avidm_gf2_param(2).unwrap();
1321        let weights = vec![1u32; 2];
1322        let ns_table = parse_ns_table(
1323            leaf_payload.byte_len().as_usize(),
1324            &leaf_payload.ns_table().encode(),
1325        );
1326        let (payload_commitment, common, shares) =
1327            AvidmGf2Scheme::ns_disperse(&avidm_param, &weights, &leaf_payload_bytes_arc, ns_table)
1328                .unwrap();
1329
1330        let (pubkey, privkey) = BLSPubKey::generated_from_seed_indexed([0; 32], 1);
1331        let mut vid = AvidmGf2DisperseShare::<SeqTypes> {
1332            view_number: ViewNumber::new(0),
1333            payload_commitment,
1334            share: shares[0].clone(),
1335            recipient_key: pubkey,
1336            epoch: Some(EpochNumber::new(0)),
1337            target_epoch: Some(EpochNumber::new(0)),
1338            common,
1339        }
1340        .to_proposal(&privkey)
1341        .unwrap()
1342        .clone();
1343        let mut quorum_proposal = QuorumProposalWrapper::<SeqTypes> {
1344            proposal: QuorumProposal2::<SeqTypes> {
1345                block_header: leaf.block_header().clone(),
1346                view_number: ViewNumber::genesis(),
1347                justify_qc: QuorumCertificate2::genesis(
1348                    &ValidatedState::default(),
1349                    &node_state,
1350                    TEST_UPGRADE,
1351                )
1352                .await,
1353                upgrade_certificate: None,
1354                view_change_evidence: None,
1355                next_drb_result: None,
1356                next_epoch_justify_qc: None,
1357                epoch: None,
1358                state_cert: None,
1359            },
1360        };
1361        let mut qc =
1362            QuorumCertificate2::genesis(&ValidatedState::default(), &node_state, TEST_UPGRADE)
1363                .await;
1364
1365        let block_payload_signature = BLSPubKey::sign(&privkey, &leaf_payload_bytes_arc)
1366            .expect("Failed to sign block payload");
1367        let mut da_proposal = Proposal {
1368            data: DaProposal2::<SeqTypes> {
1369                encoded_transactions: leaf_payload_bytes_arc.clone(),
1370                metadata: leaf_payload.ns_table().clone(),
1371                view_number: ViewNumber::new(0),
1372                epoch: Some(EpochNumber::new(0)),
1373                epoch_transition_indicator: EpochTransitionIndicator::NotInTransition,
1374            },
1375            signature: block_payload_signature,
1376            _pd: Default::default(),
1377        };
1378
1379        let vid_commitment = vid_commitment(
1380            &leaf_payload_bytes_arc,
1381            &leaf.block_header().metadata().encode(),
1382            2,
1383            TEST_UPGRADE.base,
1384        );
1385
1386        for i in 0..4 {
1387            quorum_proposal.proposal.view_number = ViewNumber::new(i);
1388            let leaf = Leaf2::from_quorum_proposal(&quorum_proposal);
1389            qc.view_number = leaf.view_number();
1390            qc.data.leaf_commit = Committable::commit(&leaf);
1391            vid.data.view_number = leaf.view_number();
1392            da_proposal.data.view_number = leaf.view_number();
1393            chain.push((leaf.clone(), qc.clone(), vid.clone(), da_proposal.clone()));
1394        }
1395
1396        // Add proposals.
1397        for (_, _, vid, da) in &chain {
1398            tracing::info!(?da, ?vid, "insert proposal");
1399            storage.append_da2(da, vid_commitment).await.unwrap();
1400            storage
1401                .append_vid(&convert_proposal(vid.clone()))
1402                .await
1403                .unwrap();
1404        }
1405
1406        // Decide 2 leaves, but fail in event processing.
1407        let leaf_chain = chain
1408            .iter()
1409            .take(2)
1410            .map(|(leaf, qc, ..)| (leaf_info(leaf.clone()), qc.clone()))
1411            .collect::<Vec<_>>();
1412        tracing::info!("decide with event handling failure");
1413        storage
1414            .append_decided_leaves(
1415                ViewNumber::new(1),
1416                leaf_chain
1417                    .iter()
1418                    .map(|(leaf, qc)| (leaf, CertificatePair::non_epoch_change(qc.clone()))),
1419                &FailConsumer,
1420            )
1421            .await
1422            .unwrap();
1423        // No garbage collection should have run.
1424        for i in 0..4 {
1425            tracing::info!(i, "check proposal availability");
1426            assert!(
1427                storage
1428                    .load_vid_share(ViewNumber::new(i))
1429                    .await
1430                    .unwrap()
1431                    .is_some()
1432            );
1433            assert!(
1434                storage
1435                    .load_da_proposal(ViewNumber::new(i))
1436                    .await
1437                    .unwrap()
1438                    .is_some()
1439            );
1440        }
1441        tracing::info!("check anchor leaf updated");
1442        assert_eq!(
1443            storage
1444                .load_anchor_leaf()
1445                .await
1446                .unwrap()
1447                .unwrap()
1448                .0
1449                .view_number(),
1450            ViewNumber::new(1)
1451        );
1452        assert_eq!(
1453            storage.load_anchor_view().await.unwrap(),
1454            ViewNumber::new(1)
1455        );
1456
1457        // Now decide remaining leaves successfully. We should now garbage collect and process a
1458        // decide event for all the leaves.
1459        let consumer = EventCollector::default();
1460        let leaf_chain = chain
1461            .iter()
1462            .skip(2)
1463            .map(|(leaf, qc, ..)| (leaf_info(leaf.clone()), qc.clone()))
1464            .collect::<Vec<_>>();
1465        tracing::info!("decide successfully");
1466        storage
1467            .append_decided_leaves(
1468                ViewNumber::new(3),
1469                leaf_chain
1470                    .iter()
1471                    .map(|(leaf, qc)| (leaf, CertificatePair::non_epoch_change(qc.clone()))),
1472                &consumer,
1473            )
1474            .await
1475            .unwrap();
1476        // Garbage collection should have run.
1477        for i in 0..4 {
1478            tracing::info!(i, "check proposal garbage collected");
1479            assert!(
1480                storage
1481                    .load_vid_share(ViewNumber::new(i))
1482                    .await
1483                    .unwrap()
1484                    .is_none()
1485            );
1486            assert!(
1487                storage
1488                    .load_da_proposal(ViewNumber::new(i))
1489                    .await
1490                    .unwrap()
1491                    .is_none()
1492            );
1493        }
1494        tracing::info!("check anchor leaf updated");
1495        assert_eq!(
1496            storage
1497                .load_anchor_leaf()
1498                .await
1499                .unwrap()
1500                .unwrap()
1501                .0
1502                .view_number(),
1503            ViewNumber::new(3)
1504        );
1505        assert_eq!(
1506            storage.load_anchor_view().await.unwrap(),
1507            ViewNumber::new(3)
1508        );
1509
1510        // Check decide event.
1511        tracing::info!("check decide event");
1512        let leaf_chain = consumer.leaf_chain().await;
1513        assert_eq!(leaf_chain.len(), 4, "{leaf_chain:#?}");
1514        for ((leaf, ..), info) in chain.iter().zip(leaf_chain.iter()) {
1515            assert_eq!(info.leaf, *leaf);
1516            let decided_vid_share = info.vid_share.as_ref().unwrap();
1517            assert_eq!(decided_vid_share.view_number(), leaf.view_number());
1518            assert!(info.leaf.block_payload().is_some());
1519        }
1520    }
1521
1522    /// Splitting a decide into `persist_decided_leaves` (persist only, no events/GC) and a later
1523    /// `process_decided_events` loses no data: multiple persists can accumulate, and one process
1524    /// pass at the latest view drains the whole backlog and runs GC.
1525    #[rstest_reuse::apply(persistence_types)]
1526    pub async fn test_deferred_decide_processing<P: TestablePersistence>(_p: PhantomData<P>) {
1527        let tmp = P::tmp_storage().await;
1528        let storage = P::connect(&tmp).await;
1529
1530        // Create a short blockchain (same setup as `test_decide_with_failing_event_consumer`).
1531        let mut chain = vec![];
1532
1533        let node_state = NodeState::mock().with_genesis_version(versions::NEW_PROTOCOL_VERSION);
1534        let leaf: Leaf2 =
1535            Leaf2::genesis(&ValidatedState::default(), &node_state, TEST_UPGRADE.base).await;
1536        let leaf_payload = leaf.block_payload().unwrap();
1537        let leaf_payload_bytes_arc = leaf_payload.encode();
1538        let avidm_param = init_avidm_gf2_param(2).unwrap();
1539        let weights = vec![1u32; 2];
1540        let ns_table = parse_ns_table(
1541            leaf_payload.byte_len().as_usize(),
1542            &leaf_payload.ns_table().encode(),
1543        );
1544        let (payload_commitment, common, shares) =
1545            AvidmGf2Scheme::ns_disperse(&avidm_param, &weights, &leaf_payload_bytes_arc, ns_table)
1546                .unwrap();
1547
1548        let (pubkey, privkey) = BLSPubKey::generated_from_seed_indexed([0; 32], 1);
1549        let mut vid = AvidmGf2DisperseShare::<SeqTypes> {
1550            view_number: ViewNumber::new(0),
1551            payload_commitment,
1552            share: shares[0].clone(),
1553            recipient_key: pubkey,
1554            epoch: Some(EpochNumber::new(0)),
1555            target_epoch: Some(EpochNumber::new(0)),
1556            common,
1557        }
1558        .to_proposal(&privkey)
1559        .unwrap()
1560        .clone();
1561        let mut quorum_proposal = QuorumProposalWrapper::<SeqTypes> {
1562            proposal: QuorumProposal2::<SeqTypes> {
1563                block_header: leaf.block_header().clone(),
1564                view_number: ViewNumber::genesis(),
1565                justify_qc: QuorumCertificate2::genesis(
1566                    &ValidatedState::default(),
1567                    &node_state,
1568                    TEST_UPGRADE,
1569                )
1570                .await,
1571                upgrade_certificate: None,
1572                view_change_evidence: None,
1573                next_drb_result: None,
1574                next_epoch_justify_qc: None,
1575                epoch: None,
1576                state_cert: None,
1577            },
1578        };
1579        let mut qc =
1580            QuorumCertificate2::genesis(&ValidatedState::default(), &node_state, TEST_UPGRADE)
1581                .await;
1582
1583        let block_payload_signature = BLSPubKey::sign(&privkey, &leaf_payload_bytes_arc)
1584            .expect("Failed to sign block payload");
1585        let mut da_proposal = Proposal {
1586            data: DaProposal2::<SeqTypes> {
1587                encoded_transactions: leaf_payload_bytes_arc.clone(),
1588                metadata: leaf_payload.ns_table().clone(),
1589                view_number: ViewNumber::new(0),
1590                epoch: Some(EpochNumber::new(0)),
1591                epoch_transition_indicator: EpochTransitionIndicator::NotInTransition,
1592            },
1593            signature: block_payload_signature,
1594            _pd: Default::default(),
1595        };
1596
1597        let vid_commitment = vid_commitment(
1598            &leaf_payload_bytes_arc,
1599            &leaf.block_header().metadata().encode(),
1600            2,
1601            TEST_UPGRADE.base,
1602        );
1603
1604        for i in 0..4 {
1605            quorum_proposal.proposal.view_number = ViewNumber::new(i);
1606            let leaf = Leaf2::from_quorum_proposal(&quorum_proposal);
1607            qc.view_number = leaf.view_number();
1608            qc.data.leaf_commit = Committable::commit(&leaf);
1609            vid.data.view_number = leaf.view_number();
1610            da_proposal.data.view_number = leaf.view_number();
1611            chain.push((leaf.clone(), qc.clone(), vid.clone(), da_proposal.clone()));
1612        }
1613
1614        // Per-view artifacts, as persisted at receive time (before any decide).
1615        for (_, _, vid, da) in &chain {
1616            storage.append_da2(da, vid_commitment).await.unwrap();
1617            storage
1618                .append_vid(&convert_proposal(vid.clone()))
1619                .await
1620                .unwrap();
1621        }
1622
1623        let consumer = EventCollector::default();
1624
1625        // Two decides with no processing in between (a lagging/coalescing processor): {0,1}, {2,3}.
1626        for (from, to) in [(0usize, 2usize), (2, 4)] {
1627            let leaf_chain = chain
1628                .iter()
1629                .take(to)
1630                .skip(from)
1631                .map(|(leaf, qc, ..)| (leaf_info(leaf.clone()), qc.clone()))
1632                .collect::<Vec<_>>();
1633            storage
1634                .persist_decided_leaves(
1635                    leaf_chain
1636                        .iter()
1637                        .map(|(leaf, qc)| (leaf, CertificatePair::non_epoch_change(qc.clone()))),
1638                    None,
1639                    &consumer,
1640                )
1641                .await
1642                .unwrap();
1643        }
1644
1645        // Persist-only: leaves saved (anchor advanced), but no events emitted and no GC.
1646        assert_eq!(
1647            storage.load_anchor_view().await.unwrap(),
1648            ViewNumber::new(3),
1649            "anchor view should reflect the latest persisted leaf"
1650        );
1651        assert!(
1652            consumer.leaf_chain().await.is_empty(),
1653            "persist_decided_leaves must not emit decide events"
1654        );
1655        for i in 0..4 {
1656            assert!(
1657                storage
1658                    .load_vid_share(ViewNumber::new(i))
1659                    .await
1660                    .unwrap()
1661                    .is_some(),
1662                "persist_decided_leaves must not garbage collect VID shares"
1663            );
1664            assert!(
1665                storage
1666                    .load_da_proposal(ViewNumber::new(i))
1667                    .await
1668                    .unwrap()
1669                    .is_some(),
1670                "persist_decided_leaves must not garbage collect DA proposals"
1671            );
1672        }
1673
1674        // A failing consumer propagates the error and leaves the cursor un-advanced: nothing is
1675        // GC'd and the range is retried below.
1676        storage
1677            .process_decided_events(ViewNumber::new(3), &FailConsumer)
1678            .await
1679            .unwrap_err();
1680        for i in 0..4 {
1681            assert!(
1682                storage
1683                    .load_da_proposal(ViewNumber::new(i))
1684                    .await
1685                    .unwrap()
1686                    .is_some(),
1687                "a failed process pass must not garbage collect anything"
1688            );
1689        }
1690
1691        // One process pass at the latest view drains the whole backlog, runs GC, and reports the
1692        // cursor it advanced to.
1693        let processed = storage
1694            .process_decided_events(ViewNumber::new(3), &consumer)
1695            .await
1696            .unwrap();
1697        assert_eq!(
1698            processed,
1699            Some(ViewNumber::new(3)),
1700            "process_decided_events should report the highest processed view"
1701        );
1702
1703        // All four leaves delivered, with payloads and VID shares reconstructed from storage.
1704        let leaf_chain = consumer.leaf_chain().await;
1705        assert_eq!(leaf_chain.len(), 4, "{leaf_chain:#?}");
1706        for ((leaf, ..), info) in chain.iter().zip(leaf_chain.iter()) {
1707            assert_eq!(info.leaf, *leaf);
1708            let decided_vid_share = info.vid_share.as_ref().unwrap();
1709            assert_eq!(decided_vid_share.view_number(), leaf.view_number());
1710            assert!(info.leaf.block_payload().is_some());
1711        }
1712
1713        // GC ran for the processed range.
1714        for i in 0..4 {
1715            assert!(
1716                storage
1717                    .load_vid_share(ViewNumber::new(i))
1718                    .await
1719                    .unwrap()
1720                    .is_none(),
1721                "process_decided_events should have garbage collected VID shares"
1722            );
1723            assert!(
1724                storage
1725                    .load_da_proposal(ViewNumber::new(i))
1726                    .await
1727                    .unwrap()
1728                    .is_none(),
1729                "process_decided_events should have garbage collected DA proposals"
1730            );
1731        }
1732
1733        // Re-processing with nothing new is a no-op.
1734        let consumer2 = EventCollector::default();
1735        storage
1736            .process_decided_events(ViewNumber::new(3), &consumer2)
1737            .await
1738            .unwrap();
1739        assert!(
1740            consumer2.leaf_chain().await.is_empty(),
1741            "re-processing an already-drained backlog must emit no events"
1742        );
1743        assert_eq!(
1744            storage.load_anchor_view().await.unwrap(),
1745            ViewNumber::new(3)
1746        );
1747    }
1748
1749    /// A chain of `n` V6 mock leaves at views `0..n` with real, consecutive block heights (the
1750    /// other mock chains in this module reuse the genesis header, so every leaf has height 0).
1751    async fn consecutive_height_chain(n: u64) -> Vec<(Leaf2, QuorumCertificate2<SeqTypes>)> {
1752        let node_state = NodeState::mock().with_genesis_version(versions::NEW_PROTOCOL_VERSION);
1753        let genesis_leaf = Leaf2::genesis(
1754            &ValidatedState::default(),
1755            &node_state,
1756            TEST_VERSIONS.test.base,
1757        )
1758        .await;
1759        let mut quorum_proposal = QuorumProposalWrapper::<SeqTypes> {
1760            proposal: QuorumProposal2::<SeqTypes> {
1761                block_header: genesis_leaf.block_header().clone(),
1762                view_number: ViewNumber::genesis(),
1763                justify_qc: QuorumCertificate2::genesis(
1764                    &ValidatedState::default(),
1765                    &node_state,
1766                    TEST_VERSIONS.test,
1767                )
1768                .await,
1769                upgrade_certificate: None,
1770                view_change_evidence: None,
1771                next_drb_result: None,
1772                next_epoch_justify_qc: None,
1773                epoch: None,
1774                state_cert: None,
1775            },
1776        };
1777        let mut qc = QuorumCertificate2::genesis(
1778            &ValidatedState::default(),
1779            &node_state,
1780            TEST_VERSIONS.test,
1781        )
1782        .await;
1783
1784        let mut chain = vec![];
1785        for i in 0..n {
1786            quorum_proposal.proposal.view_number = ViewNumber::new(i);
1787            *quorum_proposal.proposal.block_header.height_mut() = i;
1788            let leaf = Leaf2::from_quorum_proposal(&quorum_proposal);
1789            qc.view_number = leaf.view_number();
1790            qc.data.leaf_commit = Committable::commit(&leaf);
1791            chain.push((leaf, qc.clone()));
1792        }
1793        chain
1794    }
1795
1796    /// Decide the leaves of `chain` at indices (== views) `range`.
1797    async fn decide_range<P: TestablePersistence>(
1798        storage: &P,
1799        chain: &[(Leaf2, QuorumCertificate2<SeqTypes>)],
1800        range: std::ops::Range<usize>,
1801        consumer: &(impl EventConsumer + 'static),
1802    ) {
1803        let decided_view = ViewNumber::new(range.end as u64 - 1);
1804        let leaf_chain = chain[range]
1805            .iter()
1806            .map(|(leaf, qc)| (leaf_info(leaf.clone()), qc.clone()))
1807            .collect::<Vec<_>>();
1808        storage
1809            .append_decided_leaves(
1810                decided_view,
1811                leaf_chain
1812                    .iter()
1813                    .map(|(leaf, qc)| (leaf, CertificatePair::non_epoch_change(qc.clone()))),
1814                consumer,
1815            )
1816            .await
1817            .unwrap();
1818    }
1819
1820    /// Records the views delivered in decide events, in delivery order.
1821    #[derive(Clone, Debug, Default)]
1822    struct DecideViewCollector {
1823        views: Arc<RwLock<Vec<u64>>>,
1824    }
1825
1826    #[async_trait]
1827    impl EventConsumer for DecideViewCollector {
1828        async fn handle_event(&self, event: &CoordinatorEvent<SeqTypes>) -> anyhow::Result<()> {
1829            // Decide events carry their leaves newest first.
1830            if let CoordinatorEvent::NewDecide { leaf_infos, .. } = event {
1831                self.views.write().await.extend(
1832                    leaf_infos
1833                        .iter()
1834                        .rev()
1835                        .map(|info| info.leaf.view_number().u64()),
1836                );
1837            }
1838            Ok(())
1839        }
1840    }
1841
1842    async fn received_views(consumer: &DecideViewCollector) -> Vec<u64> {
1843        consumer.views.read().await.clone()
1844    }
1845
1846    /// Until a gap-fill decide arrives, event processing holds its cursor at the gap: nothing
1847    /// past it is delivered, and the pending leaves are neither skipped nor dropped.
1848    ///
1849    /// sql-only: the fs backend keeps the pop-oldest cursor and skips gaps (the consumer
1850    /// re-fetches missing blocks), trading gap-fill delivery for cheap decide passes.
1851    #[rstest::rstest]
1852    #[case::postgres(PhantomData::<crate::persistence::sql::postgres::Persistence>)]
1853    #[case::sqlite(PhantomData::<crate::persistence::sql::sqlite::Persistence>)]
1854    #[test_log::test(tokio::test(flavor = "multi_thread"))]
1855    pub async fn test_decide_gap_holds_events<P: TestablePersistence>(#[case] _p: PhantomData<P>) {
1856        let tmp = P::tmp_storage().await;
1857        let storage = P::connect(&tmp).await;
1858        let consumer = DecideViewCollector::default();
1859        let chain = consecutive_height_chain(5).await;
1860
1861        decide_range(&storage, &chain, 0..2, &consumer).await;
1862        assert_eq!(received_views(&consumer).await, vec![0, 1]);
1863
1864        // Views 3..=4 decide while view 2 is still pending.
1865        decide_range(&storage, &chain, 3..5, &consumer).await;
1866        assert_eq!(
1867            received_views(&consumer).await,
1868            vec![0, 1],
1869            "leaves past a height gap must not be delivered before the gap is filled"
1870        );
1871
1872        // Nothing was lost: the pending leaves are still in storage...
1873        assert_eq!(
1874            storage
1875                .load_anchor_leaf()
1876                .await
1877                .unwrap()
1878                .unwrap()
1879                .0
1880                .view_number(),
1881            ViewNumber::new(4)
1882        );
1883
1884        // ...and re-processing neither skips the gap nor emits duplicates.
1885        storage
1886            .process_decided_events(ViewNumber::new(4), &consumer)
1887            .await
1888            .unwrap();
1889        assert_eq!(received_views(&consumer).await, vec![0, 1]);
1890    }
1891
1892    /// Once the gap-fill decide arrives, the consumer receives the gap leaf and everything held
1893    /// up behind it, in order, exactly once each.
1894    ///
1895    /// sql-only: see [`test_decide_gap_holds_events`].
1896    #[rstest::rstest]
1897    #[case::postgres(PhantomData::<crate::persistence::sql::postgres::Persistence>)]
1898    #[case::sqlite(PhantomData::<crate::persistence::sql::sqlite::Persistence>)]
1899    #[test_log::test(tokio::test(flavor = "multi_thread"))]
1900    pub async fn test_decide_gap_fill_delivers_all<P: TestablePersistence>(
1901        #[case] _p: PhantomData<P>,
1902    ) {
1903        let tmp = P::tmp_storage().await;
1904        let storage = P::connect(&tmp).await;
1905        let consumer = DecideViewCollector::default();
1906        let chain = consecutive_height_chain(5).await;
1907
1908        decide_range(&storage, &chain, 0..2, &consumer).await;
1909        decide_range(&storage, &chain, 3..5, &consumer).await;
1910        assert_eq!(received_views(&consumer).await, vec![0, 1]);
1911
1912        // The gap-fill decide of view 2 arrives.
1913        decide_range(&storage, &chain, 2..3, &consumer).await;
1914        // A re-process pass at the watermark must not skip or duplicate anything.
1915        storage
1916            .process_decided_events(ViewNumber::new(4), &consumer)
1917            .await
1918            .unwrap();
1919
1920        assert_eq!(
1921            received_views(&consumer).await,
1922            vec![0, 1, 2, 3, 4],
1923            "after the gap fills, every leaf is delivered in order, exactly once"
1924        );
1925    }
1926
1927    type DecidedCert2s = BTreeMap<ViewNumber, Option<Certificate2<SeqTypes>>>;
1928
1929    /// Records the cert2 of each new-protocol decide event, keyed by its newest leaf's view.
1930    #[derive(Clone, Debug, Default)]
1931    struct Cert2Collector {
1932        cert2s: Arc<RwLock<DecidedCert2s>>,
1933    }
1934
1935    #[async_trait]
1936    impl EventConsumer for Cert2Collector {
1937        async fn handle_event(&self, event: &CoordinatorEvent<SeqTypes>) -> anyhow::Result<()> {
1938            if let CoordinatorEvent::NewDecide {
1939                leaf_infos, cert2, ..
1940            } = event
1941            {
1942                self.cert2s
1943                    .write()
1944                    .await
1945                    .insert(leaf_infos[0].leaf.view_number(), cert2.clone());
1946            }
1947            Ok(())
1948        }
1949    }
1950
1951    /// A decide's cert2 must reach the replayed `NewDecide`, since a single node has no peer to
1952    /// fetch it from.
1953    #[rstest_reuse::apply(persistence_types)]
1954    pub async fn test_decide_event_carries_cert2_persisted_with_leaves<P: TestablePersistence>(
1955        _p: PhantomData<P>,
1956    ) {
1957        let tmp = P::tmp_storage().await;
1958        let storage = P::connect(&tmp).await;
1959        let chain = consecutive_height_chain(2).await;
1960        let (newest, _) = &chain[1];
1961
1962        let vote_data = Vote2Data {
1963            leaf_commit: Committable::commit(newest),
1964            epoch: EpochNumber::new(0),
1965            block_number: newest.height(),
1966        };
1967        let cert2 = Certificate2::new(
1968            vote_data.clone(),
1969            vote_data.commit(),
1970            newest.view_number(),
1971            None,
1972            PhantomData,
1973        );
1974
1975        let decide = CoordinatorEvent::NewDecide {
1976            leaf_infos: chain
1977                .iter()
1978                .rev()
1979                .map(|(leaf, _)| leaf_info(leaf.clone()))
1980                .collect(),
1981            cert1: chain[1].1.clone(),
1982            cert2: Some(cert2.clone()),
1983        };
1984        storage
1985            .persist_event(&decide, &NullEventConsumer)
1986            .await
1987            .expect("a new-protocol decide is persisted");
1988
1989        let consumer = Cert2Collector::default();
1990        storage
1991            .process_decided_events(newest.view_number(), &consumer)
1992            .await
1993            .unwrap();
1994
1995        assert_eq!(
1996            consumer.cert2s.read().await.get(&newest.view_number()),
1997            Some(&Some(cert2))
1998        );
1999    }
2000
2001    #[rstest_reuse::apply(persistence_types)]
2002    pub async fn test_pruning<P: TestablePersistence>(_p: PhantomData<P>) {
2003        let tmp = P::tmp_storage().await;
2004
2005        let mut options = P::options(&tmp);
2006        options.set_view_retention(1);
2007        let storage = options.create().await.unwrap();
2008
2009        // Add some "old" data, from view 0.
2010        let leaf = Leaf::genesis(
2011            &ValidatedState::default(),
2012            &NodeState::mock(),
2013            MOCK_UPGRADE.base,
2014        )
2015        .await;
2016        let leaf_payload = leaf.block_payload().unwrap();
2017        let leaf_payload_bytes_arc = leaf_payload.encode();
2018        let avidm_param = init_avidm_param(2).unwrap();
2019        let weights = vec![1u32; 2];
2020
2021        let ns_table = parse_ns_table(
2022            leaf_payload.byte_len().as_usize(),
2023            &leaf_payload.ns_table().encode(),
2024        );
2025        let (payload_commitment, shares) =
2026            AvidMScheme::ns_disperse(&avidm_param, &weights, &leaf_payload_bytes_arc, ns_table)
2027                .unwrap();
2028
2029        let (pubkey, privkey) = BLSPubKey::generated_from_seed_indexed([0; 32], 1);
2030        let vid_share = convert_proposal(
2031            AvidMDisperseShare::<SeqTypes> {
2032                view_number: ViewNumber::new(0),
2033                payload_commitment,
2034                share: shares[0].clone(),
2035                recipient_key: pubkey,
2036                epoch: None,
2037                target_epoch: None,
2038                common: avidm_param,
2039            }
2040            .to_proposal(&privkey)
2041            .unwrap()
2042            .clone(),
2043        );
2044
2045        let quorum_proposal = QuorumProposalWrapper::<SeqTypes> {
2046            proposal: QuorumProposal2::<SeqTypes> {
2047                block_header: leaf.block_header().clone(),
2048                view_number: ViewNumber::genesis(),
2049                justify_qc: QuorumCertificate::genesis(
2050                    &ValidatedState::default(),
2051                    &NodeState::mock(),
2052                    TEST_VERSIONS.test,
2053                )
2054                .await
2055                .to_qc2(),
2056                upgrade_certificate: None,
2057                view_change_evidence: None,
2058                next_drb_result: None,
2059                next_epoch_justify_qc: None,
2060                epoch: None,
2061                state_cert: None,
2062            },
2063        };
2064        let quorum_proposal_signature =
2065            BLSPubKey::sign(&privkey, &bincode::serialize(&quorum_proposal).unwrap())
2066                .expect("Failed to sign quorum proposal");
2067        let quorum_proposal = Proposal {
2068            data: quorum_proposal,
2069            signature: quorum_proposal_signature,
2070            _pd: Default::default(),
2071        };
2072
2073        let block_payload_signature = BLSPubKey::sign(&privkey, &leaf_payload_bytes_arc)
2074            .expect("Failed to sign block payload");
2075        let da_proposal = Proposal {
2076            data: DaProposal2::<SeqTypes> {
2077                encoded_transactions: leaf_payload_bytes_arc,
2078                metadata: leaf_payload.ns_table().clone(),
2079                view_number: ViewNumber::new(0),
2080                epoch: None,
2081                epoch_transition_indicator: EpochTransitionIndicator::NotInTransition,
2082            },
2083            signature: block_payload_signature,
2084            _pd: Default::default(),
2085        };
2086
2087        let leaf2: Leaf2 = leaf.clone().into();
2088        let vote_data = Vote2Data {
2089            leaf_commit: leaf2.commit(),
2090            epoch: EpochNumber::new(0),
2091            block_number: leaf2.height(),
2092        };
2093        let cert2 = Certificate2::new(
2094            vote_data.clone(),
2095            vote_data.commit(),
2096            ViewNumber::new(0),
2097            None,
2098            PhantomData,
2099        );
2100
2101        storage
2102            .append_da2(&da_proposal, VidCommitment::V1(payload_commitment))
2103            .await
2104            .unwrap();
2105        storage.append_vid(&vid_share).await.unwrap();
2106        storage
2107            .append_quorum_proposal2(&quorum_proposal)
2108            .await
2109            .unwrap();
2110        storage
2111            .append_cert2(ViewNumber::new(0), cert2)
2112            .await
2113            .unwrap();
2114
2115        // Decide a newer view, view 1.
2116        storage
2117            .append_decided_leaves(ViewNumber::new(1), [], &NullEventConsumer)
2118            .await
2119            .unwrap();
2120
2121        // The old data is not more than the retention period (1 view) old, so it should not be
2122        // GCed.
2123        assert_eq!(
2124            storage
2125                .load_da_proposal(ViewNumber::new(0))
2126                .await
2127                .unwrap()
2128                .unwrap(),
2129            da_proposal
2130        );
2131        assert_eq!(
2132            storage
2133                .load_vid_share(ViewNumber::new(0))
2134                .await
2135                .unwrap()
2136                .unwrap(),
2137            vid_share
2138        );
2139        assert_eq!(
2140            storage
2141                .load_quorum_proposal(ViewNumber::new(0))
2142                .await
2143                .unwrap(),
2144            quorum_proposal
2145        );
2146        assert!(
2147            storage
2148                .load_cert2(ViewNumber::new(0))
2149                .await
2150                .unwrap()
2151                .is_some()
2152        );
2153
2154        // Decide an even newer view, triggering GC of the old data.
2155        storage
2156            .append_decided_leaves(ViewNumber::new(2), [], &NullEventConsumer)
2157            .await
2158            .unwrap();
2159        assert!(
2160            storage
2161                .load_da_proposal(ViewNumber::new(0))
2162                .await
2163                .unwrap()
2164                .is_none()
2165        );
2166        assert!(
2167            storage
2168                .load_vid_share(ViewNumber::new(0))
2169                .await
2170                .unwrap()
2171                .is_none()
2172        );
2173        assert!(
2174            storage
2175                .load_quorum_proposal(ViewNumber::new(0))
2176                .await
2177                .is_err()
2178        );
2179        assert!(
2180            storage
2181                .load_cert2(ViewNumber::new(0))
2182                .await
2183                .unwrap()
2184                .is_none()
2185        );
2186    }
2187
2188    async fn assert_events_eq<P: TestablePersistence>(
2189        persistence: &P,
2190        block: u64,
2191        stake_table_fetcher: &Fetcher,
2192        l1_client: &L1Client,
2193        stake_table_contract: Address,
2194    ) -> anyhow::Result<()> {
2195        // Load persisted events
2196        let (stored_l1, events) = persistence.load_events(0, block).await?;
2197        assert!(!events.is_empty());
2198        assert!(stored_l1.is_some());
2199        assert!(events.iter().all(|((l1_block, _), _)| *l1_block <= block));
2200        // Fetch events directly from the contract and compare with persisted data
2201        let contract_events = Fetcher::fetch_events_from_contract(
2202            l1_client.clone(),
2203            stake_table_contract,
2204            None,
2205            block,
2206        )
2207        .await?;
2208        assert_eq!(
2209            contract_events, events,
2210            "Events from contract and persistence do not match"
2211        );
2212
2213        // Fetch events from stake table fetcher and compare with persisted data
2214        let fetched_events = stake_table_fetcher
2215            .fetch_and_store_stake_table_events(stake_table_contract, block)
2216            .await?;
2217        assert_eq!(fetched_events, events);
2218
2219        Ok(())
2220    }
2221
2222    // test for validating stake table event fetching from persistence,
2223    // ensuring that persisted data matches the on-chain events and that event fetcher work correctly.
2224    #[rstest_reuse::apply(persistence_types)]
2225    pub async fn test_stake_table_fetching_from_persistence<P: TestablePersistence>(
2226        _p: PhantomData<P>,
2227    ) -> anyhow::Result<()> {
2228        let epoch_height = 20;
2229
2230        let network_config = TestConfigBuilder::default()
2231            .epoch_height(epoch_height)
2232            .build();
2233
2234        let anvil_provider = network_config.anvil().unwrap();
2235
2236        let query_service_port =
2237            reserve_tcp_port().expect("OS should have ephemeral ports available");
2238        let query_api_options = Options::with_port(query_service_port);
2239
2240        const NODE_COUNT: usize = 2;
2241
2242        let storage = join_all((0..NODE_COUNT).map(|_| P::tmp_storage())).await;
2243        let persistence_options: [_; NODE_COUNT] = storage
2244            .iter()
2245            .map(P::options)
2246            .collect::<Vec<_>>()
2247            .try_into()
2248            .unwrap();
2249
2250        let persistence = persistence_options[0].clone().create().await.unwrap();
2251
2252        let l1_url = network_config.l1_url();
2253
2254        let testnet_config = TestNetworkConfigBuilder::with_num_nodes()
2255            .api_config(query_api_options)
2256            .network_config(network_config.clone())
2257            .persistences(persistence_options.clone())
2258            .build();
2259
2260        //start the network
2261        let test_network = TestNetwork::new(testnet_config, TEST_UPGRADE).await;
2262
2263        let client: Client<ClientErr, SequencerApiVersion> = Client::new(
2264            format!("http://localhost:{query_service_port}")
2265                .parse()
2266                .unwrap(),
2267        );
2268        client.connect(None).await;
2269        tracing::info!(query_service_port, "server running");
2270
2271        // wait until we enter in epoch 3
2272        let _initial_blocks = client
2273            .socket("availability/stream/blocks/0")
2274            .subscribe::<BlockQueryData<SeqTypes>>()
2275            .await
2276            .unwrap()
2277            .take(40)
2278            .try_collect::<Vec<_>>()
2279            .await
2280            .unwrap();
2281        // Load initial persisted events and validate they exist.
2282        let membership_coordinator = test_network.server.node_state().coordinator;
2283
2284        let l1_client = L1Client::new(vec![l1_url]).unwrap();
2285        let node_state = test_network.server.node_state();
2286        let chain_config = node_state.chain_config;
2287        let stake_table_contract = chain_config.stake_table_contract.unwrap();
2288
2289        let current_membership = membership_coordinator.membership();
2290        {
2291            let membership_state = current_membership;
2292            let stake_table_fetcher = membership_state.fetcher();
2293
2294            let block1 = anvil_provider
2295                .get_block_number()
2296                .await
2297                .expect("latest l1 block");
2298
2299            assert_events_eq(
2300                &persistence,
2301                block1,
2302                stake_table_fetcher,
2303                &l1_client,
2304                stake_table_contract,
2305            )
2306            .await?;
2307        }
2308        let _epoch_4_blocks = client
2309            .socket("availability/stream/blocks/0")
2310            .subscribe::<BlockQueryData<SeqTypes>>()
2311            .await
2312            .unwrap()
2313            .take(65)
2314            .try_collect::<Vec<_>>()
2315            .await
2316            .unwrap();
2317        let block2 = anvil_provider
2318            .get_block_number()
2319            .await
2320            .expect("latest l1 block");
2321
2322        {
2323            let membership_state = current_membership;
2324            let stake_table_fetcher = membership_state.fetcher();
2325
2326            assert_events_eq(
2327                &persistence,
2328                block2,
2329                stake_table_fetcher,
2330                &l1_client,
2331                stake_table_contract,
2332            )
2333            .await?;
2334        }
2335        Ok(())
2336    }
2337
2338    /// Nodes still read the whole event history of mainnet's stake table contract, which went from
2339    /// V1 to V2 to V3 as the v4 chain's did.
2340    #[rstest::rstest]
2341    #[case::postgres(PhantomData::<crate::api::sql::postgres::DataSource>)]
2342    #[case::sqlite(PhantomData::<crate::api::sql::sqlite::DataSource>)]
2343    #[test_log::test(tokio::test(flavor = "multi_thread"))]
2344    async fn test_legacy_stake_table_events<D>(#[case] _d: PhantomData<D>) -> anyhow::Result<()>
2345    where
2346        D: TestableSqlDataSource,
2347        <D::Options as PersistenceOptions>::Persistence: TestablePersistence,
2348    {
2349        let chain = LegacyChain::load("v4")?;
2350        let replay = chain.replay::<D>().await?;
2351        let node_state = replay.node.node_state();
2352        let contract = node_state.chain_config.stake_table_contract.unwrap();
2353        let l1_client = L1Client::new(vec![replay.l1.endpoint_url()])?;
2354        let block = l1_client.provider.get_block_number().await?;
2355        let membership = node_state.coordinator.membership();
2356
2357        assert_events_eq(
2358            &replay.persistence().await?,
2359            block,
2360            membership.fetcher(),
2361            &l1_client,
2362            contract,
2363        )
2364        .await?;
2365
2366        let (_, events) = replay.persistence().await?.load_events(0, block).await?;
2367        let has = |f: fn(&StakeTableEvent) -> bool| events.iter().any(|(_, event)| f(event));
2368        ensure!(
2369            has(|e| matches!(e, StakeTableEvent::Register(_))),
2370            "V1 registrations"
2371        );
2372        ensure!(
2373            has(|e| matches!(e, StakeTableEvent::CommissionUpdate(_))),
2374            "V2 commissions"
2375        );
2376        ensure!(
2377            has(|e| matches!(e, StakeTableEvent::X25519KeyUpdate(_))),
2378            "V3 network config"
2379        );
2380        Ok(())
2381    }
2382
2383    #[rstest_reuse::apply(persistence_types)]
2384    pub async fn test_stake_table_background_fetching<P: TestablePersistence>(
2385        #[values(
2386            StakeTableContractVersion::V1,
2387            StakeTableContractVersion::V2,
2388            StakeTableContractVersion::V3
2389        )]
2390        stake_table_version: StakeTableContractVersion,
2391        _p: PhantomData<P>,
2392    ) -> anyhow::Result<()> {
2393        use espresso_types::v0_3::ChainConfig;
2394        use hotshot_contract_adapter::stake_table::StakeTableContractVersion;
2395
2396        let blocks_per_epoch = 10;
2397
2398        let network_config = TestConfigBuilder::<1>::default()
2399            .epoch_height(blocks_per_epoch)
2400            .build();
2401
2402        let anvil_provider = network_config.anvil().unwrap();
2403
2404        let (genesis_state, genesis_stake) = light_client_genesis_from_stake_table(
2405            &network_config.hotshot_config().hotshot_stake_table(),
2406            STAKE_TABLE_CAPACITY_FOR_TEST,
2407        )
2408        .unwrap();
2409
2410        let (_, priv_keys): (Vec<_>, Vec<_>) = (0..20)
2411            .map(|i| <PubKey as SignatureKey>::generated_from_seed_indexed([1; 32], i as u64))
2412            .unzip();
2413        let state_key_pairs = (0..20)
2414            .map(|i| StateKeyPair::generate_from_seed_indexed([2; 32], i as u64))
2415            .collect::<Vec<_>>();
2416
2417        let validators = staking_priv_keys(&priv_keys, &state_key_pairs, &[], 20);
2418
2419        let deployer = ProviderBuilder::new()
2420            .wallet(EthereumWallet::from(network_config.signer().clone()))
2421            .connect_http(network_config.l1_url().clone());
2422
2423        let mut contracts = Contracts::new();
2424        let args = DeployerArgsBuilder::default()
2425            .deployer(deployer.clone())
2426            .rpc_url(network_config.l1_url().clone())
2427            .mock_light_client(true)
2428            .genesis_lc_state(genesis_state)
2429            .genesis_st_state(genesis_stake)
2430            .blocks_per_epoch(blocks_per_epoch)
2431            .epoch_start_block(1)
2432            .exit_escrow_period(U256::from(max(
2433                blocks_per_epoch * 15 + 100,
2434                DEFAULT_EXIT_ESCROW_PERIOD_SECONDS,
2435            )))
2436            .multisig_pauser(network_config.signer().address())
2437            .token_name("Espresso".to_string())
2438            .token_symbol("ESP".to_string())
2439            .initial_token_supply(U256::from(3590000000u64))
2440            .ops_timelock_delay(U256::from(0))
2441            .ops_timelock_admin(network_config.signer().address())
2442            .ops_timelock_proposers(vec![network_config.signer().address()])
2443            .ops_timelock_executors(vec![network_config.signer().address()])
2444            .safe_exit_timelock_delay(U256::from(10))
2445            .safe_exit_timelock_admin(network_config.signer().address())
2446            .safe_exit_timelock_proposers(vec![network_config.signer().address()])
2447            .safe_exit_timelock_executors(vec![network_config.signer().address()])
2448            .build()
2449            .unwrap();
2450
2451        match stake_table_version {
2452            StakeTableContractVersion::V1 => args.deploy_to_stake_table_v1(&mut contracts).await,
2453            StakeTableContractVersion::V2 => args.deploy_to_stake_table_v2(&mut contracts).await,
2454            StakeTableContractVersion::V3 => args.deploy_to_stake_table_v3(&mut contracts).await,
2455        }
2456        .expect("contracts deployed");
2457
2458        let st_addr = contracts
2459            .address(Contract::StakeTableProxy)
2460            .expect("StakeTableProxy deployed");
2461        let l1_url = network_config.l1_url().clone();
2462
2463        let mut planned_txns = StakingTransactions::create(
2464            l1_url.clone(),
2465            &deployer,
2466            st_addr,
2467            validators,
2468            None,
2469            DelegationConfig::MultipleDelegators,
2470        )
2471        .await
2472        .expect("stake table setup failed");
2473
2474        planned_txns
2475            .apply_prerequisites()
2476            .await
2477            .expect("prerequisites failed");
2478
2479        // Ensure we have at least one stake table affecting transaction
2480        planned_txns.apply_one().await.expect("send tx failed");
2481
2482        // new block every 1s
2483        anvil_provider
2484            .anvil_set_interval_mining(1)
2485            .await
2486            .expect("interval mining");
2487
2488        // spawn a separate task
2489        // this is going to keep registering validators and multiple delegators
2490        // the interval mining is set to 1s so each transaction finalization would take atleast 1s
2491        spawn({
2492            async move {
2493                {
2494                    while let Some(receipt) =
2495                        planned_txns.apply_one().await.expect("send tx failed")
2496                    {
2497                        tracing::debug!(?receipt, "transaction finalized");
2498                    }
2499                }
2500            }
2501        });
2502
2503        let storage = P::tmp_storage().await;
2504        let persistence = P::options(&storage).create().await.unwrap();
2505
2506        let l1_client = L1ClientOptions {
2507            stake_table_update_interval: Duration::from_secs(7),
2508            l1_retry_delay: Duration::from_millis(10),
2509            l1_events_max_block_range: 10000,
2510            ..Default::default()
2511        }
2512        .connect(vec![l1_url])
2513        .unwrap();
2514        l1_client.spawn_tasks().await;
2515
2516        let fetcher = Fetcher::new(
2517            Arc::new(NullStateCatchup::default()),
2518            Arc::new(Mutex::new(persistence.clone())),
2519            l1_client.clone(),
2520            ChainConfig {
2521                stake_table_contract: Some(st_addr),
2522                base_fee: 0.into(),
2523                ..Default::default()
2524            },
2525        );
2526
2527        // sleep so that we have enough events
2528        sleep(Duration::from_secs(20)).await;
2529
2530        fetcher.spawn_update_loop().await;
2531        let mut prev_l1_block = 0;
2532        let mut prev_events_len = 0;
2533        for _i in 0..10 {
2534            // Wait for more than update interval to assert that persistence was updated
2535            // L1 update interval is 7s in this test
2536
2537            tokio::time::sleep(std::time::Duration::from_secs(8)).await;
2538
2539            let block = anvil_provider
2540                .get_block_number()
2541                .await
2542                .expect("latest l1 block");
2543
2544            let (read_offset, persisted_events) = persistence.load_events(0, block).await?;
2545            let read_offset = read_offset.unwrap();
2546            let l1_block = match read_offset {
2547                EventsPersistenceRead::Complete => block,
2548                EventsPersistenceRead::UntilL1Block(block) => block,
2549            };
2550
2551            tracing::info!("{l1_block:?}, persistence events = {persisted_events:?}.");
2552            assert!(persisted_events.len() > prev_events_len);
2553
2554            assert!(l1_block > prev_l1_block, "events not updated");
2555
2556            let contract_events =
2557                Fetcher::fetch_events_from_contract(l1_client.clone(), st_addr, None, l1_block)
2558                    .await?;
2559            assert_eq!(persisted_events, contract_events);
2560
2561            prev_l1_block = l1_block;
2562            prev_events_len = persisted_events.len();
2563        }
2564
2565        Ok(())
2566    }
2567
2568    #[rstest_reuse::apply(persistence_types)]
2569    pub async fn test_prefetch_stake_table_events<P: TestablePersistence>(
2570        _p: PhantomData<P>,
2571    ) -> anyhow::Result<()> {
2572        use espresso_types::v0_3::ChainConfig;
2573
2574        let network_config = TestConfigBuilder::<1>::default().build();
2575
2576        let (_, priv_keys): (Vec<_>, Vec<_>) = (0..1)
2577            .map(|i| <PubKey as SignatureKey>::generated_from_seed_indexed([1; 32], i as u64))
2578            .unzip();
2579        let state_key_pairs = (0..1)
2580            .map(|i| StateKeyPair::generate_from_seed_indexed([2; 32], i as u64))
2581            .collect::<Vec<_>>();
2582        let validators = staking_priv_keys(&priv_keys, &state_key_pairs, &[], 1);
2583
2584        let deployer = ProviderBuilder::new()
2585            .wallet(EthereumWallet::from(network_config.signer().clone()))
2586            .connect_http(network_config.l1_url().clone());
2587
2588        let mut contracts = Contracts::new();
2589        let (genesis_state, genesis_stake) = light_client_genesis_from_stake_table(
2590            &network_config.hotshot_config().hotshot_stake_table(),
2591            STAKE_TABLE_CAPACITY_FOR_TEST,
2592        )
2593        .unwrap();
2594        let args = DeployerArgsBuilder::default()
2595            .deployer(deployer.clone())
2596            .rpc_url(network_config.l1_url().clone())
2597            .mock_light_client(true)
2598            .genesis_lc_state(genesis_state)
2599            .genesis_st_state(genesis_stake)
2600            .blocks_per_epoch(10)
2601            .epoch_start_block(1)
2602            .exit_escrow_period(U256::from(DEFAULT_EXIT_ESCROW_PERIOD_SECONDS))
2603            .multisig_pauser(network_config.signer().address())
2604            .token_name("Espresso".to_string())
2605            .token_symbol("ESP".to_string())
2606            .initial_token_supply(U256::from(3590000000u64))
2607            .ops_timelock_delay(U256::from(0))
2608            .ops_timelock_admin(network_config.signer().address())
2609            .ops_timelock_proposers(vec![network_config.signer().address()])
2610            .ops_timelock_executors(vec![network_config.signer().address()])
2611            .safe_exit_timelock_delay(U256::from(10))
2612            .safe_exit_timelock_admin(network_config.signer().address())
2613            .safe_exit_timelock_proposers(vec![network_config.signer().address()])
2614            .safe_exit_timelock_executors(vec![network_config.signer().address()])
2615            .build()
2616            .unwrap();
2617        args.deploy_to_stake_table_v3(&mut contracts)
2618            .await
2619            .expect("contracts deployed");
2620        let st_addr = contracts
2621            .address(Contract::StakeTableProxy)
2622            .expect("StakeTableProxy deployed");
2623
2624        let mut planned_txns = StakingTransactions::create(
2625            network_config.l1_url().clone(),
2626            &deployer,
2627            st_addr,
2628            validators,
2629            None,
2630            DelegationConfig::MultipleDelegators,
2631        )
2632        .await
2633        .expect("stake table setup failed");
2634        planned_txns
2635            .apply_prerequisites()
2636            .await
2637            .expect("prerequisites failed");
2638        // At least one stake-table-affecting event, so the "no events yet" and
2639        // "events prefetched" states are distinguishable.
2640        planned_txns
2641            .apply_one()
2642            .await
2643            .expect("send tx failed")
2644            .expect("at least one registration transaction is queued");
2645
2646        let storage = P::tmp_storage().await;
2647        let persistence = P::options(&storage).create().await.unwrap();
2648
2649        let l1_client = L1ClientOptions {
2650            l1_retry_delay: Duration::from_millis(10),
2651            ..Default::default()
2652        }
2653        .connect(vec![network_config.l1_url().clone()])
2654        .unwrap();
2655        l1_client.spawn_tasks().await;
2656
2657        let fetcher = Fetcher::new(
2658            Arc::new(NullStateCatchup::default()),
2659            Arc::new(Mutex::new(persistence.clone())),
2660            l1_client.clone(),
2661            ChainConfig {
2662                stake_table_contract: Some(st_addr),
2663                base_fee: 0.into(),
2664                ..Default::default()
2665            },
2666        );
2667
2668        let (offset, events) = persistence.load_events(0, 0).await?;
2669        assert_eq!(offset, None);
2670        assert!(events.is_empty());
2671
2672        let finalized = tokio::time::timeout(Duration::from_secs(30), async {
2673            loop {
2674                if let Some(finalized) = l1_client.snapshot().await.finalized {
2675                    return finalized;
2676                }
2677                tokio::time::sleep(Duration::from_millis(50)).await;
2678            }
2679        })
2680        .await
2681        .expect("l1 client finalized a block");
2682
2683        prefetch_stake_table_events(&fetcher, &l1_client, &finalized, Some(st_addr)).await?;
2684
2685        assert_events_eq(
2686            &persistence,
2687            finalized.number,
2688            &fetcher,
2689            &l1_client,
2690            st_addr,
2691        )
2692        .await?;
2693        let (offset, _) = persistence.load_events(0, finalized.number).await?;
2694        assert_eq!(offset, Some(EventsPersistenceRead::Complete));
2695
2696        // A pre-epoch chain has no stake table contract and must not block startup.
2697        tokio::time::timeout(
2698            Duration::from_secs(5),
2699            prefetch_stake_table_events(&Fetcher::mock(), &l1_client, &finalized, None),
2700        )
2701        .await
2702        .expect("prefetch without a contract must return immediately")?;
2703
2704        Ok(())
2705    }
2706
2707    #[rstest_reuse::apply(persistence_types)]
2708    pub async fn test_membership_persistence<P: TestablePersistence>(
2709        _p: PhantomData<P>,
2710    ) -> anyhow::Result<()> {
2711        let tmp = P::tmp_storage().await;
2712        let mut opt = P::options(&tmp);
2713
2714        let storage = opt.create().await.unwrap();
2715
2716        let validator = AuthenticatedValidator::mock();
2717        let mut st = IndexMap::new();
2718        st.insert(validator.account, validator);
2719
2720        storage
2721            .store_stake(EpochNumber::new(10), st.clone(), None, None)
2722            .await?;
2723
2724        let (table, ..) = storage.load_stake(EpochNumber::new(10)).await?.unwrap();
2725        assert_eq!(st, table);
2726
2727        let val2 = AuthenticatedValidator::mock();
2728        let mut st2 = IndexMap::new();
2729        st2.insert(val2.account, val2);
2730        storage
2731            .store_stake(EpochNumber::new(11), st2.clone(), None, None)
2732            .await?;
2733
2734        let tables = storage.load_latest_stake(4).await?.unwrap();
2735        let mut iter = tables.iter();
2736        assert_eq!(
2737            Some(&(EpochNumber::new(11), (st2.clone(), None), None)),
2738            iter.next()
2739        );
2740        assert_eq!(Some(&(EpochNumber::new(10), (st, None), None)), iter.next());
2741        assert_eq!(None, iter.next());
2742
2743        for i in 0..=20 {
2744            storage
2745                .store_stake(EpochNumber::new(i), st2.clone(), None, None)
2746                .await?;
2747        }
2748
2749        let tables = storage.load_latest_stake(5).await?.unwrap();
2750        let mut iter = tables.iter();
2751        assert_eq!(
2752            Some(&(EpochNumber::new(20), (st2.clone(), None), None)),
2753            iter.next()
2754        );
2755        assert_eq!(
2756            Some(&(EpochNumber::new(19), (st2.clone(), None), None)),
2757            iter.next()
2758        );
2759        assert_eq!(
2760            Some(&(EpochNumber::new(18), (st2.clone(), None), None)),
2761            iter.next()
2762        );
2763        assert_eq!(
2764            Some(&(EpochNumber::new(17), (st2.clone(), None), None)),
2765            iter.next()
2766        );
2767        assert_eq!(
2768            Some(&(EpochNumber::new(16), (st2, None), None)),
2769            iter.next()
2770        );
2771        assert_eq!(None, iter.next());
2772
2773        Ok(())
2774    }
2775
2776    // Epoch data is stored piecemeal: `store_drb_result`, `store_epoch_root`
2777    // and `store_stake` each write only their part (one row in SQL, separate
2778    // files in FS). A loader must report data another writer created the
2779    // epoch entry for as absent — regression test for `load_stake` panicking
2780    // on a DRB-only row's NULL stake column.
2781    #[rstest_reuse::apply(persistence_types)]
2782    pub async fn test_load_partially_stored_epoch_data<P: TestablePersistence>(
2783        _p: PhantomData<P>,
2784    ) -> anyhow::Result<()> {
2785        let tmp = P::tmp_storage().await;
2786        let storage = P::connect(&tmp).await;
2787
2788        let instance_state = NodeState::mock();
2789        let validated_state = hotshot_types::traits::ValidatedState::genesis(&instance_state).0;
2790        let leaf: Leaf2 = Leaf::genesis(&validated_state, &instance_state, MOCK_UPGRADE.base)
2791            .await
2792            .into();
2793        let header = leaf.block_header().clone();
2794
2795        let validator = AuthenticatedValidator::mock();
2796        let mut stake = IndexMap::new();
2797        stake.insert(validator.account, validator);
2798
2799        // Nothing stored yet.
2800        let epoch = EpochNumber::new(1);
2801        assert!(storage.load_stake(epoch).await?.is_none());
2802        assert!(storage.load_drb_result(epoch).await?.is_none());
2803        assert!(storage.load_epoch_root(epoch).await?.is_none());
2804
2805        // A DRB result alone; the other loaders report their data as absent.
2806        storage.store_drb_result(epoch, [1; 32]).await?;
2807        assert!(storage.load_stake(epoch).await?.is_none());
2808        assert_eq!(storage.load_drb_result(epoch).await?, Some([1; 32]));
2809        assert!(storage.load_epoch_root(epoch).await?.is_none());
2810
2811        // A stake table alone, likewise.
2812        let epoch2 = EpochNumber::new(2);
2813        storage
2814            .store_stake(epoch2, stake.clone(), None, None)
2815            .await?;
2816        assert!(storage.load_stake(epoch2).await?.is_some());
2817        assert!(storage.load_drb_result(epoch2).await?.is_none());
2818        assert!(storage.load_epoch_root(epoch2).await?.is_none());
2819
2820        // With everything stored for one epoch, each loader returns its
2821        // value and none has clobbered the others.
2822        storage.store_epoch_root(epoch, header.clone()).await?;
2823        storage
2824            .store_stake(epoch, stake.clone(), None, None)
2825            .await?;
2826        let (table, ..) = storage.load_stake(epoch).await?.unwrap();
2827        assert_eq!(stake, table);
2828        assert_eq!(storage.load_drb_result(epoch).await?, Some([1; 32]));
2829        assert_eq!(storage.load_epoch_root(epoch).await?, Some(header));
2830
2831        Ok(())
2832    }
2833
2834    #[rstest_reuse::apply(persistence_types)]
2835    pub async fn test_delete_stake_tables<P: TestablePersistence>(
2836        _p: PhantomData<P>,
2837    ) -> anyhow::Result<()> {
2838        let tmp = P::tmp_storage().await;
2839        let mut opt = P::options(&tmp);
2840        let storage = opt.create().await.unwrap();
2841
2842        let event1 = StakeTableEvent::Delegate(Delegated {
2843            delegator: Address::ZERO,
2844            validator: Address::ZERO,
2845            amount: U256::from(100),
2846        });
2847        let event2 = StakeTableEvent::Delegate(Delegated {
2848            delegator: Address::ZERO,
2849            validator: Address::ZERO,
2850            amount: U256::from(200),
2851        });
2852
2853        let l1_block = 42u64;
2854        let events: Vec<(EventKey, StakeTableEvent)> =
2855            vec![((l1_block, 0), event1), ((l1_block, 1), event2)];
2856
2857        storage.store_events(l1_block, events.clone()).await?;
2858
2859        let (read_offset, loaded_events) = storage.load_events(0_u64, l1_block).await?;
2860        assert!(read_offset.is_some());
2861        assert_eq!(loaded_events.len(), 2);
2862        assert_eq!(loaded_events, events);
2863
2864        // Store some validators
2865        let v = RegisteredValidator::mock();
2866        let mut vmap = IndexMap::new();
2867        vmap.insert(v.account, v);
2868        storage
2869            .store_all_validators(EpochNumber::new(1), vmap.clone())
2870            .await?;
2871
2872        let loaded = storage
2873            .load_all_validators(EpochNumber::new(1), 0, 10)
2874            .await?;
2875        assert_eq!(loaded.len(), 1);
2876
2877        storage.delete_stake_tables().await?;
2878
2879        // Events cleared
2880        let (read_offset, loaded_events) = storage.load_events(0_u64, l1_block).await?;
2881        assert!(read_offset.is_none());
2882        assert!(loaded_events.is_empty());
2883
2884        // Validators cleared
2885        let loaded = storage
2886            .load_all_validators(EpochNumber::new(1), 0, 10)
2887            .await
2888            .unwrap_or_default();
2889        assert!(loaded.is_empty());
2890
2891        let event3 = StakeTableEvent::Delegate(Delegated {
2892            delegator: Address::ZERO,
2893            validator: Address::ZERO,
2894            amount: U256::from(300),
2895        });
2896        let new_events: Vec<(EventKey, StakeTableEvent)> = vec![((l1_block, 0), event3)];
2897        storage.store_events(l1_block, new_events.clone()).await?;
2898
2899        let (read_offset, loaded_events) = storage.load_events(0_u64, l1_block).await?;
2900        assert!(read_offset.is_some());
2901        assert_eq!(loaded_events.len(), 1);
2902        assert_eq!(loaded_events, new_events);
2903
2904        Ok(())
2905    }
2906
2907    #[rstest_reuse::apply(persistence_types)]
2908    pub async fn test_store_and_load_all_validators<P: TestablePersistence>(
2909        _p: PhantomData<P>,
2910    ) -> anyhow::Result<()> {
2911        let tmp = P::tmp_storage().await;
2912        let mut opt = P::options(&tmp);
2913        let storage = opt.create().await.unwrap();
2914
2915        let mut vmap1 = IndexMap::new();
2916        for _i in 0..25 {
2917            let v = RegisteredValidator::mock();
2918            vmap1.insert(v.account, v);
2919        }
2920        storage
2921            .store_all_validators(EpochNumber::new(10), vmap1.clone())
2922            .await?;
2923
2924        let mut expected_all: Vec<_> = vmap1.clone().into_values().collect();
2925        expected_all.sort_by_key(|v| v.account);
2926
2927        // Load all
2928        let loaded_all = storage
2929            .load_all_validators(EpochNumber::new(10), 0, 100)
2930            .await?;
2931        // SQLite returns a different ordered list even though there is an `ORDER BY address ASC` clause
2932        assert_eq!(expected_all, loaded_all);
2933
2934        // Load first 10
2935        let loaded_first_10 = storage
2936            .load_all_validators(EpochNumber::new(10), 0, 10)
2937            .await?;
2938
2939        assert_eq!(expected_all[..10], loaded_first_10);
2940
2941        // Load next 10
2942        let loaded_next_10 = storage
2943            .load_all_validators(EpochNumber::new(10), 10, 10)
2944            .await?;
2945
2946        assert_eq!(expected_all[10..20], loaded_next_10);
2947
2948        // Load remaining 5
2949        let loaded_last_5 = storage
2950            .load_all_validators(EpochNumber::new(10), 20, 10)
2951            .await?;
2952
2953        assert_eq!(expected_all[20..], loaded_last_5);
2954
2955        // offset beyond size should return empty
2956        let loaded_empty = storage
2957            .load_all_validators(EpochNumber::new(10), 100, 10)
2958            .await?;
2959        assert!(loaded_empty.is_empty());
2960
2961        // epoch 11
2962        let validator2 = RegisteredValidator::mock();
2963        let mut vmap2 = IndexMap::new();
2964        vmap2.insert(validator2.account, validator2.clone());
2965
2966        storage
2967            .store_all_validators(EpochNumber::new(11), vmap2.clone())
2968            .await?;
2969
2970        let mut expected_epoch11: Vec<_> = vmap2.clone().into_values().collect();
2971        expected_epoch11.sort_by_key(|v| v.account);
2972
2973        let loaded2 = storage
2974            .load_all_validators(EpochNumber::new(11), 0, 100)
2975            .await?;
2976
2977        assert_eq!(expected_epoch11, loaded2);
2978
2979        // Epoch 10 still there
2980        let loaded1_again = storage
2981            .load_all_validators(EpochNumber::new(10), 0, 100)
2982            .await?;
2983
2984        assert_eq!(expected_all, loaded1_again);
2985
2986        Ok(())
2987    }
2988
2989    /// Non-consecutive decided leaves are delivered as separate events, and only the event of the
2990    /// directly finalized leaf carries its cert2.
2991    #[rstest_reuse::apply(persistence_types)]
2992    pub async fn test_non_consecutive_decide<P: TestablePersistence>(_p: PhantomData<P>) {
2993        let tmp = P::tmp_storage().await;
2994        let storage = P::connect(&tmp).await;
2995
2996        let node_state = NodeState::mock().with_genesis_version(versions::NEW_PROTOCOL_VERSION);
2997        let genesis_leaf: Leaf2 =
2998            Leaf2::genesis(&ValidatedState::default(), &node_state, TEST_UPGRADE.base).await;
2999        let mut quorum_proposal = QuorumProposalWrapper::<SeqTypes> {
3000            proposal: QuorumProposal2::<SeqTypes> {
3001                epoch: None,
3002                block_header: genesis_leaf.block_header().clone(),
3003                view_number: genesis_leaf.view_number(),
3004                justify_qc: QuorumCertificate2::genesis(
3005                    &ValidatedState::default(),
3006                    &node_state,
3007                    TEST_UPGRADE,
3008                )
3009                .await,
3010                upgrade_certificate: None,
3011                view_change_evidence: None,
3012                next_drb_result: None,
3013                next_epoch_justify_qc: None,
3014                state_cert: None,
3015            },
3016        };
3017
3018        let leaf0 = Leaf2::from_quorum_proposal(&quorum_proposal);
3019
3020        quorum_proposal.proposal.view_number = ViewNumber::new(2);
3021        *quorum_proposal.proposal.block_header.height_mut() = 2;
3022        quorum_proposal.proposal.justify_qc.view_number = ViewNumber::new(1);
3023        let leaf2 = Leaf2::from_quorum_proposal(&quorum_proposal);
3024
3025        let newest_view = ViewNumber::new(2 + DECIDE_GAP_FILL_HORIZON);
3026        quorum_proposal.proposal.view_number = newest_view;
3027        *quorum_proposal.proposal.block_header.height_mut() = newest_view.u64();
3028        let newest_leaf = Leaf2::from_quorum_proposal(&quorum_proposal);
3029
3030        let mut qc0 = leaf0.justify_qc().clone();
3031        qc0.data.leaf_commit = Committable::commit(&leaf0);
3032
3033        let mut qc2 = leaf2.justify_qc().clone();
3034        qc2.view_number += 1;
3035        qc2.data.leaf_commit = Committable::commit(&leaf2);
3036
3037        let mut newest_qc = newest_leaf.justify_qc().clone();
3038        newest_qc.view_number = newest_view;
3039        newest_qc.data.leaf_commit = Committable::commit(&newest_leaf);
3040
3041        let cert2 = {
3042            let data = Vote2Data {
3043                leaf_commit: Committable::commit(&leaf2),
3044                epoch: EpochNumber::new(1),
3045                block_number: leaf2.height(),
3046            };
3047            Certificate2::new(
3048                data.clone(),
3049                data.commit(),
3050                leaf2.view_number(),
3051                None,
3052                PhantomData,
3053            )
3054        };
3055
3056        // Decide the first leaf, but fail to generate a decide event.
3057        storage
3058            .append_decided_leaves(
3059                ViewNumber::new(0),
3060                [(
3061                    &leaf_info(leaf0.clone()),
3062                    CertificatePair::non_epoch_change(qc0),
3063                )],
3064                &FailConsumer,
3065            )
3066            .await
3067            .unwrap();
3068
3069        // Far enough ahead that the gap below `leaf2` falls outside `DECIDE_GAP_FILL_HORIZON`, so
3070        // `leaf2` is delivered rather than held for a gap-fill decide.
3071        storage
3072            .persist_decided_leaves(
3073                [(
3074                    &leaf_info(newest_leaf),
3075                    CertificatePair::non_epoch_change(newest_qc),
3076                )],
3077                None,
3078                &NullEventConsumer,
3079            )
3080            .await
3081            .unwrap();
3082        storage
3083            .append_cert2(leaf2.view_number(), cert2.clone())
3084            .await
3085            .unwrap();
3086        let consumer = EventCollector::default();
3087        storage
3088            .append_decided_leaves(
3089                ViewNumber::new(2),
3090                [(
3091                    &leaf_info(leaf2.clone()),
3092                    CertificatePair::non_epoch_change(qc2),
3093                )],
3094                &consumer,
3095            )
3096            .await
3097            .unwrap();
3098
3099        let events = consumer.events.read().await;
3100        assert_eq!(events.len(), 2, "{events:#?}");
3101
3102        let CoordinatorEvent::NewDecide {
3103            leaf_infos: leaf_infos0,
3104            cert2: cert2_0,
3105            ..
3106        } = &events[0]
3107        else {
3108            panic!("expected decide event, got {:?}", events[0]);
3109        };
3110        assert_eq!(leaf_infos0.len(), 1);
3111        assert_eq!(leaf_infos0[0].leaf, leaf0);
3112        assert_eq!(*cert2_0, None);
3113
3114        let CoordinatorEvent::NewDecide {
3115            leaf_infos: leaf_infos2,
3116            cert2: cert2_2,
3117            ..
3118        } = &events[1]
3119        else {
3120            panic!("expected decide event, got {:?}", events[1]);
3121        };
3122        assert_eq!(leaf_infos2.len(), 1);
3123        assert_eq!(leaf_infos2[0].leaf, leaf2);
3124        assert_eq!(cert2_2.as_ref(), Some(&cert2));
3125    }
3126
3127    /// Without the query module, storage keeps what consensus needs to run and restart, and
3128    /// drops what only a decide event consumer would read.
3129    #[rstest_reuse::apply(persistence_types)]
3130    pub async fn test_consensus_only_decide<P: TestablePersistence>(_p: PhantomData<P>) {
3131        let tmp = P::tmp_storage().await;
3132
3133        let node_state = NodeState::mock().with_genesis_version(versions::NEW_PROTOCOL_VERSION);
3134        let leaf: Leaf2 =
3135            Leaf2::genesis(&ValidatedState::default(), &node_state, TEST_UPGRADE.base).await;
3136        let leaf_payload = leaf.block_payload().unwrap();
3137        let leaf_payload_bytes_arc = leaf_payload.encode();
3138        let avidm_param = init_avidm_gf2_param(2).unwrap();
3139        let weights = Vec::from([1u32; 2]);
3140        let ns_table = parse_ns_table(
3141            leaf_payload.byte_len().as_usize(),
3142            &leaf_payload.ns_table().encode(),
3143        );
3144        let (payload_commitment, common, shares) =
3145            AvidmGf2Scheme::ns_disperse(&avidm_param, &weights, &leaf_payload_bytes_arc, ns_table)
3146                .unwrap();
3147
3148        let (pubkey, privkey) = BLSPubKey::generated_from_seed_indexed([0; 32], 1);
3149        let mut vid = AvidmGf2DisperseShare::<SeqTypes> {
3150            view_number: ViewNumber::new(0),
3151            payload_commitment,
3152            share: shares[0].clone(),
3153            recipient_key: pubkey,
3154            epoch: Some(EpochNumber::new(0)),
3155            target_epoch: Some(EpochNumber::new(0)),
3156            common,
3157        }
3158        .to_proposal(&privkey)
3159        .unwrap();
3160        let mut da_proposal = Proposal {
3161            data: DaProposal2::<SeqTypes> {
3162                encoded_transactions: leaf_payload_bytes_arc.clone(),
3163                metadata: leaf_payload.ns_table().clone(),
3164                view_number: ViewNumber::new(0),
3165                epoch: Some(EpochNumber::new(0)),
3166                epoch_transition_indicator: EpochTransitionIndicator::NotInTransition,
3167            },
3168            signature: BLSPubKey::sign(&privkey, &leaf_payload_bytes_arc).unwrap(),
3169            _pd: Default::default(),
3170        };
3171        let vid_commitment = vid_commitment(
3172            &leaf_payload_bytes_arc,
3173            &leaf.block_header().metadata().encode(),
3174            2,
3175            TEST_UPGRADE.base,
3176        );
3177        let mut quorum_proposal = QuorumProposalWrapper::<SeqTypes> {
3178            proposal: QuorumProposal2::<SeqTypes> {
3179                block_header: leaf.block_header().clone(),
3180                view_number: ViewNumber::genesis(),
3181                justify_qc: QuorumCertificate2::genesis(
3182                    &ValidatedState::default(),
3183                    &node_state,
3184                    TEST_UPGRADE,
3185                )
3186                .await,
3187                upgrade_certificate: None,
3188                view_change_evidence: None,
3189                next_drb_result: None,
3190                next_epoch_justify_qc: None,
3191                epoch: None,
3192                state_cert: None,
3193            },
3194        };
3195        let mut qc =
3196            QuorumCertificate2::genesis(&ValidatedState::default(), &node_state, TEST_UPGRADE)
3197                .await;
3198
3199        let mut chain = Vec::new();
3200        for i in 0..4 {
3201            quorum_proposal.proposal.view_number = ViewNumber::new(i);
3202            *quorum_proposal.proposal.block_header.height_mut() = i;
3203            let leaf = Leaf2::from_quorum_proposal(&quorum_proposal);
3204            qc.view_number = leaf.view_number();
3205            qc.data.leaf_commit = Committable::commit(&leaf);
3206            vid.data.view_number = leaf.view_number();
3207            da_proposal.data.view_number = leaf.view_number();
3208            chain.push((
3209                leaf,
3210                CertificatePair::non_epoch_change(qc.clone()),
3211                convert_proposal(vid.clone()),
3212                da_proposal.clone(),
3213            ));
3214        }
3215        let state_cert = LightClientStateUpdateCertificateV2::<SeqTypes> {
3216            epoch: EpochNumber::new(1),
3217            light_client_state: LightClientState {
3218                view_number: 2,
3219                ..Default::default()
3220            },
3221            next_stake_table_state: StakeTableState::default(),
3222            signatures: Vec::new(),
3223            auth_root: Default::default(),
3224        };
3225
3226        let mut opt = P::options(&tmp);
3227        opt.set_consensus_only();
3228        let storage = opt.create().await.unwrap();
3229        for (_, _, vid, da) in &chain {
3230            storage.append_vid(vid).await.unwrap();
3231            storage.append_da2(da, vid_commitment).await.unwrap();
3232        }
3233        storage.add_state_cert(state_cert.clone()).await.unwrap();
3234        for i in 0..4 {
3235            assert_eq!(
3236                storage.load_da_proposal(ViewNumber::new(i)).await.unwrap(),
3237                None
3238            );
3239        }
3240
3241        let consumer = EventCollector::default();
3242        let leaf_chain = chain[..3]
3243            .iter()
3244            .map(|(leaf, qc, ..)| (leaf_info(leaf.clone()), qc.clone()))
3245            .collect::<Vec<_>>();
3246        storage
3247            .append_decided_leaves(
3248                ViewNumber::new(2),
3249                leaf_chain.iter().map(|(info, qc)| (info, qc.clone())),
3250                &consumer,
3251            )
3252            .await
3253            .unwrap();
3254
3255        assert!(consumer.leaf_chain().await.is_empty());
3256        let (anchor_leaf, anchor_qc, ..) = &chain[2];
3257        assert_eq!(
3258            storage.load_anchor_leaf().await.unwrap(),
3259            Some((anchor_leaf.clone(), anchor_qc.clone()))
3260        );
3261        assert_eq!(
3262            storage.get_state_cert_by_epoch(1).await.unwrap(),
3263            Some(state_cert.clone())
3264        );
3265        for i in 0..2 {
3266            assert_eq!(
3267                storage.load_vid_share(ViewNumber::new(i)).await.unwrap(),
3268                None
3269            );
3270        }
3271        // Consensus can still decide view 3, so pruning must not reach past the decided view.
3272        assert!(
3273            storage
3274                .load_vid_share(ViewNumber::new(3))
3275                .await
3276                .unwrap()
3277                .is_some()
3278        );
3279        drop(storage);
3280
3281        let mut opt = P::options(&tmp);
3282        opt.set_consensus_only();
3283        let storage = opt.create().await.unwrap();
3284        let (initializer, anchor_view) = storage
3285            .load_consensus_state(
3286                NodeState::mock(),
3287                Upgrade::trivial(versions::EPOCH_REWARD_VERSION),
3288            )
3289            .await
3290            .unwrap();
3291        assert_eq!(initializer.anchor_leaf(), anchor_leaf);
3292        assert_eq!(anchor_view, Some(ViewNumber::new(2)));
3293        assert_eq!(storage.load_state_cert().await.unwrap(), Some(state_cert));
3294    }
3295}
espresso_node/persistence.rs

espresso_node/
persistence.rs
