drive/cache/
protocol_version.rs

1use crate::drive::Drive;
2use crate::error::cache::CacheError;
3use crate::error::Error;
4use dpp::util::deserializer::ProtocolVersion;
5use grovedb::TransactionArg;
6use nohash_hasher::IntMap;
7use platform_version::version::drive_versions::DriveVersion;
8
9/// ProtocolVersion cache that handles both global and block data
10#[derive(Default)]
11pub struct ProtocolVersionsCache {
12    /// The current global cache for protocol versions
13    // TODO: If we persist this in the state and it should be loaded for correct
14    //  use then it's not actually the cache. Move out of cache because it's confusing
15    pub global_cache: IntMap<ProtocolVersion, u64>,
16    block_cache: IntMap<ProtocolVersion, u64>,
17    loaded: bool,
18    is_global_cache_blocked: bool,
19}
20
21#[cfg(feature = "server")]
22impl ProtocolVersionsCache {
23    /// Create a new ProtocolVersionsCache instance
24    pub fn new() -> Self {
25        Self::default()
26    }
27
28    /// Load the protocol versions cache from disk if needed
29    pub fn load_if_needed(
30        &mut self,
31        drive: &Drive,
32        transaction: TransactionArg,
33        drive_version: &DriveVersion,
34    ) -> Result<(), Error> {
35        if !self.loaded {
36            self.global_cache = drive.fetch_versions_with_counter(transaction, drive_version)?;
37            self.loaded = true;
38        };
39        Ok(())
40    }
41
42    /// Whether the global cache has been loaded from Drive.
43    ///
44    /// Until it has, the global cache is empty rather than a mirror of the persisted counters,
45    /// and any tally over it would count zero votes.
46    pub fn is_loaded(&self) -> bool {
47        self.loaded
48    }
49
50    /// Sets the protocol version to the block cache
51    pub fn set_block_cache_version_count(&mut self, version: ProtocolVersion, count: u64) {
52        self.block_cache.insert(version, count);
53    }
54
55    /// Tries to get a version from block cache if present
56    /// if block cache doesn't have the version set
57    /// then it tries get the version from global cache
58    pub fn get(&self, version: &ProtocolVersion) -> Result<Option<&u64>, Error> {
59        if self.is_global_cache_blocked {
60            return Err(Error::Cache(CacheError::GlobalCacheIsBlocked));
61        }
62
63        let counter = if let Some(count) = self.block_cache.get(version) {
64            Some(count)
65        } else {
66            self.global_cache.get(version)
67        };
68
69        Ok(counter)
70    }
71
72    /// Disable the global cache to do not allow get counters
73    /// If global cache is blocked then [get] will return an error
74    pub fn block_global_cache(&mut self) {
75        self.is_global_cache_blocked = true;
76    }
77
78    /// Unblock the global cache
79    /// This function enables the normal behaviour of [get] function
80    pub fn unblock_global_cache(&mut self) {
81        self.is_global_cache_blocked = false;
82    }
83
84    /// Merge block cache to global cache
85    pub fn merge_block_cache(&mut self) {
86        self.global_cache.extend(self.block_cache.drain());
87    }
88
89    /// Clears the global cache
90    pub fn clear_global_cache(&mut self) {
91        self.global_cache.clear();
92    }
93
94    /// Clear block cache
95    pub fn clear_block_cache(&mut self) {
96        self.block_cache.clear()
97    }
98
99    /// Collect versions passing threshold
100    pub fn versions_passing_threshold(&self, required_upgraded_hpmns: u64) -> Vec<ProtocolVersion> {
101        let mut cache = self.global_cache.clone();
102
103        cache.extend(self.block_cache.iter());
104        cache
105            .into_iter()
106            .filter_map(|(protocol_version, count)| {
107                if count >= required_upgraded_hpmns {
108                    Some(protocol_version)
109                } else {
110                    None
111                }
112            })
113            .collect::<Vec<ProtocolVersion>>()
114    }
115}
116
117#[cfg(feature = "server")]
118#[cfg(test)]
119mod tests {
120    use super::*;
121    use crate::config::DriveConfig;
122    use dpp::version::PlatformVersion;
123    use std::path::Path;
124    use tempfile::TempDir;
125
126    /// Opens a Drive at `path`, commits votes for `protocol_version + 1` from three validators
127    /// and for `protocol_version + 2` from a fourth, stores the protocol version the way a
128    /// committed block does, and returns the counters as a node that never restarted holds them
129    /// once the block is finalized.
130    fn persist_votes(
131        path: &Path,
132        platform_version: &PlatformVersion,
133    ) -> IntMap<ProtocolVersion, u64> {
134        let (drive, _) =
135            Drive::open(path, Some(DriveConfig::default())).expect("expected to open Drive");
136        drive
137            .create_initial_state_structure(None, platform_version)
138            .expect("expected to create the initial state structure");
139
140        let next_version = platform_version.protocol_version + 1;
141        let transaction = drive.grove.start_transaction();
142        for validator in 1..=3u8 {
143            drive
144                .update_validator_proposed_app_version(
145                    [validator; 32],
146                    next_version,
147                    Some(&transaction),
148                    &platform_version.drive,
149                )
150                .expect("expected to record the vote");
151        }
152        drive
153            .update_validator_proposed_app_version(
154                [4; 32],
155                next_version + 1,
156                Some(&transaction),
157                &platform_version.drive,
158            )
159            .expect("expected to record the vote");
160        drive
161            .store_current_protocol_version(platform_version.protocol_version, Some(&transaction))
162            .expect("expected to store the protocol version");
163        drive
164            .grove
165            .commit_transaction(transaction)
166            .unwrap()
167            .expect("expected to commit the votes");
168
169        let mut counter = drive.cache.protocol_versions_counter.write();
170        counter.merge_block_cache();
171        counter.global_cache.clone()
172    }
173
174    #[test]
175    fn should_start_a_fresh_drive_with_nothing_loaded() {
176        let tempdir = TempDir::new().expect("expected a temporary directory");
177
178        let (drive, protocol_version) = Drive::open(tempdir.path(), Some(DriveConfig::default()))
179            .expect("expected to open Drive");
180
181        assert!(protocol_version.is_none());
182        assert!(!drive.cache.protocol_versions_counter.read().is_loaded());
183    }
184
185    #[test]
186    fn should_reopen_a_drive_with_the_persisted_votes_loaded() {
187        let tempdir = TempDir::new().expect("expected a temporary directory");
188        let platform_version = PlatformVersion::latest();
189        let next_version = platform_version.protocol_version + 1;
190
191        let warm_counters = persist_votes(tempdir.path(), platform_version);
192        assert_eq!(warm_counters.get(&next_version), Some(&3));
193
194        let (reopened, stored_version) = Drive::open(tempdir.path(), Some(DriveConfig::default()))
195            .expect("expected to reopen Drive");
196        assert_eq!(
197            stored_version.map(|version| version.protocol_version),
198            Some(platform_version.protocol_version)
199        );
200
201        let counter = reopened.cache.protocol_versions_counter.read();
202        assert!(
203            counter.is_loaded(),
204            "a reopened Drive must present the persisted votes without waiting for the first vote of a block"
205        );
206        assert_eq!(counter.global_cache, warm_counters);
207        assert_eq!(counter.versions_passing_threshold(3), vec![next_version]);
208    }
209
210    #[test]
211    fn should_load_the_persisted_votes_into_a_cold_cache_through_the_transaction() {
212        let tempdir = TempDir::new().expect("expected a temporary directory");
213        let platform_version = PlatformVersion::latest();
214        let next_version = platform_version.protocol_version + 1;
215
216        let warm_counters = persist_votes(tempdir.path(), platform_version);
217        let (reopened, _) = Drive::open(tempdir.path(), Some(DriveConfig::default()))
218            .expect("expected to reopen Drive");
219
220        // A cache that was never loaded, as after `drop_cache`, tallies zero votes.
221        let mut cold_cache = ProtocolVersionsCache::new();
222        assert!(!cold_cache.is_loaded());
223        assert!(cold_cache.versions_passing_threshold(1).is_empty());
224
225        let transaction = reopened.grove.start_transaction();
226        cold_cache
227            .load_if_needed(&reopened, Some(&transaction), &platform_version.drive)
228            .expect("expected to load the persisted votes");
229
230        assert!(cold_cache.is_loaded());
231        assert_eq!(cold_cache.global_cache, warm_counters);
232        assert_eq!(cold_cache.versions_passing_threshold(3), vec![next_version]);
233    }
234}
drive/cache/protocol_version.rs

drive/cache/
protocol_version.rs
