1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
//! Transaction Cleaner
//!
//! Provides cleanup functionality for expired and stale transactions
use std::collections::HashSet;
use std::sync::Arc;
use dashmap::DashMap;
use super::mvcc::VersionManager;
use crate::context::TransactionContext;
use crate::error::TransactionError;
use crate::types::{TransactionId, TransactionStats};
use linkrs_core::types::Timestamp;
/// Transaction Cleaner
///
/// Provides cleanup functionality for expired and stale transactions
///
/// Expiry detection and timestamp reaping live here; the abort itself always
/// runs through the manager's canonical abort protocol (via the callback in
/// `cleanup_expired_transactions_with`), so abort semantics cannot diverge
/// between cleanup and explicit rollback paths.
pub struct TransactionCleaner {
version_manager: Arc<VersionManager>,
stats: Arc<TransactionStats>,
}
impl TransactionCleaner {
pub fn new(version_manager: Arc<VersionManager>, stats: Arc<TransactionStats>) -> Self {
Self {
version_manager,
stats,
}
}
/// Cleanup expired transactions, delegating the abort to the provided
/// callback.
///
/// This is the manager-facing cleanup path. It reaps orphaned write
/// timestamps whose owning path vanished, then aborts every expired or
/// idle-timeout transaction through the callback so the caller keeps its
/// canonical abort semantics (full protocol, sink, events).
pub fn cleanup_expired_transactions_with<F>(
&self,
active_transactions: &DashMap<TransactionId, Arc<TransactionContext>>,
mut abort: F,
) where
F: FnMut(TransactionId) -> Result<(), TransactionError>,
{
// Safety net: reap write timestamps whose owning path vanished (orphaned
// write). Timestamps owned by live write transactions are excluded so a
// transaction still within its configured timeout is never reaped.
let owned: HashSet<Timestamp> = active_transactions
.iter()
.filter(|entry| !entry.value().read_only)
.map(|entry| entry.value().timestamp())
.collect();
let reaped = self
.version_manager
.reap_expired_write_timestamps(self.version_manager.write_reap_timeout(), &owned);
if reaped > 0 {
log::warn!("Reaped {reaped} orphaned write timestamp(s) older than timeout");
}
// A live long-running transaction is protected from reaping by the
// `owned` set above, but its `Pending` slot still pins the read
// frontier behind it. Report it separately so operators can tell
// "frontier stalled by a live transaction" apart from "orphan reaped".
let warn_after = self.version_manager.write_reap_timeout() / 2;
for (ts, age) in self.version_manager.pending_write_ages() {
if owned.contains(&ts) && age > warn_after {
log::warn!(
"Long-running write transaction holds timestamp {ts} for {age:?}; \
read frontier cannot advance past it until commit or abort"
);
}
}
let expired: Vec<(TransactionId, bool)> = active_transactions
.iter()
.filter(|entry| {
entry.value().state().can_execute()
&& (entry.value().is_expired() || entry.value().is_idle_timeout())
})
.map(|entry| (*entry.key(), entry.value().is_expired()))
.collect();
for (txn_id, timed_out) in expired {
if timed_out {
self.stats.record_timeout();
}
if let Err(error) = abort(txn_id) {
log::error!(
"Transaction {} could not complete the cleanup protocol: {}",
txn_id,
error
);
self.stats.increment_cleanup_failure();
}
}
}
}
impl Default for TransactionCleaner {
fn default() -> Self {
Self::new(
Arc::new(VersionManager::new()),
Arc::new(TransactionStats::new()),
)
}
}