reifydb_cdc/storage/
cached.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::collections::Bound;
5
6use reifydb_core::{
7	common::CommitVersion,
8	interface::cdc::{Cdc, CdcBatch},
9};
10use reifydb_value::value::datetime::DateTime;
11
12use super::{CdcStorage, CdcStorageResult, DropBeforeResult, normalize_range_inclusive, recent_cache::RecentCdcCache};
13
14#[derive(Clone)]
15pub struct CachedCdcStorage<S: CdcStorage> {
16	inner: S,
17	cache: RecentCdcCache,
18}
19
20impl<S: CdcStorage> CachedCdcStorage<S> {
21	pub fn new(inner: S, capacity: usize) -> Self {
22		Self {
23			inner,
24			cache: RecentCdcCache::new(capacity),
25		}
26	}
27
28	pub fn inner(&self) -> &S {
29		&self.inner
30	}
31}
32
33impl<S: CdcStorage> CdcStorage for CachedCdcStorage<S> {
34	fn write(&self, cdc: &Cdc) -> CdcStorageResult<()> {
35		self.inner.write(cdc)?;
36		self.cache.insert(cdc);
37		Ok(())
38	}
39
40	fn read(&self, version: CommitVersion) -> CdcStorageResult<Option<Cdc>> {
41		if let Some(cdc) = self.cache.get(version) {
42			return Ok(Some((*cdc).clone()));
43		}
44		self.inner.read(version)
45	}
46
47	fn read_range(
48		&self,
49		start: Bound<CommitVersion>,
50		end: Bound<CommitVersion>,
51		batch_size: u64,
52	) -> CdcStorageResult<CdcBatch> {
53		if let Some((lo_inc, hi_inc)) = normalize_range_inclusive(start, end)
54			&& let Some((items, has_more)) = self.cache.try_serve_range(lo_inc, hi_inc, batch_size as usize)
55		{
56			return Ok(CdcBatch {
57				items,
58				has_more,
59			});
60		}
61		self.inner.read_range(start, end, batch_size)
62	}
63
64	fn count(&self, version: CommitVersion) -> CdcStorageResult<usize> {
65		self.inner.count(version)
66	}
67
68	fn min_version(&self) -> CdcStorageResult<Option<CommitVersion>> {
69		self.inner.min_version()
70	}
71
72	fn max_version(&self) -> CdcStorageResult<Option<CommitVersion>> {
73		self.inner.max_version()
74	}
75
76	fn drop_before(&self, version: CommitVersion, limit: usize) -> CdcStorageResult<DropBeforeResult> {
77		self.inner.drop_before(version, limit)
78	}
79
80	fn vacuum(&self) -> CdcStorageResult<()> {
81		self.inner.vacuum()
82	}
83
84	fn find_ttl_cutoff(&self, cutoff: DateTime) -> CdcStorageResult<Option<CommitVersion>> {
85		self.inner.find_ttl_cutoff(cutoff)
86	}
87}
88
89#[cfg(test)]
90mod tests {
91	use std::collections::Bound;
92
93	use reifydb_core::{common::CommitVersion, interface::cdc::Cdc};
94	use reifydb_value::value::datetime::DateTime;
95
96	use super::*;
97	use crate::storage::memory::MemoryCdcStorage;
98
99	fn cv(n: u64) -> CommitVersion {
100		CommitVersion(n)
101	}
102
103	fn cdc(version: u64) -> Cdc {
104		Cdc::new(cv(version), DateTime::default(), Vec::new(), Vec::new())
105	}
106
107	#[test]
108	fn write_is_persisted_to_inner_and_served_from_cache() {
109		let cached = CachedCdcStorage::new(MemoryCdcStorage::new(), 16);
110		cached.write(&cdc(1)).unwrap();
111		// inner has it durably
112		assert!(cached.inner().read(cv(1)).unwrap().is_some());
113		// and the cache serves the read
114		assert_eq!(cached.read(cv(1)).unwrap().unwrap().version, cv(1));
115	}
116
117	#[test]
118	fn read_range_served_from_cache_when_covered() {
119		let cached = CachedCdcStorage::new(MemoryCdcStorage::new(), 16);
120		for v in 1..=5 {
121			cached.write(&cdc(v)).unwrap();
122		}
123		let batch = cached.read_range(Bound::Excluded(cv(1)), Bound::Included(cv(4)), 100).unwrap();
124		assert_eq!(batch.items.iter().map(|c| c.version).collect::<Vec<_>>(), vec![cv(2), cv(3), cv(4)]);
125		assert!(!batch.has_more);
126	}
127
128	#[test]
129	fn read_range_falls_back_to_inner_when_below_cache_window() {
130		// Capacity 2 keeps only versions {4,5}; a request starting at 1 is not covered, so the
131		// decorator must fall through to the backend, which still has the full history.
132		let inner = MemoryCdcStorage::new();
133		let cached = CachedCdcStorage::new(inner, 2);
134		for v in 1..=5 {
135			cached.write(&cdc(v)).unwrap();
136		}
137		let batch = cached.read_range(Bound::Included(cv(1)), Bound::Included(cv(5)), 100).unwrap();
138		assert_eq!(batch.items.len(), 5, "fallback must serve the full range from the backend");
139	}
140}
reifydb_cdc/storage/cached.rs

reifydb_cdc/storage/
cached.rs
