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

iota_core/epoch/
committee_store.rs
