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 optional zstd compression of freezer values through
15//! [Config::freezer_value_compression]. Keep the choice between `None` and `Some(_)` fixed for the
16//! life of the archive. Only the compression level may change between initializations when
17//! compression is 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::new(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 ///
121 /// Fixed for the life of the archive's data. Sections and their commit records are addressed
122 /// by `index / items_per_section`, and [Archive::init] fails when a committed section no
123 /// longer matches this value.
124 pub items_per_section: NonZeroU64,
125
126 /// The amount of bytes that can be buffered for the freezer key journal before being
127 /// written to a [commonware_runtime::Blob].
128 pub freezer_key_write_buffer: NonZeroUsize,
129
130 /// The amount of bytes that can be buffered for the freezer value journal before being
131 /// written to a [commonware_runtime::Blob].
132 pub freezer_value_write_buffer: NonZeroUsize,
133
134 /// The amount of bytes that can be buffered for the ordinal journal before being
135 /// written to a [commonware_runtime::Blob].
136 pub ordinal_write_buffer: NonZeroUsize,
137
138 /// The buffer size to use when replaying a [commonware_runtime::Blob].
139 pub replay_buffer: NonZeroUsize,
140
141 /// The [commonware_codec::Codec] configuration to use for the value stored in the archive.
142 pub codec_config: C,
143}
144
145#[cfg(test)]
146mod tests {
147 use super::*;
148 use crate::archive::{Archive as ArchiveTrait, Error, Identifier};
149 use commonware_cryptography::{Hasher, Sha256, sha256::Digest};
150 use commonware_runtime::{
151 Runner, Supervisor as _,
152 buffer::paged::CacheRef,
153 deterministic::{self, FaultConfig, PartialWriteMode, WriteConfig},
154 };
155 use commonware_utils::{NZU16, NZU64, NZUsize, Widen, probability};
156 use std::num::{NonZeroU16, NonZeroU64};
157
158 const PAGE_SIZE: NonZeroU16 = NZU16!(1024);
159 const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(10);
160
161 /// Build a configuration whose small sections spread a few records across several freezer
162 /// and ordinal blobs.
163 fn config(items_per_section: u64) -> Config<()> {
164 Config {
165 metadata_partition: "destroy-metadata".into(),
166 freezer_table_partition: "destroy-freezer-table".into(),
167 freezer_table_initial_size: 4,
168 freezer_table_resize_frequency: 2,
169 freezer_table_resize_chunk_size: 2,
170 freezer_key_partition: "destroy-freezer-key".into(),
171 freezer_key_page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
172 freezer_value_partition: "destroy-freezer-value".into(),
173 freezer_value_target_size: 64,
174 freezer_value_compression: None,
175 ordinal_partition: "destroy-ordinal".into(),
176 items_per_section: NonZeroU64::new(items_per_section).unwrap(),
177 freezer_key_write_buffer: NZUsize!(1024),
178 freezer_value_write_buffer: NZUsize!(1024),
179 ordinal_write_buffer: NZUsize!(1024),
180 replay_buffer: NZUsize!(1024),
181 codec_config: (),
182 }
183 }
184
185 /// Reopening with a different `items_per_section` fails instead of addressing committed
186 /// sections by the new size.
187 #[test]
188 fn test_init_rejects_changed_items_per_section() {
189 let executor = deterministic::Runner::default();
190 executor.start(|context| async move {
191 let key = |index: u64| Sha256::hash(&[&index.to_be_bytes()]);
192
193 // A partial section records its size in its bitmap.
194 let archive = Archive::init(context.child("sparse"), config(4))
195 .await
196 .unwrap();
197 let archive = archive.put(0, key(0), &0).await.unwrap();
198 let archive = archive.put_sync(5, key(5), &50).await.unwrap();
199 drop(archive);
200 let result =
201 Archive::<_, Digest, i32>::init(context.child("sparse_grown"), config(8)).await;
202 assert!(matches!(result, Err(Error::RecordCorrupted)), "{result:?}");
203 let result =
204 Archive::<_, Digest, i32>::init(context.child("sparse_shrunk"), config(2)).await;
205 assert!(matches!(result, Err(Error::RecordCorrupted)), "{result:?}");
206
207 // A complete section records its size only through the length of its ordinal blob.
208 Archive::<_, Digest, i32>::init(context.child("reset"), config(4))
209 .await
210 .unwrap()
211 .destroy()
212 .await
213 .unwrap();
214 let mut archive = Archive::init(context.child("full"), config(8))
215 .await
216 .unwrap();
217 for index in 0..8u64 {
218 archive = archive
219 .put(index, key(index), &(index as i32))
220 .await
221 .unwrap();
222 }
223 let archive = archive.sync().await.unwrap();
224 drop(archive);
225 let result =
226 Archive::<_, Digest, i32>::init(context.child("full_shrunk"), config(4)).await;
227 assert!(matches!(result, Err(Error::Ordinal(_))), "{result:?}");
228 let result =
229 Archive::<_, Digest, i32>::init(context.child("full_grown"), config(16)).await;
230 assert!(matches!(result, Err(Error::Ordinal(_))), "{result:?}");
231
232 // The original size still opens the complete section.
233 let archive = Archive::<_, Digest, i32>::init(context.child("full_same"), config(8))
234 .await
235 .unwrap();
236 for index in 0..8u64 {
237 assert_eq!(
238 archive.get(Identifier::Index(index)).await.unwrap(),
239 Some(index as i32)
240 );
241 }
242 });
243 }
244
245 /// A destroy interrupted at any point leaves an archive that reopens at its latest commit or
246 /// empty, with index and key views agreeing, and that a retried destroy empties.
247 #[test]
248 fn test_interrupted_destroy_reopens() {
249 let entries = || -> Vec<(u64, Digest, i32)> {
250 (0..11u64)
251 .map(|index| {
252 (
253 index,
254 Sha256::hash(&[&index.to_be_bytes()]),
255 index as i32 * 7,
256 )
257 })
258 .collect()
259 };
260 let mut interrupted = 0;
261 for seed in 0..32u64 {
262 // Retain every unsynced write across the crash, as a runtime that flushes a reopened
263 // blob's pending writes would.
264 let faults = FaultConfig::default().write(WriteConfig {
265 failure_rate: probability!(0.0),
266 retention_rate: probability!(1.0),
267 mode: PartialWriteMode::Prefix,
268 });
269 let runner = deterministic::Runner::new(
270 deterministic::Config::default()
271 .with_seed(seed)
272 .with_storage_fault_config(faults),
273 );
274 let (stopped, checkpoint) = runner.start_and_recover(|context| async move {
275 let cfg = config(4);
276 let entries = entries();
277
278 // Commit twice so both metadata copies hold a generation, then leave a put
279 // uncommitted.
280 let mut archive = Archive::init(context.child("first"), cfg).await.unwrap();
281 for (index, key, value) in &entries[..5] {
282 archive = archive.put(*index, *key, value).await.unwrap();
283 }
284 archive = archive.sync().await.unwrap();
285 for (index, key, value) in &entries[5..10] {
286 archive = archive.put(*index, *key, value).await.unwrap();
287 }
288 archive = archive.sync().await.unwrap();
289 let (index, key, value) = &entries[10];
290 let archive = archive.put(*index, *key, value).await.unwrap();
291
292 // Fail a sample of removals so destroy stops at varying depths, from the first
293 // metadata blob through the freezer's own blobs.
294 context.storage_fault_config().write().remove_rate = Some(probability!(0.1));
295 let result = archive.destroy().await;
296 context.storage_fault_config().write().remove_rate = None;
297 result.is_err()
298 });
299 if !stopped {
300 continue;
301 }
302 interrupted += 1;
303 deterministic::Runner::from(checkpoint).start(|context| async move {
304 let cfg = config(4);
305 let entries = entries();
306
307 // The reopened archive holds exactly the entries of the latest commit or none,
308 // with index and key views agreeing.
309 let archive = Archive::<_, Digest, i32>::init(context.child("second"), cfg.clone())
310 .await
311 .unwrap();
312 let mut present = Vec::new();
313 for (index, key, value) in &entries {
314 let by_index = archive.get(Identifier::Index(*index)).await.unwrap();
315 assert!(by_index.is_none_or(|stored| stored == *value));
316 assert_eq!(
317 archive.get(Identifier::Key(key)).await.unwrap(),
318 by_index,
319 "seed {seed} index {index}"
320 );
321 assert_eq!(
322 archive.has(Identifier::Index(*index)).await.unwrap(),
323 by_index.is_some()
324 );
325 assert_eq!(
326 archive.has(Identifier::Key(key)).await.unwrap(),
327 by_index.is_some()
328 );
329 if by_index.is_some() {
330 present.push(*index);
331 }
332 }
333 assert!(
334 [0, 10].contains(&present.len())
335 && present.iter().copied().eq(0..Widen::widen(present.len())),
336 "seed {seed} reopened with {present:?}"
337 );
338
339 // A retried destroy completes and leaves nothing behind.
340 archive.destroy().await.unwrap();
341 let archive = Archive::<_, Digest, i32>::init(context.child("third"), cfg)
342 .await
343 .unwrap();
344 assert_eq!(archive.first_index(), None);
345 });
346 }
347 assert!(interrupted > 0, "no seed interrupted destroy");
348 }
349
350 #[test]
351 fn test_unclean_shutdown() {
352 let executor = deterministic::Runner::default();
353 executor.start(|context| async move {
354 let cfg = Config {
355 metadata_partition: "test-metadata2".into(),
356 freezer_table_partition: "test-freezer-table2".into(),
357 freezer_table_initial_size: 8192, // Must be power of 2
358 freezer_table_resize_frequency: 4,
359 freezer_table_resize_chunk_size: 8192,
360 freezer_key_partition: "test-freezer-key2".into(),
361 freezer_key_page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
362 freezer_value_partition: "test-freezer-value2".into(),
363 freezer_value_target_size: 1024 * 1024,
364 freezer_value_compression: Some(3),
365 ordinal_partition: "test-ordinal2".into(),
366 items_per_section: NZU64!(512),
367 freezer_key_write_buffer: NZUsize!(1024),
368 freezer_value_write_buffer: NZUsize!(1024),
369 ordinal_write_buffer: NZUsize!(1024),
370 replay_buffer: NZUsize!(1024),
371 codec_config: (),
372 };
373
374 // First initialization
375 let archive: Archive<_, Digest, i32> =
376 Archive::init(context.child("first"), cfg.clone())
377 .await
378 .unwrap();
379 drop(archive);
380
381 // Second initialization
382 let mut archive = Archive::init(context.child("second"), cfg.clone())
383 .await
384 .unwrap();
385
386 // Add some data
387 let key1 = Sha256::hash(&[b"key1"]);
388 let key2 = Sha256::hash(&[b"key2"]);
389 archive = archive.put(1, key1, &2000).await.unwrap();
390 archive = archive.put(2, key2, &2001).await.unwrap();
391
392 // Sync archive to save the checkpoint
393 let archive = archive.sync().await.unwrap();
394 drop(archive);
395
396 // Re-initialize archive (should load from checkpoint)
397 let archive = Archive::init(context.child("third"), cfg).await.unwrap();
398
399 // Verify data persisted
400 assert_eq!(
401 archive
402 .get(crate::archive::Identifier::Key(&key1))
403 .await
404 .unwrap(),
405 Some(2000)
406 );
407 assert_eq!(
408 archive
409 .get(crate::archive::Identifier::Key(&key2))
410 .await
411 .unwrap(),
412 Some(2001)
413 );
414 });
415 }
416
417 #[test]
418 fn test_sync_empty_archive_then_restart() {
419 let executor = deterministic::Runner::default();
420 executor.start(|context| async move {
421 let cfg = Config {
422 metadata_partition: "empty-metadata".into(),
423 freezer_table_partition: "empty-freezer-table".into(),
424 freezer_table_initial_size: 8192,
425 freezer_table_resize_frequency: 4,
426 freezer_table_resize_chunk_size: 8192,
427 freezer_key_partition: "empty-freezer-key".into(),
428 freezer_key_page_cache: CacheRef::new(PAGE_SIZE, PAGE_CACHE_SIZE),
429 freezer_value_partition: "empty-freezer-value".into(),
430 freezer_value_target_size: 1024 * 1024,
431 freezer_value_compression: Some(3),
432 ordinal_partition: "empty-ordinal".into(),
433 items_per_section: NZU64!(512),
434 freezer_key_write_buffer: NZUsize!(1024),
435 freezer_value_write_buffer: NZUsize!(1024),
436 ordinal_write_buffer: NZUsize!(1024),
437 replay_buffer: NZUsize!(1024),
438 codec_config: (),
439 };
440
441 // Initialize archive, sync without writing anything, then drop
442 let archive: Archive<_, Digest, i32> =
443 Archive::init(context.child("first"), cfg.clone())
444 .await
445 .unwrap();
446 let archive = archive.sync().await.unwrap();
447 drop(archive);
448
449 // Re-initializing must not fail with SectionOutOfRange(0).
450 let archive: Archive<_, Digest, i32> =
451 Archive::init(context.child("second"), cfg.clone())
452 .await
453 .unwrap();
454
455 // Write data after restart to confirm archive is functional
456 let key = Sha256::hash(&[b"after-restart"]);
457 let archive = archive.put_sync(0, key, &42).await.unwrap();
458 drop(archive);
459
460 // Third init to verify persistence
461 let archive: Archive<_, Digest, i32> =
462 Archive::init(context.child("third"), cfg).await.unwrap();
463 assert_eq!(
464 archive
465 .get(crate::archive::Identifier::Key(&key))
466 .await
467 .unwrap(),
468 Some(42)
469 );
470 });
471 }
472}