iota_core/epoch/
committee_store.rs

1// Copyright (c) Mysten Labs, Inc.
2// Modifications Copyright (c) 2024 IOTA Stiftung
3// SPDX-License-Identifier: Apache-2.0
4
5use std::{collections::HashMap, path::PathBuf, sync::Arc};
6
7use iota_types::{
8    committee::{Committee, EpochId},
9    error::{IotaError, IotaResult},
10};
11use parking_lot::RwLock;
12use typed_store::{
13    DBMapUtils, Map,
14    rocks::{DBMap, DBOptions, MetricConf, default_db_options},
15    rocksdb::Options,
16};
17
18pub struct CommitteeStore {
19    tables: CommitteeStoreTables,
20    cache: RwLock<HashMap<EpochId, Arc<Committee>>>,
21}
22
23#[derive(DBMapUtils)]
24pub struct CommitteeStoreTables {
25    /// Map from each epoch ID to the committee information.
26    #[default_options_override_fn = "committee_table_default_config"]
27    committee_map: DBMap<EpochId, Committee>,
28}
29
30// These functions are used to initialize the DB tables
31fn committee_table_default_config() -> DBOptions {
32    default_db_options().optimize_for_point_lookup(64)
33}
34
35impl CommitteeStore {
36    /// Open the on-disk tables at `path` into an empty-cache store, without
37    /// touching the genesis committee.
38    fn open_tables(path: PathBuf, db_options: Option<Options>) -> Self {
39        let tables = CommitteeStoreTables::open_tables_read_write(
40            path,
41            MetricConf::new("committee"),
42            db_options,
43            None,
44        );
45        Self {
46            tables,
47            cache: RwLock::new(HashMap::new()),
48        }
49    }
50
51    pub fn new(path: PathBuf, genesis_committee: &Committee, db_options: Option<Options>) -> Self {
52        let store = Self::open_tables(path, db_options);
53        if store
54            .database_is_empty()
55            .expect("CommitteeStore initialization failed")
56        {
57            store
58                .init_genesis_committee(genesis_committee.clone())
59                .expect("Init genesis committee data must not fail");
60        }
61        store
62    }
63
64    pub fn new_for_testing(genesis_committee: &Committee) -> Self {
65        let path = iota_common::tempdir().keep();
66        Self::new(path, genesis_committee, None)
67    }
68
69    /// Open an existing committee store whose genesis committee is already
70    /// persisted (e.g. a restored or synced node's store). Unlike [`Self::new`]
71    /// it takes no genesis committee — it errors if the store has none, rather
72    /// than initializing one.
73    pub fn open(path: PathBuf, db_options: Option<Options>) -> IotaResult<Self> {
74        let store = Self::open_tables(path, db_options);
75        if store.database_is_empty()? {
76            return Err(IotaError::Storage(
77                "committee store has no genesis committee".to_string(),
78            ));
79        }
80        Ok(store)
81    }
82
83    pub fn init_genesis_committee(&self, genesis_committee: Committee) -> IotaResult {
84        assert_eq!(genesis_committee.epoch, 0);
85        self.tables.committee_map.insert(&0, &genesis_committee)?;
86        self.cache.write().insert(0, Arc::new(genesis_committee));
87        Ok(())
88    }
89
90    pub fn insert_new_committee(&self, new_committee: &Committee) -> IotaResult {
91        if let Some(old_committee) = self.get_committee(&new_committee.epoch)? {
92            // If somehow we already have this committee in the store, they must be the
93            // same.
94            assert_eq!(&*old_committee, new_committee);
95        } else {
96            self.tables
97                .committee_map
98                .insert(&new_committee.epoch, new_committee)?;
99            self.cache
100                .write()
101                .insert(new_committee.epoch, Arc::new(new_committee.clone()));
102        }
103        Ok(())
104    }
105
106    pub fn get_committee(&self, epoch_id: &EpochId) -> IotaResult<Option<Arc<Committee>>> {
107        if let Some(committee) = self.cache.read().get(epoch_id) {
108            return Ok(Some(committee.clone()));
109        }
110        let committee = self.tables.committee_map.get(epoch_id)?;
111        let committee = committee.map(Arc::new);
112        if let Some(committee) = committee.as_ref() {
113            self.cache.write().insert(*epoch_id, committee.clone());
114        }
115        Ok(committee)
116    }
117
118    // todo - make use of cache or remove this method
119    pub fn get_latest_committee(&self) -> IotaResult<Committee> {
120        Ok(self
121            .tables
122            .committee_map
123            .safe_range_iter_reversed(..)
124            .next()
125            .transpose()?
126            // unwrap safe because we guarantee there is at least a genesis epoch
127            // when initializing the store.
128            .unwrap()
129            .1)
130    }
131    /// Return the committee specified by `epoch`. If `epoch` is `None`, return
132    /// the latest committee.
133    // todo - make use of cache or remove this method
134    pub fn get_or_latest_committee(&self, epoch: Option<EpochId>) -> IotaResult<Committee> {
135        Ok(match epoch {
136            Some(epoch) => self
137                .get_committee(&epoch)?
138                .ok_or(IotaError::MissingCommitteeAtEpoch(epoch))
139                .map(|c| Committee::clone(&*c))?,
140            None => self.get_latest_committee()?,
141        })
142    }
143
144    fn database_is_empty(&self) -> IotaResult<bool> {
145        Ok(self
146            .tables
147            .committee_map
148            .safe_iter()
149            .next()
150            .transpose()?
151            .is_none())
152    }
153}
154
155#[cfg(test)]
156mod tests {
157    use iota_types::committee::Committee;
158
159    use super::*;
160
161    #[tokio::test]
162    async fn open_reads_existing_genesis_and_rejects_empty() {
163        let dir = iota_common::tempdir();
164        let path = dir.path().to_path_buf();
165        let (genesis_committee, _) = Committee::new_simple_test_committee();
166
167        // A fresh directory has no genesis committee yet.
168        assert!(CommitteeStore::open(path.clone(), None).is_err());
169
170        // `new` initializes the genesis committee; `open` then reads it back
171        // without being handed one.
172        {
173            let _store = CommitteeStore::new(path.clone(), &genesis_committee, None);
174        }
175        let opened = CommitteeStore::open(path, None).expect("store has a genesis committee");
176        assert_eq!(
177            *opened.get_committee(&0).unwrap().unwrap(),
178            genesis_committee,
179        );
180    }
181}
iota_core/epoch/committee_store.rs

iota_core/epoch/
committee_store.rs
