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

commonware_storage/cache/
mod.rs
