|
| 1 | +// SPDX-License-Identifier: BUSL-1.1 |
| 2 | + |
| 3 | +//! Regression test: an abandoned NATIVE-protocol transaction's Data-Plane |
| 4 | +//! staging overlays (`txn_overlays` / `graph_txn_overlays`, keyed by |
| 5 | +//! `txn_id`) leaked forever if the connection ended (abrupt disconnect, |
| 6 | +//! idle/absolute timeout, or clean EOF) without a `COMMIT`/`ROLLBACK` -- |
| 7 | +//! nothing dispatched `MetaOp::DropTxnOverlay` for it. `NativeSession::run` |
| 8 | +//! now wraps the frame loop and reclaims any still-open transaction's |
| 9 | +//! overlay via `lifecycle::run_rollback` on every exit path. |
| 10 | +//! |
| 11 | +//! This test proves reclamation (rather than just "row not visible", which |
| 12 | +//! cannot distinguish reclaimed from leaked-but-invisible) via the |
| 13 | +//! `active_txn_overlays` gauge on `SystemMetrics`: it rises when the |
| 14 | +//! transaction's first staged write materializes the overlay, and must |
| 15 | +//! fall back to baseline once the abandoned connection is torn down. |
| 16 | +
|
| 17 | +mod common; |
| 18 | + |
| 19 | +use std::sync::atomic::Ordering; |
| 20 | +use std::time::{Duration, Instant}; |
| 21 | + |
| 22 | +use common::native_harness::{NativeTestServer, do_handshake, send_sql}; |
| 23 | + |
| 24 | +use nodedb_types::protocol::HelloFrame; |
| 25 | +use nodedb_types::protocol::opcodes::ResponseStatus; |
| 26 | + |
| 27 | +/// Poll `active_txn_overlays` until `pred` is satisfied or `deadline` elapses. |
| 28 | +/// Returns the last observed value; callers assert against it so a timeout |
| 29 | +/// panics with the actual value instead of a bare "timed out". |
| 30 | +async fn poll_gauge( |
| 31 | + server: &NativeTestServer, |
| 32 | + deadline: Duration, |
| 33 | + pred: impl Fn(u64) -> bool, |
| 34 | +) -> u64 { |
| 35 | + let metrics = server |
| 36 | + .shared |
| 37 | + .system_metrics |
| 38 | + .as_ref() |
| 39 | + .expect("system_metrics must be wired into the native harness"); |
| 40 | + let start = Instant::now(); |
| 41 | + loop { |
| 42 | + let value = metrics.active_txn_overlays.load(Ordering::Relaxed); |
| 43 | + if pred(value) || start.elapsed() >= deadline { |
| 44 | + return value; |
| 45 | + } |
| 46 | + tokio::time::sleep(Duration::from_millis(20)).await; |
| 47 | + } |
| 48 | +} |
| 49 | + |
| 50 | +#[tokio::test] |
| 51 | +async fn native_abandoned_txn_overlay_reclaimed_on_teardown() { |
| 52 | + let server = NativeTestServer::start().await; |
| 53 | + |
| 54 | + let baseline = server |
| 55 | + .shared |
| 56 | + .system_metrics |
| 57 | + .as_ref() |
| 58 | + .expect("system_metrics must be wired into the native harness") |
| 59 | + .active_txn_overlays |
| 60 | + .load(Ordering::Relaxed); |
| 61 | + assert_eq!(baseline, 0, "gauge must start at zero: {baseline}"); |
| 62 | + |
| 63 | + { |
| 64 | + let (mut stream, _ack) = do_handshake(server.addr, &HelloFrame::current()) |
| 65 | + .await |
| 66 | + .expect("handshake"); |
| 67 | + |
| 68 | + let create_resp = send_sql( |
| 69 | + &mut stream, |
| 70 | + 1, |
| 71 | + "CREATE COLLECTION native_txn_overlay_teardown (id STRING PRIMARY KEY, n INT) \ |
| 72 | + WITH (engine='document_schemaless')", |
| 73 | + ) |
| 74 | + .await; |
| 75 | + assert_ne!( |
| 76 | + create_resp.status, |
| 77 | + ResponseStatus::Error, |
| 78 | + "CREATE must succeed: {create_resp:?}" |
| 79 | + ); |
| 80 | + |
| 81 | + let begin_resp = send_sql(&mut stream, 2, "BEGIN").await; |
| 82 | + assert_ne!( |
| 83 | + begin_resp.status, |
| 84 | + ResponseStatus::Error, |
| 85 | + "BEGIN must succeed: {begin_resp:?}" |
| 86 | + ); |
| 87 | + |
| 88 | + let staged_insert = send_sql( |
| 89 | + &mut stream, |
| 90 | + 3, |
| 91 | + "INSERT INTO native_txn_overlay_teardown (id, n) VALUES ('a', 1)", |
| 92 | + ) |
| 93 | + .await; |
| 94 | + assert_ne!( |
| 95 | + staged_insert.status, |
| 96 | + ResponseStatus::Error, |
| 97 | + "in-tx INSERT must succeed: {staged_insert:?}" |
| 98 | + ); |
| 99 | + |
| 100 | + // The staged write must have materialized the overlay, bumping the |
| 101 | + // gauge above baseline -- confirms there is something to reclaim. |
| 102 | + let after_stage = poll_gauge(&server, Duration::from_secs(5), |v| v > baseline).await; |
| 103 | + assert!( |
| 104 | + after_stage > baseline, |
| 105 | + "staged write must raise active_txn_overlays above baseline {baseline}, \ |
| 106 | + got {after_stage}" |
| 107 | + ); |
| 108 | + |
| 109 | + // Abruptly abandon the connection -- no COMMIT/ROLLBACK. Dropping the |
| 110 | + // raw TcpStream (returned by `do_handshake`) closes the socket, |
| 111 | + // triggering the server-side EOF/error teardown path. |
| 112 | + drop(stream); |
| 113 | + } |
| 114 | + |
| 115 | + // The abandoned transaction's overlay must be reclaimed on teardown, |
| 116 | + // bringing the gauge back down to baseline. |
| 117 | + let after_teardown = poll_gauge(&server, Duration::from_secs(5), |v| v == baseline).await; |
| 118 | + assert_eq!( |
| 119 | + after_teardown, baseline, |
| 120 | + "abandoned txn overlay must be reclaimed on connection teardown, \ |
| 121 | + active_txn_overlays still {after_teardown} (baseline {baseline})" |
| 122 | + ); |
| 123 | + |
| 124 | + // Belt-and-suspenders: a fresh autocommit connection must not see the |
| 125 | + // staged row -- it was never committed. |
| 126 | + let (mut stream2, _ack2) = do_handshake(server.addr, &HelloFrame::current()) |
| 127 | + .await |
| 128 | + .expect("handshake"); |
| 129 | + let check = send_sql( |
| 130 | + &mut stream2, |
| 131 | + 1, |
| 132 | + "SELECT * FROM native_txn_overlay_teardown WHERE id = 'a'", |
| 133 | + ) |
| 134 | + .await; |
| 135 | + assert_ne!( |
| 136 | + check.status, |
| 137 | + ResponseStatus::Error, |
| 138 | + "post-teardown SELECT must succeed: {check:?}" |
| 139 | + ); |
| 140 | + assert_eq!( |
| 141 | + check.rows.as_ref().map(Vec::len).unwrap_or(0), |
| 142 | + 0, |
| 143 | + "the never-committed staged row must not be visible: {check:?}" |
| 144 | + ); |
| 145 | + |
| 146 | + server.shutdown().await; |
| 147 | +} |
0 commit comments