feat(node): persisted vote store

Cargo.toml

@@ -142,6 +142,7 @@ rand = "0.8"
 rand_chacha = "0.3"
 regex = "1"
 reqwest = { version = "0.11.13", features = ["json"] }
+rocksdb = { version = "0.21", features = ["multi-threaded-cf"] }
 sha2 = "0.10"
 serde = { version = "1", features = ["derive"] }
 serde_bytes = "0.11"

fendermint/app/src/cmd/run.rs

@@ -31,11 +31,12 @@ use fendermint_vm_topdown::proxy::{
     IPCProviderProxy, IPCProviderProxyWithLatency, ParentQueryProxy,
 };
 use fendermint_vm_topdown::syncer::poll::ParentPoll;
-use fendermint_vm_topdown::syncer::store::{InMemoryParentViewStore, ParentViewStore};
+use fendermint_vm_topdown::syncer::store::{ParentViewStore, PersistedParentViewStore};
 use fendermint_vm_topdown::syncer::{ParentPoller, ParentSyncerConfig, TopDownSyncEvent};
 use fendermint_vm_topdown::vote::error::Error;
 use fendermint_vm_topdown::vote::gossip::{GossipReceiver, GossipSender};
 use fendermint_vm_topdown::vote::payload::Vote;
+use fendermint_vm_topdown::vote::store::PersistedVoteStore;
 use fendermint_vm_topdown::vote::VoteConfig;
 use fendermint_vm_topdown::{Checkpoint, TopdownClient};
 use fvm_shared::address::{current_network, Address, Network};
@@ -254,7 +255,7 @@ async fn run(settings: Settings) -> anyhow::Result<()> {
             state_hist_size: settings.db.state_hist_size,
             halt_height: settings.halt_height,
         },
