commonware_storage/cache/
mod.rs

1//! A prunable cache for ordered data with index-based lookups.
2//!
3//! Data is stored in [crate::journal::segmented::variable::Journal] (an append-only log) and the location of
4//! written data is tracked in-memory by index to enable **single-read lookups** for cached data.
5//!
6//! Unlike [crate::archive::Archive], the [Cache] is optimized for simplicity and does
7//! not support key-based lookups (only index-based access is provided). This makes it ideal for
8//! caching sequential data where you know the exact index of the item you want to retrieve.
9//!
10//! # Memory Overhead
11//!
12//! [Cache] maintains a single in-memory map to track the location of each index item. The memory
13//! used to track each item is `8 + 4 + 4` bytes (where `8` is the index, `4` is the offset, and
14//! `4` is the length). This results in approximately `16` bytes of memory overhead per cached item.
15//!
16//! # Pruning
17//!
18//! [Cache] supports pruning up to a minimum `index` using the `prune` method. After `prune` is
19//! called on a `section`, all interaction with a `section` less than the pruned `section` will
20//! return an error. The pruning granularity is determined by `items_per_blob` in the configuration.
21//!
22//! # Single Operation Reads
23//!
24//! To enable single operation reads (i.e. reading all of an item in a single call to
25//! [commonware_runtime::Blob]), [Cache] stores the length of each item in its in-memory index.
26//! This ensures that reading a cached item requires only one disk operation.
27//!
28//! # Compression
29//!
30//! [Cache] supports compressing data before storing it on disk. This can be enabled by setting
31//! the `compression` field in the `Config` struct to a valid `zstd` compression level. This setting
32//! can be changed between initializations of [Cache], however, it must remain populated if any
33//! data was written with compression enabled.
34//!
35//! # Querying for Gaps
36//!
37//! [Cache] tracks gaps in the index space to enable the caller to efficiently fetch unknown keys
38//! using `next_gap`. This is a very common pattern when syncing blocks in a blockchain.
39//!
40//! # Example
41//!
42//! ```rust
43//! use commonware_runtime::{Spawner, Runner, deterministic, buffer::paged::CacheRef};
44//! use commonware_storage::cache::{Cache, Config};
45//! use commonware_utils::{NZUsize, NZU16, NZU64};
46//!
47//! let executor = deterministic::Runner::default();
48//! executor.start(|context| async move {
49//!     // Create a cache
50//!     let cfg = Config {
51//!         partition: "cache".into(),
52//!         compression: Some(3),
53//!         codec_config: (),
54//!         items_per_blob: NZU64!(1024),
55//!         write_buffer: NZUsize!(1024 * 1024),
56//!         replay_buffer: NZUsize!(4096),
57//!         page_cache: CacheRef::from_pooler(&context, NZU16!(1024), NZUsize!(10)),
58//!     };
59//!     let mut cache = Cache::init(context, cfg).await.unwrap();
60//!
61//!     // Put data at index
62//!     cache.put(1, 100u32).await.unwrap();
63//!
64//!     // Get data by index
65//!     let data: Option<u32> = cache.get(1).await.unwrap();
66//!     assert_eq!(data, Some(100));
67//!
68//!     // Check for gaps in the index space
69//!     cache.put(10, 200u32).await.unwrap();
70//!     let (current_end, start_next) = cache.next_gap(5);
71//!     assert!(current_end.is_none());
72//!     assert_eq!(start_next, Some(10));
73//!
74//!     // Sync the cache
75//!     cache.sync().await.unwrap();
76//! });
77//! ```
78
79use commonware_runtime::buffer::paged::CacheRef;
80use std::num::{NonZeroU64, NonZeroUsize};
81use thiserror::Error;
82
83#[cfg(all(test, feature = "arbitrary"))]
84mod conformance;
85mod storage;
86pub use storage::Cache;
87
88/// Errors that can occur when interacting with the cache.
89#[derive(Debug, Error)]
90pub enum Error {
91    #[error("journal error: {0}")]
92    Journal(#[from] crate::journal::Error),
93    #[error("record corrupted")]
94    RecordCorrupted,
95    #[error("already pruned to: {0}")]
96    AlreadyPrunedTo(u64),
97    #[error("record too large")]
98    RecordTooLarge,
99}
100
101/// Configuration for [Cache] storage.
102#[derive(Clone)]
103pub struct Config<C> {
104    /// The partition to use for the cache's [crate::journal] storage.
105    pub partition: String,
106
107    /// The compression level to use for the cache's [crate::journal] storage.
108    pub compression: Option<u8>,
109
110    /// The [commonware_codec::Codec] configuration to use for the value stored in the cache.
111    pub codec_config: C,
112
113    /// The number of items per section (the granularity of pruning).
114    pub items_per_blob: NonZeroU64,
115
116    /// The amount of bytes that can be buffered in a section before being written to a
117    /// [commonware_runtime::Blob].
118    pub write_buffer: NonZeroUsize,
119
120    /// The buffer size to use when replaying a [commonware_runtime::Blob].
121    pub replay_buffer: NonZeroUsize,
122
123    /// The page cache to use for the underlying [crate::journal] storage.
124    pub page_cache: CacheRef,
125}
126
127#[cfg(test)]
128mod tests {
129    use super::*;
130    use crate::journal::Error as JournalError;
131    use commonware_macros::{test_group, test_traced};
132    use commonware_runtime::{
133        deterministic, telemetry::metrics::has_metric_value, Metrics as _, Runner, Supervisor as _,
134    };
135    use commonware_utils::{NZUsize, NZU16, NZU64};
136    use rand::RngExt as _;
137    use std::{collections::BTreeMap, num::NonZeroU16};
138
139    const DEFAULT_ITEMS_PER_BLOB: u64 = 65536;
140    const DEFAULT_WRITE_BUFFER: usize = 1024;
141    const DEFAULT_REPLAY_BUFFER: usize = 4096;
142    const PAGE_SIZE: NonZeroU16 = NZU16!(1024);
143    const PAGE_CACHE_SIZE: NonZeroUsize = NZUsize!(10);
144
145    #[test_traced]
146    fn test_cache_compression_then_none() {
147        // Initialize the deterministic context
148        let executor = deterministic::Runner::default();
149        executor.start(|context| async move {
150            // Initialize the cache
151            let cfg = Config {
152                partition: "test-partition".into(),
153                codec_config: (),
154                compression: Some(3),
155                write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
156                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
157                items_per_blob: NZU64!(DEFAULT_ITEMS_PER_BLOB),
158                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
159            };
160            let mut cache = Cache::init(context.child("first"), cfg.clone())
161                .await
162                .expect("Failed to initialize cache");
163
164            // Put the data
165            let index = 1u64;
166            let data = 1;
167            cache.put(index, data).await.expect("Failed to put data");
168
169            // Sync and drop the cache
170            cache.sync().await.expect("Failed to sync cache");
171            drop(cache);
172
173            // Initialize the cache again without compression
174            let cfg = Config {
175                partition: "test-partition".into(),
176                codec_config: (),
177                compression: None,
178                write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
179                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
180                items_per_blob: NZU64!(DEFAULT_ITEMS_PER_BLOB),
181                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
182            };
183            let result = Cache::<_, i32>::init(context.child("second"), cfg.clone()).await;
184            assert!(matches!(
185                result,
186                Err(Error::Journal(JournalError::Codec(_)))
187            ));
188        });
189    }
190
191    #[test_traced]
192    fn test_cache_prune() {
193        // Initialize the deterministic context
194        let executor = deterministic::Runner::default();
195        executor.start(|context| async move {
196            // Initialize the cache
197            let cfg = Config {
198                partition: "test-partition".into(),
199                codec_config: (),
200                compression: None,
201                write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
202                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
203                items_per_blob: NZU64!(1), // no mask - each item is its own section
204                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
205            };
206            let mut cache = Cache::init(context.child("storage"), cfg.clone())
207                .await
208                .expect("Failed to initialize cache");
209
210            // Insert multiple items across different sections
211            let items = vec![(1u64, 1), (2u64, 2), (3u64, 3), (4u64, 4), (5u64, 5)];
212            for (index, data) in &items {
213                cache.put(*index, *data).await.expect("Failed to put data");
214            }
215            assert_eq!(cache.first(), Some(1));
216
217            // Check metrics
218            let buffer = context.encode();
219            assert!(has_metric_value(&buffer, "items_tracked", 5));
220
221            // Prune sections less than 3
222            cache.prune(3).await.expect("Failed to prune");
223
224            // Ensure items 1 and 2 are no longer present
225            for (index, data) in items {
226                let retrieved = cache.get(index).await.expect("Failed to get data");
227                if index < 3 {
228                    assert!(retrieved.is_none());
229                } else {
230                    assert_eq!(retrieved.expect("Data not found"), data);
231                }
232            }
233            assert_eq!(cache.first(), Some(3));
234
235            // Check metrics
236            let buffer = context.encode();
237            assert!(has_metric_value(&buffer, "items_tracked", 3));
238
239            // Try to prune older section
240            cache.prune(2).await.expect("Failed to prune");
241            assert_eq!(cache.first(), Some(3));
242
243            // Try to prune current section again
244            cache.prune(3).await.expect("Failed to prune");
245            assert_eq!(cache.first(), Some(3));
246
247            // Try to put older index
248            let result = cache.put(1, 1).await;
249            assert!(matches!(result, Err(Error::AlreadyPrunedTo(3))));
250        });
251    }
252
253    fn test_cache_restart(num_items: usize) -> String {
254        // Initialize the deterministic context
255        let executor = deterministic::Runner::default();
256        executor.start(|mut context| async move {
257            // Initialize the cache
258            let items_per_blob = 256u64;
259            let cfg = Config {
260                partition: "test-partition".into(),
261                codec_config: (),
262                compression: None,
263                write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
264                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
265                items_per_blob: NZU64!(items_per_blob),
266                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
267            };
268            let mut cache = Cache::init(
269                context.child("init").with_attribute("index", 1),
270                cfg.clone(),
271            )
272            .await
273            .expect("Failed to initialize cache");
274
275            // Insert multiple items
276            let mut items = BTreeMap::new();
277            while items.len() < num_items {
278                let index = items.len() as u64;
279                let mut data = [0u8; 1024];
280                context.fill(&mut data);
281                items.insert(index, data);
282
283                cache.put(index, data).await.expect("Failed to put data");
284            }
285
286            // Ensure all items can be retrieved
287            for (index, data) in &items {
288                let retrieved = cache
289                    .get(*index)
290                    .await
291                    .expect("Failed to get data")
292                    .expect("Data not found");
293                assert_eq!(retrieved, *data);
294            }
295
296            // Check metrics
297            let buffer = context.encode();
298            assert!(has_metric_value(&buffer, "items_tracked", num_items));
299
300            // Sync and drop the cache
301            cache.sync().await.expect("Failed to sync cache");
302            drop(cache);
303
304            // Reinitialize the cache
305            let cfg = Config {
306                partition: "test-partition".into(),
307                codec_config: (),
308                compression: None,
309                write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
310                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
311                items_per_blob: NZU64!(items_per_blob),
312                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
313            };
314            let mut cache = Cache::<_, [u8; 1024]>::init(
315                context.child("init").with_attribute("index", 2),
316                cfg.clone(),
317            )
318            .await
319            .expect("Failed to initialize cache");
320
321            // Ensure all items can be retrieved
322            for (index, data) in &items {
323                let retrieved = cache
324                    .get(*index)
325                    .await
326                    .expect("Failed to get data")
327                    .expect("Data not found");
328                assert_eq!(&retrieved, data);
329            }
330
331            // Prune first half
332            let min = (items.len() / 2) as u64;
333            cache.prune(min).await.expect("Failed to prune");
334
335            // Ensure all items can be retrieved that haven't been pruned
336            let min = (min / items_per_blob) * items_per_blob;
337            let mut removed = 0;
338            for (index, data) in items {
339                if index >= min {
340                    let retrieved = cache
341                        .get(index)
342                        .await
343                        .expect("Failed to get data")
344                        .expect("Data not found");
345                    assert_eq!(retrieved, data);
346                } else {
347                    let retrieved = cache.get(index).await.expect("Failed to get data");
348                    assert!(retrieved.is_none());
349                    removed += 1;
350                }
351            }
352
353            // Check metrics
354            let buffer = context.encode();
355            assert!(has_metric_value(
356                &buffer,
357                "items_tracked",
358                num_items - removed
359            ));
360
361            context.auditor().state()
362        })
363    }
364
365    #[test_group("slow")]
366    #[test_traced]
367    fn test_cache_many_items_and_restart() {
368        test_cache_restart(100_000);
369    }
370
371    #[test_group("slow")]
372    #[test_traced]
373    fn test_determinism() {
374        let state1 = test_cache_restart(5_000);
375        let state2 = test_cache_restart(5_000);
376        assert_eq!(state1, state2);
377    }
378
379    #[test_traced]
380    fn test_cache_next_gap() {
381        let executor = deterministic::Runner::default();
382        executor.start(|context| async move {
383            let cfg = Config {
384                partition: "test-partition".into(),
385                codec_config: (),
386                compression: None,
387                write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
388                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
389                items_per_blob: NZU64!(DEFAULT_ITEMS_PER_BLOB),
390                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
391            };
392            let mut cache = Cache::init(context.child("storage"), cfg.clone())
393                .await
394                .expect("Failed to initialize cache");
395
396            // Check first
397            assert_eq!(cache.first(), None);
398
399            // Insert values with gaps
400            cache.put(1, 1).await.unwrap();
401            cache.put(10, 10).await.unwrap();
402            cache.put(11, 11).await.unwrap();
403            cache.put(14, 14).await.unwrap();
404
405            // Check gaps
406            let (current_end, start_next) = cache.next_gap(0);
407            assert!(current_end.is_none());
408            assert_eq!(start_next, Some(1));
409            assert_eq!(cache.first(), Some(1));
410
411            let (current_end, start_next) = cache.next_gap(1);
412            assert_eq!(current_end, Some(1));
413            assert_eq!(start_next, Some(10));
414
415            let (current_end, start_next) = cache.next_gap(10);
416            assert_eq!(current_end, Some(11));
417            assert_eq!(start_next, Some(14));
418
419            let (current_end, start_next) = cache.next_gap(11);
420            assert_eq!(current_end, Some(11));
421            assert_eq!(start_next, Some(14));
422
423            let (current_end, start_next) = cache.next_gap(12);
424            assert!(current_end.is_none());
425            assert_eq!(start_next, Some(14));
426
427            let (current_end, start_next) = cache.next_gap(14);
428            assert_eq!(current_end, Some(14));
429            assert!(start_next.is_none());
430        });
431    }
432
433    #[test_traced]
434    fn test_cache_missing_items() {
435        let executor = deterministic::Runner::default();
436        executor.start(|context| async move {
437            let cfg = Config {
438                partition: "test-partition".into(),
439                codec_config: (),
440                compression: None,
441                write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
442                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
443                items_per_blob: NZU64!(DEFAULT_ITEMS_PER_BLOB),
444                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
445            };
446            let mut cache = Cache::init(context.child("storage"), cfg.clone())
447                .await
448                .expect("Failed to initialize cache");
449
450            // Test 1: Empty cache - should return no items
451            assert_eq!(cache.first(), None);
452            assert_eq!(cache.missing_items(0, 5), Vec::<u64>::new());
453            assert_eq!(cache.missing_items(100, 10), Vec::<u64>::new());
454
455            // Test 2: Insert values with gaps
456            cache.put(1, 1).await.unwrap();
457            cache.put(2, 2).await.unwrap();
458            cache.put(5, 5).await.unwrap();
459            cache.put(6, 6).await.unwrap();
460            cache.put(10, 10).await.unwrap();
461
462            // Test 3: Find missing items from the beginning
463            assert_eq!(cache.missing_items(0, 5), vec![0, 3, 4, 7, 8]);
464            assert_eq!(cache.missing_items(0, 6), vec![0, 3, 4, 7, 8, 9]);
465            assert_eq!(cache.missing_items(0, 7), vec![0, 3, 4, 7, 8, 9]);
466
467            // Test 4: Find missing items from within a gap
468            assert_eq!(cache.missing_items(3, 3), vec![3, 4, 7]);
469            assert_eq!(cache.missing_items(4, 2), vec![4, 7]);
470
471            // Test 5: Find missing items from within a range
472            assert_eq!(cache.missing_items(1, 3), vec![3, 4, 7]);
473            assert_eq!(cache.missing_items(2, 4), vec![3, 4, 7, 8]);
474            assert_eq!(cache.missing_items(5, 2), vec![7, 8]);
475
476            // Test 6: Find missing items after the last range (no more gaps)
477            assert_eq!(cache.missing_items(11, 5), Vec::<u64>::new());
478            assert_eq!(cache.missing_items(100, 10), Vec::<u64>::new());
479
480            // Test 7: Large gap scenario
481            cache.put(1000, 1000).await.unwrap();
482
483            // Gap between 10 and 1000
484            let items = cache.missing_items(11, 10);
485            assert_eq!(items, vec![11, 12, 13, 14, 15, 16, 17, 18, 19, 20]);
486
487            // Request more items than available in gap
488            let items = cache.missing_items(990, 15);
489            assert_eq!(
490                items,
491                vec![990, 991, 992, 993, 994, 995, 996, 997, 998, 999]
492            );
493
494            // Test 8: After syncing (data should remain consistent)
495            cache.sync().await.unwrap();
496            assert_eq!(cache.missing_items(0, 5), vec![0, 3, 4, 7, 8]);
497            assert_eq!(cache.missing_items(3, 3), vec![3, 4, 7]);
498
499            // Test 9: Cross-section boundary scenario
500            cache.put(DEFAULT_ITEMS_PER_BLOB - 1, 99).await.unwrap();
501            cache.put(DEFAULT_ITEMS_PER_BLOB + 1, 101).await.unwrap();
502
503            // Find missing items across section boundary
504            let items = cache.missing_items(DEFAULT_ITEMS_PER_BLOB - 2, 5);
505            assert_eq!(
506                items,
507                vec![DEFAULT_ITEMS_PER_BLOB - 2, DEFAULT_ITEMS_PER_BLOB]
508            );
509        });
510    }
511
512    #[test_traced]
513    fn test_cache_intervals_after_restart() {
514        let executor = deterministic::Runner::default();
515        executor.start(|context| async move {
516            let cfg = Config {
517                partition: "test-partition".into(),
518                codec_config: (),
519                compression: None,
520                write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
521                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
522                items_per_blob: NZU64!(DEFAULT_ITEMS_PER_BLOB),
523                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
524            };
525
526            // Insert data and sync
527            {
528                let mut cache = Cache::init(context.child("first"), cfg.clone())
529                    .await
530                    .expect("Failed to initialize cache");
531
532                cache.put(0, 0).await.expect("Failed to put data");
533                cache.put(100, 100).await.expect("Failed to put data");
534                cache.put(1000, 1000).await.expect("Failed to put data");
535
536                cache.sync().await.expect("Failed to sync cache");
537            }
538
539            // Reopen and verify intervals are preserved
540            {
541                let cache = Cache::<_, i32>::init(context.child("second"), cfg.clone())
542                    .await
543                    .expect("Failed to initialize cache");
544
545                // Check gaps are preserved
546                let (current_end, start_next) = cache.next_gap(0);
547                assert_eq!(current_end, Some(0));
548                assert_eq!(start_next, Some(100));
549
550                let (current_end, start_next) = cache.next_gap(100);
551                assert_eq!(current_end, Some(100));
552                assert_eq!(start_next, Some(1000));
553
554                // Check missing items
555                let items = cache.missing_items(1, 5);
556                assert_eq!(items, vec![1, 2, 3, 4, 5]);
557            }
558        });
559    }
560
561    #[test_traced]
562    fn test_cache_intervals_with_pruning() {
563        let executor = deterministic::Runner::default();
564        executor.start(|context| async move {
565            let cfg = Config {
566                partition: "test-partition".into(),
567                codec_config: (),
568                compression: None,
569                write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
570                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
571                items_per_blob: NZU64!(100), // Smaller sections for easier testing
572                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
573            };
574            let mut cache = Cache::init(context.child("storage"), cfg.clone())
575                .await
576                .expect("Failed to initialize cache");
577
578            // Insert values across multiple sections
579            cache.put(50, 50).await.unwrap();
580            cache.put(150, 150).await.unwrap();
581            cache.put(250, 250).await.unwrap();
582            cache.put(350, 350).await.unwrap();
583
584            // Check gaps before pruning
585            let (current_end, start_next) = cache.next_gap(0);
586            assert!(current_end.is_none());
587            assert_eq!(start_next, Some(50));
588
589            // Prune sections less than 200
590            cache.prune(200).await.expect("Failed to prune");
591
592            // Check that pruned indices are not accessible
593            assert!(!cache.has(50));
594            assert!(!cache.has(150));
595
596            // Check gaps after pruning - should not include pruned ranges
597            let (current_end, start_next) = cache.next_gap(200);
598            assert!(current_end.is_none());
599            assert_eq!(start_next, Some(250));
600
601            // Missing items should not include pruned ranges
602            let items = cache.missing_items(200, 5);
603            assert_eq!(items, vec![200, 201, 202, 203, 204]);
604
605            // Verify remaining data is still accessible
606            assert!(cache.has(250));
607            assert!(cache.has(350));
608            assert_eq!(cache.get(250).await.unwrap(), Some(250));
609            assert_eq!(cache.get(350).await.unwrap(), Some(350));
610        });
611    }
612
613    #[test_traced]
614    fn test_cache_sparse_indices() {
615        let executor = deterministic::Runner::default();
616        executor.start(|context| async move {
617            let cfg = Config {
618                partition: "test-partition".into(),
619                codec_config: (),
620                compression: None,
621                write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
622                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
623                items_per_blob: NZU64!(100), // Smaller sections for testing
624                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
625            };
626            let mut cache = Cache::init(context.child("storage"), cfg.clone())
627                .await
628                .expect("Failed to initialize cache");
629
630            // Insert sparse values
631            let indices = vec![
632                (0u64, 0),
633                (99u64, 99),   // End of first section
634                (100u64, 100), // Start of second section
635                (500u64, 500), // Start of sixth section
636            ];
637
638            for (index, value) in &indices {
639                cache.put(*index, *value).await.expect("Failed to put data");
640            }
641
642            // Check that intermediate indices don't exist
643            assert!(!cache.has(1));
644            assert!(!cache.has(50));
645            assert!(!cache.has(101));
646            assert!(!cache.has(499));
647
648            // Verify gap detection works correctly
649            let (current_end, start_next) = cache.next_gap(50);
650            assert!(current_end.is_none());
651            assert_eq!(start_next, Some(99));
652
653            let (current_end, start_next) = cache.next_gap(99);
654            assert_eq!(current_end, Some(100));
655            assert_eq!(start_next, Some(500));
656
657            // Sync and verify
658            cache.sync().await.expect("Failed to sync");
659
660            for (index, value) in &indices {
661                let retrieved = cache
662                    .get(*index)
663                    .await
664                    .expect("Failed to get data")
665                    .expect("Data not found");
666                assert_eq!(retrieved, *value);
667            }
668        });
669    }
670
671    #[test_traced]
672    fn test_cache_intervals_edge_cases() {
673        let executor = deterministic::Runner::default();
674        executor.start(|context| async move {
675            let cfg = Config {
676                partition: "test-partition".into(),
677                codec_config: (),
678                compression: None,
679                write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
680                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
681                items_per_blob: NZU64!(DEFAULT_ITEMS_PER_BLOB),
682                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
683            };
684            let mut cache = Cache::init(context.child("storage"), cfg.clone())
685                .await
686                .expect("Failed to initialize cache");
687
688            // Test edge case: single item
689            cache.put(42, 42).await.unwrap();
690
691            let (current_end, start_next) = cache.next_gap(42);
692            assert_eq!(current_end, Some(42));
693            assert!(start_next.is_none());
694
695            let (current_end, start_next) = cache.next_gap(41);
696            assert!(current_end.is_none());
697            assert_eq!(start_next, Some(42));
698
699            let (current_end, start_next) = cache.next_gap(43);
700            assert!(current_end.is_none());
701            assert!(start_next.is_none());
702
703            // Test edge case: consecutive items
704            cache.put(43, 43).await.unwrap();
705            cache.put(44, 44).await.unwrap();
706
707            let (current_end, start_next) = cache.next_gap(42);
708            assert_eq!(current_end, Some(44));
709            assert!(start_next.is_none());
710
711            // Test edge case: boundary values
712            cache.put(u64::MAX - 1, 999).await.unwrap();
713
714            let (current_end, start_next) = cache.next_gap(u64::MAX - 2);
715            assert!(current_end.is_none());
716            assert_eq!(start_next, Some(u64::MAX - 1));
717
718            let (current_end, start_next) = cache.next_gap(u64::MAX - 1);
719            assert_eq!(current_end, Some(u64::MAX - 1));
720            assert!(start_next.is_none());
721        });
722    }
723
724    #[test_traced]
725    fn test_cache_intervals_duplicate_inserts() {
726        let executor = deterministic::Runner::default();
727        executor.start(|context| async move {
728            let cfg = Config {
729                partition: "test-partition".into(),
730                codec_config: (),
731                compression: None,
732                write_buffer: NZUsize!(DEFAULT_WRITE_BUFFER),
733                replay_buffer: NZUsize!(DEFAULT_REPLAY_BUFFER),
734                items_per_blob: NZU64!(DEFAULT_ITEMS_PER_BLOB),
735                page_cache: CacheRef::from_pooler(&context, PAGE_SIZE, PAGE_CACHE_SIZE),
736            };
737            let mut cache = Cache::init(context.child("storage"), cfg.clone())
738                .await
739                .expect("Failed to initialize cache");
740
741            // Insert initial value
742            cache.put(10, 10).await.unwrap();
743            assert!(cache.has(10));
744            assert_eq!(cache.get(10).await.unwrap(), Some(10));
745
746            // Try to insert duplicate - should be no-op
747            cache.put(10, 20).await.unwrap();
748            assert!(cache.has(10));
749            assert_eq!(cache.get(10).await.unwrap(), Some(10)); // Should still be original value
750
751            // Verify intervals are correct
752            let (current_end, start_next) = cache.next_gap(10);
753            assert_eq!(current_end, Some(10));
754            assert!(start_next.is_none());
755
756            // Insert adjacent values
757            cache.put(9, 9).await.unwrap();
758            cache.put(11, 11).await.unwrap();
759
760            // Verify intervals updated correctly
761            let (current_end, start_next) = cache.next_gap(9);
762            assert_eq!(current_end, Some(11));
763            assert!(start_next.is_none());
764        });
765    }
766}
commonware_storage/cache/mod.rs

commonware_storage/cache/
mod.rs
