Fetch Memberships from l1

.config/nextest.toml

@@ -5,10 +5,13 @@ default-filter = 'not (test(slow_) | package(tests))'
 retries = 2
 
 [[profile.default.overrides]]
-# This test is fast if it works (usually about 150ms) but sometimes hangs,
-# especially when running concurrently with other tests. Kill it after a while
-# and retry up to 10 times.
-filter = 'test(test_process_client_handling_stream_subscribe_node_identity)'
+# These tests are fast if they work (usually about 150ms) but sometimes they
+# hang, especially when running Concurrently with other tests. Kill after a
+# while and retry up to 10 times to avoid failing the test suite.
+filter = """
+test(test_process_client_handling_stream_subscribe_node_identity) |
+test(test_process_client_handling_stream_subscribe_voters)
+"""
 slow-timeout = { period = "2s", terminate-after = 1 }
 retries  = 10
 

.env

@@ -58,10 +58,10 @@ ESPRESSO_BUILDER_ETH_ACCOUNT_INDEX=8
 ESPRESSO_DEPLOYER_ACCOUNT_INDEX=9
 
 # Contracts
-ESPRESSO_SEQUENCER_LIGHT_CLIENT_PROXY_ADDRESS=0xf7cd8fa9b94db2aa972023b379c7f72c65e4de9d
+ESPRESSO_SEQUENCER_LIGHT_CLIENT_PROXY_ADDRESS=0xed1db453c3156ff3155a97ad217b3087d5dc5f6e
 ESPRESSO_SEQUENCER_LIGHTCLIENT_ADDRESS=$ESPRESSO_SEQUENCER_LIGHT_CLIENT_PROXY_ADDRESS
 ESPRESSO_SEQUENCER_PERMISSIONED_PROVER=0x14dc79964da2c08b23698b3d3cc7ca32193d9955
-SPRESSO_SEQUENCER_PERMISSIONED_STAKE_TABLE_ADDRESS=0x8ce361602b935680e8dec218b820ff5056beb7af
+ESPRESSO_SEQUENCER_PERMISSIONED_STAKE_TABLE_ADDRESS=0x8ce361602b935680e8dec218b820ff5056beb7af
 
 # Example sequencer demo private keys
 ESPRESSO_DEMO_SEQUENCER_STAKING_PRIVATE_KEY_0=BLS_SIGNING_KEY~lNDh4Pn-pTAyzyprOAFdXHwhrKhEwqwtMtkD3CZF4x3o

sequencer/Cargo.toml

@@ -48,7 +48,6 @@ async-lock = { workspace = true }
 async-once-cell = { workspace = true }
 async-trait = { workspace = true }
 bincode = { workspace = true }
-todo_by = "0.3"
 
 # CDN imports
 cdn-broker = { workspace = true }
@@ -117,6 +116,7 @@ tagged-base64 = { workspace = true }
 tempfile = { workspace = true, optional = true }
 tide-disco = { workspace = true }
 time = { workspace = true }
+todo_by = "0.3"
 tokio = { workspace = true }
 toml = { workspace = true }
 tracing = { workspace = true }

sequencer/src/api.rs

@@ -7,8 +7,8 @@ use data_source::{CatchupDataSource, StakeTableDataSource, SubmitDataSource};
 use derivative::Derivative;
 use espresso_types::{
     retain_accounts, v0::traits::SequencerPersistence, v0_99::ChainConfig, AccountQueryData,
-    BlockMerkleTree, FeeAccount, FeeAccountProof, FeeMerkleTree, NodeState, PubKey,
-    StaticCommittee, Transaction, ValidatedState,
+    BlockMerkleTree, FeeAccount, FeeAccountProof, FeeMerkleTree, NodeState, PubKey, Transaction,
+    ValidatedState,
 };
 use futures::{
     future::{BoxFuture, Future, FutureExt},
@@ -187,7 +187,7 @@ impl<N: ConnectedNetwork<PubKey>, V: Versions, P: SequencerPersistence>
             self.consensus().await.read().await.cur_epoch().await
         };
 