-        db,
+        db.clone(),
         state_store,
         interpreter,
         ChainEnv {
@@ -291,13 +292,15 @@ async fn run(settings: Settings) -> anyhow::Result<()> {
         let parent_proxy = Arc::new(IPCProviderProxyWithLatency::new(make_ipc_provider_proxy(
             &settings,
         )?));
-        let parent_view_store = InMemoryParentViewStore::new();
-
+        let parent_view_store =
+            PersistedParentViewStore::new(db.db.clone(), "topdown-parent-view".to_string())?;
+        let vote_store = PersistedVoteStore::new(db.db, "topdown-vote".to_string())?;
         let gossip = ipld_gossip_client
             .ok_or_else(|| anyhow!("topdown enabled but ipld is not, enable ipld first"))?;
 
         let client = run_topdown(
             parent_view_store.clone(),
+            vote_store,
             app_parent_finality_query,
             config,
             validator

fendermint/vm/topdown/Cargo.toml

@@ -23,6 +23,7 @@ ipc-api = { workspace = true }
 ipc-provider = { workspace = true }
 libp2p = { workspace = true }
 num-traits = { workspace = true }
+rocksdb = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 tendermint-rpc = { workspace = true }
@@ -36,11 +37,13 @@ num-rational = { workspace = true }
 
 
 multihash = { workspace = true }
+tempfile = { workspace = true }
 
 fendermint_vm_genesis = { path = "../genesis" }
 fendermint_vm_event = { path = "../event" }
 fendermint_tracing = { path = "../../tracing" }
 fendermint_crypto = { path = "../../crypto" }
+fendermint_rocksdb = { path = "../../rocksdb" }
 
 ipc-observability = { workspace = true }
 

fendermint/vm/topdown/src/launch.rs

@@ -2,13 +2,13 @@
 // SPDX-License-Identifier: Apache-2.0, MIT
 
 use crate::proxy::ParentQueryProxy;
-use crate::syncer::store::InMemoryParentViewStore;
+use crate::syncer::store::PersistedParentViewStore;
 use crate::syncer::{
     start_polling_reactor, ParentPoller, ParentSyncerConfig, ParentSyncerReactorClient,
 };
 use crate::vote::gossip::{GossipReceiver, GossipSender};
 use crate::vote::payload::PowerUpdates;
-use crate::vote::store::InMemoryVoteStore;
+use crate::vote::store::PersistedVoteStore;
 use crate::vote::{StartVoteReactorParams, VoteReactorClient};
 use crate::{BlockHeight, Checkpoint, Config, TopdownClient, TopdownProposal};
 use anyhow::anyhow;
@@ -30,8 +30,10 @@ use std::time::Duration;
 ///     - signs/certifies and broadcast topdown observation to p2p peers
 ///     - listens to certified topdown observation from p2p
 ///     - aggregate peer certified observations into a quorum certificate for commitment in fendermint
+#[allow(clippy::too_many_arguments)]
 pub async fn run_topdown<CheckpointQuery, GossipTx, GossipRx, Poller, ParentClient>(
-    store: InMemoryParentViewStore,
+    parent_view_store: PersistedParentViewStore,
+    vote_store: PersistedVoteStore,
     query: CheckpointQuery,
     config: Config,
     validator_key: SecretKey,
@@ -49,7 +51,7 @@ where
     ParentClient: ParentQueryProxy + Send + Sync + 'static,
 {
     let (syncer_client, syncer_rx) =
-        ParentSyncerReactorClient::new(config.syncer.request_channel_size, store);
+        ParentSyncerReactorClient::new(config.syncer.request_channel_size, parent_view_store);
     let (voting_client, voting_rx) = VoteReactorClient::new(config.voting.req_channel_buffer_size);
 
     let syncer_client_cloned = syncer_client.clone();
@@ -92,7 +94,7 @@ where
                     .expect("should query latest chain block"),
                 gossip_tx: gossip.0,
                 gossip_rx: gossip.1,
-                vote_store: InMemoryVoteStore::default(),
+                vote_store,
                 internal_event_listener: internal_event_rx,
             },
         )

fendermint/vm/topdown/src/lib.rs

@@ -21,7 +21,7 @@ use std::fmt::{Display, Formatter};
 
 pub use crate::cache::{SequentialAppendError, SequentialKeyCache, ValueIter};
 use crate::observation::Observation;
-use crate::syncer::store::InMemoryParentViewStore;
+use crate::syncer::store::PersistedParentViewStore;
 use crate::syncer::{ParentSyncerConfig, ParentSyncerReactorClient};
 use crate::vote::payload::PowerUpdates;
 use crate::vote::{VoteConfig, VoteReactorClient};
@@ -55,7 +55,7 @@ pub struct TopdownProposal {
 
 #[derive(Clone)]
 pub struct TopdownClient {
-    syncer: ParentSyncerReactorClient<InMemoryParentViewStore>,
+    syncer: ParentSyncerReactorClient<PersistedParentViewStore>,
     voting: VoteReactorClient,
 }
 

fendermint/vm/topdown/src/syncer/error.rs

@@ -25,4 +25,6 @@ pub enum Error {
     CannotCommitObservationAtNullBlock(BlockHeight),
     #[error("Missing block view at height {0} for target observation height {0}")]
     MissingBlockView(BlockHeight, BlockHeight),
+    #[error("persistent parent view store error: {0}")]
+    PersistentParentViewStore(String),
 }

fendermint/vm/topdown/src/syncer/mod.rs

@@ -51,13 +51,13 @@ pub struct ParentSyncerConfig {
 }
 
 #[derive(Clone)]
-pub struct ParentSyncerReactorClient<S> {
+pub struct ParentSyncerReactorClient<S: Clone> {
     tx: mpsc::Sender<ParentSyncerRequest>,
     checkpoint: Arc<Mutex<Checkpoint>>,
     store: S,
 }
 
-impl<S: ParentViewStore + Send + Sync> ParentSyncerReactorClient<S> {
+impl<S: ParentViewStore + Send + Sync + Clone> ParentSyncerReactorClient<S> {
     pub fn new(
         request_channel_size: usize,
         store: S,

fendermint/vm/topdown/src/syncer/payload.rs

@@ -9,8 +9,9 @@ use ipc_api::cross::IpcEnvelope;
 use ipc_api::staking::StakingChangeRequest;
 use multihash::Code;
 use multihash::MultihashDigest;
+use serde::{Deserialize, Serialize};
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct ParentBlockViewPayload {
     pub parent_hash: BlockHash,
     /// Encodes cross-net messages.
@@ -19,7 +20,7 @@ pub struct ParentBlockViewPayload {
     pub validator_changes: Vec<StakingChangeRequest>,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Serialize, Deserialize)]
 pub struct ParentBlockView {
     pub parent_height: BlockHeight,
     /// If the payload is None, this means the parent height is a null block

fendermint/vm/topdown/src/syncer/store.rs → fendermint/vm/topdown/src/syncer/store/mod.rs

@@ -1,13 +1,16 @@
 // Copyright 2022-2024 Protocol Labs
 // SPDX-License-Identifier: Apache-2.0, MIT
 
+mod persisted;
+
 use crate::syncer::error::Error;
 use crate::syncer::payload::ParentBlockView;
 use crate::{BlockHeight, SequentialKeyCache};
+pub use persisted::PersistedParentViewStore;
 use std::sync::{Arc, RwLock};
 
 /// Stores the parent view observed of the current node
-pub trait ParentViewStore {
+pub trait ParentViewStore: Clone {
     /// Store a newly observed parent view
     fn store(&self, view: ParentBlockView) -> Result<(), Error>;
 

fendermint/vm/topdown/src/syncer/store/persisted.rs

@@ -0,0 +1,92 @@
+// Copyright 2022-2024 Protocol Labs
+// SPDX-License-Identifier: Apache-2.0, MIT
+
+use crate::syncer::error::Error;
+use crate::syncer::payload::ParentBlockView;
+use crate::syncer::store::{InMemoryParentViewStore, ParentViewStore};
+use crate::BlockHeight;
+use rocksdb::{BoundColumnFamily, IteratorMode, OptimisticTransactionDB};
+use std::sync::Arc;
+
+#[derive(Clone)]
+pub struct PersistedParentViewStore {
+    cache: InMemoryParentViewStore,
+    db: Arc<OptimisticTransactionDB>,
+    ns: String,
+}
+
+impl PersistedParentViewStore {
+    pub fn new(db: Arc<OptimisticTransactionDB>, ns: String) -> Result<Self, Error> {
+        // All namespaces are pre-created during open.
+        if db.cf_handle(&ns).is_none() {
+            return Err(Error::PersistentParentViewStore(format!(
+                "namespace {ns} does not exist"
+            )));
+        }
+
+        let memory = InMemoryParentViewStore::default();
+
+        let cf = get_cf(&db, &ns)?;
+        let iter = db.iterator_cf(&cf, IteratorMode::Start);
+        for item in iter {
+            let (_, value) = item.map_err(|e| Error::PersistentParentViewStore(format!("{e}")))?;
+
+            let view = fvm_ipld_encoding::from_slice(value.as_ref()).map_err(|e| {
+                Error::PersistentParentViewStore(format!("cannot convert value to vote: {e}"))
+            })?;
+
+            memory.store(view)?;
+        }
+        drop(cf);
+
+        Ok(Self {
+            cache: memory,
+            db,
+            ns,
+        })
+    }
+}
+
+impl ParentViewStore for PersistedParentViewStore {
+    fn store(&self, view: ParentBlockView) -> Result<(), Error> {
+        let height = view.parent_height;
+        let bytes = fvm_ipld_encoding::to_vec(&view)
+            .map_err(|e| Error::PersistentParentViewStore(format!("{e}")))?;
+
+        let cf = get_cf(self.db.as_ref(), self.ns.as_str())?;
+        self.db
+            .put_cf(&cf, height.to_be_bytes(), bytes)
+            .map_err(|e| {
+                Error::PersistentParentViewStore(format!("cannot store parent view {e}"))
+            })?;
+        self.cache.store(view)
+    }
+
+    fn get(&self, height: BlockHeight) -> Result<Option<ParentBlockView>, Error> {
+        self.cache.get(height)
+    }
+
+    fn purge(&self, height: BlockHeight) -> Result<(), Error> {
+        let cf = get_cf(self.db.as_ref(), self.ns.as_str())?;
+        self.db.delete_cf(&cf, height.to_be_bytes()).map_err(|e| {
+            Error::PersistentParentViewStore(format!("cannot remove block height {e}"))
+        })?;
+        self.cache.purge(height)
+    }
+
+    fn min_parent_view_height(&self) -> Result<Option<BlockHeight>, Error> {
+        self.cache.min_parent_view_height()
+    }
+
+    fn max_parent_view_height(&self) -> Result<Option<BlockHeight>, Error> {
+        self.cache.max_parent_view_height()
+    }
+}
+
+pub(crate) fn get_cf<'a>(
+    db: &'a OptimisticTransactionDB,
+    ns: &'a str,
+) -> Result<Arc<BoundColumnFamily<'a>>, Error> {
+    db.cf_handle(ns)
+        .ok_or_else(|| Error::PersistentParentViewStore(format!("namespace {ns} does not exist")))
+}

fendermint/vm/topdown/src/vote/error.rs

@@ -31,4 +31,7 @@ pub enum Error {
 
     #[error("received unexpected gossip event {0}")]
     UnexpectedGossipEvent(String),
+
+    #[error("persistent vote store error: {0}")]
+    PersistentVoteStore(String),
 }

fendermint/vm/topdown/src/vote/store.rs → fendermint/vm/topdown/src/vote/store/mod.rs

@@ -1,13 +1,16 @@
 // Copyright 2022-2024 Protocol Labs
 // SPDX-License-Identifier: Apache-2.0, MIT
 
+mod persisted;
+
 use crate::observation::Observation;
 use crate::vote::error::Error;
 use crate::vote::payload::{PowerTable, Vote};
 use crate::vote::Weight;
 use crate::BlockHeight;
 use fendermint_crypto::quorum::ECDSACertificate;
 use fendermint_vm_genesis::ValidatorKey;
+pub use persisted::PersistedVoteStore;
 use std::collections::btree_map::Entry;
 use std::collections::{BTreeMap, HashMap};