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

commonware_storage/cache/
mod.rs