-        <StaticCommittee as Membership<SeqTypes>>::stake_table(
+        <SeqTypes as NodeType>::Membership::stake_table(
             &self.consensus().await.read().await.memberships,
             epoch,
         )

sequencer/src/context.rs

@@ -3,7 +3,7 @@ use async_lock::RwLock;
 use derivative::Derivative;
 use espresso_types::{
     v0::traits::{EventConsumer as PersistenceEventConsumer, SequencerPersistence},
-    NodeState, PubKey, StaticCommittee, Transaction, ValidatedState,
+    EpochCommittees, NodeState, PubKey, Transaction, ValidatedState,
 };
 use futures::{
     future::{join_all, Future},
@@ -80,7 +80,7 @@ impl<N: ConnectedNetwork<PubKey>, P: SequencerPersistence, V: Versions> Sequence
     pub async fn init(
         network_config: NetworkConfig<PubKey>,
         validator_config: ValidatorConfig<<SeqTypes as NodeType>::SignatureKey>,
-        membership: StaticCommittee,
+        membership: EpochCommittees,
         instance_state: NodeState,
         persistence: P,
         network: Arc<N>,

sequencer/src/genesis.rs

@@ -418,7 +418,7 @@ mod test {
                 fee_recipient: FeeAccount::default(),
                 fee_contract: Some(Address::default()),
                 bid_recipient: None,
-                stake_table_contract: Some(Address::default())
+                stake_table_contract: None
             }
         );
         assert_eq!(

sequencer/src/lib.rs

@@ -15,13 +15,12 @@ mod message_compat_tests;
 use anyhow::Context;
 use catchup::StatePeers;
 use context::SequencerContext;
-use espresso_types::StaticCommittee;
+use espresso_types::EpochCommittees;
 use espresso_types::{
     traits::EventConsumer, BackoffParams, L1ClientOptions, NodeState, PubKey, SeqTypes,
     SolverAuctionResultsProvider, ValidatedState,
 };
 use genesis::L1Finalized;
-use hotshot_types::traits::election::Membership;
 use proposal_fetcher::ProposalFetcherConfig;
 use std::sync::Arc;
 use tokio::select;
@@ -476,14 +475,15 @@ pub async fn init_node<P: SequencerPersistence, V: Versions>(
         node_id: node_index,
         upgrades: genesis.upgrades,
         current_version: V::Base::VERSION,
+        epoch_height: None,
     };
 
     // Create the HotShot membership
-    let membership = StaticCommittee::new_stake(
+    let membership = EpochCommittees::new_stake(
         network_config.config.known_nodes_with_stake.clone(),
         network_config.config.known_nodes_with_stake.clone(),
         &instance_state,
-        Default::default(),
+        network_config.config.epoch_height,
     );
 
     // Initialize the Libp2p network
@@ -960,12 +960,15 @@ pub mod testing {
             )
             .with_current_version(V::Base::version())
             .with_genesis(state)
+            .with_epoch_height(config.epoch_height)
             .with_upgrades(upgrades);
 
             // Create the HotShot membership
-            let membership = StaticCommittee::new(
+            let membership = EpochCommittees::new_stake(
                 config.known_nodes_with_stake.clone(),
                 config.known_nodes_with_stake.clone(),
+                &node_state,
+                100,
             );
 
             tracing::info!(

sequencer/src/message_compat_tests.rs

@@ -37,8 +37,7 @@ use hotshot_types::{
         ViewSyncFinalizeVote, ViewSyncPreCommitData, ViewSyncPreCommitVote,
     },
     traits::{
-        election::Membership, node_implementation::ConsensusTime, signature_key::SignatureKey,
-        BlockPayload, EncodeBytes,
+        node_implementation::ConsensusTime, signature_key::SignatureKey, BlockPayload, EncodeBytes,
     },
     vid::vid_scheme,
 };
@@ -52,14 +51,15 @@ use vbs::{
 
 #[cfg(feature = "testing")]
 async fn test_message_compat<Ver: StaticVersionType>(_ver: Ver) {
-    use espresso_types::{Leaf, Payload, SeqTypes, StaticCommittee, Transaction};
+    use espresso_types::{EpochCommittees, Leaf, Payload, SeqTypes, Transaction};
     use hotshot_example_types::node_types::TestVersions;
     use hotshot_types::PeerConfig;
 
     let (sender, priv_key) = PubKey::generated_from_seed_indexed(Default::default(), 0);
     let signature = PubKey::sign(&priv_key, &[]).unwrap();
     let committee = vec![PeerConfig::default()]; /* one committee member, necessary to generate a VID share */
-    let membership = StaticCommittee::new(committee.clone(), committee);
+    let membership =
+        EpochCommittees::new_stake(committee.clone(), committee, &NodeState::default(), 10);
     let upgrade_data = UpgradeProposalData {
         old_version: Version { major: 0, minor: 1 },
         new_version: Version { major: 1, minor: 0 },

types/Cargo.toml

@@ -11,6 +11,7 @@ testing = []
 anyhow = { workspace = true }
 ark-serialize = { workspace = true }
 async-broadcast = { workspace = true }
+async-lock = { workspace = true }
 async-trait = { workspace = true }
 base64-bytes = { workspace = true }
 bincode = { workspace = true }

types/src/v0/impls/instance_state.rs

@@ -24,6 +24,7 @@ pub struct NodeState {
     pub genesis_header: GenesisHeader,
     pub genesis_state: ValidatedState,
     pub l1_genesis: Option<L1BlockInfo>,
+    pub epoch_height: Option<u64>,
 
     /// Map containing all planned and executed upgrades.
     ///
@@ -64,6 +65,7 @@ impl NodeState {
             l1_genesis: None,
             upgrades: Default::default(),
             current_version,
+            epoch_height: None,
         }
     }
 
@@ -130,6 +132,13 @@ impl NodeState {
         self.current_version = ver;
         self
     }
+
+    // TODO remove following `Memberships` trait update:
+    // https://github.com/EspressoSystems/HotShot/issues/3966
+    pub fn with_epoch_height(mut self, epoch_height: u64) -> Self {
+        self.epoch_height = Some(epoch_height);
+        self
+    }
 }
 
 // This allows us to turn on `Default` on InstanceState trait

types/src/v0/impls/l1.rs

@@ -13,14 +13,15 @@ use futures::{
     future::Future,
     stream::{self, StreamExt},
 };
-use hotshot_types::traits::metrics::Metrics;
+use hotshot_types::{data::EpochNumber, traits::metrics::Metrics};
 use lru::LruCache;
 use serde::{de::DeserializeOwned, Serialize};
 use std::{
     cmp::{min, Ordering},
+    collections::BTreeMap,
     fmt::Debug,
     num::NonZeroUsize,
-    sync::Arc,
+    sync::{self, Arc},
     time::Duration,
 };
 use tokio::{
@@ -33,8 +34,7 @@ use url::Url;
 
 use super::{L1BlockInfo, L1ClientMetrics, L1State, L1UpdateTask, RpcClient};
 use crate::{
-    v0::impls::stake_table::StakeTables, FeeInfo, L1Client, L1ClientOptions, L1Event,
-    L1ReconnectTask, L1Snapshot,
+    v0_3::StakeTables, FeeInfo, L1Client, L1ClientOptions, L1Event, L1ReconnectTask, L1Snapshot,
 };
 
 impl PartialOrd for L1BlockInfo {
@@ -344,6 +344,7 @@ impl L1Client {
             provider: Arc::new(provider),
             events_max_block_range: opt.l1_events_max_block_range,
             state: Arc::new(Mutex::new(L1State::new(opt.l1_blocks_cache_size))),
+            stake_table_state: Arc::new(sync::RwLock::new(BTreeMap::new())),
             sender,
             receiver: receiver.deactivate(),
             update_task: Default::default(),
@@ -376,6 +377,39 @@ impl L1Client {
             *update_task = Some(spawn(self.update_loop()));
         }
     }
+    pub async fn update_membership(
+        &self,
+        contract: Address,
+        l1_block_number: u64,
+        epoch: EpochNumber,
+    ) {
+        let retry_delay = self.retry_delay;
+        let state = self.stake_table_state.clone();
+
+        let span = tracing::warn_span!("L1 client memberships update");
+
+        async move {
+            loop {
+                match self.get_stake_table(contract, l1_block_number).await {
+                    Err(err) => {
+                        tracing::warn!(
+                            ?epoch,
+                            ?l1_block_number,
+                            "error fetching stake table from l1. err {err}"
+                        );
+                    }
+                    Ok(stake_tables) => {
+                        let mut state = state.write().unwrap();
+                        let _ = state.insert(epoch, stake_tables);
+                    }
+                }
+
+                sleep(retry_delay).await;
+            }
+        }
+        .instrument(span)
+        .await
+    }
 
     /// Shut down background tasks associated with this L1 client.
     ///
@@ -530,6 +564,16 @@ impl L1Client {
         self.state.lock().await.snapshot
     }
 
+    // TODO remove after `Memberships` trait update on hotshot
+    // https://github.com/EspressoSystems/HotShot/issues/3966
+    pub fn stake_table(&self, epoch: &EpochNumber) -> StakeTables {
+        if let Some(stake_tables) = self.stake_table_state.read().unwrap().get(epoch) {
+            stake_tables.clone()
+        } else {
+            StakeTables::new(vec![].into(), vec![].into())
+        }
+    }
+
     /// Wait until the highest L1 block number reaches at least `number`.
     ///
     /// This function does not return any information about the block, since the block is not
@@ -785,20 +829,23 @@ impl L1Client {
     }
 
     /// Get `StakeTable` at block height.
-    pub async fn get_stake_table(&self, block: u64, address: Address) -> StakeTables {
-        // TODO epoch size may need to be passed in as well
-        // TODO here or in memberships check if we have fetched table this epoch
-        let stake_table_contract = PermissionedStakeTable::new(address, self.provider.clone());
+    pub async fn get_stake_table(
+        &self,
+        contract: Address,
+        block: u64,
+    ) -> anyhow::Result<StakeTables> {
+        // TODO stake_table_address needs to be passed in to L1Client
+        // before update loop starts.
+        let stake_table_contract = PermissionedStakeTable::new(contract, self.provider.clone());
 
         let events = stake_table_contract
             .stakers_updated_filter()
             .from_block(0)
             .to_block(block)
             .query()
-            .await
-            .unwrap();
+            .await?;
 
-        StakeTables::from_l1_events(events.clone())
+        Ok(StakeTables::from_l1_events(events.clone()))
     }
 }
 
@@ -1301,14 +1348,15 @@ mod test {
 
         let new_nodes: Vec<NodeInfo> = vec![node.into()];
         let updater = stake_table_contract.update(v, new_nodes);
-        updater.send().await?;
+        updater.send().await?.await?;
 
         let block = client.get_block(BlockNumber::Latest).await?.unwrap();
         let nodes = l1_client
-            .get_stake_table(block.number.unwrap().as_u64(), address)
-            .await;
+            .get_stake_table(address, block.number.unwrap().as_u64())
+            .await
+            .unwrap();
 
-        let result = nodes.consensus_stake_table.0[0].clone();
+        let result = nodes.stake_table.0[0].clone();
         assert_eq!(result.stake_amount.as_u64(), 1);
         Ok(())
     }

types/src/v0/impls/mod.rs

@@ -15,7 +15,7 @@ mod transaction;
 pub use auction::SolverAuctionResultsProvider;
 pub use fee_info::{retain_accounts, FeeError};
 pub use instance_state::NodeState;
-pub use stake_table::StaticCommittee;
+pub use stake_table::*;
 pub use state::{
     get_l1_deposits, BuilderValidationError, ProposalValidationError, StateValidationError,
     ValidatedState,