Skip to content

Commit cc7bef1

Browse files
farhan-syahfarhan-syah
committed
feat(data): expose active_txn_overlays gauge for staged transaction overlays
Route every site that creates a per-transaction value/TTL or graph staging overlay through new txn_overlay_mut / graph_txn_overlay_mut choke points, so a live count of staged-but-uncommitted overlays can be tracked exactly. Decrement the gauge wherever overlays are torn down (DropTxnOverlay, Calvin synthetic overlay cleanup) and surface it as a Prometheus metric, giving visibility into abandoned-transaction overlay leaks.
1 parent f8f019d commit cc7bef1

19 files changed

Lines changed: 179 additions & 103 deletions

‎nodedb/src/control/metrics/prometheus/engines.rs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -185,6 +185,12 @@ impl SystemMetrics {
185185
"Per-core arena memory bytes",
186186
self.arena_memory_bytes.load(Ordering::Relaxed),
187187
);
188+
gauge(
189+
out,
190+
"nodedb_active_txn_overlays",
191+
"Live per-transaction staging overlays across all Data-Plane cores",
192+
self.active_txn_overlays.load(Ordering::Relaxed),
193+
);
188194

189195
// ── Contention ──
190196
counter(

‎nodedb/src/control/metrics/system/fields.rs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -105,6 +105,12 @@ pub struct SystemMetrics {
105105
pub io_uring_completions: AtomicU64,
106106
pub tpc_utilization_ratio: AtomicU64,
107107
pub arena_memory_bytes: AtomicU64,
108+
/// Current number of live per-transaction staging overlays across all
109+
/// Data-Plane cores (txn_overlays + graph_txn_overlays entries). A gauge:
110+
/// rises on first staged write of a transaction, falls when the overlay is
111+
/// dropped on COMMIT/ROLLBACK/teardown. A persistently non-zero idle value
112+
/// indicates leaked abandoned-transaction overlays.
113+
pub active_txn_overlays: AtomicU64,
108114

109115
// ── Contention ──
110116
pub mmap_major_faults: AtomicU64,

‎nodedb/src/data/executor/dispatch/meta.rs‎

Lines changed: 47 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -282,8 +282,14 @@ impl CoreLoop {
282282
// memtable) before this dispatches, so the empty-check fails and
283283
// the engine correctly stays registered with its committed rows.
284284
MetaOp::DropTxnOverlay { txn_id } => {
285-
self.txn_overlays.remove(txn_id);
286-
self.graph_txn_overlays.remove(txn_id);
285+
let removed = u64::from(self.txn_overlays.remove(txn_id).is_some())
286+
+ u64::from(self.graph_txn_overlays.remove(txn_id).is_some());
287+
if removed > 0
288+
&& let Some(m) = &self.metrics
289+
{
290+
m.active_txn_overlays
291+
.fetch_sub(removed, std::sync::atomic::Ordering::Relaxed);
292+
}
287293
if let Some(created) = self.txn_created_columnar_engines.remove(txn_id) {
288294
for engine_key in created {
289295
let still_empty = self
@@ -596,4 +602,43 @@ mod txn_created_columnar_engine_tests {
596602
"rolling back a staged insert into a pre-existing engine must not drop it"
597603
);
598604
}
605+
606+
#[test]
607+
fn active_txn_overlays_gauge_tracks_overlay_lifecycle() {
608+
let (mut core, _dir) = make_core();
609+
let metrics = std::sync::Arc::new(crate::control::metrics::SystemMetrics::new());
610+
core.metrics = Some(metrics.clone());
611+
let txn_id = TxnId::new(99);
612+
613+
let gauge = || {
614+
metrics
615+
.active_txn_overlays
616+
.load(std::sync::atomic::Ordering::Relaxed)
617+
};
618+
619+
// Idle: nothing staged, gauge sits at zero.
620+
assert_eq!(gauge(), 0, "gauge must start at zero");
621+
622+
// First materialization of the value/TTL overlay bumps the gauge to 1.
623+
let _ = core.txn_overlay_mut(txn_id);
624+
assert_eq!(gauge(), 1, "first overlay creation must bump the gauge");
625+
626+
// A second access to the SAME transaction's overlay must NOT double-count.
627+
let _ = core.txn_overlay_mut(txn_id);
628+
assert_eq!(gauge(), 1, "re-accessing an existing overlay must not bump");
629+
630+
// The parallel GRAPH overlay for the same txn is a distinct entry: 2.
631+
let _ = core.graph_txn_overlay_mut(txn_id);
632+
assert_eq!(gauge(), 2, "the graph overlay is a distinct tracked entry");
633+
634+
// DropTxnOverlay removes both entries and decrements by the exact count.
635+
let task = make_task();
636+
let resp = core.dispatch_meta(&task, TID, &MetaOp::DropTxnOverlay { txn_id });
637+
assert_eq!(resp.status, Status::Ok);
638+
assert_eq!(
639+
gauge(),
640+
0,
641+
"dropping both overlays must return the gauge to zero"
642+
);
643+
}
599644
}

‎nodedb/src/data/executor/handlers/control/calvin_overlay_stage.rs‎

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -176,15 +176,27 @@ impl CoreLoop {
176176
/// Idempotent: a missing key (already removed, or the id derivation
177177
/// itself failing) is a silent no-op — the same shape as the
178178
/// `commit_pending` removal it accompanies.
179+
///
180+
/// Mirrors `MetaOp::DropTxnOverlay`'s gauge accounting exactly: both maps
181+
/// were populated (if at all) via the `txn_overlay_mut` /
182+
/// `graph_txn_overlay_mut` choke points, which bump `active_txn_overlays`
183+
/// on first creation, so removal here must decrement by the same count
184+
/// or the gauge drifts upward forever on every Calvin-staged transaction.
179185
pub(in crate::data::executor) fn drop_calvin_synthetic_overlay(
180186
&mut self,
181187
epoch: u64,
182188
position: u32,
183189
vshard: u32,
184190
) {
185191
if let Ok(synthetic_txn_id) = calvin_synthetic_txn_id(epoch, position, vshard) {
186-
self.txn_overlays.remove(&synthetic_txn_id);
187-
self.graph_txn_overlays.remove(&synthetic_txn_id);
192+
let removed = u64::from(self.txn_overlays.remove(&synthetic_txn_id).is_some())
193+
+ u64::from(self.graph_txn_overlays.remove(&synthetic_txn_id).is_some());
194+
if removed > 0
195+
&& let Some(m) = &self.metrics
196+
{
197+
m.active_txn_overlays
198+
.fetch_sub(removed, std::sync::atomic::Ordering::Relaxed);
199+
}
188200
}
189201
}
190202
}

‎nodedb/src/data/executor/handlers/control/calvin_overlay_stage_bulk.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -86,7 +86,7 @@ impl CoreLoop {
8686
predicted_sorted.sort_unstable();
8787
let doc_ids = ollp_predicted_doc_ids(predicted);
8888

89-
let overlay = self.txn_overlays.entry(txn_id).or_default();
89+
let overlay = self.txn_overlay_mut(txn_id);
9090
for (surrogate, doc_id) in predicted_sorted.into_iter().zip(doc_ids) {
9191
overlay.insert_tombstone(coll_key.clone(), surrogate, &doc_id);
9292
}

‎nodedb/src/data/executor/handlers/transaction/mod.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@
22

33
mod batch;
44
pub mod overlay;
5+
mod overlay_gauge;
56
mod resolve;
67
pub(in crate::data::executor) mod stage_write;
78
mod sub_plan;

‎nodedb/src/data/executor/handlers/transaction/overlay_gauge.rs‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
// SPDX-License-Identifier: BUSL-1.1
2+
3+
//! Choke-point accessors for the per-transaction staging overlays.
4+
//!
5+
//! Every site that first materializes a transaction's overlay must go through
6+
//! `txn_overlay_mut` / `graph_txn_overlay_mut` so the shared
7+
//! `active_txn_overlays` gauge stays exact. The gauge is decremented in
8+
//! lockstep when the overlays are dropped (see `MetaOp::DropTxnOverlay`).
9+
10+
use std::sync::atomic::Ordering;
11+
12+
use crate::data::executor::core_loop::CoreLoop;
13+
use crate::data::executor::handlers::transaction::overlay::{GraphTxnOverlay, TxnOverlay};
14+
use crate::types::TxnId;
15+
16+
impl CoreLoop {
17+
/// Get-or-create this transaction's staging overlay, bumping the
18+
/// `active_txn_overlays` gauge on FIRST creation. The single creation choke
19+
/// point for `txn_overlays` — every staging site must go through here so the
20+
/// gauge stays exact.
21+
pub(in crate::data::executor) fn txn_overlay_mut(&mut self, txn_id: TxnId) -> &mut TxnOverlay {
22+
if !self.txn_overlays.contains_key(&txn_id)
23+
&& let Some(m) = &self.metrics
24+
{
25+
m.active_txn_overlays.fetch_add(1, Ordering::Relaxed);
26+
}
27+
self.txn_overlays.entry(txn_id).or_default()
28+
}
29+
30+
/// Get-or-create this transaction's GRAPH staging overlay, bumping the
31+
/// `active_txn_overlays` gauge on FIRST creation. The single creation choke
32+
/// point for `graph_txn_overlays` — every staging site must go through here
33+
/// so the gauge stays exact.
34+
pub(in crate::data::executor) fn graph_txn_overlay_mut(
35+
&mut self,
36+
txn_id: TxnId,
37+
) -> &mut GraphTxnOverlay {
38+
if !self.graph_txn_overlays.contains_key(&txn_id)
39+
&& let Some(m) = &self.metrics
40+
{
41+
m.active_txn_overlays.fetch_add(1, Ordering::Relaxed);
42+
}
43+
self.graph_txn_overlays.entry(txn_id).or_default()
44+
}
45+
}

‎nodedb/src/data/executor/handlers/transaction/resolve/entry.rs‎

Lines changed: 23 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -670,7 +670,7 @@ mod tests {
670670
// non-zero TTL leaves in the overlay: value + `ExpireAt`).
671671
let expire_at = 1_700_000_000_000u64;
672672
{
673-
let overlay = core.txn_overlays.entry(txn).or_default();
673+
let overlay = core.txn_overlay_mut(txn);
674674
overlay.insert_put(coll_key("sessions"), 7, &hex_key(b"s1"), b"v1".to_vec());
675675
overlay.set_ttl(
676676
coll_key("sessions"),
@@ -699,12 +699,8 @@ mod tests {
699699
let task = make_task();
700700
let txn = TxnId::new(3);
701701

702-
core.txn_overlays.entry(txn).or_default().insert_put(
703-
coll_key("kvc"),
704-
9,
705-
&hex_key(b"k9"),
706-
b"body".to_vec(),
707-
);
702+
core.txn_overlay_mut(txn)
703+
.insert_put(coll_key("kvc"), 9, &hex_key(b"k9"), b"body".to_vec());
708704

709705
let resp = core.execute_resolve_txn(&task, TID, txn, &[kv_write_plan("kvc")]);
710706
let redo = decode_redo(&resp);
@@ -732,11 +728,8 @@ mod tests {
732728
let task = make_task();
733729
let txn = TxnId::new(4);
734730

735-
core.txn_overlays.entry(txn).or_default().insert_tombstone(
736-
coll_key("kvc"),
737-
11,
738-
&hex_key(b"gone"),
739-
);
731+
core.txn_overlay_mut(txn)
732+
.insert_tombstone(coll_key("kvc"), 11, &hex_key(b"gone"));
740733

741734
let resp = core.execute_resolve_txn(
742735
&task,
@@ -782,7 +775,7 @@ mod tests {
782775
.get(DatabaseId::DEFAULT.as_u64(), TID, "kvc", b"k", now);
783776
assert_eq!(before.as_deref(), Some(b"base".as_slice()));
784777

785-
core.txn_overlays.entry(txn).or_default().insert_put(
778+
core.txn_overlay_mut(txn).insert_put(
786779
coll_key("kvc"),
787780
1,
788781
&hex_key(b"k"),
@@ -811,11 +804,8 @@ mod tests {
811804
// instead of raising the old typed error. (The RETURNING projection is a
812805
// response-shape concern the Control Plane already discards inside a
813806
// transaction; it leaves no separate post-image to preserve.)
814-
core.txn_overlays.entry(txn).or_default().insert_tombstone(
815-
coll_key("notes"),
816-
surrogate,
817-
"gone",
818-
);
807+
core.txn_overlay_mut(txn)
808+
.insert_tombstone(coll_key("notes"), surrogate, "gone");
819809

820810
let doc_plan = PhysicalPlan::Document(DocumentOp::PointDelete {
821811
collection: "notes".to_string(),
@@ -1028,7 +1018,7 @@ mod tests {
10281018
// Two rows staged per-surrogate exactly as the bulk-update staging path
10291019
// leaves them; a RETURNING clause does not change the overlay contents.
10301020
{
1031-
let overlay = src.txn_overlays.entry(txn).or_default();
1021+
let overlay = src.txn_overlay_mut(txn);
10321022
overlay.insert_put(coll_key("notes"), 1, "u1", schemaless_body("bob"));
10331023
overlay.insert_put(coll_key("notes"), 2, "u2", schemaless_body("bob"));
10341024
}
@@ -1216,7 +1206,7 @@ mod tests {
12161206
let surrogate = 7u32;
12171207
let row_key = surrogate_to_doc_id(Surrogate::new(surrogate));
12181208

1219-
src.txn_overlays.entry(txn).or_default().insert_put(
1209+
src.txn_overlay_mut(txn).insert_put(
12201210
coll_key("sdocs"),
12211211
surrogate,
12221212
"row1",
@@ -1266,12 +1256,8 @@ mod tests {
12661256
let row_key = surrogate_to_doc_id(Surrogate::new(surrogate));
12671257
let body = schemaless_body("alice");
12681258

1269-
src.txn_overlays.entry(txn).or_default().insert_put(
1270-
coll_key("notes"),
1271-
surrogate,
1272-
"userpk",
1273-
body.clone(),
1274-
);
1259+
src.txn_overlay_mut(txn)
1260+
.insert_put(coll_key("notes"), surrogate, "userpk", body.clone());
12751261

12761262
let resp = src.execute_resolve_txn(&task, TID, txn, &[doc_put_plan("notes")]);
12771263
let redo = decode_redo(&resp);
@@ -1301,11 +1287,8 @@ mod tests {
13011287
let surrogate = 11u32;
13021288
let row_key = surrogate_to_doc_id(Surrogate::new(surrogate));
13031289

1304-
src.txn_overlays.entry(txn).or_default().insert_tombstone(
1305-
coll_key("notes"),
1306-
surrogate,
1307-
"gone",
1308-
);
1290+
src.txn_overlay_mut(txn)
1291+
.insert_tombstone(coll_key("notes"), surrogate, "gone");
13091292

13101293
let delete_plan = PhysicalPlan::Document(DocumentOp::PointDelete {
13111294
collection: "notes".to_string(),
@@ -1424,7 +1407,7 @@ mod tests {
14241407
.expect("get");
14251408
assert_eq!(before.as_deref(), Some(schemaless_body("base").as_slice()));
14261409

1427-
core.txn_overlays.entry(txn).or_default().insert_put(
1410+
core.txn_overlay_mut(txn).insert_put(
14281411
coll_key("notes"),
14291412
surrogate,
14301413
"userpk",
@@ -1455,7 +1438,7 @@ mod tests {
14551438
let doc_row_key = surrogate_to_doc_id(Surrogate::new(doc_surrogate));
14561439

14571440
{
1458-
let overlay = src.txn_overlays.entry(txn).or_default();
1441+
let overlay = src.txn_overlay_mut(txn);
14591442
overlay.insert_put(coll_key("kvc"), 1, &hex_key(b"k"), b"V".to_vec());
14601443
overlay.insert_put(
14611444
coll_key("notes"),
@@ -1512,7 +1495,7 @@ mod tests {
15121495

15131496
let expire_at = crate::engine::kv::current_ms() + 3_600_000;
15141497
{
1515-
let overlay = src.txn_overlays.entry(txn).or_default();
1498+
let overlay = src.txn_overlay_mut(txn);
15161499
overlay.insert_put(coll_key("kvc"), 1, &hex_key(b"live"), b"V".to_vec());
15171500
overlay.set_ttl(
15181501
coll_key("kvc"),
@@ -1861,7 +1844,7 @@ mod tests {
18611844
let doc_row_key = surrogate_to_doc_id(Surrogate::new(doc_surrogate));
18621845

18631846
{
1864-
let overlay = src.txn_overlays.entry(txn).or_default();
1847+
let overlay = src.txn_overlay_mut(txn);
18651848
overlay.insert_put(
18661849
coll_key("notes"),
18671850
doc_surrogate,
@@ -1966,7 +1949,7 @@ mod tests {
19661949
let txn = TxnId::new(38);
19671950

19681951
{
1969-
let overlay = src.graph_txn_overlays.entry(txn).or_default();
1952+
let overlay = src.graph_txn_overlay_mut(txn);
19701953
overlay.stage_edge_put(coll_key("g"), "c", "l", "z", vec![]);
19711954
overlay.stage_edge_put(coll_key("g"), "a", "l", "x", vec![]);
19721955
overlay.stage_edge_put(coll_key("g"), "b", "l", "y", vec![]);
@@ -2349,12 +2332,8 @@ mod tests {
23492332
let txn = TxnId::new(47);
23502333

23512334
// Stage the KV write into the overlay (overlay-driven serializer).
2352-
src.txn_overlays.entry(txn).or_default().insert_put(
2353-
coll_key("kvc"),
2354-
1,
2355-
&hex_key(b"k"),
2356-
b"V".to_vec(),
2357-
);
2335+
src.txn_overlay_mut(txn)
2336+
.insert_put(coll_key("kvc"), 1, &hex_key(b"k"), b"V".to_vec());
23582337

23592338
let mut row = std::collections::HashMap::new();
23602339
row.insert("a".to_string(), nodedb_types::Value::Integer(7));
@@ -2698,12 +2677,8 @@ mod tests {
26982677
let txn = TxnId::new(44);
26992678
let surrogate = 13u32;
27002679

2701-
src.txn_overlays.entry(txn).or_default().insert_put(
2702-
coll_key("kvc"),
2703-
1,
2704-
&hex_key(b"k"),
2705-
b"V".to_vec(),
2706-
);
2680+
src.txn_overlay_mut(txn)
2681+
.insert_put(coll_key("kvc"), 1, &hex_key(b"k"), b"V".to_vec());
27072682

27082683
let plans = [
27092684
kv_write_plan("kvc"),

‎nodedb/src/data/executor/handlers/transaction/stage_write/dispatch.rs‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -333,7 +333,7 @@ impl CoreLoop {
333333
limit: MAX_TXN_OVERLAY_BYTES,
334334
});
335335
}
336-
self.txn_overlays.entry(ctx.txn_id).or_default().insert_put(
336+
self.txn_overlay_mut(ctx.txn_id).insert_put(
337337
ctx.coll_key.clone(),
338338
ctx.surrogate.0,
339339
&ctx.document_id,

‎nodedb/src/data/executor/handlers/transaction/stage_write/stage_bulk_delete.rs‎

Lines changed: 1 addition & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -89,9 +89,7 @@ impl CoreLoop {
8989
let Ok(surrogate) = u32::from_str_radix(row_key, 16) else {
9090
continue;
9191
};
92-
self.txn_overlays
93-
.entry(txn_id)
94-
.or_default()
92+
self.txn_overlay_mut(txn_id)
9593
.insert_tombstone(coll_key.clone(), surrogate, row_key);
9694
affected += 1;
9795
}

0 commit comments

Comments
 (0)