reifydb_store_multi/flush/

engine.rs

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2026 ReifyDB

#[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"
				);
			}
		}
	}
}