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
//! Transaction monitoring and statistics
use std::time::Duration;
use super::TransactionManager;
use crate::error::TransactionError;
use crate::types::*;
impl TransactionManager {
/// Get active transaction list
pub fn list_active_transactions(&self) -> Vec<TransactionInfo> {
self.monitor
.list_active_transactions(&self.active_transactions)
}
/// Get transaction info
pub fn get_transaction_info(&self, txn_id: TransactionId) -> Option<TransactionInfo> {
self.monitor
.get_transaction_info(&self.active_transactions, txn_id)
}
/// List active and recovery-required transactions for administration.
pub fn list_transactions(&self) -> Vec<TransactionInfo> {
self.monitor.list_transactions(&self.active_transactions)
}
/// Retry delivery of pending synchronization outbox entries.
pub fn retry_outbox_projection(&self) -> Result<usize, TransactionError> {
self.sync_manager
.as_ref()
.ok_or_else(|| TransactionError::sync_failed("Sync manager is not configured"))?
.retry_outbox_sync()
.map_err(|error| TransactionError::sync_failed(error.to_string()))
}
/// Recover transactions whose data was durably persisted but whose
/// post-commit finalization (commit_sink finalize or undo-log cleanup)
/// left them in an incomplete state.
///
/// Called once at startup (after WAL replay) and can also be invoked
/// on demand by an administrator.
///
/// Recovery runs in two stages: first, same-process pendings that still
/// have a live transaction context are re-driven through
/// `recover_pending_finalization` (retrying storage finalization,
/// allocating the commit timestamp, and completing the commit); then the
/// remaining queue plus any sidecar records are accounted for at the
/// storage level via `RecoveryManager::recover`.
///
/// Returns the number of recovered commits.
pub fn startup_recovery(&self) -> Result<usize, TransactionError> {
let mut recovered = 0usize;
for txn_id in self.recovery.pending_txn_ids() {
match self.recover_pending_finalization(txn_id) {
Ok(()) => recovered += 1,
Err(error) => {
log::warn!(
"startup recovery could not re-drive transaction {}: {}; \
left queued for the next attempt",
txn_id,
error
);
}
}
}
recovered += self.recovery.recover(self.commit_sink.as_deref())?;
Ok(recovered)
}
/// Mirror the recovery pending queue to a sidecar file so
/// unfinalized-commit intent survives a process crash.
pub fn set_recovery_sidecar_path(&self, path: impl Into<std::path::PathBuf>) {
self.recovery.set_sidecar_path(path);
}
/// Get statistics
pub fn stats(&self) -> &TransactionStats {
self.monitor.stats()
}
/// Return resource gauges used by monitoring and administrative tooling.
pub fn resource_metrics(&self) -> TransactionResourceMetrics {
let mut staged_wal_bytes = 0;
let mut undo_bytes = 0;
for entry in self.active_transactions.iter() {
let context = entry.value();
staged_wal_bytes += context.staged_bytes();
undo_bytes += context.undo_log_len() as u64;
}
let metrics = TransactionResourceMetrics {
active_snapshots: self.version_manager.snapshot_tracker().active_count() as u64,
pending_writes: self.version_manager.pending_count(),
committed_frontier_lag: self
.version_manager
.write_timestamp()
.saturating_sub(self.version_manager.read_timestamp()),
staged_wal_bytes,
undo_bytes,
checkpoint_drain_time: Duration::ZERO,
};
metrics
}
}