#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
use std::collections::{HashMap, HashSet};
use std::sync::{Arc, OnceLock};
#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
use reifydb_codec::key::encoded::EncodedKey;
#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
use reifydb_core::event::metric::{MultiEviction, MultiPersist, MultiSweptEvent};
#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
use reifydb_core::lifecycle::progress::Progress;
#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
use reifydb_core::{
common::CommitVersion,
default,
interface::store::{EntryKind, storage_key},
};
use reifydb_core::{event::EventBus, lifecycle::watermark::EvictionWatermark};
use reifydb_runtime::{
context::clock::Clock,
sync::{mutex::Mutex, rwlock::RwLock},
};
#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
use reifydb_store_commit::TierBatch;
use reifydb_store_commit::store::CommitStore;
#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
use reifydb_store_commit::store::EvictedVersion;
#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
use reifydb_value::byte_size::ByteSize;
#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
use reifydb_value::{reifydb_assertions, util::cowvec::CowVec};
#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
use tracing::{debug, error, instrument, warn};
#[cfg(all(test, feature = "sqlite", not(target_arch = "wasm32")))]
use crate::tier::TierStorage;
use crate::{
flush::ObjectPersistence,
tier::{persistent::MultiPersistentTier, point::MultiPointTier, range::MultiRangeTier},
};
#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
pub const FLUSH_BYTE_BUDGET: ByteSize = if default::TESTING {
default::store::MULTI_FLUSH_BUDGET_TESTING
} else {
default::store::MULTI_FLUSH_BUDGET
};
#[derive(Default)]
pub struct FlushEngineState {
#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
resume_from: Option<EntryKind>,
}
#[allow(dead_code)]
pub struct FlushEngine {
commit: CommitStore,
persistent: MultiPersistentTier,
persistence: Arc<OnceLock<Arc<dyn ObjectPersistence>>>,
eviction_watermark: Arc<RwLock<Option<Arc<dyn EvictionWatermark>>>>,
point: Option<MultiPointTier>,
range: Option<MultiRangeTier>,
clock: Clock,
event_bus: EventBus,
sweep_lock: Mutex<FlushEngineState>,
}
#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
type EvictablePersist = Vec<(EncodedKey, CommitVersion, Option<CowVec<u8>>)>;
#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
type EvictableDrop = Vec<EvictedVersion>;
#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
type EvictablePartition = (EvictablePersist, EvictableDrop);
#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
pub struct SweepOutcome {
pub progress: Progress,
pub reclaimed: u64,
pub backlog: u64,
}
#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
impl FlushEngine {
#[allow(clippy::too_many_arguments)]
pub fn new(
commit: CommitStore,
persistent: MultiPersistentTier,
persistence: Arc<OnceLock<Arc<dyn ObjectPersistence>>>,
eviction_watermark: Arc<RwLock<Option<Arc<dyn EvictionWatermark>>>>,
clock: Clock,
event_bus: EventBus,
) -> Self {
Self {
commit,
persistent,
persistence,
eviction_watermark,
point: None,
range: None,
clock,
event_bus,
sweep_lock: Mutex::new(FlushEngineState::default()),
}
}
pub fn with_point(mut self, point: Option<MultiPointTier>) -> Self {
self.point = point;
self
}
pub fn with_range(mut self, range: Option<MultiRangeTier>) -> Self {
self.range = range;
self
}
pub fn sweep_slice(&self, budget: ByteSize) -> SweepOutcome {
let mut state = self.sweep_lock.lock();
let (progress, reclaimed) = match self.eviction_cutoff() {
Some(cutoff) => self.sweep_once(&mut state, cutoff, budget),
None => (Progress::Exhausted, 0),
};
SweepOutcome {
progress,
reclaimed,
backlog: self.buffered_entries(),
}
}
fn buffered_entries(&self) -> u64 {
self.commit
.list_all_entry_kinds()
.map(|kinds| {
kinds.iter().map(|kind| self.commit.estimated_current_count(*kind).unwrap_or(0)).sum()
})
.unwrap_or(0)
}
pub fn flush_pending(&self) {
let mut guard = self.sweep_lock.lock();
if let Some(cutoff) = self.eviction_cutoff() {
while self.sweep_once(&mut guard, cutoff, FLUSH_BYTE_BUDGET).0.is_yielded() {}
}
}
pub fn flush_all(&self) {
let mut guard = self.sweep_lock.lock();
while self.sweep_once(&mut guard, CommitVersion(u64::MAX), FLUSH_BYTE_BUDGET).0.is_yielded() {}
}
fn eviction_cutoff(&self) -> Option<CommitVersion> {
let cutoff = self.eviction_watermark.read().as_ref()?.watermark();
if cutoff.0 == 0 {
return None;
}
Some(cutoff)
}
fn is_persistent_object(&self, kind: EntryKind) -> bool {
match kind {
EntryKind::Source(storage, _) | EntryKind::PartitionedSource(storage, _) => {
self.persistence.get().map(|provider| provider.is_persistent(storage)).unwrap_or(true)
}
EntryKind::Multi => true,
}
}
#[cfg(test)]
fn sweep(&self, cutoff: CommitVersion) {
let mut guard = self.sweep_lock.lock();
while self.sweep_once(&mut guard, cutoff, FLUSH_BYTE_BUDGET).0.is_yielded() {}
}
#[instrument(name = "store::multi::flush::sweep_once", level = "debug", skip_all)]
fn sweep_once(&self, state: &mut FlushEngineState, cutoff: CommitVersion, budget: ByteSize) -> (Progress, u64) {
let Some(mut entry_kinds) = self.list_evictable_kinds() else {
return (Progress::Exhausted, 0);
};
if let Some(resume) = state.resume_from
&& let Some(position) = entry_kinds.iter().position(|kind| *kind == resume)
{
entry_kinds.rotate_left(position);
}
state.resume_from = None;
let mut remaining = budget;
let mut more = false;
let mut plan: Vec<(EntryKind, bool, EvictablePartition)> = Vec::new();
let mut batches: HashMap<CommitVersion, TierBatch> = HashMap::new();
for kind in entry_kinds {
if remaining == ByteSize::ZERO {
more = true;
state.resume_from = Some(kind);
break;
}
let (to_persist, to_drop, consumed, kind_more) =
self.collect_evictable(kind, cutoff, remaining);
if to_persist.is_empty() && to_drop.is_empty() {
continue;
}
remaining = remaining.saturating_sub(consumed);
more |= kind_more;
let persistent_object = self.is_persistent_object(kind);
if persistent_object {
for (key, version, value) in &to_persist {
batches.entry(*version)
.or_default()
.entry(kind)
.or_default()
.push((key.clone(), value.clone()));
}
}
plan.push((kind, persistent_object, (to_persist, to_drop)));
}
if plan.is_empty() {
return (Progress::Exhausted, 0);
}
let accepted = if batches.values().any(|batch| !batch.is_empty()) {
match self.persistent.persist_sweep(batches.into_iter().collect()) {
Ok(accepted) => accepted,
Err(e) => {
error!(error = %e, "flush sweep: persist failed, aborting slice");
return (Progress::Exhausted, 0);
}
}
} else {
Vec::new()
};
let persisted = accepted.len();
let accepted_keys: HashSet<&[u8]> = accepted.iter().map(|k| k.as_slice()).collect();
let mut evictions: Vec<MultiEviction> = Vec::new();
let mut persists: Vec<MultiPersist> = Vec::new();
let mut dropped = 0usize;
for (kind, persistent_object, (to_persist, to_drop)) in plan {
self.refresh_read_tier(kind, persistent_object, &to_persist, &to_drop, &accepted_keys);
if persistent_object {
for (key, _, value) in &to_persist {
if accepted_keys.contains(key.as_slice()) {
persists.push(MultiPersist {
key: key.clone(),
value_bytes: ByteSize::from_bytes(
value.as_ref().map(|v| v.len() as u64).unwrap_or(0),
),
});
}
}
}
for evicted in &to_drop {
evictions.push(MultiEviction {
key: evicted.key.clone(),
value_bytes: evicted.value_bytes,
current: evicted.current,
});
}
if let Some(count) = self.drop_from_commit(kind, to_drop) {
dropped += count;
}
}
if !evictions.is_empty() || !persists.is_empty() {
self.event_bus.emit(MultiSweptEvent::new(evictions, persists, cutoff));
}
if persisted > 0 || dropped > 0 {
debug!(cutoff = cutoff.0, persisted, dropped, more, "flush sweep slice completed");
}
let progress = if more {
Progress::Yielded
} else {
Progress::Exhausted
};
(progress, dropped as u64)
}
#[inline]
fn list_evictable_kinds(&self) -> Option<Vec<EntryKind>> {
match self.commit.list_entry_kinds_by_oldest_pending() {
Ok(v) => Some(v),
Err(e) => {
warn!(error = %e, "flush sweep: list_entry_kinds_by_oldest_pending failed");
None
}
}
}
#[inline]
fn collect_evictable(
&self,
kind: EntryKind,
cutoff: CommitVersion,
budget: ByteSize,
) -> (EvictablePersist, EvictableDrop, ByteSize, bool) {
self.commit.collect_evictable_below(kind, cutoff, budget)
}
#[inline]
#[instrument(name = "store::multi::flush::refresh_read_tier", level = "debug", skip_all, fields(persist_count = to_persist.len(), drop_count = to_drop.len()))]
fn refresh_read_tier(
&self,
table: EntryKind,
persistent_object: bool,
to_persist: &[(EncodedKey, CommitVersion, Option<CowVec<u8>>)],
to_drop: &[EvictedVersion],
accepted: &HashSet<&[u8]>,
) {
if self.point.is_none() && self.range.is_none() {
return;
}
if persistent_object {
for (key, version, value) in to_persist {
if accepted.contains(key.as_slice()) {
if let Some(range) = &self.range {
range.insert(table, key.clone(), *version, value.clone());
}
if let Some(point) = &self.point {
point.insert(
table,
storage_key(key).1,
key.clone(),
*version,
value.clone(),
);
}
} else {
if let Some(range) = &self.range {
range.invalidate(table, key);
}
if let Some(point) = &self.point {
point.invalidate(table, storage_key(key).1, key);
}
}
}
} else {
for evicted in to_drop {
if let Some(range) = &self.range {
range.invalidate(table, &evicted.key);
}
if let Some(point) = &self.point {
point.invalidate(table, storage_key(&evicted.key).1, &evicted.key);
}
}
}
}
#[inline]
fn drop_from_commit(&self, kind: EntryKind, to_drop: EvictableDrop) -> Option<usize> {
let drop_count = to_drop.len();
reifydb_assertions! {
assert!(
drop_count > 0,
"sweep must only reach drop_from_commit with a non-empty drop set; an empty drop \
issues a no-op commit-buffer drop and lets the dropped counter run for zero work \
(kind={kind:?})"
);
}
let mut batches: HashMap<EntryKind, Vec<(EncodedKey, CommitVersion)>> = HashMap::new();
batches.insert(kind, to_drop.into_iter().map(|e| (e.key, e.version)).collect());
if let Err(e) = self.commit.compact(batches) {
warn!(?kind, error = %e, "flush sweep: commit buffer drop failed");
return None;
}
Some(drop_count)
}
}
#[cfg(all(test, feature = "sqlite", not(target_arch = "wasm32")))]
mod tests {
use std::{
collections::hash_map::DefaultHasher,
hash::{Hash, Hasher},
};
use reifydb_core::{
event::EventListener,
interface::{
catalog::{id::TableId, storage::StorageId},
store::EntryLayout,
},
key::row::RowKey,
};
use reifydb_runtime::{actor::system::ActorSystem, shutdown::Shutdown};
use reifydb_sqlite::SqliteTempPathGuard;
use reifydb_store_commit::VersionedGetResult;
use reifydb_value::{util::cowvec::CowVec, value::row_number::RowNumber};
use super::*;
use crate::tier::point::MultiPointConfig;
fn ek(s: &str) -> EncodedKey {
let mut hasher = DefaultHasher::new();
s.hash(&mut hasher);
RowKey::encoded(StorageId::table(TableId(1)), RowNumber(hasher.finish()))
}
fn val(s: &str) -> CowVec<u8> {
CowVec::new(s.as_bytes().to_vec())
}
fn write(buffer: &CommitStore, kind: EntryKind, key: &EncodedKey, version: u64, value: &str) {
buffer.set(CommitVersion(version), HashMap::from([(kind, vec![(key.clone(), Some(val(value)))])]))
.unwrap();
}
fn budget_for(keys: &[String], value: &str) -> ByteSize {
let storage = CommitStore::new();
for key in keys {
storage.set(
CommitVersion(1),
HashMap::from([(EntryKind::Multi, vec![(ek(key), Some(val(value)))])]),
)
.unwrap();
}
let (_, _, consumed, _) = storage.collect_evictable_below(
EntryKind::Multi,
CommitVersion(1),
ByteSize::from_bytes(u64::MAX),
);
consumed
}
struct StaticWatermark(CommitVersion);
impl EvictionWatermark for StaticWatermark {
fn watermark(&self) -> CommitVersion {
self.0
}
}
struct AllPersistent;
impl ObjectPersistence for AllPersistent {
fn is_persistent(&self, _storage: StorageId) -> bool {
true
}
}
struct NonePersistent;
impl ObjectPersistence for NonePersistent {
fn is_persistent(&self, _storage: StorageId) -> bool {
false
}
}
fn build_engine(
persistence: Arc<dyn ObjectPersistence>,
watermark: Option<CommitVersion>,
) -> (FlushEngine, SqliteTempPathGuard) {
let buffer = CommitStore::new();
let (persistent, guard) = MultiPersistentTier::sqlite_in_memory();
let persistence_lock: Arc<OnceLock<Arc<dyn ObjectPersistence>>> = Arc::new(OnceLock::new());
let _ = persistence_lock.set(persistence);
let watermark_lock: Arc<RwLock<Option<Arc<dyn EvictionWatermark>>>> = Arc::new(RwLock::new(None));
if let Some(w) = watermark {
*watermark_lock.write() = Some(Arc::new(StaticWatermark(w)));
}
(
FlushEngine::new(
buffer,
persistent,
persistence_lock,
watermark_lock,
Clock::Real,
testing_event_bus(),
),
guard,
)
}
fn testing_event_bus() -> EventBus {
EventBus::new(&ActorSystem::testing(Clock::testing()).spawner())
}
#[derive(Clone, Default)]
struct SweepCollector {
events: Arc<Mutex<Vec<MultiSweptEvent>>>,
}
impl EventListener<MultiSweptEvent> for SweepCollector {
fn on(&self, event: &MultiSweptEvent) {
self.events.lock().push(event.clone());
}
}
fn build_engine_watching_sweeps(
persistence: Arc<dyn ObjectPersistence>,
watermark: CommitVersion,
) -> (FlushEngine, SqliteTempPathGuard, SweepCollector) {
let buffer = CommitStore::new();
let (persistent, guard) = MultiPersistentTier::sqlite_in_memory();
let persistence_lock: Arc<OnceLock<Arc<dyn ObjectPersistence>>> = Arc::new(OnceLock::new());
let _ = persistence_lock.set(persistence);
let watermark_lock: Arc<RwLock<Option<Arc<dyn EvictionWatermark>>>> = Arc::new(RwLock::new(None));
*watermark_lock.write() = Some(Arc::new(StaticWatermark(watermark)));
let event_bus = testing_event_bus();
let collector = SweepCollector::default();
event_bus.register::<MultiSweptEvent, _>(collector.clone());
(
FlushEngine::new(buffer, persistent, persistence_lock, watermark_lock, Clock::Real, event_bus),
guard,
collector,
)
}
#[test]
fn a_sweep_reports_every_version_it_evicted_from_the_commit_buffer() {
let (engine, _guard, collector) =
build_engine_watching_sweeps(Arc::new(AllPersistent), CommitVersion(2));
let kind = EntryKind::Source(StorageId::table(TableId(1)), EntryLayout::Row);
let key = ek("k");
write(&engine.commit, kind, &key, 1, "v1");
write(&engine.commit, kind, &key, 2, "v2");
engine.sweep(CommitVersion(2));
engine.event_bus.wait_for_completion();
let events = collector.events.lock().clone();
assert_eq!(events.len(), 1, "one sweep slice must report exactly once");
let evictions = events[0].evictions();
assert_eq!(evictions.len(), 2, "both versions left the buffer and both must be accounted");
let current: Vec<&MultiEviction> = evictions.iter().filter(|e| e.current).collect();
assert_eq!(current.len(), 1, "exactly one of the two was the live version");
assert_eq!(
current[0].value_bytes,
ByteSize::from_bytes(2),
"the evicted bytes must be the value's own, not a placeholder"
);
let superseded: Vec<&MultiEviction> = evictions.iter().filter(|e| !e.current).collect();
assert_eq!(superseded.len(), 1, "v1 was superseded by v2 and is discarded, not persisted");
let persists = events[0].persists();
assert_eq!(persists.len(), 1, "only the latest version below the cutoff reaches the persistent tier");
assert_eq!(persists[0].value_bytes, ByteSize::from_bytes(2));
}
#[test]
fn a_sweep_that_persists_nothing_still_reports_what_it_discarded() {
let (engine, _guard, collector) =
build_engine_watching_sweeps(Arc::new(NonePersistent), CommitVersion(2));
let kind = EntryKind::Source(StorageId::table(TableId(1)), EntryLayout::Row);
let key = ek("k");
write(&engine.commit, kind, &key, 1, "v1");
write(&engine.commit, kind, &key, 2, "v2");
engine.sweep(CommitVersion(2));
engine.event_bus.wait_for_completion();
let events = collector.events.lock().clone();
assert_eq!(events.len(), 1);
assert_eq!(events[0].evictions().len(), 2, "the discarded versions are still reported");
assert!(events[0].persists().is_empty(), "a non-persistent object persists nothing");
}
fn build_engine_with_point(
persistence: Arc<dyn ObjectPersistence>,
watermark: CommitVersion,
point: MultiPointTier,
) -> (FlushEngine, SqliteTempPathGuard) {
let buffer = CommitStore::new();
let (persistent, guard) = MultiPersistentTier::sqlite_in_memory();
let persistence_lock: Arc<OnceLock<Arc<dyn ObjectPersistence>>> = Arc::new(OnceLock::new());
let _ = persistence_lock.set(persistence);
let watermark_lock: Arc<RwLock<Option<Arc<dyn EvictionWatermark>>>> = Arc::new(RwLock::new(None));
*watermark_lock.write() = Some(Arc::new(StaticWatermark(watermark)));
(
FlushEngine::new(
buffer,
persistent,
persistence_lock,
watermark_lock,
Clock::Real,
testing_event_bus(),
)
.with_point(Some(point)),
guard,
)
}
#[test]
fn eviction_cutoff_is_none_without_watermark() {
let (actor, _guard) = build_engine(Arc::new(AllPersistent), None);
assert!(actor.eviction_cutoff().is_none(), "no watermark set => no eviction");
}
#[test]
fn eviction_cutoff_is_none_at_zero() {
let (actor, _guard) = build_engine(Arc::new(AllPersistent), Some(CommitVersion(0)));
assert!(actor.eviction_cutoff().is_none());
}
#[test]
fn a_pinned_cutoff_reports_the_entries_it_could_not_release() {
let (actor, _guard) = build_engine(Arc::new(AllPersistent), Some(CommitVersion(1)));
let kind = EntryKind::Source(StorageId::Table(TableId(1)), EntryLayout::Row);
for i in 0..8u64 {
write(&actor.commit, kind, &ek(&format!("k{i}")), 10 + i, "v");
}
let outcome = actor.sweep_slice(FLUSH_BYTE_BUDGET);
assert_eq!(outcome.reclaimed, 0, "nothing is below the pinned cutoff, so nothing can be reclaimed");
assert_eq!(
outcome.backlog, 8,
"the entries the cutoff could not release must still be reported, or a pinned floor \
looks exactly like an idle one"
);
assert!(
outcome.progress.is_exhausted(),
"budget exhaustion must not be the backlog signal: a pinned cutoff collects nothing and \
therefore never reports more work to do"
);
}
#[test]
fn a_cutoff_that_can_release_reports_what_it_reclaimed() {
let (actor, _guard) = build_engine(Arc::new(AllPersistent), Some(CommitVersion(20)));
let kind = EntryKind::Source(StorageId::Table(TableId(1)), EntryLayout::Row);
for i in 0..8u64 {
write(&actor.commit, kind, &ek(&format!("k{i}")), 10 + i, "v");
}
let outcome = actor.sweep_slice(FLUSH_BYTE_BUDGET);
assert!(outcome.reclaimed > 0, "entries below the cutoff must count as work done");
assert_eq!(outcome.backlog, 0, "a drained buffer reports no backlog");
}
#[test]
fn sweep_persists_then_evicts_persistent_object_below_watermark() {
let (actor, _guard) = build_engine(Arc::new(AllPersistent), Some(CommitVersion(2)));
let kind = EntryKind::Source(StorageId::Table(TableId(1)), EntryLayout::Row);
let key = ek("k");
write(&actor.commit, kind, &key, 1, "v1");
write(&actor.commit, kind, &key, 2, "v2");
write(&actor.commit, kind, &key, 3, "v3");
actor.sweep(CommitVersion(2));
assert!(
matches!(
actor.commit.get(kind, key.as_ref(), CommitVersion(2)).unwrap(),
VersionedGetResult::NotFound
),
"v2 must be gone from the buffer after eviction"
);
assert!(
matches!(
actor.persistent.get(kind, key.as_ref(), CommitVersion(2)).unwrap(),
VersionedGetResult::Value { .. }
),
"v2 must survive in the persistent tier"
);
assert_eq!(
actor.commit.get(kind, key.as_ref(), CommitVersion(3)).unwrap().value().as_deref(),
Some(b"v3".as_slice()),
"v3 (> cutoff) must stay in the buffer"
);
}
#[test]
fn sweep_evicts_non_persistent_object_without_persisting() {
let (actor, _guard) = build_engine(Arc::new(NonePersistent), Some(CommitVersion(2)));
let kind = EntryKind::Source(StorageId::Table(TableId(7)), EntryLayout::Row);
let key = ek("ephemeral");
write(&actor.commit, kind, &key, 1, "v1");
write(&actor.commit, kind, &key, 2, "v2");
write(&actor.commit, kind, &key, 3, "v3");
actor.sweep(CommitVersion(2));
assert!(
matches!(
actor.commit.get(kind, key.as_ref(), CommitVersion(2)).unwrap(),
VersionedGetResult::NotFound
),
"non-persistent object must still be evicted below the watermark"
);
assert!(
matches!(
actor.persistent.get(kind, key.as_ref(), CommitVersion(2)).unwrap(),
VersionedGetResult::NotFound
),
"non-persistent object must NOT be written to the persistent tier"
);
assert_eq!(
actor.commit.get(kind, key.as_ref(), CommitVersion(3)).unwrap().value().as_deref(),
Some(b"v3".as_slice()),
"v3 (> cutoff) must stay resident even for a non-persistent object"
);
}
#[test]
fn sweep_keeps_everything_when_all_above_watermark() {
let (actor, _guard) = build_engine(Arc::new(AllPersistent), Some(CommitVersion(1)));
let kind = EntryKind::Source(StorageId::Table(TableId(3)), EntryLayout::Row);
let key = ek("k");
write(&actor.commit, kind, &key, 5, "v5");
actor.sweep(CommitVersion(1));
assert_eq!(
actor.commit.get(kind, key.as_ref(), CommitVersion(5)).unwrap().value().as_deref(),
Some(b"v5".as_slice()),
"a version above the watermark must never be evicted"
);
assert!(
matches!(
actor.persistent.get(kind, key.as_ref(), CommitVersion(5)).unwrap(),
VersionedGetResult::NotFound
),
"nothing below the watermark => nothing persisted"
);
}
#[test]
fn sweep_seeds_evicted_keys_into_the_read_tier() {
let point = MultiPointTier::new(MultiPointConfig::testing()).unwrap();
let (actor, _guard) = build_engine_with_point(Arc::new(AllPersistent), CommitVersion(2), point.clone());
let kind = EntryKind::Source(StorageId::Table(TableId(11)), EntryLayout::Row);
let key = ek("k");
write(&actor.commit, kind, &key, 1, "v1");
write(&actor.commit, kind, &key, 2, "v2");
point.insert(kind, storage_key(&key).1, key.clone(), CommitVersion(2), Some(val("stale")));
actor.sweep(CommitVersion(2));
match point.get(kind, storage_key(&key).1, &key, CommitVersion(2)) {
VersionedGetResult::Value {
value,
..
} => assert_eq!(
value.as_ref(),
val("v2").as_ref(),
"the point tier must hold the persisted value, not the stale one"
),
other => panic!("the sweep must seed the evicted key into the point tier, got {other:?}"),
}
}
#[test]
fn sweep_seeds_a_delete_of_a_persisted_row_into_the_read_tier() {
let point = MultiPointTier::new(MultiPointConfig::testing()).unwrap();
let (actor, _guard) = build_engine_with_point(Arc::new(AllPersistent), CommitVersion(2), point.clone());
let kind = EntryKind::Source(StorageId::Table(TableId(21)), EntryLayout::Row);
let key = ek("k");
actor.persistent
.set(CommitVersion(1), HashMap::from([(kind, vec![(key.clone(), Some(val("v1")))])]))
.unwrap();
point.insert(kind, storage_key(&key).1, key.clone(), CommitVersion(1), Some(val("v1")));
write(&actor.commit, kind, &key, 1, "v1");
actor.commit.set(CommitVersion(2), HashMap::from([(kind, vec![(key.clone(), None)])])).unwrap();
actor.sweep(CommitVersion(2));
assert!(
matches!(
point.get(kind, storage_key(&key).1, &key, CommitVersion(2)),
VersionedGetResult::Tombstone
),
"an evicted delete that matched a persisted row must be seeded into the point tier as a \
definitive miss, never left holding the value it deleted"
);
}
#[test]
fn sweep_invalidates_rejected_key_but_seeds_accepted() {
let point = MultiPointTier::new(MultiPointConfig::testing()).unwrap();
let (actor, _guard) = build_engine_with_point(Arc::new(AllPersistent), CommitVersion(2), point.clone());
let kind = EntryKind::Source(StorageId::Table(TableId(22)), EntryLayout::Row);
let rejected = ek("rejected");
let accepted = ek("accepted");
actor.persistent
.set(CommitVersion(3), HashMap::from([(kind, vec![(rejected.clone(), Some(val("high")))])]))
.unwrap();
point.insert(kind, storage_key(&rejected).1, rejected.clone(), CommitVersion(2), Some(val("stale")));
write(&actor.commit, kind, &rejected, 2, "low");
write(&actor.commit, kind, &accepted, 2, "b");
actor.sweep(CommitVersion(2));
assert!(
matches!(
point.get(kind, storage_key(&rejected).1, &rejected, CommitVersion(2)),
VersionedGetResult::NotFound
),
"a guard-rejected key must be invalidated in the point tier so reads fall through to the newer \
persisted value, never serving the stale entry"
);
match point.get(kind, storage_key(&accepted).1, &accepted, CommitVersion(2)) {
VersionedGetResult::Value {
value,
..
} => assert_eq!(value.as_ref(), val("b").as_ref(), "the accepted key must be seeded"),
other => panic!("the accepted key must be seeded into the point tier, got {other:?}"),
}
}
#[test]
fn sweep_seed_respects_read_tier_downgrade_guard() {
let point = MultiPointTier::new(MultiPointConfig::testing()).unwrap();
let (actor, _guard) = build_engine_with_point(Arc::new(AllPersistent), CommitVersion(2), point.clone());
let kind = EntryKind::Source(StorageId::Table(TableId(23)), EntryLayout::Row);
let key = ek("k");
point.insert(kind, storage_key(&key).1, key.clone(), CommitVersion(5), Some(val("newer")));
write(&actor.commit, kind, &key, 2, "older");
actor.sweep(CommitVersion(2));
match point.get(kind, storage_key(&key).1, &key, CommitVersion(5)) {
VersionedGetResult::Value {
value,
..
} => assert_eq!(
value.as_ref(),
val("newer").as_ref(),
"the older seeded value must not overwrite a newer resident point-tier entry"
),
other => panic!("the newer point-tier entry must survive the sweep's seed, got {other:?}"),
}
}
#[test]
fn sweep_invalidates_ephemeral_object_in_read_tier() {
let point = MultiPointTier::new(MultiPointConfig::testing()).unwrap();
let (actor, _guard) =
build_engine_with_point(Arc::new(NonePersistent), CommitVersion(2), point.clone());
let kind = EntryKind::Source(StorageId::Table(TableId(24)), EntryLayout::Row);
let key = ek("k");
point.insert(kind, storage_key(&key).1, key.clone(), CommitVersion(2), Some(val("stale")));
write(&actor.commit, kind, &key, 2, "v2");
actor.sweep(CommitVersion(2));
assert!(
matches!(
point.get(kind, storage_key(&key).1, &key, CommitVersion(2)),
VersionedGetResult::NotFound
),
"an ephemeral (persistent:false) object must be invalidated in the point tier, never seeded"
);
assert!(
matches!(
actor.persistent.get(kind, key.as_ref(), CommitVersion(2)).unwrap(),
VersionedGetResult::NotFound
),
"an ephemeral object must not be persisted"
);
}
#[test]
fn sweep_seeds_accepted_keys_across_version_buckets() {
let point = MultiPointTier::new(MultiPointConfig::testing()).unwrap();
let (actor, _guard) = build_engine_with_point(Arc::new(AllPersistent), CommitVersion(4), point.clone());
let kind = EntryKind::Source(StorageId::Table(TableId(25)), EntryLayout::Row);
let a = ek("a");
let b = ek("b");
write(&actor.commit, kind, &a, 1, "a1");
write(&actor.commit, kind, &a, 2, "a2");
write(&actor.commit, kind, &b, 3, "b3");
write(&actor.commit, kind, &b, 4, "b4");
actor.sweep(CommitVersion(4));
match point.get(kind, storage_key(&a).1, &a, CommitVersion(4)) {
VersionedGetResult::Value {
value,
..
} => assert_eq!(value.as_ref(), val("a2").as_ref(), "a's latest-<=W (v2) must be seeded"),
other => panic!("key a must be seeded across version buckets, got {other:?}"),
}
match point.get(kind, storage_key(&b).1, &b, CommitVersion(4)) {
VersionedGetResult::Value {
value,
..
} => assert_eq!(value.as_ref(), val("b4").as_ref(), "b's latest-<=W (v4) must be seeded"),
other => panic!("key b must be seeded across version buckets, got {other:?}"),
}
}
#[test]
fn sweep_removes_the_persisted_row_so_deleted_keys_stay_deleted_after_eviction() {
let (actor, _guard) = build_engine(Arc::new(AllPersistent), Some(CommitVersion(2)));
let kind = EntryKind::Source(StorageId::Table(TableId(12)), EntryLayout::Row);
let key = ek("k");
actor.persistent
.set(CommitVersion(1), HashMap::from([(kind, vec![(key.clone(), Some(val("v1")))])]))
.unwrap();
write(&actor.commit, kind, &key, 1, "v1");
actor.commit.set(CommitVersion(2), HashMap::from([(kind, vec![(key.clone(), None)])])).unwrap();
actor.sweep(CommitVersion(2));
assert!(
matches!(
actor.commit.get(kind, key.as_ref(), CommitVersion(2)).unwrap(),
VersionedGetResult::NotFound
),
"both versions are gone from the buffer"
);
assert!(
matches!(
actor.persistent.get(kind, key.as_ref(), CommitVersion(2)).unwrap(),
VersionedGetResult::NotFound
),
"the persisted row must be gone - leaving it behind resurrects v1 once the buffer drops it"
);
}
#[test]
fn sweep_evicts_below_and_keeps_above_across_multiple_keys() {
let (actor, _guard) = build_engine(Arc::new(AllPersistent), Some(CommitVersion(2)));
let kind = EntryKind::Source(StorageId::Table(TableId(13)), EntryLayout::Row);
let cold = ek("cold");
let hot = ek("hot");
write(&actor.commit, kind, &cold, 1, "cold1");
write(&actor.commit, kind, &hot, 4, "hot4");
actor.sweep(CommitVersion(2));
assert!(
matches!(
actor.commit.get(kind, cold.as_ref(), CommitVersion(2)).unwrap(),
VersionedGetResult::NotFound
),
"cold (v1 <= cutoff) must be evicted from the buffer"
);
assert!(
matches!(
actor.persistent.get(kind, cold.as_ref(), CommitVersion(2)).unwrap(),
VersionedGetResult::Value { .. }
),
"cold must survive in persistent"
);
assert_eq!(
actor.commit.get(kind, hot.as_ref(), CommitVersion(4)).unwrap().value().as_deref(),
Some(b"hot4".as_slice()),
"hot (v4 > cutoff) must stay resident in the buffer"
);
assert!(
matches!(
actor.persistent.get(kind, hot.as_ref(), CommitVersion(4)).unwrap(),
VersionedGetResult::NotFound
),
"hot must not be persisted - it is above the watermark"
);
}
#[test]
fn flush_all_persists_every_key_regardless_of_watermark() {
let (actor, _guard) = build_engine(Arc::new(AllPersistent), Some(CommitVersion(1)));
let kind = EntryKind::Source(StorageId::Table(TableId(101)), EntryLayout::Row);
let cold = ek("cold");
let hot = ek("hot");
write(&actor.commit, kind, &cold, 2, "cold2");
write(&actor.commit, kind, &hot, 50, "hot50");
actor.sweep(CommitVersion(u64::MAX));
assert_eq!(
actor.persistent.get(kind, cold.as_ref(), CommitVersion(u64::MAX)).unwrap().value().as_deref(),
Some(b"cold2".as_slice()),
"a key committed above the watermark must be persisted by a full flush"
);
assert_eq!(
actor.persistent.get(kind, hot.as_ref(), CommitVersion(u64::MAX)).unwrap().value().as_deref(),
Some(b"hot50".as_slice()),
"the latest committed value of every key must survive a full flush"
);
assert!(
matches!(
actor.commit.get(kind, hot.as_ref(), CommitVersion(u64::MAX)).unwrap(),
VersionedGetResult::NotFound
),
"a full flush drains the buffer after persisting"
);
}
#[test]
fn sweep_aborts_and_keeps_buffer_when_persist_fails() {
let (actor, _guard) = build_engine(Arc::new(AllPersistent), Some(CommitVersion(2)));
let row_kind = EntryKind::Source(StorageId::Table(TableId(31)), EntryLayout::Row);
let dict_kind = EntryKind::Multi;
let row_key = ek("row-referencing-id-7");
let dict_key = ek("dictionary-entry-7");
write(&actor.commit, row_kind, &row_key, 1, "id=7");
write(&actor.commit, dict_kind, &dict_key, 1, "entry-7");
actor.persistent.shutdown();
actor.sweep(CommitVersion(2));
assert_eq!(
actor.commit.get(row_kind, row_key.as_ref(), CommitVersion(2)).unwrap().value().as_deref(),
Some(b"id=7".as_slice()),
"a failed persist must leave the row write in the commit buffer, not drop the only copy"
);
assert_eq!(
actor.commit.get(dict_kind, dict_key.as_ref(), CommitVersion(2)).unwrap().value().as_deref(),
Some(b"entry-7".as_slice()),
"a failed persist must leave the dictionary write in the commit buffer, not drop the only copy"
);
}
#[test]
fn persist_sweep_errors_when_storage_is_shut_down() {
let (persistent, _guard) = MultiPersistentTier::sqlite_in_memory();
persistent.shutdown();
let kind = EntryKind::Source(StorageId::Table(TableId(32)), EntryLayout::Row);
let batches = vec![(CommitVersion(1), HashMap::from([(kind, vec![(ek("k"), Some(val("v")))])]))];
assert!(
persistent.persist_sweep(batches).is_err(),
"a shut-down persistent tier must refuse the sweep loudly so the buffer is not dropped"
);
}
#[test]
fn sweep_persists_all_kinds_and_versions_together() {
let (actor, _guard) = build_engine(Arc::new(AllPersistent), Some(CommitVersion(3)));
let row_kind = EntryKind::Source(StorageId::Table(TableId(33)), EntryLayout::Row);
let dict_kind = EntryKind::Multi;
let row_key = ek("row-referencing-id-9");
let dict_key = ek("dictionary-entry-9");
write(&actor.commit, row_kind, &row_key, 3, "id=9");
write(&actor.commit, dict_kind, &dict_key, 2, "entry-9");
actor.sweep(CommitVersion(3));
assert_eq!(
actor.persistent.get(row_kind, row_key.as_ref(), CommitVersion(3)).unwrap().value().as_deref(),
Some(b"id=9".as_slice()),
"the row write must be durable after the sweep"
);
assert_eq!(
actor.persistent
.get(dict_kind, dict_key.as_ref(), CommitVersion(3))
.unwrap()
.value()
.as_deref(),
Some(b"entry-9".as_slice()),
"the dictionary write committed at an earlier version must be durable in the same sweep"
);
assert!(
matches!(
actor.commit.get(row_kind, row_key.as_ref(), CommitVersion(3)).unwrap(),
VersionedGetResult::NotFound
),
"a persisted row write must be drained from the buffer"
);
assert!(
matches!(
actor.commit.get(dict_kind, dict_key.as_ref(), CommitVersion(3)).unwrap(),
VersionedGetResult::NotFound
),
"a persisted dictionary write must be drained from the buffer"
);
}
#[test]
fn flush_all_removes_the_persisted_row_for_a_delete_above_the_watermark() {
let (actor, _guard) = build_engine(Arc::new(AllPersistent), Some(CommitVersion(1)));
let kind = EntryKind::Source(StorageId::Table(TableId(102)), EntryLayout::Row);
let key = ek("k");
actor.persistent
.set(CommitVersion(5), HashMap::from([(kind, vec![(key.clone(), Some(val("v5")))])]))
.unwrap();
write(&actor.commit, kind, &key, 5, "v5");
actor.commit.set(CommitVersion(9), HashMap::from([(kind, vec![(key.clone(), None)])])).unwrap();
actor.sweep(CommitVersion(u64::MAX));
assert!(
matches!(
actor.persistent.get(kind, key.as_ref(), CommitVersion(u64::MAX)).unwrap(),
VersionedGetResult::NotFound
),
"a delete committed above the watermark must remove the persisted row, not resurrect v5"
);
}
#[test]
fn sweep_persists_multi_kind_entries() {
let (actor, _guard) = build_engine(Arc::new(AllPersistent), Some(CommitVersion(10)));
let kind = EntryKind::Multi;
let key = ek("dictionary-entry");
write(&actor.commit, kind, &key, 5, "mint-id-7");
actor.sweep(CommitVersion(10));
assert!(
matches!(
actor.persistent.get(kind, key.as_ref(), CommitVersion(10)).unwrap(),
VersionedGetResult::Value { .. }
),
"a Multi entry committed below the watermark must reach the persistent tier; \
dictionary entries and CDC checkpoints live in this keyspace and are lost on restart if it does not"
);
}
#[test]
fn a_kind_behind_the_budget_prefix_is_still_swept_under_sustained_writes() {
const KINDS: u64 = 40;
const KEYS_PER_ROUND: u64 = 20;
const KEYS_PER_SLICE: u64 = 40;
const ROUNDS: u64 = 60;
const FIRST_VERSION: u64 = 1;
let budget = budget_for(&(0..KEYS_PER_SLICE).map(|i| format!("v1-k{i}")).collect::<Vec<_>>(), "x");
let (actor, _guard) = build_engine(Arc::new(AllPersistent), Some(CommitVersion(1_000_000)));
let kinds: Vec<EntryKind> = (0..KINDS)
.map(|i| EntryKind::Source(StorageId::Table(TableId(i + 1)), EntryLayout::Row))
.collect();
let chronological_key = |version: u64, key: u64| {
let sequence = version * KEYS_PER_ROUND + key;
RowKey::encoded(StorageId::table(TableId(1)), RowNumber(u64::MAX - sequence))
};
let round_writes = |version: u64| {
for kind in &kinds {
for key in 0..KEYS_PER_ROUND {
write(&actor.commit, *kind, &chronological_key(version, key), version, "x");
}
}
};
round_writes(FIRST_VERSION);
let mut exhausted = 0;
for round in 0..ROUNDS {
if actor.sweep_slice(budget).progress.is_yielded() {
exhausted += 1;
}
round_writes(FIRST_VERSION + 1 + round);
}
assert_eq!(
exhausted, ROUNDS,
"the budget must run out every slice for this to exercise starvation at all"
);
let oldest = actor.commit.oldest_pending_version().expect("writes are still pending");
assert!(
oldest.0 > FIRST_VERSION,
"the oldest pending version is still {} after {ROUNDS} slices, so at least one of the \
{KINDS} kinds was never swept once; that kind pins the durable frontier at the first \
write, which clamps the tombstone reap cutoff to zero and leaves every tombstone in the \
persistent tier undeletable",
oldest.0
);
}
#[test]
fn a_kind_that_sorts_behind_a_deeper_backlog_is_still_reached_by_the_sweep() {
const HOT_KINDS: u64 = 3;
const KEYS_PER_ROUND: u64 = 40;
const KEYS_PER_SLICE: u64 = 30;
const ROUNDS: u64 = 80;
const COLD_FIRST_VERSION: u64 = 50;
let budget = budget_for(&(0..KEYS_PER_SLICE).map(|i| format!("v1-k{i}")).collect::<Vec<_>>(), "x");
let (actor, _guard) = build_engine(Arc::new(AllPersistent), Some(CommitVersion(1_000_000)));
let hot: Vec<EntryKind> = (0..HOT_KINDS)
.map(|i| EntryKind::Source(StorageId::Table(TableId(i + 1)), EntryLayout::Row))
.collect();
let cold = EntryKind::Source(StorageId::Table(TableId(HOT_KINDS + 1)), EntryLayout::Row);
for round in 1..=ROUNDS {
for kind in &hot {
for key in 0..KEYS_PER_ROUND {
write(&actor.commit, *kind, &ek(&format!("v{round}-k{key}")), round, "x");
}
}
if round == COLD_FIRST_VERSION {
write(&actor.commit, cold, &ek("cold-key"), round, "x");
}
actor.sweep_slice(budget);
}
assert!(
actor.commit.oldest_pending_for(hot[0]).is_some(),
"the hot kinds must stay backlogged, otherwise the budget never ran out and this exercises nothing"
);
assert_eq!(
actor.commit.oldest_pending_for(cold),
None,
"the single write to the cold kind is still pending after {} slices, so the sweep never reached past the hot kinds sorted ahead of it",
ROUNDS - COLD_FIRST_VERSION
);
}
#[test]
fn a_slice_cut_off_mid_kind_resumes_past_it_instead_of_restarting_at_the_head() {
const DEEP_KEYS: u64 = 400;
const KEYS_PER_SLICE: u64 = 10;
const SLICES: u64 = 20;
const COLD_VERSION: u64 = 50;
let budget = budget_for(&(0..KEYS_PER_SLICE).map(|i| format!("deep-k{i}")).collect::<Vec<_>>(), "x");
let (actor, _guard) = build_engine(Arc::new(AllPersistent), Some(CommitVersion(1_000_000)));
let deep = EntryKind::Source(StorageId::Table(TableId(1)), EntryLayout::Row);
let cold = EntryKind::Source(StorageId::Table(TableId(2)), EntryLayout::Row);
for key in 0..DEEP_KEYS {
write(&actor.commit, deep, &ek(&format!("deep-k{key}")), 1, "x");
}
write(&actor.commit, cold, &ek("cold-k"), COLD_VERSION, "x");
for _ in 0..SLICES {
actor.sweep_slice(budget);
}
assert_eq!(
actor.commit.oldest_pending_for(deep),
Some(CommitVersion(1)),
"the deep kind must still hold the oldest pending version after every slice, otherwise it \
stopped ranking ahead of the cold kind and this never exercised the cursor at all"
);
assert_eq!(
actor.commit.oldest_pending_for(cold),
None,
"the cold write is still pending after {SLICES} slices, so every slice restarted at the deep \
kind the ranking puts first and never resumed past the kind it was cut off in"
);
}
#[test]
fn a_slice_with_no_resume_point_serves_the_kind_holding_the_oldest_pending_write() {
const KINDS: u64 = 6;
const OLD_VERSION: u64 = 1;
const YOUNG_VERSION: u64 = 100;
let budget = budget_for(&["k".to_string()], "v");
for oldest in 0..KINDS {
let (actor, _guard) = build_engine(Arc::new(AllPersistent), Some(CommitVersion(1_000_000)));
let kinds: Vec<EntryKind> = (0..KINDS)
.map(|i| EntryKind::Source(StorageId::Table(TableId(i + 1)), EntryLayout::Row))
.collect();
for (index, kind) in kinds.iter().enumerate() {
let version = if index as u64 == oldest {
OLD_VERSION
} else {
YOUNG_VERSION
};
write(&actor.commit, *kind, &ek("k"), version, "v");
}
actor.sweep_slice(budget);
assert_eq!(
actor.commit.oldest_pending_for(kinds[oldest as usize]),
None,
"a budget covering exactly one entry must buy the kind holding the oldest pending \
write, but the kind at version {OLD_VERSION} was left pending while a younger kind \
took the slice; the durable frontier is the minimum over kinds, so serving anything \
but the oldest cannot advance it"
);
for (index, kind) in kinds.iter().enumerate() {
if index as u64 == oldest {
continue;
}
assert_eq!(
actor.commit.oldest_pending_for(*kind),
Some(CommitVersion(YOUNG_VERSION)),
"the budget covered one entry, so no kind younger than the oldest may have \
been served in the same slice"
);
}
}
}
}