commonware_storage/metadata/
mod.rs

1//! A key-value store optimized for atomically committing a small collection of metadata.
2//!
3//! [Metadata] is a key-value store optimized for tracking a small collection of metadata
4//! that allows multiple updates to be committed in a single batch. It is commonly used with
5//! a variety of other underlying storage systems to persist application state across restarts.
6//!
7//! # Format
8//!
9//! Data stored in [Metadata] is serialized as a sequence of key-value pairs in either a
10//! "left" or "right" blob:
11//!
12//! ```text
13//! +---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
14//! | 0 | 1 |    ...    | 8 | 9 |10 |11 |12 |13 |14 |15 |16 |  ...  |50 |...|90 |91 |92 |93 |
15//! +---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
16//! |    Version (u64)  |      Key1     |              Value1           |...|  CRC32(u32)   |
17//! +---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+---+
18//! ```
19//!
20//! _To ensure the integrity of the data, a CRC32 checksum is appended to the end of the blob.
21//! This ensures that partial writes are detected before any data is relied on._
22//!
23//! # Atomic Updates
24//!
25//! To provide support for atomic updates, [Metadata] maintains two blobs: a "left" and a "right"
26//! blob. When a new update is committed, it is written to the "older" of the two blobs (indicated
27//! by the version persisted). Writes to [commonware_runtime::Blob] are not atomic and may only
28//! complete partially, so we only overwrite the "newer" blob once the "older" blob has been synced
29//! (otherwise, we would not be guaranteed to recover the latest complete state from disk on
30//! restart as half of a blob could be old data and half new data).
31//!
32//! # Delta Writes
33//!
34//! If the set of keys and the length of values are stable, [Metadata] will only write an update's
35//! delta to disk (rather than rewriting the entire metadata). This makes [Metadata] a great choice
36//! for maintaining even large collections of data (with the majority rarely modified).
37//!
38//! # Example
39//!
40//! ```rust
41//! use commonware_runtime::{Spawner, Runner, deterministic};
42//! use commonware_storage::metadata::{Metadata, Config};
43//! use commonware_utils::sequence::U64;
44//!
45//! let executor = deterministic::Runner::default();
46//! executor.start(|context| async move {
47//!     // Create a store
48//!     let mut metadata = Metadata::init(context, Config {
49//!         partition: "partition".into(),
50//!         codec_config: ((0..).into(), ()),
51//!     }).await.unwrap();
52//!
53//!     // Store metadata
54//!     metadata.put(U64::new(1), b"hello".to_vec());
55//!     metadata.put(U64::new(2), b"world".to_vec());
56//!
57//!     // Sync the metadata store (batch write changes)
58//!     metadata.sync().await.unwrap();
59//!
60//!     // Retrieve some metadata
61//!     let value = metadata.get(&U64::new(1)).unwrap();
62//!
63//! });
64//! ```
65
66#[cfg(all(test, feature = "arbitrary"))]
67mod conformance;
68mod storage;
69pub use storage::Metadata;
70use thiserror::Error;
71
72/// Errors that can occur when interacting with [Metadata].
73#[derive(Debug, Error)]
74pub enum Error {
75    #[error("runtime error: {0}")]
76    Runtime(#[from] commonware_runtime::Error),
77}
78
79/// Configuration for [Metadata] storage.
80#[derive(Clone)]
81pub struct Config<C> {
82    /// The [commonware_runtime::Storage] partition to use for storing metadata.
83    pub partition: String,
84
85    /// The codec configuration to use for the value stored in the metadata.
86    pub codec_config: C,
87}
88
89#[cfg(test)]
90mod tests {
91    use super::*;
92    use commonware_formatting::hex;
93    use commonware_macros::{test_group, test_traced};
94    use commonware_runtime::{deterministic, Blob, Metrics as _, Runner, Storage, Supervisor as _};
95    use commonware_utils::sequence::U64;
96    use rand::{Rng, RngExt as _};
97
98    #[test_traced]
99    fn test_put_get_clear() {
100        // Initialize the deterministic context
101        let executor = deterministic::Runner::default();
102        executor.start(|context| async move {
103            // Create a metadata store
104            let cfg = Config {
105                partition: "test".into(),
106                codec_config: ((0..).into(), ()),
107            };
108            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg)
109                .await
110                .unwrap();
111
112            // Get a key that doesn't exist
113            let key = U64::new(42);
114            let value = metadata.get(&key);
115            assert!(value.is_none());
116
117            // Check metrics
118            let buffer = context.encode();
119            assert!(buffer.contains("first_sync_rewrites_total 0"));
120            assert!(buffer.contains("first_sync_overwrites_total 0"));
121            assert!(buffer.contains("first_keys 0"));
122
123            // Put a key
124            let hello = b"hello".to_vec();
125            metadata.put(key.clone(), hello.clone());
126
127            // Get the key
128            let value = metadata.get(&key).unwrap();
129            assert_eq!(value, &hello);
130
131            // Check metrics
132            let buffer = context.encode();
133            assert!(buffer.contains("first_sync_rewrites_total 0"));
134            assert!(buffer.contains("first_sync_overwrites_total 0"));
135            assert!(buffer.contains("first_keys 1"));
136
137            // Sync the metadata store
138            metadata.sync().await.unwrap();
139
140            // Check metrics
141            let buffer = context.encode();
142            assert!(buffer.contains("first_sync_rewrites_total 1"));
143            assert!(buffer.contains("first_sync_overwrites_total 0"));
144            assert!(buffer.contains("first_keys 1"));
145
146            // Reopen the metadata store
147            drop(metadata);
148            let cfg = Config {
149                partition: "test".into(),
150                codec_config: ((0..).into(), ()),
151            };
152            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
153                .await
154                .unwrap();
155
156            // Check metrics
157            let buffer = context.encode();
158            assert!(buffer.contains("second_sync_rewrites_total 0"));
159            assert!(buffer.contains("second_sync_overwrites_total 0"));
160            assert!(buffer.contains("second_keys 1"));
161
162            // Get the key
163            let value = metadata.get(&key).unwrap();
164            assert_eq!(value, &hello);
165
166            // Test clearing the metadata store
167            metadata.clear();
168            let value = metadata.get(&key);
169            assert!(value.is_none());
170
171            // Check metrics
172            let buffer = context.encode();
173            assert!(buffer.contains("second_sync_rewrites_total 0"));
174            assert!(buffer.contains("second_sync_overwrites_total 0"));
175            assert!(buffer.contains("second_keys 0"));
176
177            metadata.destroy().await.unwrap();
178        });
179    }
180
181    #[test_traced]
182    fn test_put_returns_previous_value() {
183        let executor = deterministic::Runner::default();
184        executor.start(|context| async move {
185            let cfg = Config {
186                partition: "test".into(),
187                codec_config: ((0..).into(), ()),
188            };
189            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg)
190                .await
191                .unwrap();
192
193            let key = U64::new(42);
194
195            // First put returns None (no previous value)
196            let previous = metadata.put(key.clone(), b"first".to_vec());
197            assert!(previous.is_none());
198
199            // Second put returns the previous value
200            let previous = metadata.put(key.clone(), b"second".to_vec());
201            assert_eq!(previous, Some(b"first".to_vec()));
202
203            // Third put returns the previous value
204            let previous = metadata.put(key.clone(), b"third".to_vec());
205            assert_eq!(previous, Some(b"second".to_vec()));
206
207            // Current value is the latest
208            assert_eq!(metadata.get(&key), Some(&b"third".to_vec()));
209
210            // Different key returns None
211            let other_key = U64::new(99);
212            let previous = metadata.put(other_key.clone(), b"other".to_vec());
213            assert!(previous.is_none());
214
215            // Sync and verify persistence
216            metadata.sync().await.unwrap();
217            drop(metadata);
218
219            let cfg = Config {
220                partition: "test".into(),
221                codec_config: ((0..).into(), ()),
222            };
223            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
224                .await
225                .unwrap();
226
227            // After restart, put still returns previous value
228            let previous = metadata.put(key.clone(), b"fourth".to_vec());
229            assert_eq!(previous, Some(b"third".to_vec()));
230
231            metadata.destroy().await.unwrap();
232        });
233    }
234
235    #[test_traced]
236    fn test_multi_sync() {
237        // Initialize the deterministic context
238        let executor = deterministic::Runner::default();
239        executor.start(|context| async move {
240            // Create a metadata store
241            let cfg = Config {
242                partition: "test".into(),
243                codec_config: ((0..).into(), ()),
244            };
245            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg)
246                .await
247                .unwrap();
248
249            // Put a key
250            let key = U64::new(42);
251            let hello = b"hello".to_vec();
252            metadata.put(key.clone(), hello.clone());
253
254            // Sync the metadata store
255            metadata.sync().await.unwrap();
256
257            // Check metrics
258            let buffer = context.encode();
259            assert!(buffer.contains("first_sync_rewrites_total 1"));
260            assert!(buffer.contains("first_sync_overwrites_total 0"));
261            assert!(buffer.contains("first_keys 1"));
262
263            // Put an overlapping key and a new key
264            let world = b"world".to_vec();
265            metadata.put(key.clone(), world.clone());
266            let key2 = U64::new(43);
267            let foo = b"foo".to_vec();
268            metadata.put(key2.clone(), foo.clone());
269
270            // Sync the metadata store
271            metadata.sync().await.unwrap();
272
273            // Check metrics
274            let buffer = context.encode();
275            assert!(buffer.contains("first_sync_rewrites_total 2"));
276            assert!(buffer.contains("first_sync_overwrites_total 0"));
277            assert!(buffer.contains("first_keys 2"));
278
279            // Reopen the metadata store
280            drop(metadata);
281            let cfg = Config {
282                partition: "test".into(),
283                codec_config: ((0..).into(), ()),
284            };
285            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
286                .await
287                .unwrap();
288
289            // Check metrics
290            let buffer = context.encode();
291            assert!(buffer.contains("second_sync_rewrites_total 0"));
292            assert!(buffer.contains("second_sync_overwrites_total 0"));
293            assert!(buffer.contains("second_keys 2"));
294
295            // Get the key
296            let value = metadata.get(&key).unwrap();
297            assert_eq!(value, &world);
298            let value = metadata.get(&key2).unwrap();
299            assert_eq!(value, &foo);
300
301            // Remove the key
302            metadata.remove(&key);
303
304            // Sync the metadata store
305            metadata.sync().await.unwrap();
306
307            // Check metrics
308            let buffer = context.encode();
309            assert!(buffer.contains("second_sync_rewrites_total 1"));
310            assert!(buffer.contains("second_sync_overwrites_total 0"));
311            assert!(buffer.contains("second_keys 1"));
312
313            // Reopen the metadata store
314            drop(metadata);
315            let cfg = Config {
316                partition: "test".into(),
317                codec_config: ((0..).into(), ()),
318            };
319            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("third"), cfg)
320                .await
321                .unwrap();
322
323            // Check metrics
324            let buffer = context.encode();
325            assert!(buffer.contains("third_sync_rewrites_total 0"));
326            assert!(buffer.contains("third_sync_overwrites_total 0"));
327            assert!(buffer.contains("third_keys 1"));
328
329            // Get the key
330            let value = metadata.get(&key);
331            assert!(value.is_none());
332            let value = metadata.get(&key2).unwrap();
333            assert_eq!(value, &foo);
334
335            metadata.destroy().await.unwrap();
336        });
337    }
338
339    #[test_traced]
340    fn test_recover_corrupted_one() {
341        // Initialize the deterministic context
342        let executor = deterministic::Runner::default();
343        executor.start(|context| async move {
344            // Create a metadata store
345            let cfg = Config {
346                partition: "test".into(),
347                codec_config: ((0..).into(), ()),
348            };
349            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg)
350                .await
351                .unwrap();
352
353            // Put a key
354            let key = U64::new(42);
355            let hello = b"hello".to_vec();
356            metadata.put(key.clone(), hello.clone());
357
358            // Sync the metadata store
359            metadata.sync().await.unwrap();
360
361            // Put an overlapping key and a new key
362            let world = b"world".to_vec();
363            metadata.put(key.clone(), world.clone());
364            let key2 = U64::new(43);
365            let foo = b"foo".to_vec();
366            metadata.put(key2, foo.clone());
367
368            // Sync the metadata store
369            metadata.sync().await.unwrap();
370            drop(metadata);
371
372            // Corrupt the metadata store
373            let (blob, _) = context.open("test", b"left").await.unwrap();
374            blob.write_at_sync(0, b"corrupted".to_vec()).await.unwrap();
375
376            // Reopen the metadata store
377            let cfg = Config {
378                partition: "test".into(),
379                codec_config: ((0..).into(), ()),
380            };
381            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
382                .await
383                .unwrap();
384
385            // Get the key (falls back to non-corrupt)
386            let value = metadata.get(&key).unwrap();
387            assert_eq!(value, &hello);
388
389            metadata.destroy().await.unwrap();
390        });
391    }
392
393    #[test_traced]
394    fn test_recover_corrupted_both() {
395        // Initialize the deterministic context
396        let executor = deterministic::Runner::default();
397        executor.start(|context| async move {
398            // Create a metadata store
399            let cfg = Config {
400                partition: "test".into(),
401                codec_config: ((0..).into(), ()),
402            };
403            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg)
404                .await
405                .unwrap();
406
407            // Put a key
408            let key = U64::new(42);
409            let hello = b"hello".to_vec();
410            metadata.put(key.clone(), hello.clone());
411
412            // Sync the metadata store
413            metadata.sync().await.unwrap();
414
415            // Put an overlapping key and a new key
416            let world = b"world".to_vec();
417            metadata.put(key.clone(), world.clone());
418            let key2 = U64::new(43);
419            let foo = b"foo".to_vec();
420            metadata.put(key2, foo.clone());
421
422            // Sync the metadata store
423            metadata.sync().await.unwrap();
424            drop(metadata);
425
426            // Corrupt the metadata store
427            let (blob, _) = context.open("test", b"left").await.unwrap();
428            blob.write_at_sync(0, b"corrupted".to_vec()).await.unwrap();
429            let (blob, _) = context.open("test", b"right").await.unwrap();
430            blob.write_at_sync(0, b"corrupted".to_vec()).await.unwrap();
431
432            // Reopen the metadata store
433            let cfg = Config {
434                partition: "test".into(),
435                codec_config: ((0..).into(), ()),
436            };
437            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
438                .await
439                .unwrap();
440
441            // Get the key (falls back to non-corrupt)
442            let value = metadata.get(&key);
443            assert!(value.is_none());
444
445            // Check metrics
446            let buffer = context.encode();
447            assert!(buffer.contains("second_sync_rewrites_total 0"));
448            assert!(buffer.contains("second_sync_overwrites_total 0"));
449            assert!(buffer.contains("second_keys 0"));
450
451            metadata.destroy().await.unwrap();
452        });
453    }
454
455    #[test_traced]
456    fn test_recover_corrupted_truncate() {
457        // Initialize the deterministic context
458        let executor = deterministic::Runner::default();
459        executor.start(|context| async move {
460            // Create a metadata store
461            let cfg = Config {
462                partition: "test".into(),
463                codec_config: ((0..).into(), ()),
464            };
465            let mut metadata = Metadata::init(context.child("first"), cfg).await.unwrap();
466
467            // Put a key
468            let key = U64::new(42);
469            let hello = b"hello".to_vec();
470            metadata.put(key.clone(), hello.clone());
471
472            // Sync the metadata store
473            metadata.sync().await.unwrap();
474
475            // Put an overlapping key and a new key
476            let world = b"world".to_vec();
477            metadata.put(key.clone(), world.clone());
478            let key2 = U64::new(43);
479            let foo = b"foo".to_vec();
480            metadata.put(key2, foo.clone());
481
482            // Sync the metadata store
483            metadata.sync().await.unwrap();
484            drop(metadata);
485
486            // Corrupt the metadata store
487            let (blob, len) = context.open("test", b"left").await.unwrap();
488            blob.resize(len - 8).await.unwrap();
489            blob.sync().await.unwrap();
490
491            // Reopen the metadata store
492            let cfg = Config {
493                partition: "test".into(),
494                codec_config: ((0..).into(), ()),
495            };
496            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
497                .await
498                .unwrap();
499
500            // Get the key (falls back to non-corrupt)
501            let value = metadata.get(&key).unwrap();
502            assert_eq!(value, &hello);
503
504            metadata.destroy().await.unwrap();
505        });
506    }
507
508    #[test_traced]
509    fn test_recover_corrupted_short() {
510        // Initialize the deterministic context
511        let executor = deterministic::Runner::default();
512        executor.start(|context| async move {
513            // Create a metadata store
514            let cfg = Config {
515                partition: "test".into(),
516                codec_config: ((0..).into(), ()),
517            };
518            let mut metadata = Metadata::init(context.child("first"), cfg).await.unwrap();
519
520            // Put a key
521            let key = U64::new(42);
522            let hello = b"hello".to_vec();
523            metadata.put(key.clone(), hello.clone());
524
525            // Sync the metadata store
526            metadata.sync().await.unwrap();
527
528            // Put an overlapping key and a new key
529            let world = b"world".to_vec();
530            metadata.put(key.clone(), world.clone());
531            let key2 = U64::new(43);
532            let foo = b"foo".to_vec();
533            metadata.put(key2, foo.clone());
534
535            // Sync the metadata store
536            metadata.sync().await.unwrap();
537            drop(metadata);
538
539            // Corrupt the metadata store
540            let (blob, _) = context.open("test", b"left").await.unwrap();
541            blob.resize(5).await.unwrap();
542            blob.sync().await.unwrap();
543
544            // Reopen the metadata store
545            let cfg = Config {
546                partition: "test".into(),
547                codec_config: ((0..).into(), ()),
548            };
549            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
550                .await
551                .unwrap();
552
553            // Get the key (falls back to non-corrupt)
554            let value = metadata.get(&key).unwrap();
555            assert_eq!(value, &hello);
556
557            metadata.destroy().await.unwrap();
558        });
559    }
560
561    #[test_traced]
562    fn test_unclean_shutdown() {
563        // Initialize the deterministic context
564        let executor = deterministic::Runner::default();
565        executor.start(|context| async move {
566            let key = U64::new(42);
567            let hello = b"hello".to_vec();
568            {
569                // Create a metadata store
570                let cfg = Config {
571                    partition: "test".into(),
572                    codec_config: ((0..).into(), ()),
573                };
574                let mut metadata = Metadata::init(context.child("first"), cfg).await.unwrap();
575
576                // Put a key
577                metadata.put(key.clone(), hello.clone());
578
579                // Drop metadata before sync
580            }
581
582            // Reopen the metadata store
583            let cfg = Config {
584                partition: "test".into(),
585                codec_config: ((0..).into(), ()),
586            };
587            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
588                .await
589                .unwrap();
590
591            // Get the key
592            let value = metadata.get(&key);
593            assert!(value.is_none());
594
595            // Check metrics
596            let buffer = context.encode();
597            assert!(buffer.contains("second_sync_rewrites_total 0"));
598            assert!(buffer.contains("second_sync_overwrites_total 0"));
599            assert!(buffer.contains("second_keys 0"));
600
601            metadata.destroy().await.unwrap();
602        });
603    }
604
605    #[test_traced]
606    #[should_panic(expected = "usize value is larger than u32")]
607    fn test_value_too_big_error() {
608        // Initialize the deterministic context
609        let executor = deterministic::Runner::default();
610        executor.start(|context| async move {
611            // Create a metadata store
612            let cfg = Config {
613                partition: "test".into(),
614                codec_config: ((0..).into(), ()),
615            };
616            let mut metadata = Metadata::init(context.child("storage"), cfg).await.unwrap();
617
618            // Create a value that exceeds u32::MAX bytes
619            let value = vec![0u8; (u32::MAX as usize) + 1];
620            metadata.put(U64::new(1), value);
621
622            // Assert
623            metadata.sync().await.unwrap();
624        });
625    }
626
627    #[test_traced]
628    fn test_delta_writes() {
629        // Initialize the deterministic context
630        let executor = deterministic::Runner::default();
631        executor.start(|context| async move {
632            // Create a metadata store
633            let cfg = Config {
634                partition: "test".into(),
635                codec_config: ((0..).into(), ()),
636            };
637            let mut metadata = Metadata::init(context.child("storage"), cfg).await.unwrap();
638
639            // Put initial keys
640            for i in 0..100 {
641                metadata.put(U64::new(i), vec![i as u8; 100]);
642            }
643
644            // First sync - should write everything to the first blob
645            //
646            // 100 keys * (8 bytes for key + 1 byte for len + 100 bytes for value) + 8 bytes for version + 4 bytes for checksum
647            metadata.sync().await.unwrap();
648            let buffer = context.encode();
649            assert!(buffer.contains("sync_rewrites_total 1"), "{buffer}");
650            assert!(buffer.contains("sync_overwrites_total 0"), "{buffer}");
651            assert!(
652                buffer.contains("runtime_storage_write_bytes_total 10912"),
653                "{buffer}",
654            );
655
656            // Modify just one key
657            metadata.put(U64::new(51), vec![0xff; 100]);
658
659            // Sync again - should write everything to the second blob
660            metadata.sync().await.unwrap();
661            let buffer = context.encode();
662            assert!(buffer.contains("sync_rewrites_total 2"), "{buffer}");
663            assert!(buffer.contains("sync_overwrites_total 0"), "{buffer}");
664            assert!(
665                buffer.contains("runtime_storage_write_bytes_total 21824"),
666                "{buffer}",
667            );
668
669            // Sync again - should write only diff from the first blob
670            //
671            // 1 byte for len + 100 bytes for value + 8 byte for version + 4 bytes for checksum
672            metadata.sync().await.unwrap();
673            let buffer = context.encode();
674            assert!(buffer.contains("sync_rewrites_total 2"), "{buffer}");
675            assert!(buffer.contains("sync_overwrites_total 1"), "{buffer}");
676            assert!(
677                buffer.contains("runtime_storage_write_bytes_total 21937"),
678                "{buffer}",
679            );
680
681            // Sync again - both blobs already contain the latest state
682            metadata.sync().await.unwrap();
683            let buffer = context.encode();
684            assert!(buffer.contains("sync_rewrites_total 2"), "{buffer}");
685            assert!(buffer.contains("sync_overwrites_total 1"), "{buffer}");
686            assert!(
687                buffer.contains("runtime_storage_write_bytes_total 21937"),
688                "{buffer}",
689            );
690
691            // Remove a key - should rewrite everything
692            //
693            // 99 keys * (8 bytes for key + 1 bytes for len + 100 bytes for value) + 8 bytes for version + 4 bytes for checksum
694            metadata.remove(&U64::new(51));
695            metadata.sync().await.unwrap();
696            let buffer = context.encode();
697            assert!(buffer.contains("sync_rewrites_total 3"), "{buffer}");
698            assert!(buffer.contains("sync_overwrites_total 1"), "{buffer}");
699            assert!(
700                buffer.contains("runtime_storage_write_bytes_total 32740"),
701                "{buffer}"
702            );
703
704            // Sync again - should also rewrite
705            metadata.sync().await.unwrap();
706            let buffer = context.encode();
707            assert!(buffer.contains("sync_rewrites_total 4"), "{buffer}");
708            assert!(buffer.contains("sync_overwrites_total 1"), "{buffer}");
709            assert!(
710                buffer.contains("runtime_storage_write_bytes_total 43543"),
711                "{buffer}"
712            );
713
714            // Modify in-place - should overwrite
715            //
716            // 1 byte for len + 100 bytes for value + 8 byte for version + 4 bytes for checksum
717            metadata.put(U64::new(50), vec![0xff; 100]);
718            metadata.sync().await.unwrap();
719            let buffer = context.encode();
720            assert!(buffer.contains("sync_rewrites_total 4"), "{buffer}");
721            assert!(buffer.contains("sync_overwrites_total 2"), "{buffer}");
722            assert!(
723                buffer.contains("runtime_storage_write_bytes_total 43656"),
724                "{buffer}"
725            );
726
727            // Clean up
728            metadata.destroy().await.unwrap();
729        });
730    }
731
732    #[test_traced]
733    fn test_multi_key_overwrites() {
734        let executor = deterministic::Runner::default();
735        executor.start(|context| async move {
736            let cfg = Config {
737                partition: "test".into(),
738                codec_config: ((0..).into(), ()),
739            };
740            let mut metadata =
741                Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg.clone())
742                    .await
743                    .unwrap();
744
745            // Put initial keys and populate both blobs
746            for i in 0..100 {
747                metadata.put(U64::new(i), vec![i as u8; 100]);
748            }
749            metadata.sync().await.unwrap();
750            metadata.sync().await.unwrap();
751            let buffer = context.encode();
752            assert!(buffer.contains("first_sync_rewrites_total 2"), "{buffer}");
753            assert!(
754                buffer.contains("runtime_storage_write_bytes_total 21824"),
755                "{buffer}",
756            );
757
758            // Modify several keys with same-size values
759            for i in [10u64, 11, 12, 50, 98, 99] {
760                metadata.put(U64::new(i), vec![0xAA; 100]);
761            }
762
763            // Sync writes one delta per modified value.
764            //
765            // 6 * (1 byte for len + 100 bytes for value) + 8 bytes for version
766            // + 4 bytes for checksum.
767            metadata.sync().await.unwrap();
768            let buffer = context.encode();
769            assert!(buffer.contains("first_sync_rewrites_total 2"), "{buffer}");
770            assert!(buffer.contains("first_sync_overwrites_total 1"), "{buffer}");
771            assert!(
772                buffer.contains("runtime_storage_write_bytes_total 22442"),
773                "{buffer}",
774            );
775
776            // Sync again - the same deltas propagate to the other blob
777            metadata.sync().await.unwrap();
778            let buffer = context.encode();
779            assert!(buffer.contains("first_sync_overwrites_total 2"), "{buffer}");
780            assert!(
781                buffer.contains("runtime_storage_write_bytes_total 23060"),
782                "{buffer}",
783            );
784
785            // Sync again - both blobs already contain the latest state
786            metadata.sync().await.unwrap();
787            let buffer = context.encode();
788            assert!(buffer.contains("first_sync_rewrites_total 2"), "{buffer}");
789            assert!(buffer.contains("first_sync_overwrites_total 2"), "{buffer}");
790            assert!(
791                buffer.contains("runtime_storage_write_bytes_total 23060"),
792                "{buffer}",
793            );
794
795            // Mix a same-size update with a size-changing update. The overwrite
796            // scan updates the mirror for the smaller key before the size change
797            // forces a rewrite, which must discard that partial mutation.
798            metadata.put(U64::new(20), vec![0xBB; 100]);
799            metadata.put(U64::new(30), vec![0xCC; 150]);
800            metadata.sync().await.unwrap();
801            metadata.sync().await.unwrap();
802            let buffer = context.encode();
803            assert!(buffer.contains("first_sync_rewrites_total 4"), "{buffer}");
804            assert!(buffer.contains("first_sync_overwrites_total 2"), "{buffer}");
805
806            // Restart the metadata store
807            drop(metadata);
808            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
809                .await
810                .unwrap();
811
812            // Verify every value survived exactly
813            for i in 0..100u64 {
814                let expected = match i {
815                    10 | 11 | 12 | 50 | 98 | 99 => vec![0xAA; 100],
816                    20 => vec![0xBB; 100],
817                    30 => vec![0xCC; 150],
818                    _ => vec![i as u8; 100],
819                };
820                assert_eq!(metadata.get(&U64::new(i)).unwrap(), &expected, "key {i}");
821            }
822
823            metadata.destroy().await.unwrap();
824        });
825    }
826
827    #[test_traced]
828    fn test_sync_with_no_changes() {
829        let executor = deterministic::Runner::default();
830        executor.start(|context| async move {
831            let cfg = Config {
832                partition: "test".into(),
833                codec_config: ((0..).into(), ()),
834            };
835            let mut metadata =
836                Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg.clone())
837                    .await
838                    .unwrap();
839
840            // Put initial data
841            metadata
842                .put_sync(U64::new(1), b"hello".to_vec())
843                .await
844                .unwrap();
845
846            // Sync again with no changes. This still rewrites because only one blob
847            // has the new key order.
848            metadata.sync().await.unwrap();
849            let buffer = context.encode();
850            assert!(buffer.contains("sync_rewrites_total 2"));
851            assert!(buffer.contains("sync_overwrites_total 0"));
852
853            // Sync again - both blobs already contain the latest state
854            metadata.sync().await.unwrap();
855            let buffer = context.encode();
856            assert!(buffer.contains("sync_rewrites_total 2"));
857            assert!(buffer.contains("sync_overwrites_total 0"));
858
859            // Sync again - should remain a no-op
860            metadata.sync().await.unwrap();
861            let buffer = context.encode();
862            assert!(buffer.contains("sync_rewrites_total 2"));
863            assert!(buffer.contains("sync_overwrites_total 0"));
864
865            // Restart the metadata store and verify the no-op left durable state
866            drop(metadata);
867            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
868                .await
869                .unwrap();
870            assert_eq!(metadata.get(&U64::new(1)).unwrap(), b"hello");
871
872            metadata.destroy().await.unwrap();
873        });
874    }
875
876    #[test_traced]
877    fn test_get_mut_marks_modified() {
878        let executor = deterministic::Runner::default();
879        executor.start(|context| async move {
880            let cfg = Config {
881                partition: "test".into(),
882                codec_config: ((0..).into(), ()),
883            };
884            let mut metadata =
885                Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg.clone())
886                    .await
887                    .unwrap();
888
889            // Put initial data
890            metadata
891                .put_sync(U64::new(1), b"hello".to_vec())
892                .await
893                .unwrap();
894
895            // Sync again to ensure both blobs are populated
896            metadata.sync().await.unwrap();
897
898            // Use get_mut to modify value
899            let value = metadata.get_mut(&U64::new(1)).unwrap();
900            value[0] = b'H';
901
902            // Sync should detect the modification and do a rewrite (due to recent key_order_changed)
903            metadata.sync().await.unwrap();
904            let buffer = context.encode();
905            assert!(buffer.contains("first_sync_rewrites_total 2"));
906            assert!(buffer.contains("first_sync_overwrites_total 1"));
907
908            // Restart the metadata store
909            drop(metadata);
910            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
911                .await
912                .unwrap();
913
914            // Verify the change persisted
915            let value = metadata.get(&U64::new(1)).unwrap();
916            assert_eq!(value[0], b'H');
917
918            metadata.destroy().await.unwrap();
919        });
920    }
921
922    #[test_traced]
923    fn test_mixed_operation_sequences() {
924        let executor = deterministic::Runner::default();
925        executor.start(|context| async move {
926            let cfg = Config {
927                partition: "test".into(),
928                codec_config: ((0..).into(), ()),
929            };
930            let mut metadata =
931                Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg.clone())
932                    .await
933                    .unwrap();
934
935            let key = U64::new(1);
936
937            // Test: put -> remove -> put same key
938            metadata.put(key.clone(), b"first".to_vec());
939            metadata.remove(&key);
940            metadata
941                .put_sync(key.clone(), b"second".to_vec())
942                .await
943                .unwrap();
944            let value = metadata.get(&key).unwrap();
945            assert_eq!(value, b"second");
946
947            // Test: put -> get_mut -> remove -> put
948            metadata.put(key.clone(), b"third".to_vec());
949            let value = metadata.get_mut(&key).unwrap();
950            value[0] = b'T';
951            metadata.remove(&key);
952            metadata
953                .put_sync(key.clone(), b"fourth".to_vec())
954                .await
955                .unwrap();
956            let value = metadata.get(&key).unwrap();
957            assert_eq!(value, b"fourth");
958
959            // Restart the metadata store
960            drop(metadata);
961            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
962                .await
963                .unwrap();
964
965            // Verify the changes persisted
966            let value = metadata.get(&key).unwrap();
967            assert_eq!(value, b"fourth");
968
969            metadata.destroy().await.unwrap();
970        });
971    }
972
973    #[test_traced]
974    fn test_overwrite_vs_rewrite() {
975        let executor = deterministic::Runner::default();
976        executor.start(|context| async move {
977            let cfg = Config {
978                partition: "test".into(),
979                codec_config: ((0..).into(), ()),
980            };
981            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("storage"), cfg)
982                .await
983                .unwrap();
984
985            // Set up initial data
986            metadata.put(U64::new(1), vec![1; 10]);
987            metadata.put(U64::new(2), vec![2; 10]);
988            metadata.sync().await.unwrap();
989
990            // Same size modification before both blobs are populated
991            metadata.put(U64::new(1), vec![0xFF; 10]);
992            metadata.sync().await.unwrap();
993            let buffer = context.encode();
994            assert!(buffer.contains("sync_rewrites_total 2"));
995            assert!(buffer.contains("sync_overwrites_total 0"));
996
997            // Let key order stabilize with another sync
998            metadata.sync().await.unwrap();
999            let buffer = context.encode();
1000            assert!(buffer.contains("sync_rewrites_total 2"));
1001            assert!(buffer.contains("sync_overwrites_total 1"));
1002
1003            // Same size modification after both blobs are populated - should overwrite
1004            metadata.put(U64::new(1), vec![0xAA; 10]);
1005            metadata.sync().await.unwrap();
1006            let buffer = context.encode();
1007            assert!(buffer.contains("sync_rewrites_total 2"));
1008            assert!(buffer.contains("sync_overwrites_total 2"));
1009
1010            // Different size modification - should rewrite
1011            metadata.put(U64::new(1), vec![0xFF; 20]);
1012            metadata.sync().await.unwrap();
1013            let buffer = context.encode();
1014            assert!(buffer.contains("sync_rewrites_total 3"));
1015            assert!(buffer.contains("sync_overwrites_total 2"));
1016
1017            // Add new key - should rewrite (key order changed)
1018            metadata.put(U64::new(3), vec![3; 10]);
1019            metadata.sync().await.unwrap();
1020            let buffer = context.encode();
1021            assert!(buffer.contains("sync_rewrites_total 4"));
1022            assert!(buffer.contains("sync_overwrites_total 2"));
1023
1024            // Stabilize key order
1025            metadata.sync().await.unwrap();
1026            let buffer = context.encode();
1027            assert!(buffer.contains("sync_rewrites_total 5"));
1028            assert!(buffer.contains("sync_overwrites_total 2"));
1029
1030            // Modify existing key with same size - should overwrite after stabilized
1031            metadata.put(U64::new(2), vec![0xAA; 10]);
1032            metadata.sync().await.unwrap();
1033            let buffer = context.encode();
1034            assert!(buffer.contains("sync_rewrites_total 5"));
1035            assert!(buffer.contains("sync_overwrites_total 3"));
1036
1037            metadata.destroy().await.unwrap();
1038        });
1039    }
1040
1041    #[test_traced]
1042    fn test_blob_resize() {
1043        let executor = deterministic::Runner::default();
1044        executor.start(|context| async move {
1045            let cfg = Config {
1046                partition: "test".into(),
1047                codec_config: ((0..).into(), ()),
1048            };
1049            let mut metadata =
1050                Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg.clone())
1051                    .await
1052                    .unwrap();
1053
1054            // Start with large data
1055            for i in 0..10 {
1056                metadata.put(U64::new(i), vec![i as u8; 100]);
1057            }
1058            metadata.sync().await.unwrap();
1059
1060            // Stabilize key order
1061            metadata.sync().await.unwrap();
1062            let buffer = context.encode();
1063            assert!(buffer.contains("first_sync_rewrites_total 2"));
1064            assert!(buffer.contains("first_sync_overwrites_total 0"));
1065
1066            // Remove most data to make blob smaller
1067            for i in 1..10 {
1068                metadata.remove(&U64::new(i));
1069            }
1070            metadata.sync().await.unwrap();
1071
1072            // Verify the remaining data is still accessible
1073            let value = metadata.get(&U64::new(0)).unwrap();
1074            assert_eq!(value.len(), 100);
1075            assert_eq!(value[0], 0);
1076
1077            // Check that sync properly handles blob resizing
1078            let buffer = context.encode();
1079            assert!(buffer.contains("first_sync_rewrites_total 3"));
1080            assert!(buffer.contains("first_sync_overwrites_total 0"));
1081
1082            // Restart the metadata store
1083            drop(metadata);
1084            let metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
1085                .await
1086                .unwrap();
1087
1088            // Verify the changes persisted
1089            let value = metadata.get(&U64::new(0)).unwrap();
1090            assert_eq!(value.len(), 100);
1091            assert_eq!(value[0], 0);
1092
1093            // Verify the removed keys are not present
1094            for i in 1..10 {
1095                assert!(metadata.get(&U64::new(i)).is_none());
1096            }
1097
1098            metadata.destroy().await.unwrap();
1099        });
1100    }
1101
1102    #[test_traced]
1103    fn test_clear_and_repopulate() {
1104        let executor = deterministic::Runner::default();
1105        executor.start(|context| async move {
1106            let cfg = Config {
1107                partition: "test".into(),
1108                codec_config: ((0..).into(), ()),
1109            };
1110            let mut metadata =
1111                Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg.clone())
1112                    .await
1113                    .unwrap();
1114
1115            // Initial data
1116            metadata.put(U64::new(1), b"first".to_vec());
1117            metadata
1118                .put_sync(U64::new(2), b"second".to_vec())
1119                .await
1120                .unwrap();
1121
1122            // Clear everything
1123            metadata.clear();
1124            metadata.sync().await.unwrap();
1125
1126            // Verify empty
1127            assert!(metadata.get(&U64::new(1)).is_none());
1128            assert!(metadata.get(&U64::new(2)).is_none());
1129
1130            // Restart the metadata store
1131            drop(metadata);
1132            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
1133                .await
1134                .unwrap();
1135
1136            // Verify the changes persisted
1137            assert!(metadata.get(&U64::new(1)).is_none());
1138            assert!(metadata.get(&U64::new(2)).is_none());
1139
1140            // Repopulate with different data
1141            metadata.put(U64::new(3), b"third".to_vec());
1142            metadata
1143                .put_sync(U64::new(4), b"fourth".to_vec())
1144                .await
1145                .unwrap();
1146
1147            // Verify new data
1148            assert_eq!(metadata.get(&U64::new(3)).unwrap(), b"third");
1149            assert_eq!(metadata.get(&U64::new(4)).unwrap(), b"fourth");
1150            assert!(metadata.get(&U64::new(1)).is_none());
1151            assert!(metadata.get(&U64::new(2)).is_none());
1152
1153            metadata.destroy().await.unwrap();
1154        });
1155    }
1156
1157    fn test_metadata_operations_and_restart(num_operations: usize) -> String {
1158        let executor = deterministic::Runner::default();
1159        executor.start(|mut context| async move {
1160            let cfg = Config {
1161                partition: "test-determinism".into(),
1162                codec_config: ((0..).into(), ()),
1163            };
1164            let mut metadata =
1165                Metadata::<_, U64, Vec<u8>>::init(context.child("storage"), cfg.clone())
1166                    .await
1167                    .unwrap();
1168
1169            // Perform a series of deterministic operations
1170            for i in 0..num_operations {
1171                let key = U64::new(i as u64);
1172                let mut value = vec![0u8; 64];
1173                context.fill_bytes(&mut value);
1174                metadata.put(key, value);
1175
1176                // Sync occasionally
1177                if context.random_bool(0.1) {
1178                    metadata.sync().await.unwrap();
1179                }
1180
1181                // Update some existing keys
1182                if context.random_bool(0.1) {
1183                    let selected_index = context.random_range(0..=i);
1184                    let update_key = U64::new(selected_index as u64);
1185                    let mut new_value = vec![0u8; 64];
1186                    context.fill_bytes(&mut new_value);
1187                    metadata.put(update_key, new_value);
1188                }
1189
1190                // Remove some keys
1191                if context.random_bool(0.1) {
1192                    let selected_index = context.random_range(0..=i);
1193                    let remove_key = U64::new(selected_index as u64);
1194                    metadata.remove(&remove_key);
1195                }
1196
1197                // Use get_mut occasionally
1198                if context.random_bool(0.1) {
1199                    let selected_index = context.random_range(0..=i);
1200                    let mut_key = U64::new(selected_index as u64);
1201                    if let Some(value) = metadata.get_mut(&mut_key) {
1202                        if !value.is_empty() {
1203                            value[0] = value[0].wrapping_add(1);
1204                        }
1205                    }
1206                }
1207            }
1208            metadata.sync().await.unwrap();
1209
1210            // Destroy the metadata store
1211            metadata.destroy().await.unwrap();
1212
1213            context.auditor().state()
1214        })
1215    }
1216
1217    #[test_group("slow")]
1218    #[test_traced]
1219    fn test_determinism() {
1220        let state1 = test_metadata_operations_and_restart(1_000);
1221        let state2 = test_metadata_operations_and_restart(1_000);
1222        assert_eq!(state1, state2);
1223    }
1224
1225    #[test_traced]
1226    fn test_keys_iterator() {
1227        // Initialize the deterministic context
1228        let executor = deterministic::Runner::default();
1229        executor.start(|context| async move {
1230            // Create a metadata store
1231            let cfg = Config {
1232                partition: "test".into(),
1233                codec_config: ((0..).into(), ()),
1234            };
1235            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("storage"), cfg)
1236                .await
1237                .unwrap();
1238
1239            // Add some keys with different prefixes
1240            metadata.put(U64::new(0x1000), b"value1".to_vec());
1241            metadata.put(U64::new(0x1001), b"value2".to_vec());
1242            metadata.put(U64::new(0x1002), b"value3".to_vec());
1243            metadata.put(U64::new(0x2000), b"value4".to_vec());
1244            metadata.put(U64::new(0x2001), b"value5".to_vec());
1245            metadata.put(U64::new(0x3000), b"value6".to_vec());
1246
1247            // Test iterating over all keys
1248            let all_keys: Vec<_> = metadata.keys().cloned().collect();
1249            assert_eq!(all_keys.len(), 6);
1250            assert!(all_keys.contains(&U64::new(0x1000)));
1251            assert!(all_keys.contains(&U64::new(0x3000)));
1252
1253            // Test iterating with prefix 0x10
1254            let prefix = hex!("0x00000000000010");
1255            let prefix_keys: Vec<_> = metadata
1256                .keys()
1257                .filter(|k| k.as_ref().starts_with(&prefix))
1258                .cloned()
1259                .collect();
1260            assert_eq!(prefix_keys.len(), 3);
1261            assert!(prefix_keys.contains(&U64::new(0x1000)));
1262            assert!(prefix_keys.contains(&U64::new(0x1001)));
1263            assert!(prefix_keys.contains(&U64::new(0x1002)));
1264            assert!(!prefix_keys.contains(&U64::new(0x2000)));
1265
1266            // Test iterating with prefix 0x20
1267            let prefix = hex!("0x00000000000020");
1268            let prefix_keys: Vec<_> = metadata
1269                .keys()
1270                .filter(|k| k.as_ref().starts_with(&prefix))
1271                .cloned()
1272                .collect();
1273            assert_eq!(prefix_keys.len(), 2);
1274            assert!(prefix_keys.contains(&U64::new(0x2000)));
1275            assert!(prefix_keys.contains(&U64::new(0x2001)));
1276
1277            // Test with non-matching prefix
1278            let prefix = hex!("0x00000000000040");
1279            let prefix_keys: Vec<_> = metadata
1280                .keys()
1281                .filter(|k| k.as_ref().starts_with(&prefix))
1282                .cloned()
1283                .collect();
1284            assert_eq!(prefix_keys.len(), 0);
1285
1286            metadata.destroy().await.unwrap();
1287        });
1288    }
1289
1290    #[test_traced]
1291    fn test_retain() {
1292        // Initialize the deterministic context
1293        let executor = deterministic::Runner::default();
1294        executor.start(|context| async move {
1295            // Create a metadata store
1296            let cfg = Config {
1297                partition: "test".into(),
1298                codec_config: ((0..).into(), ()),
1299            };
1300            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("first"), cfg)
1301                .await
1302                .unwrap();
1303
1304            // Add some keys with different prefixes
1305            metadata.put(U64::new(0x1000), b"value1".to_vec());
1306            metadata.put(U64::new(0x1001), b"value2".to_vec());
1307            metadata.put(U64::new(0x1002), b"value3".to_vec());
1308            metadata.put(U64::new(0x2000), b"value4".to_vec());
1309            metadata.put(U64::new(0x2001), b"value5".to_vec());
1310            metadata.put(U64::new(0x3000), b"value6".to_vec());
1311
1312            // Check initial metrics
1313            let buffer = context.encode();
1314            assert!(buffer.contains("first_keys 6"));
1315
1316            // Remove keys with prefix 0x10
1317            let prefix = hex!("0x00000000000010");
1318            metadata.retain(|k, _| !k.as_ref().starts_with(&prefix));
1319
1320            // Check metrics after removal
1321            let buffer = context.encode();
1322            assert!(buffer.contains("first_keys 3"));
1323
1324            // Verify remaining keys
1325            assert!(metadata.get(&U64::new(0x1000)).is_none());
1326            assert!(metadata.get(&U64::new(0x1001)).is_none());
1327            assert!(metadata.get(&U64::new(0x1002)).is_none());
1328            assert!(metadata.get(&U64::new(0x2000)).is_some());
1329            assert!(metadata.get(&U64::new(0x2001)).is_some());
1330            assert!(metadata.get(&U64::new(0x3000)).is_some());
1331
1332            // Sync and reopen to ensure persistence
1333            metadata.sync().await.unwrap();
1334            drop(metadata);
1335            let cfg = Config {
1336                partition: "test".into(),
1337                codec_config: ((0..).into(), ()),
1338            };
1339            let mut metadata = Metadata::<_, U64, Vec<u8>>::init(context.child("second"), cfg)
1340                .await
1341                .unwrap();
1342
1343            // Verify keys are still removed after restart
1344            assert!(metadata.get(&U64::new(0x1000)).is_none());
1345            assert!(metadata.get(&U64::new(0x2000)).is_some());
1346            assert_eq!(metadata.keys().count(), 3);
1347
1348            // Remove non-existing prefix
1349            let prefix = hex!("0x00000000000040");
1350            metadata.retain(|k, _| !k.as_ref().starts_with(&prefix));
1351
1352            // Remove all remaining keys
1353            metadata.retain(|_, _| false);
1354            assert_eq!(metadata.keys().count(), 0);
1355
1356            metadata.destroy().await.unwrap();
1357        });
1358    }
1359}
commonware_storage/metadata/mod.rs

commonware_storage/metadata/
mod.rs
