reifydb_store_multi/flush/
actor.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
5use std::collections::{HashMap, HashSet};
6use std::sync::{Arc, OnceLock};
7
8#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
9use reifydb_codec::key::encoded::EncodedKey;
10#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
11use reifydb_core::{common::CommitVersion, interface::store::EntryKind};
12#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
13use reifydb_runtime::actor::timers::TimerHandle;
14#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
15use reifydb_runtime::actor::{
16	context::Context,
17	mailbox::ActorRef,
18	system::{ActorConfig, ActorSpawner},
19	traits::{Actor, Directive},
20};
21use reifydb_runtime::sync::{rwlock::RwLock, waiter::WaiterHandle};
22use reifydb_value::value::{datetime::DateTime, duration::Duration};
23#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
24use reifydb_value::{reifydb_assertions, util::cowvec::CowVec};
25#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
26use tracing::{debug, error, warn};
27
28#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
29use crate::tier::{TierBatch, TierStorage};
30use crate::{
31	flush::ShapePersistence,
32	gc::EvictionWatermark,
33	tier::{commit::buffer::MultiCommitBufferTier, persistent::MultiPersistentTier, read::MultiReadBufferTier},
34};
35
36#[derive(Clone)]
37pub enum FlushMessage {
38	Tick(DateTime),
39	Shutdown,
40
41	SetInterval(Duration),
42
43	FlushPending {
44		waiter: Arc<WaiterHandle>,
45	},
46
47	FlushAll {
48		waiter: Arc<WaiterHandle>,
49	},
50}
51
52#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
53pub struct FlushActorState {
54	timer_handle: Option<TimerHandle>,
55}
56
57#[allow(dead_code)]
58pub struct FlushActor {
59	commit: MultiCommitBufferTier,
60	persistent: MultiPersistentTier,
61	flush_interval: Duration,
62	persistence: Arc<OnceLock<Arc<dyn ShapePersistence>>>,
63	eviction_watermark: Arc<RwLock<Option<Arc<dyn EvictionWatermark>>>>,
64	read: Option<MultiReadBufferTier>,
65}
66
67#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
68type EvictablePartition = (Vec<(EncodedKey, CommitVersion, Option<CowVec<u8>>)>, Vec<(EncodedKey, CommitVersion)>);
69
70#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
71impl FlushActor {
72	pub fn new(
73		commit: MultiCommitBufferTier,
74		persistent: MultiPersistentTier,
75		flush_interval: Duration,
76		persistence: Arc<OnceLock<Arc<dyn ShapePersistence>>>,
77		eviction_watermark: Arc<RwLock<Option<Arc<dyn EvictionWatermark>>>>,
78		read: Option<MultiReadBufferTier>,
79	) -> Self {
80		Self {
81			commit,
82			persistent,
83			flush_interval,
84			persistence,
85			eviction_watermark,
86			read,
87		}
88	}
89
90	pub fn spawn(
91		spawner: &ActorSpawner,
92		commit: MultiCommitBufferTier,
93		persistent: MultiPersistentTier,
94		flush_interval: Duration,
95		persistence: Arc<OnceLock<Arc<dyn ShapePersistence>>>,
96		eviction_watermark: Arc<RwLock<Option<Arc<dyn EvictionWatermark>>>>,
97		read: Option<MultiReadBufferTier>,
98	) -> ActorRef<FlushMessage> {
99		let actor = Self::new(commit, persistent, flush_interval, persistence, eviction_watermark, read);
100		spawner.spawn_coordination("persistent-flush", actor).actor_ref().clone()
101	}
102
103	fn eviction_cutoff(&self) -> Option<CommitVersion> {
104		let cutoff = self.eviction_watermark.read().as_ref()?.watermark();
105		if cutoff.0 == 0 {
106			return None;
107		}
108		Some(cutoff)
109	}
110
111	fn is_persistent_shape(&self, kind: EntryKind) -> bool {
112		match kind {
113			EntryKind::Source(shape) | EntryKind::PartitionedSource(shape) => {
114				self.persistence.get().map(|provider| provider.is_persistent(shape)).unwrap_or(true)
115			}
116			_ => true,
117		}
118	}
119
120	fn sweep(&self, cutoff: CommitVersion) {
121		let Some(entry_kinds) = self.list_evictable_kinds() else {
122			return;
123		};
124
125		let mut plan: Vec<(EntryKind, bool, EvictablePartition)> = Vec::new();
126		let mut batches: HashMap<CommitVersion, TierBatch> = HashMap::new();
127		for kind in entry_kinds {
128			let (to_persist, to_drop) = self.collect_evictable(kind, cutoff);
129			if to_drop.is_empty() {
130				continue;
131			}
132			let persistent_shape = self.is_persistent_shape(kind);
133			if persistent_shape {
134				for (key, version, value) in &to_persist {
135					batches.entry(*version)
136						.or_default()
137						.entry(kind)
138						.or_default()
139						.push((key.clone(), value.clone()));
140				}
141			}
142			plan.push((kind, persistent_shape, (to_persist, to_drop)));
143		}
144		if plan.is_empty() {
145			return;
146		}
147
148		let accepted = if batches.values().any(|batch| !batch.is_empty()) {
149			match self.persistent.persist_sweep(batches.into_iter().collect()) {
150				Ok(accepted) => accepted,
151				Err(e) => {
152					error!(error = %e, "flush sweep: persist failed, aborting sweep");
153					return;
154				}
155			}
156		} else {
157			Vec::new()
158		};
159		let persisted = accepted.len();
160
161		let mut dropped = 0usize;
162		for (kind, persistent_shape, (to_persist, to_drop)) in plan {
163			self.refresh_read_tier(persistent_shape, &to_persist, &to_drop, &accepted);
164			if let Some(count) = self.drop_from_commit(kind, to_drop) {
165				dropped += count;
166			}
167		}
168
169		self.checkpoint_and_maintain(cutoff, persisted, dropped);
170	}
171
172	#[inline]
173	fn list_evictable_kinds(&self) -> Option<Vec<EntryKind>> {
174		match self.commit.list_all_entry_kinds() {
175			Ok(v) => Some(v),
176			Err(e) => {
177				warn!(error = %e, "flush sweep: list_all_entry_kinds failed");
178				None
179			}
180		}
181	}
182
183	#[inline]
184	fn collect_evictable(&self, kind: EntryKind, cutoff: CommitVersion) -> EvictablePartition {
185		match &self.commit {
186			MultiCommitBufferTier::Memory(s) => s.collect_evictable_below(kind, cutoff),
187		}
188	}
189
190	#[inline]
191	fn refresh_read_tier(
192		&self,
193		persistent_shape: bool,
194		to_persist: &[(EncodedKey, CommitVersion, Option<CowVec<u8>>)],
195		to_drop: &[(EncodedKey, CommitVersion)],
196		accepted: &[EncodedKey],
197	) {
198		let Some(read) = &self.read else {
199			return;
200		};
201		if persistent_shape {
202			let accepted: HashSet<&[u8]> = accepted.iter().map(|k| k.as_slice()).collect();
203			for (key, version, value) in to_persist {
204				if accepted.contains(key.as_slice()) {
205					read.insert(key.clone(), *version, value.clone());
206				} else {
207					read.invalidate(key);
208				}
209			}
210		} else {
211			for (key, _) in to_drop {
212				read.invalidate(key);
213			}
214		}
215	}
216
217	#[inline]
218	fn drop_from_commit(&self, kind: EntryKind, to_drop: Vec<(EncodedKey, CommitVersion)>) -> Option<usize> {
219		let drop_count = to_drop.len();
220		reifydb_assertions! {
221			assert!(
222				drop_count > 0,
223				"sweep must only reach drop_from_commit with a non-empty drop set; an empty drop \
224				 issues a no-op commit-buffer drop and lets the dropped counter run for zero work \
225				 (kind={kind:?})"
226			);
227		}
228		let mut batches: HashMap<EntryKind, Vec<(EncodedKey, CommitVersion)>> = HashMap::new();
229		batches.insert(kind, to_drop);
230		if let Err(e) = self.commit.drop(batches) {
231			warn!(?kind, error = %e, "flush sweep: commit buffer drop failed");
232			return None;
233		}
234		Some(drop_count)
235	}
236
237	#[inline]
238	fn checkpoint_and_maintain(&self, cutoff: CommitVersion, persisted: usize, dropped: usize) {
239		if persisted > 0 || dropped > 0 {
240			debug!(cutoff = cutoff.0, persisted, dropped, "flush sweep completed");
241			if persisted > 0
242				&& let Err(e) = self.persistent.maybe_checkpoint()
243			{
244				warn!(error = %e, "flush sweep: checkpoint failed");
245			}
246			self.commit.maintenance();
247		}
248	}
249}
250
251#[cfg(all(feature = "sqlite", not(target_arch = "wasm32")))]
252impl Actor for FlushActor {
253	type State = FlushActorState;
254	type Message = FlushMessage;
255
256	fn init(&self, ctx: &Context<FlushMessage>) -> FlushActorState {
257		debug!("Persistent flush actor started");
258		let timer_handle = ctx.schedule_tick(self.flush_interval.to_std(), |nanos| {
259			FlushMessage::Tick(DateTime::from_nanos(nanos))
260		});
261		FlushActorState {
262			timer_handle: Some(timer_handle),
263		}
264	}
265
266	fn handle(&self, state: &mut FlushActorState, msg: FlushMessage, ctx: &Context<FlushMessage>) -> Directive {
267		if ctx.is_cancelled() {
268			if let Some(cutoff) = self.eviction_cutoff() {
269				self.sweep(cutoff);
270			}
271			return Directive::Stop;
272		}
273		match msg {
274			FlushMessage::Tick(_) => {
275				if let Some(cutoff) = self.eviction_cutoff() {
276					self.sweep(cutoff);
277				}
278			}
279			FlushMessage::SetInterval(interval) => {
280				if let Some(handle) = state.timer_handle.take() {
281					handle.cancel();
282				}
283				state.timer_handle = Some(ctx.schedule_tick(interval.to_std(), |nanos| {
284					FlushMessage::Tick(DateTime::from_nanos(nanos))
285				}));
286			}
287			FlushMessage::Shutdown => {
288				debug!("Persistent flush actor shutting down");
289				if let Some(cutoff) = self.eviction_cutoff() {
290					self.sweep(cutoff);
291				}
292				return Directive::Stop;
293			}
294			FlushMessage::FlushPending {
295				waiter,
296			} => {
297				if let Some(cutoff) = self.eviction_cutoff() {
298					self.sweep(cutoff);
299				}
300				waiter.notify();
301			}
302			FlushMessage::FlushAll {
303				waiter,
304			} => {
305				self.sweep(CommitVersion(u64::MAX));
306				waiter.notify();
307			}
308		}
309		Directive::Continue
310	}
311
312	fn post_stop(&self) {
313		debug!("Persistent flush actor stopped");
314	}
315
316	fn config(&self) -> ActorConfig {
317		ActorConfig::new().mailbox_capacity(4096)
318	}
319}
320
321#[cfg(all(test, feature = "sqlite", not(target_arch = "wasm32")))]
322mod tests {
323	use reifydb_core::interface::catalog::{id::TableId, shape::ShapeId};
324	use reifydb_runtime::shutdown::Shutdown;
325	use reifydb_sqlite::SqliteTempPathGuard;
326	use reifydb_value::util::cowvec::CowVec;
327
328	use super::*;
329	use crate::tier::{VersionedGetResult, read::ReadBufferConfig};
330
331	fn ek(s: &str) -> EncodedKey {
332		EncodedKey::new(s.as_bytes().to_vec())
333	}
334
335	fn val(s: &str) -> CowVec<u8> {
336		CowVec::new(s.as_bytes().to_vec())
337	}
338
339	fn write(buffer: &MultiCommitBufferTier, kind: EntryKind, key: &EncodedKey, version: u64, value: &str) {
340		buffer.set(CommitVersion(version), HashMap::from([(kind, vec![(key.clone(), Some(val(value)))])]))
341			.unwrap();
342	}
343
344	struct StaticWatermark(CommitVersion);
345
346	impl EvictionWatermark for StaticWatermark {
347		fn watermark(&self) -> CommitVersion {
348			self.0
349		}
350	}
351
352	struct AllPersistent;
353
354	impl ShapePersistence for AllPersistent {
355		fn is_persistent(&self, _shape: ShapeId) -> bool {
356			true
357		}
358	}
359
360	struct NonePersistent;
361
362	impl ShapePersistence for NonePersistent {
363		fn is_persistent(&self, _shape: ShapeId) -> bool {
364			false
365		}
366	}
367
368	fn build_actor(
369		persistence: Arc<dyn ShapePersistence>,
370		watermark: Option<CommitVersion>,
371	) -> (FlushActor, SqliteTempPathGuard) {
372		let buffer = MultiCommitBufferTier::memory();
373		let (persistent, guard) = MultiPersistentTier::sqlite_in_memory();
374		let persistence_lock: Arc<OnceLock<Arc<dyn ShapePersistence>>> = Arc::new(OnceLock::new());
375		let _ = persistence_lock.set(persistence);
376		let watermark_lock: Arc<RwLock<Option<Arc<dyn EvictionWatermark>>>> = Arc::new(RwLock::new(None));
377		if let Some(w) = watermark {
378			*watermark_lock.write() = Some(Arc::new(StaticWatermark(w)));
379		}
380		(
381			FlushActor::new(
382				buffer,
383				persistent,
384				Duration::from_seconds(5).unwrap(),
385				persistence_lock,
386				watermark_lock,
387				None,
388			),
389			guard,
390		)
391	}
392
393	fn build_actor_with_read(
394		persistence: Arc<dyn ShapePersistence>,
395		watermark: CommitVersion,
396		read: MultiReadBufferTier,
397	) -> (FlushActor, SqliteTempPathGuard) {
398		let buffer = MultiCommitBufferTier::memory();
399		let (persistent, guard) = MultiPersistentTier::sqlite_in_memory();
400		let persistence_lock: Arc<OnceLock<Arc<dyn ShapePersistence>>> = Arc::new(OnceLock::new());
401		let _ = persistence_lock.set(persistence);
402		let watermark_lock: Arc<RwLock<Option<Arc<dyn EvictionWatermark>>>> = Arc::new(RwLock::new(None));
403		*watermark_lock.write() = Some(Arc::new(StaticWatermark(watermark)));
404		(
405			FlushActor::new(
406				buffer,
407				persistent,
408				Duration::from_seconds(5).unwrap(),
409				persistence_lock,
410				watermark_lock,
411				Some(read),
412			),
413			guard,
414		)
415	}
416
417	#[test]
418	fn eviction_cutoff_is_none_without_watermark() {
419		let (actor, _guard) = build_actor(Arc::new(AllPersistent), None);
420		assert!(actor.eviction_cutoff().is_none(), "no watermark set => no eviction");
421	}
422
423	#[test]
424	fn eviction_cutoff_is_none_at_zero() {
425		let (actor, _guard) = build_actor(Arc::new(AllPersistent), Some(CommitVersion(0)));
426		assert!(actor.eviction_cutoff().is_none());
427	}
428
429	#[test]
430	fn sweep_persists_then_evicts_persistent_shape_below_watermark() {
431		let (actor, _guard) = build_actor(Arc::new(AllPersistent), Some(CommitVersion(2)));
432		let kind = EntryKind::Source(ShapeId::Table(TableId(1)));
433		let key = ek("k");
434		write(&actor.commit, kind, &key, 1, "v1");
435		write(&actor.commit, kind, &key, 2, "v2");
436		write(&actor.commit, kind, &key, 3, "v3");
437
438		actor.sweep(CommitVersion(2));
439
440		assert!(
441			matches!(
442				actor.commit.get(kind, key.as_ref(), CommitVersion(2)).unwrap(),
443				VersionedGetResult::NotFound
444			),
445			"v2 must be gone from the buffer after eviction"
446		);
447		assert!(
448			matches!(
449				actor.persistent.get(kind, key.as_ref(), CommitVersion(2)).unwrap(),
450				VersionedGetResult::Value { .. }
451			),
452			"v2 must survive in the persistent tier"
453		);
454
455		assert_eq!(
456			actor.commit.get(kind, key.as_ref(), CommitVersion(3)).unwrap().value().as_deref(),
457			Some(b"v3".as_slice()),
458			"v3 (> cutoff) must stay in the buffer"
459		);
460	}
461
462	#[test]
463	fn sweep_evicts_non_persistent_shape_without_persisting() {
464		let (actor, _guard) = build_actor(Arc::new(NonePersistent), Some(CommitVersion(2)));
465		let kind = EntryKind::Source(ShapeId::Table(TableId(7)));
466		let key = ek("ephemeral");
467		write(&actor.commit, kind, &key, 1, "v1");
468		write(&actor.commit, kind, &key, 2, "v2");
469		write(&actor.commit, kind, &key, 3, "v3");
470
471		actor.sweep(CommitVersion(2));
472
473		assert!(
474			matches!(
475				actor.commit.get(kind, key.as_ref(), CommitVersion(2)).unwrap(),
476				VersionedGetResult::NotFound
477			),
478			"non-persistent shape must still be evicted below the watermark"
479		);
480		assert!(
481			matches!(
482				actor.persistent.get(kind, key.as_ref(), CommitVersion(2)).unwrap(),
483				VersionedGetResult::NotFound
484			),
485			"non-persistent shape must NOT be written to the persistent tier"
486		);
487		assert_eq!(
488			actor.commit.get(kind, key.as_ref(), CommitVersion(3)).unwrap().value().as_deref(),
489			Some(b"v3".as_slice()),
490			"v3 (> cutoff) must stay resident even for a non-persistent shape"
491		);
492	}
493
494	#[test]
495	fn sweep_keeps_everything_when_all_above_watermark() {
496		let (actor, _guard) = build_actor(Arc::new(AllPersistent), Some(CommitVersion(1)));
497		let kind = EntryKind::Source(ShapeId::Table(TableId(3)));
498		let key = ek("k");
499		write(&actor.commit, kind, &key, 5, "v5");
500
501		actor.sweep(CommitVersion(1));
502
503		assert_eq!(
504			actor.commit.get(kind, key.as_ref(), CommitVersion(5)).unwrap().value().as_deref(),
505			Some(b"v5".as_slice()),
506			"a version above the watermark must never be evicted"
507		);
508		assert!(
509			matches!(
510				actor.persistent.get(kind, key.as_ref(), CommitVersion(5)).unwrap(),
511				VersionedGetResult::NotFound
512			),
513			"nothing below the watermark => nothing persisted"
514		);
515	}
516
517	#[test]
518	fn sweep_seeds_evicted_keys_into_the_read_tier() {
519		let read = MultiReadBufferTier::new(ReadBufferConfig {
520			resident_pages: 16,
521			..Default::default()
522		});
523		let (actor, _guard) = build_actor_with_read(Arc::new(AllPersistent), CommitVersion(2), read.clone());
524		let kind = EntryKind::Source(ShapeId::Table(TableId(11)));
525		let key = ek("k");
526		write(&actor.commit, kind, &key, 1, "v1");
527		write(&actor.commit, kind, &key, 2, "v2");
528
529		read.insert(key.clone(), CommitVersion(2), Some(val("stale")));
530
531		actor.sweep(CommitVersion(2));
532
533		match read.get(&key, CommitVersion(2)) {
534			VersionedGetResult::Value {
535				value,
536				..
537			} => assert_eq!(
538				value.as_ref(),
539				val("v2").as_ref(),
540				"the read tier must hold the persisted value, not the stale one"
541			),
542			other => panic!("the sweep must seed the evicted key into the read tier, got {other:?}"),
543		}
544	}
545
546	#[test]
547	fn sweep_seeds_tombstone_into_read_tier() {
548		let read = MultiReadBufferTier::new(ReadBufferConfig {
549			resident_pages: 16,
550			..Default::default()
551		});
552		let (actor, _guard) = build_actor_with_read(Arc::new(AllPersistent), CommitVersion(2), read.clone());
553		let kind = EntryKind::Source(ShapeId::Table(TableId(21)));
554		let key = ek("k");
555		write(&actor.commit, kind, &key, 1, "v1");
556		actor.commit.set(CommitVersion(2), HashMap::from([(kind, vec![(key.clone(), None)])])).unwrap();
557
558		actor.sweep(CommitVersion(2));
559
560		assert!(
561			matches!(read.get(&key, CommitVersion(2)), VersionedGetResult::Tombstone),
562			"an evicted tombstone must be seeded into the read tier as a definitive miss, not left absent \
563			 (which would fall through and risk resurrecting an older value)"
564		);
565	}
566
567	#[test]
568	fn sweep_invalidates_rejected_key_but_seeds_accepted() {
569		let read = MultiReadBufferTier::new(ReadBufferConfig {
570			resident_pages: 16,
571			..Default::default()
572		});
573		let (actor, _guard) = build_actor_with_read(Arc::new(AllPersistent), CommitVersion(2), read.clone());
574		let kind = EntryKind::Source(ShapeId::Table(TableId(22)));
575		let rejected = ek("rejected");
576		let accepted = ek("accepted");
577
578		actor.persistent
579			.set(CommitVersion(3), HashMap::from([(kind, vec![(rejected.clone(), Some(val("high")))])]))
580			.unwrap();
581
582		read.insert(rejected.clone(), CommitVersion(2), Some(val("stale")));
583
584		write(&actor.commit, kind, &rejected, 2, "low");
585		write(&actor.commit, kind, &accepted, 2, "b");
586
587		actor.sweep(CommitVersion(2));
588
589		assert!(
590			matches!(read.get(&rejected, CommitVersion(2)), VersionedGetResult::NotFound),
591			"a guard-rejected key must be invalidated in the read tier so reads fall through to the newer \
592			 persisted value, never serving the stale entry"
593		);
594		match read.get(&accepted, CommitVersion(2)) {
595			VersionedGetResult::Value {
596				value,
597				..
598			} => assert_eq!(value.as_ref(), val("b").as_ref(), "the accepted key must be seeded"),
599			other => panic!("the accepted key must be seeded into the read tier, got {other:?}"),
600		}
601	}
602
603	#[test]
604	fn sweep_seed_respects_read_tier_downgrade_guard() {
605		let read = MultiReadBufferTier::new(ReadBufferConfig {
606			resident_pages: 16,
607			..Default::default()
608		});
609		let (actor, _guard) = build_actor_with_read(Arc::new(AllPersistent), CommitVersion(2), read.clone());
610		let kind = EntryKind::Source(ShapeId::Table(TableId(23)));
611		let key = ek("k");
612
613		read.insert(key.clone(), CommitVersion(5), Some(val("newer")));
614
615		write(&actor.commit, kind, &key, 2, "older");
616		actor.sweep(CommitVersion(2));
617
618		match read.get(&key, CommitVersion(5)) {
619			VersionedGetResult::Value {
620				value,
621				..
622			} => assert_eq!(
623				value.as_ref(),
624				val("newer").as_ref(),
625				"the older seeded value must not overwrite a newer resident read-tier entry"
626			),
627			other => panic!("the newer read-tier entry must survive the sweep's seed, got {other:?}"),
628		}
629	}
630
631	#[test]
632	fn sweep_invalidates_ephemeral_shape_in_read_tier() {
633		let read = MultiReadBufferTier::new(ReadBufferConfig {
634			resident_pages: 16,
635			..Default::default()
636		});
637		let (actor, _guard) = build_actor_with_read(Arc::new(NonePersistent), CommitVersion(2), read.clone());
638		let kind = EntryKind::Source(ShapeId::Table(TableId(24)));
639		let key = ek("k");
640
641		read.insert(key.clone(), CommitVersion(2), Some(val("stale")));
642		write(&actor.commit, kind, &key, 2, "v2");
643
644		actor.sweep(CommitVersion(2));
645
646		assert!(
647			matches!(read.get(&key, CommitVersion(2)), VersionedGetResult::NotFound),
648			"an ephemeral (persistent:false) shape must be invalidated in the read tier, never seeded"
649		);
650		assert!(
651			matches!(
652				actor.persistent.get(kind, key.as_ref(), CommitVersion(2)).unwrap(),
653				VersionedGetResult::NotFound
654			),
655			"an ephemeral shape must not be persisted"
656		);
657	}
658
659	#[test]
660	fn sweep_seeds_accepted_keys_across_version_buckets() {
661		let read = MultiReadBufferTier::new(ReadBufferConfig {
662			resident_pages: 16,
663			..Default::default()
664		});
665		let (actor, _guard) = build_actor_with_read(Arc::new(AllPersistent), CommitVersion(4), read.clone());
666		let kind = EntryKind::Source(ShapeId::Table(TableId(25)));
667		let a = ek("a");
668		let b = ek("b");
669		write(&actor.commit, kind, &a, 1, "a1");
670		write(&actor.commit, kind, &a, 2, "a2");
671		write(&actor.commit, kind, &b, 3, "b3");
672		write(&actor.commit, kind, &b, 4, "b4");
673
674		actor.sweep(CommitVersion(4));
675
676		match read.get(&a, CommitVersion(4)) {
677			VersionedGetResult::Value {
678				value,
679				..
680			} => assert_eq!(value.as_ref(), val("a2").as_ref(), "a's latest-<=W (v2) must be seeded"),
681			other => panic!("key a must be seeded across version buckets, got {other:?}"),
682		}
683		match read.get(&b, CommitVersion(4)) {
684			VersionedGetResult::Value {
685				value,
686				..
687			} => assert_eq!(value.as_ref(), val("b4").as_ref(), "b's latest-<=W (v4) must be seeded"),
688			other => panic!("key b must be seeded across version buckets, got {other:?}"),
689		}
690	}
691
692	#[test]
693	fn sweep_persists_tombstone_so_deleted_keys_stay_deleted_after_eviction() {
694		let (actor, _guard) = build_actor(Arc::new(AllPersistent), Some(CommitVersion(2)));
695		let kind = EntryKind::Source(ShapeId::Table(TableId(12)));
696		let key = ek("k");
697		write(&actor.commit, kind, &key, 1, "v1");
698		actor.commit.set(CommitVersion(2), HashMap::from([(kind, vec![(key.clone(), None)])])).unwrap();
699
700		actor.sweep(CommitVersion(2));
701
702		assert!(
703			matches!(
704				actor.commit.get(kind, key.as_ref(), CommitVersion(2)).unwrap(),
705				VersionedGetResult::NotFound
706			),
707			"both versions are gone from the buffer"
708		);
709		assert!(
710			matches!(
711				actor.persistent.get(kind, key.as_ref(), CommitVersion(2)).unwrap(),
712				VersionedGetResult::Tombstone
713			),
714			"the persisted latest value must be the tombstone - the row must not resurrect"
715		);
716	}
717
718	#[test]
719	fn sweep_evicts_below_and_keeps_above_across_multiple_keys() {
720		let (actor, _guard) = build_actor(Arc::new(AllPersistent), Some(CommitVersion(2)));
721		let kind = EntryKind::Source(ShapeId::Table(TableId(13)));
722		let cold = ek("cold");
723		let hot = ek("hot");
724		write(&actor.commit, kind, &cold, 1, "cold1");
725		write(&actor.commit, kind, &hot, 4, "hot4");
726
727		actor.sweep(CommitVersion(2));
728
729		assert!(
730			matches!(
731				actor.commit.get(kind, cold.as_ref(), CommitVersion(2)).unwrap(),
732				VersionedGetResult::NotFound
733			),
734			"cold (v1 <= cutoff) must be evicted from the buffer"
735		);
736		assert!(
737			matches!(
738				actor.persistent.get(kind, cold.as_ref(), CommitVersion(2)).unwrap(),
739				VersionedGetResult::Value { .. }
740			),
741			"cold must survive in persistent"
742		);
743		assert_eq!(
744			actor.commit.get(kind, hot.as_ref(), CommitVersion(4)).unwrap().value().as_deref(),
745			Some(b"hot4".as_slice()),
746			"hot (v4 > cutoff) must stay resident in the buffer"
747		);
748		assert!(
749			matches!(
750				actor.persistent.get(kind, hot.as_ref(), CommitVersion(4)).unwrap(),
751				VersionedGetResult::NotFound
752			),
753			"hot must not be persisted - it is above the watermark"
754		);
755	}
756
757	#[test]
758	fn flush_all_persists_every_key_regardless_of_watermark() {
759		let (actor, _guard) = build_actor(Arc::new(AllPersistent), Some(CommitVersion(1)));
760		let kind = EntryKind::Source(ShapeId::Table(TableId(101)));
761		let cold = ek("cold");
762		let hot = ek("hot");
763		write(&actor.commit, kind, &cold, 2, "cold2");
764		write(&actor.commit, kind, &hot, 50, "hot50");
765
766		actor.sweep(CommitVersion(u64::MAX));
767
768		assert_eq!(
769			actor.persistent.get(kind, cold.as_ref(), CommitVersion(u64::MAX)).unwrap().value().as_deref(),
770			Some(b"cold2".as_slice()),
771			"a key committed above the watermark must be persisted by a full flush"
772		);
773		assert_eq!(
774			actor.persistent.get(kind, hot.as_ref(), CommitVersion(u64::MAX)).unwrap().value().as_deref(),
775			Some(b"hot50".as_slice()),
776			"the latest committed value of every key must survive a full flush"
777		);
778		assert!(
779			matches!(
780				actor.commit.get(kind, hot.as_ref(), CommitVersion(u64::MAX)).unwrap(),
781				VersionedGetResult::NotFound
782			),
783			"a full flush drains the buffer after persisting"
784		);
785	}
786
787	#[test]
788	fn sweep_aborts_and_keeps_buffer_when_persist_fails() {
789		let (actor, _guard) = build_actor(Arc::new(AllPersistent), Some(CommitVersion(2)));
790		let row_kind = EntryKind::Source(ShapeId::Table(TableId(31)));
791		let dict_kind = EntryKind::Multi;
792		let row_key = ek("row-referencing-id-7");
793		let dict_key = ek("dictionary-entry-7");
794		write(&actor.commit, row_kind, &row_key, 1, "id=7");
795		write(&actor.commit, dict_kind, &dict_key, 1, "entry-7");
796
797		actor.persistent.shutdown();
798		actor.sweep(CommitVersion(2));
799
800		assert_eq!(
801			actor.commit.get(row_kind, row_key.as_ref(), CommitVersion(2)).unwrap().value().as_deref(),
802			Some(b"id=7".as_slice()),
803			"a failed persist must leave the row write in the commit buffer, not drop the only copy"
804		);
805		assert_eq!(
806			actor.commit.get(dict_kind, dict_key.as_ref(), CommitVersion(2)).unwrap().value().as_deref(),
807			Some(b"entry-7".as_slice()),
808			"a failed persist must leave the dictionary write in the commit buffer, not drop the only copy"
809		);
810	}
811
812	#[test]
813	fn persist_sweep_errors_when_storage_is_shut_down() {
814		let (persistent, _guard) = MultiPersistentTier::sqlite_in_memory();
815		persistent.shutdown();
816
817		let kind = EntryKind::Source(ShapeId::Table(TableId(32)));
818		let batches = vec![(CommitVersion(1), HashMap::from([(kind, vec![(ek("k"), Some(val("v")))])]))];
819		assert!(
820			persistent.persist_sweep(batches).is_err(),
821			"a shut-down persistent tier must refuse the sweep loudly so the buffer is not dropped"
822		);
823	}
824
825	#[test]
826	fn sweep_persists_all_kinds_and_versions_together() {
827		let (actor, _guard) = build_actor(Arc::new(AllPersistent), Some(CommitVersion(3)));
828		let row_kind = EntryKind::Source(ShapeId::Table(TableId(33)));
829		let dict_kind = EntryKind::Multi;
830		let row_key = ek("row-referencing-id-9");
831		let dict_key = ek("dictionary-entry-9");
832		write(&actor.commit, row_kind, &row_key, 3, "id=9");
833		write(&actor.commit, dict_kind, &dict_key, 2, "entry-9");
834
835		actor.sweep(CommitVersion(3));
836
837		assert_eq!(
838			actor.persistent.get(row_kind, row_key.as_ref(), CommitVersion(3)).unwrap().value().as_deref(),
839			Some(b"id=9".as_slice()),
840			"the row write must be durable after the sweep"
841		);
842		assert_eq!(
843			actor.persistent
844				.get(dict_kind, dict_key.as_ref(), CommitVersion(3))
845				.unwrap()
846				.value()
847				.as_deref(),
848			Some(b"entry-9".as_slice()),
849			"the dictionary write committed at an earlier version must be durable in the same sweep"
850		);
851		assert!(
852			matches!(
853				actor.commit.get(row_kind, row_key.as_ref(), CommitVersion(3)).unwrap(),
854				VersionedGetResult::NotFound
855			),
856			"a persisted row write must be drained from the buffer"
857		);
858		assert!(
859			matches!(
860				actor.commit.get(dict_kind, dict_key.as_ref(), CommitVersion(3)).unwrap(),
861				VersionedGetResult::NotFound
862			),
863			"a persisted dictionary write must be drained from the buffer"
864		);
865	}
866
867	#[test]
868	fn flush_all_persists_latest_tombstone_above_watermark() {
869		let (actor, _guard) = build_actor(Arc::new(AllPersistent), Some(CommitVersion(1)));
870		let kind = EntryKind::Source(ShapeId::Table(TableId(102)));
871		let key = ek("k");
872		write(&actor.commit, kind, &key, 5, "v5");
873		actor.commit.set(CommitVersion(9), HashMap::from([(kind, vec![(key.clone(), None)])])).unwrap();
874
875		actor.sweep(CommitVersion(u64::MAX));
876
877		assert!(
878			matches!(
879				actor.persistent.get(kind, key.as_ref(), CommitVersion(u64::MAX)).unwrap(),
880				VersionedGetResult::Tombstone
881			),
882			"a delete committed above the watermark must persist as a tombstone, not resurrect"
883		);
884	}
885}
reifydb_store_multi/flush/actor.rs

reifydb_store_multi/flush/
actor.rs
