reifydb_engine/
watermark.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_cdc::consume::watermark::compute_watermark;
5use reifydb_core::common::CommitVersion;
6use reifydb_store_multi::gc::{EvictionWatermark, historical::QueryWatermark};
7use reifydb_transaction::transaction::Transaction;
8use reifydb_value::value::identity::IdentityId;
9
10use crate::engine::StandardEngine;
11
12impl QueryWatermark for StandardEngine {
13	fn effective_gc_cutoff(&self) -> CommitVersion {
14		let qdu = self.query_done_until();
15		let lease_min = self.multi().leases().min_active().unwrap_or(CommitVersion(u64::MAX));
16		qdu.min(lease_min).min(self.multi().consumer_watermark())
17	}
18}
19
20impl EvictionWatermark for StandardEngine {
21	fn watermark(&self) -> CommitVersion {
22		self.effective_gc_cutoff().min(self.consumer_watermark())
23	}
24}
25
26impl StandardEngine {
27	pub fn consumer_watermark(&self) -> CommitVersion {
28		let mut txn = match self.begin_query(IdentityId::system()) {
29			Ok(txn) => txn,
30			Err(_) => return CommitVersion(0),
31		};
32		match compute_watermark(&mut Transaction::Query(&mut txn)) {
33			Ok(Some(v)) => v,
34			Ok(None) => CommitVersion(u64::MAX),
35			Err(_) => CommitVersion(0),
36		}
37	}
38}
39
40#[cfg(test)]
41mod tests {
42	use reifydb_core::common::CommitVersion;
43	use reifydb_store_multi::gc::historical::QueryWatermark;
44
45	use crate::test_harness::TestEngine;
46
47	// The historical-GC cutoff must be lowered by the CDC consumer watermark so the version store
48	// retains snapshots a lagging consumer (e.g. a subscription worker draining a backlog) still
49	// needs to read. This is the storage half of the TXN_012 fix; the lease-acquire half is covered
50	// in the transaction crate's write tests.
51	#[test]
52	fn effective_gc_cutoff_is_lowered_by_the_consumer_watermark() {
53		let t = TestEngine::new();
54
55		// Advance the query watermark to a known positive baseline. A bare engine sits at version 0,
56		// so without this the cutoff would be 0 and there would be nothing to lower below.
57		t.multi().advance_version_to(CommitVersion(100));
58
59		// Nothing pins the consumer term by default (u64::MAX), so the cutoff is just the query/lease
60		// watermark.
61		let baseline = t.effective_gc_cutoff();
62		assert!(baseline.0 >= 100, "precondition: the query watermark is advanced to the baseline");
63
64		// A consumer lagging just below the baseline must pull the cutoff down to its own position so
65		// the version it has not consumed yet is retained.
66		let lagging = CommitVersion(baseline.0 - 1);
67		t.multi().set_consumer_watermark(lagging);
68		assert_eq!(
69			t.effective_gc_cutoff(),
70			lagging,
71			"the consumer watermark must lower the historical-GC cutoff to the consumer position"
72		);
73
74		// Restoring the term to u64::MAX makes it inert: the cutoff returns to the query watermark.
75		// This guards against a 0 default, which would have pinned all history.
76		t.multi().set_consumer_watermark(CommitVersion(u64::MAX));
77		assert!(
78			t.effective_gc_cutoff().0 >= baseline.0,
79			"a u64::MAX consumer watermark must not pin the cutoff below the query watermark"
80		);
81	}
82}
reifydb_engine/watermark.rs

reifydb_engine/
watermark.rs
