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;
11use tracing::error;
12
13use crate::engine::StandardEngine;
14
15impl QueryWatermark for StandardEngine {
16	fn effective_gc_cutoff(&self) -> CommitVersion {
17		let qdu = self.query_done_until();
18		let lease_min = self.multi().leases().min_active().unwrap_or(CommitVersion(u64::MAX));
19		qdu.min(lease_min)
20	}
21}
22
23impl EvictionWatermark for StandardEngine {
24	fn watermark(&self) -> CommitVersion {
25		self.effective_gc_cutoff().min(self.consumer_watermark())
26	}
27}
28
29impl StandardEngine {
30	pub fn consumer_watermark(&self) -> CommitVersion {
31		let mut txn = match self.begin_query(IdentityId::system()) {
32			Ok(txn) => txn,
33			Err(err) => {
34				error!(error = %err, "cannot begin a query for the pinning watermark, holding cdc retention at version zero");
35				return CommitVersion(0);
36			}
37		};
38		match compute_pinning_watermark(&mut Transaction::Query(&mut txn), Some(&*self.checkpoint_floor())) {
39			Ok(Some(v)) => v,
40			Ok(None) => CommitVersion(u64::MAX),
41			Err(err) => {
42				error!(error = %err, "pinning watermark unavailable, holding cdc retention at version zero");
43				CommitVersion(0)
44			}
45		}
46	}
47}
48
49#[cfg(test)]
50mod tests {
51	use reifydb_core::{common::CommitVersion, lifecycle::watermark::QueryWatermark};
52	use reifydb_runtime::shutdown::Shutdown;
53	use reifydb_test_harness::engine::TestEngine;
54
55	#[test]
56	fn a_closed_operator_store_holds_the_consumer_watermark_at_zero_instead_of_releasing_retention() {
57		// an unreadable floor must never read as "nothing to pin". u64::MAX is the no-pin answer and
58		// it lets cdc reap every version the flows still need, so a closed store has to invert it to 0.
59		let t = TestEngine::new();
60
61		assert_eq!(
62			t.inner().consumer_watermark(),
63			CommitVersion(u64::MAX),
64			"a healthy store with no checkpoint rows pins nothing, which is what makes the closed case \
65			 below a real inversion rather than a no-op"
66		);
67
68		t.inner().operator_state().shutdown().unwrap();
69
70		assert_eq!(
71			t.inner().consumer_watermark(),
72			CommitVersion(0),
73			"once the floor cannot be read the engine must hold retention at zero; answering u64::MAX \
74			 here releases every cdc entry the flows have not replayed yet"
75		);
76	}
77
78	#[test]
79	fn effective_gc_cutoff_is_lowered_by_a_held_lease_and_nothing_else() {
80		// Only a held lease may pin the cutoff. A lagging consumer without one pinning it is
81		// an unbounded stall, so the pin has to end when the lease is dropped.
82		let t = TestEngine::new();
83
84		// Leased before the advance; acquiring after would be rejected as evicted, which is the
85		// overtaken signal rather than the pin under test.
86		let lagging = CommitVersion(50);
87		let lease = t.multi().acquire_version_lease(lagging).expect("leasing at the current head must succeed");
88
89		// A bare engine sits at version 0, so without a positive baseline there is nothing the
90		// lease could lower the cutoff below.
91		t.multi().advance_version_to(CommitVersion(100));
92
93		assert_eq!(
94			t.effective_gc_cutoff(),
95			lagging,
96			"a held lease must lower the historical-GC cutoff to the leased version"
97		);
98
99		drop(lease);
100		assert!(
101			t.effective_gc_cutoff().0 >= 100,
102			"with no lease held, nothing may pin the cutoff below the query watermark"
103		);
104	}
105}
reifydb_engine/watermark.rs

reifydb_engine/
watermark.rs
