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.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 metadata.
86    pub metadata_partition: String,
87
88    /// The partition to use for the archive's freezer table.
89    pub freezer_table_partition: String,
90
91    /// The size of the archive's freezer table.
92    pub freezer_table_initial_size: u32,
93
94    /// The number of items added to the freezer table before it is resized.
95    pub freezer_table_resize_frequency: u8,
96
97    /// The number of items to move during each resize operation (many may be required to complete a resize).
98    pub freezer_table_resize_chunk_size: u32,
99
100    /// The partition to use for the archive's freezer keys.
101    pub freezer_key_partition: String,
102
103    /// The page cache to use for the archive's freezer keys.
104    pub freezer_key_page_cache: CacheRef,
105
106    /// The partition to use for the archive's freezer values.
107    pub freezer_value_partition: String,
108
109    /// The target size of the archive's freezer value sections.
110    pub freezer_value_target_size: u64,
111
112    /// The compression level to use for the archive's freezer values.
113    pub freezer_value_compression: Option<u8>,
114
115    /// The partition to use for the archive's ordinal.
116    pub ordinal_partition: String,
117
118    /// The number of items per section.
119    pub items_per_section: NonZeroU64,
120
121    /// The amount of bytes that can be buffered for the freezer key journal before being
122    /// written to a [commonware_runtime::Blob].
123    pub freezer_key_write_buffer: NonZeroUsize,
124
125    /// The amount of bytes that can be buffered for the freezer value journal before being
126    /// written to a [commonware_runtime::Blob].
127    pub freezer_value_write_buffer: NonZeroUsize,
128
129    /// The amount of bytes that can be buffered for the ordinal journal before being
130    /// written to a [commonware_runtime::Blob].
131    pub ordinal_write_buffer: NonZeroUsize,
132
133    /// The buffer size to use when replaying a [commonware_runtime::Blob].
134    pub replay_buffer: NonZeroUsize,
135
136    /// The [commonware_codec::Codec] configuration to use for the value stored in the archive.
137    pub codec_config: C,
138}
139
140#[cfg(test)]
141mod tests {
142    use super::*;
143    use crate::archive::Archive as ArchiveTrait;
144    use commonware_cryptography::{sha256::Digest, Hasher, Sha256};
145    use commonware_runtime::{buffer::paged::CacheRef, deterministic, Runner, Supervisor as _};
146    use commonware_utils::{NZUsize, NZU16, NZU64};
147    use std::num::NonZeroU16;
148
149    const PAGE_SIZE: NonZeroU16 = NZU16!(1024);
150    const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(10);
151
152    #[test]
153    fn test_unclean_shutdown() {
154        let executor = deterministic::Runner::default();
155        executor.start(|context| async move {
156            let cfg = Config {
157                metadata_partition: "test-metadata2".into(),
158                freezer_table_partition: "test-freezer-table2".into(),
159                freezer_table_initial_size: 8192, // Must be power of 2
160                freezer_table_resize_frequency: 4,
161                freezer_table_resize_chunk_size: 8192,
162                freezer_key_partition: "test-freezer-key2".into(),
163                freezer_key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
164                freezer_value_partition: "test-freezer-value2".into(),
165                freezer_value_target_size: 1024 * 1024,
166                freezer_value_compression: Some(3),
167                ordinal_partition: "test-ordinal2".into(),
168                items_per_section: NZU64!(512),
169                freezer_key_write_buffer: NZUsize!(1024),
170                freezer_value_write_buffer: NZUsize!(1024),
171                ordinal_write_buffer: NZUsize!(1024),
172                replay_buffer: NZUsize!(1024),
173                codec_config: (),
174            };
175
176            // First initialization
177            let archive: Archive<_, Digest, i32> =
178                Archive::init(context.child("first"), cfg.clone())
179                    .await
180                    .unwrap();
181            drop(archive);
182
183            // Second initialization
184            let mut archive = Archive::init(context.child("second"), cfg.clone())
185                .await
186                .unwrap();
187
188            // Add some data
189            let key1 = Sha256::hash(b"key1");
190            let key2 = Sha256::hash(b"key2");
191            archive.put(1, key1, 2000).await.unwrap();
192            archive.put(2, key2, 2001).await.unwrap();
193
194            // Sync archive to save the checkpoint
195            archive.sync().await.unwrap();
196            drop(archive);
197
198            // Re-initialize archive (should load from checkpoint)
199            let archive = Archive::init(context.child("third"), cfg).await.unwrap();
200
201            // Verify data persisted
202            assert_eq!(
203                archive
204                    .get(crate::archive::Identifier::Key(&key1))
205                    .await
206                    .unwrap(),
207                Some(2000)
208            );
209            assert_eq!(
210                archive
211                    .get(crate::archive::Identifier::Key(&key2))
212                    .await
213                    .unwrap(),
214                Some(2001)
215            );
216        });
217    }
218
219    #[test]
220    fn test_sync_empty_archive_then_restart() {
221        let executor = deterministic::Runner::default();
222        executor.start(|context| async move {
223            let cfg = Config {
224                metadata_partition: "empty-metadata".into(),
225                freezer_table_partition: "empty-freezer-table".into(),
226                freezer_table_initial_size: 8192,
227                freezer_table_resize_frequency: 4,
228                freezer_table_resize_chunk_size: 8192,
229                freezer_key_partition: "empty-freezer-key".into(),
230                freezer_key_page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
231                freezer_value_partition: "empty-freezer-value".into(),
232                freezer_value_target_size: 1024 * 1024,
233                freezer_value_compression: Some(3),
234                ordinal_partition: "empty-ordinal".into(),
235                items_per_section: NZU64!(512),
236                freezer_key_write_buffer: NZUsize!(1024),
237                freezer_value_write_buffer: NZUsize!(1024),
238                ordinal_write_buffer: NZUsize!(1024),
239                replay_buffer: NZUsize!(1024),
240                codec_config: (),
241            };
242
243            // Initialize archive, sync without writing anything, then drop
244            let mut archive: Archive<_, Digest, i32> =
245                Archive::init(context.child("first"), cfg.clone())
246                    .await
247                    .unwrap();
248            archive.sync().await.unwrap();
249            drop(archive);
250
251            // Re-initialize -- should not fail with SectionOutOfRange(0)
252            let mut archive: Archive<_, Digest, i32> =
253                Archive::init(context.child("second"), cfg.clone())
254                    .await
255                    .unwrap();
256
257            // Write data after restart to confirm archive is functional
258            let key = Sha256::hash(b"after-restart");
259            archive.put_sync(0, key, 42).await.unwrap();
260            drop(archive);
261
262            // Third init to verify persistence
263            let archive: Archive<_, Digest, i32> =
264                Archive::init(context.child("third"), cfg).await.unwrap();
265            assert_eq!(
266                archive
267                    .get(crate::archive::Identifier::Key(&key))
268                    .await
269                    .unwrap(),
270                Some(42)
271            );
272        });
273    }
274}
commonware_storage/archive/immutable/mod.rs

commonware_storage/archive/immutable/
mod.rs
