commonware_storage/archive/immutable/
mod.rs

1//! An immutable key-value store for ordered data with a minimal memory footprint.
2//!
3//! Data is stored in a [crate::freezer::Freezer] and a [crate::ordinal::Ordinal] to enable
4//! lookups by both index and key with minimal memory overhead.
5//!
6//! # Uniqueness
7//!
8//! [Archive] assumes all stored indices are unique. Writing to an occupied index is a no-op.
9//! If the same key is associated with multiple indices, there is no guarantee which value will
10//! be returned.
11//!
12//! # Compression
13//!
14//! [Archive] supports compressing data before storing it on disk. This can be enabled by setting
15//! the `compression` field in the `Config` struct to a valid `zstd` compression level. This setting
16//! can be changed between initializations of [Archive], however, it must remain populated if any
17//! data was written with compression enabled.
18//!
19//! # Durability and Recovery
20//!
21//! `put` updates the underlying [crate::freezer::Freezer] and [crate::ordinal::Ordinal]
22//! eagerly, but data is not committed until `sync` succeeds. Sync first makes the freezer
23//! and ordinal data durable, then commits metadata that names the freezer checkpoint and ordinal
24//! section bits. On restart, this metadata is the source of truth: lower-layer data not described by
25//! metadata is treated as uncommitted and may be removed during initialization. If no freezer
26//! checkpoint has been committed yet, initialization starts from an empty archive.
27//!
28//! # Querying for Gaps
29//!
30//! [Archive] tracks gaps in the index space to enable the caller to efficiently fetch unknown keys
31//! using `next_gap`. This is a very common pattern when syncing blocks in a blockchain.
32//!
33//! # Example
34//!
35//! ```rust
36//! use commonware_runtime::{Spawner, Runner, deterministic, buffer::paged::CacheRef};
37//! use commonware_cryptography::{Hasher as _, Sha256};
38//! use commonware_storage::{
39//!     archive::{
40//!         Archive as _,
41//!         immutable::{Archive, Config},
42//!     },
43//! };
44//! use commonware_utils::{NZUsize, NZU16, NZU64};
45//!
46//! let executor = deterministic::Runner::default();
47//! executor.start(|context| async move {
48//!     // Create an archive
49//!     let cfg = Config {
50//!         metadata_partition: "metadata".into(),
51//!         freezer_table_partition: "freezer-table".into(),
52//!         freezer_table_initial_size: 65_536,
53//!         freezer_table_resize_frequency: 4,
54//!         freezer_table_resize_chunk_size: 16_384,
55//!         freezer_key_partition: "freezer-key".into(),
56//!         freezer_key_page_cache: CacheRef::from_pooler(&context, NZU16!(1024), NZUsize!(10)),
57//!         freezer_value_partition: "freezer-value".into(),
58//!         freezer_value_target_size: 1024,
59//!         freezer_value_compression: Some(3),
60//!         ordinal_partition: "ordinal".into(),
61//!         items_per_section: NZU64!(1024),
62//!         freezer_key_write_buffer: NZUsize!(1024),
63//!         freezer_value_write_buffer: NZUsize!(1024),
64//!         ordinal_write_buffer: NZUsize!(1024),
65//!         replay_buffer: NZUsize!(1024),
66//!         codec_config: (),
67//!     };
68//!     let mut archive = Archive::init(context, cfg).await.unwrap();
69//!
70//!     // Put a key
71//!     archive = archive.put(1, Sha256::hash(&[b"data"]), 10).await.unwrap();
72//!
73//!     // Sync the archive
74//!     archive.sync().await.unwrap();
75//! });
76
77mod storage;
78use commonware_runtime::buffer::paged::CacheRef;
79use std::num::{NonZeroU64, NonZeroUsize};
80pub use storage::Archive;
81
82/// Configuration for [Archive] storage.
83#[derive(Clone)]
84pub struct Config<C> {
85    /// The partition to use for the archive's commit records: the freezer checkpoint and
86    /// ordinal bitmaps that recovery uses to keep both stores consistent.
87    pub metadata_partition: String,
88
89    /// The partition to use for the archive's freezer table.
90    pub freezer_table_partition: String,
91
92    /// The size of the archive's freezer table.
93    pub freezer_table_initial_size: u32,
94
95    /// The number of items added to the freezer table before it is resized.
96    pub freezer_table_resize_frequency: u8,
97
98    /// The number of items to move during each resize operation (many may be required to complete a resize).
99    pub freezer_table_resize_chunk_size: u32,
100
101    /// The partition to use for the archive's freezer keys.
102    pub freezer_key_partition: String,
103
104    /// The page cache to use for the archive's freezer keys.
105    pub freezer_key_page_cache: CacheRef,
106
107    /// The partition to use for the archive's freezer values.
108    pub freezer_value_partition: String,
109
110    /// The target size of the archive's freezer value sections.
111    pub freezer_value_target_size: u64,
112
113    /// The compression level to use for the archive's freezer values.
114    pub freezer_value_compression: Option<u8>,
115
116    /// The partition to use for the archive's ordinal.
117    pub ordinal_partition: String,
118
119    /// The number of items per section.
120    pub items_per_section: NonZeroU64,
121
122    /// The amount of bytes that can be buffered for the freezer key journal before being
123    /// written to a [commonware_runtime::Blob].
124    pub freezer_key_write_buffer: NonZeroUsize,
125
126    /// The amount of bytes that can be buffered for the freezer value journal before being
127    /// written to a [commonware_runtime::Blob].
128    pub freezer_value_write_buffer: NonZeroUsize,
129
130    /// The amount of bytes that can be buffered for the ordinal journal before being
131    /// written to a [commonware_runtime::Blob].
132    pub ordinal_write_buffer: NonZeroUsize,
133
134    /// The buffer size to use when replaying a [commonware_runtime::Blob].
135    pub replay_buffer: NonZeroUsize,
136
137    /// The [commonware_codec::Codec] configuration to use for the value stored in the archive.
138    pub codec_config: C,
139}
140
141#[cfg(test)]
142mod tests {
143    use super::*;
144    use crate::archive::Archive as ArchiveTrait;
145    use commonware_cryptography::{Hasher, Sha256, sha256::Digest};
146    use commonware_runtime::{Runner, Supervisor as _, buffer::paged::CacheRef, deterministic};
147    use commonware_utils::{NZU16, NZU64, NZUsize};
148    use std::num::NonZeroU16;
149
150    const PAGE_SIZE: NonZeroU16 = NZU16!(1024);
151    const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(10);
152
153    #[test]
154    fn test_unclean_shutdown() {
155        let executor = deterministic::Runner::default();
156        executor.start(|context| async move {
157            let cfg = Config {
158                metadata_partition: "test-metadata2".into(),
159                freezer_table_partition: "test-freezer-table2".into(),
160                freezer_table_initial_size: 8192, // Must be power of 2
161                freezer_table_resize_frequency: 4,
162                freezer_table_resize_chunk_size: 8192,
163                freezer_key_partition: "test-freezer-key2".into(),
164                freezer_key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
165                freezer_value_partition: "test-freezer-value2".into(),
166                freezer_value_target_size: 1024 * 1024,
167                freezer_value_compression: Some(3),
168                ordinal_partition: "test-ordinal2".into(),
169                items_per_section: NZU64!(512),
170                freezer_key_write_buffer: NZUsize!(1024),
171                freezer_value_write_buffer: NZUsize!(1024),
172                ordinal_write_buffer: NZUsize!(1024),
173                replay_buffer: NZUsize!(1024),
174                codec_config: (),
175            };
176
177            // First initialization
178            let archive: Archive<_, Digest, i32> =
179                Archive::init(context.child("first"), cfg.clone())
180                    .await
181                    .unwrap();
182            drop(archive);
183
184            // Second initialization
185            let mut archive = Archive::init(context.child("second"), cfg.clone())
186                .await
187                .unwrap();
188
189            // Add some data
190            let key1 = Sha256::hash(&[b"key1"]);
191            let key2 = Sha256::hash(&[b"key2"]);
192            archive = archive.put(1, key1, 2000).await.unwrap();
193            archive = archive.put(2, key2, 2001).await.unwrap();
194
195            // Sync archive to save the checkpoint
196            let archive = archive.sync().await.unwrap();
197            drop(archive);
198
199            // Re-initialize archive (should load from checkpoint)
200            let archive = Archive::init(context.child("third"), cfg).await.unwrap();
201
202            // Verify data persisted
203            assert_eq!(
204                archive
205                    .get(crate::archive::Identifier::Key(&key1))
206                    .await
207                    .unwrap(),
208                Some(2000)
209            );
210            assert_eq!(
211                archive
212                    .get(crate::archive::Identifier::Key(&key2))
213                    .await
214                    .unwrap(),
215                Some(2001)
216            );
217        });
218    }
219
220    #[test]
221    fn test_sync_empty_archive_then_restart() {
222        let executor = deterministic::Runner::default();
223        executor.start(|context| async move {
224            let cfg = Config {
225                metadata_partition: "empty-metadata".into(),
226                freezer_table_partition: "empty-freezer-table".into(),
227                freezer_table_initial_size: 8192,
228                freezer_table_resize_frequency: 4,
229                freezer_table_resize_chunk_size: 8192,
230                freezer_key_partition: "empty-freezer-key".into(),
231                freezer_key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
232                freezer_value_partition: "empty-freezer-value".into(),
233                freezer_value_target_size: 1024 * 1024,
234                freezer_value_compression: Some(3),
235                ordinal_partition: "empty-ordinal".into(),
236                items_per_section: NZU64!(512),
237                freezer_key_write_buffer: NZUsize!(1024),
238                freezer_value_write_buffer: NZUsize!(1024),
239                ordinal_write_buffer: NZUsize!(1024),
240                replay_buffer: NZUsize!(1024),
241                codec_config: (),
242            };
243
244            // Initialize archive, sync without writing anything, then drop
245            let archive: Archive<_, Digest, i32> =
246                Archive::init(context.child("first"), cfg.clone())
247                    .await
248                    .unwrap();
249            let archive = archive.sync().await.unwrap();
250            drop(archive);
251
252            // Re-initialize -- should not fail with SectionOutOfRange(0)
253            let archive: Archive<_, Digest, i32> =
254                Archive::init(context.child("second"), cfg.clone())
255                    .await
256                    .unwrap();
257
258            // Write data after restart to confirm archive is functional
259            let key = Sha256::hash(&[b"after-restart"]);
260            let archive = archive.put_sync(0, key, 42).await.unwrap();
261            drop(archive);
262
263            // Third init to verify persistence
264            let archive: Archive<_, Digest, i32> =
265                Archive::init(context.child("third"), cfg).await.unwrap();
266            assert_eq!(
267                archive
268                    .get(crate::archive::Identifier::Key(&key))
269                    .await
270                    .unwrap(),
271                Some(42)
272            );
273        });
274    }
275}
commonware_storage/archive/immutable/mod.rs

commonware_storage/archive/immutable/
mod.rs
