reifydb_engine/
watermark.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_cdc::consume::watermark::compute_pinning_watermark;
5use reifydb_core::{
6	common::CommitVersion,
7	lifecycle::watermark::{EvictionWatermark, QueryWatermark},
8};
9use reifydb_transaction::transaction::Transaction;
10use reifydb_value::value::identity::IdentityId;
11
12use crate::engine::StandardEngine;
13
14impl QueryWatermark for StandardEngine {
15	fn effective_gc_cutoff(&self) -> CommitVersion {
16		let qdu = self.query_done_until();
17		let lease_min = self.multi().leases().min_active().unwrap_or(CommitVersion(u64::MAX));
18		qdu.min(lease_min)
19	}
20}
21
22impl EvictionWatermark for StandardEngine {
23	fn watermark(&self) -> CommitVersion {
24		self.effective_gc_cutoff().min(self.consumer_watermark())
25	}
26}
27
28impl StandardEngine {
29	pub fn consumer_watermark(&self) -> CommitVersion {
30		let mut txn = match self.begin_query(IdentityId::system()) {
31			Ok(txn) => txn,
32			Err(_) => return CommitVersion(0),
33		};
34		match compute_pinning_watermark(&mut Transaction::Query(&mut txn), Some(&*self.checkpoint_floor())) {
35			Ok(Some(v)) => v,
36			Ok(None) => CommitVersion(u64::MAX),
37			Err(_) => CommitVersion(0),
38		}
39	}
40}
41
42#[cfg(test)]
43mod tests {
44	use reifydb_core::{common::CommitVersion, lifecycle::watermark::QueryWatermark};
45	use reifydb_test_harness::engine::TestEngine;
46
47	#[test]
48	fn effective_gc_cutoff_is_lowered_by_a_held_lease_and_nothing_else() {
49		// Only a held lease may pin the cutoff. A lagging consumer without one pinning it is
50		// an unbounded stall, so the pin has to end when the lease is dropped.
51		let t = TestEngine::new();
52
53		// Leased before the advance; acquiring after would be rejected as evicted, which is the
54		// overtaken signal rather than the pin under test.
55		let lagging = CommitVersion(50);
56		let lease = t.multi().acquire_version_lease(lagging).expect("leasing at the current head must succeed");
57
58		// A bare engine sits at version 0, so without a positive baseline there is nothing the
59		// lease could lower the cutoff below.
60		t.multi().advance_version_to(CommitVersion(100));
61
62		assert_eq!(
63			t.effective_gc_cutoff(),
64			lagging,
65			"a held lease must lower the historical-GC cutoff to the leased version"
66		);
67
68		drop(lease);
69		assert!(
70			t.effective_gc_cutoff().0 >= 100,
71			"with no lease held, nothing may pin the cutoff below the query watermark"
72		);
73	}
74}
reifydb_engine/watermark.rs

reifydb_engine/
watermark.rs
