From 3f135c75177aabae8cd1f3de0198db503a0eac26 Mon Sep 17 00:00:00 2001 From: Jepson Taylor <130782313+jepsontaylor@users.noreply.github.com> Date: Fri, 25 Sep 2026 20:07:00 +0000 Subject: [PATCH] Release row locks when an open transaction is dropped. Commit and rollback already drain those locks before the transaction closes. Drop now aborts an unclosed transaction and unlocks anything still held, including when commit returns before that drain. Record bytes and durability are unchanged. Claim: AGENT_CHAT.md 2026-09-25T20:05Z fix/txn-drop-row-locks. Acceptance: cargo test -p redlinedb-kernel --offline --test engine_tests row_lock_is_released --- AGENT_CHAT.md | 5 ++++ crates/kernel/src/engine/mod.rs | 2 +- crates/kernel/src/engine/recovery.rs | 4 ++-- crates/kernel/src/engine/runtime.rs | 4 +++- crates/kernel/src/engine/tx.rs | 36 +++++++++++++++++++++++++--- crates/kernel/tests/engine_tests.rs | 32 +++++++++++++++++++++++++ 6 files changed, 76 insertions(+), 7 deletions(-) diff --git a/AGENT_CHAT.md b/AGENT_CHAT.md index f287c0b03..9c3562b1b 100644 --- a/AGENT_CHAT.md +++ b/AGENT_CHAT.md @@ -1,5 +1,10 @@ # RedlineDB Agent Chat +## 2026-09-25T20:05Z fix/txn-drop-row-locks + +Claim: canonical checkout `/home/ubuntu/redlineDB`, branch `fix/txn-drop-row-locks` from `origin/main` `a58bbb649f893b0ed00c6a4a0c46e44358d5bcb1`. Dropping an open transaction aborts it and releases the row locks it still holds. Commit and rollback already release those locks before the transaction closes, and that order stays. No speed ratio. No worktree. + + ## 2026-09-25T19:40Z perf/positional-wal-batch Claim: canonical checkout `/home/ubuntu/redlineDB`, branch `perf/positional-wal-batch` from `origin/main` `4c622a4fad6211133cd2bcd962375bca38170d0d`. Page and WAL bytes use positional read/write. The WAL writer emits one write for a contiguous run of records and splits the run at a segment boundary. WAL record bytes and durability are unchanged. `wal_pipeline` stays off. No speed ratio. No worktree. diff --git a/crates/kernel/src/engine/mod.rs b/crates/kernel/src/engine/mod.rs index d8b2ea67a..084e6aee5 100644 --- a/crates/kernel/src/engine/mod.rs +++ b/crates/kernel/src/engine/mod.rs @@ -191,7 +191,7 @@ pub struct Engine { heap: PageBackedHeap, catalog: CatalogManager, catalog_store: CatalogStore, - locks: RowLockManager, + locks: Arc, wal: Arc, /// Phase 11 Wave 0: engine-level aggregator for the new /// telemetry counters. Lives next to `wal` because it is the diff --git a/crates/kernel/src/engine/recovery.rs b/crates/kernel/src/engine/recovery.rs index 376e2c5d5..08f190a98 100644 --- a/crates/kernel/src/engine/recovery.rs +++ b/crates/kernel/src/engine/recovery.rs @@ -113,7 +113,7 @@ impl Engine { } let buffer = Arc::clone(&buffer); let phase11_counters = Arc::new(Phase11Counters::default()); - let locks = RowLockManager::new(config.lock_shards, config.busy_timeout); + let locks = Arc::new(RowLockManager::new(config.lock_shards, config.busy_timeout)); // Wave 1A-F: pipe Phase11 telemetry into the row-lock manager so // contended-acquire waits land in `lock_wait_us_buckets`. Same // Arc piped into the WAL coordinator so `wal_batch_size_buckets` @@ -266,7 +266,7 @@ impl Engine { let recovered_catalog = recover_catalog_snapshot(&scan_report.records, target)?; let catalog = CatalogManager::new(recovered_catalog.unwrap_or(initial_catalog)); let phase11_counters = Arc::new(Phase11Counters::default()); - let locks = RowLockManager::new(config.lock_shards, config.busy_timeout); + let locks = Arc::new(RowLockManager::new(config.lock_shards, config.busy_timeout)); // Wave 1A-F: same telemetry pipe on the open path. locks.set_phase11_counters(Arc::clone(&phase11_counters)); wal.set_phase11_counters(Arc::clone(&phase11_counters)); diff --git a/crates/kernel/src/engine/runtime.rs b/crates/kernel/src/engine/runtime.rs index 21481cc57..1c6c83fa5 100644 --- a/crates/kernel/src/engine/runtime.rs +++ b/crates/kernel/src/engine/runtime.rs @@ -33,7 +33,9 @@ impl Engine { if isolation == Isolation::Serializable { return Err(Error::UnsupportedIsolation); } - Ok(self.txs.begin_txn(isolation)) + let mut tx = self.txs.begin_txn(isolation); + tx.attach_row_lock_manager(Arc::clone(&self.locks)); + Ok(tx) } pub fn reserve_begin_lock(&self, tx: &mut Txn) -> Result<()> { diff --git a/crates/kernel/src/engine/tx.rs b/crates/kernel/src/engine/tx.rs index fcc2dedd2..9d41e6e0f 100644 --- a/crates/kernel/src/engine/tx.rs +++ b/crates/kernel/src/engine/tx.rs @@ -1,6 +1,7 @@ use crate::catalog::IndexId as CatalogIndexId; use crate::catalog::SchemaSnapshot; use crate::engine::lock::RowKey; +use crate::engine::lock::RowLockManager; use crate::format::{Csn, TxId}; use crate::index::BtreeIndex; use crate::txn::{Isolation, Snapshot}; @@ -25,6 +26,9 @@ pub struct Txn { row_locks: Vec, /// Membership for `has_row_lock`. The vec alone is O(n) per probe. row_lock_set: HashSet, + /// Present for engine transactions so `Drop` can release locks that + /// commit and rollback did not already drain. + row_lock_manager: Option>, open: bool, lifecycle: Option>, /// Lane A5-triggers: depth counter for nested trigger fires. The SQL @@ -56,6 +60,7 @@ impl Txn { pending_index_handles: Vec::new(), row_locks: Vec::new(), row_lock_set: HashSet::new(), + row_lock_manager: None, open: true, lifecycle: Some(lifecycle), trigger_depth: 0, @@ -147,6 +152,27 @@ impl Txn { self.row_lock_set.clear(); self.row_locks.drain(..) } + + pub(crate) fn attach_row_lock_manager(&mut self, locks: Arc) { + self.row_lock_manager = Some(locks); + } +} + +impl Drop for Txn { + fn drop(&mut self) { + // Abort before unlock so a waiter does not observe this transaction + // as still in progress. Commit and rollback already closed it. + if let Some(lifecycle) = &self.lifecycle { + lifecycle.abort_if_open(); + } + let Some(locks) = self.row_lock_manager.take() else { + return; + }; + let tx_id = self.id; + for key in self.drain_row_locks() { + locks.unlock(key.rel_id, key.row_id, tx_id); + } + } } #[derive(Debug)] @@ -170,10 +196,8 @@ impl TxnLifecycle { inner.unregister_active(self.tx_id); } } -} -impl Drop for TxnLifecycle { - fn drop(&mut self) { + fn abort_if_open(&self) { if !self.closed.swap(true, Ordering::SeqCst) && let Some(inner) = self.inner.upgrade() { @@ -183,6 +207,12 @@ impl Drop for TxnLifecycle { } } +impl Drop for TxnLifecycle { + fn drop(&mut self) { + self.abort_if_open(); + } +} + #[cfg(test)] mod tests { use super::*; diff --git a/crates/kernel/tests/engine_tests.rs b/crates/kernel/tests/engine_tests.rs index 62ccf8ab6..25c37df55 100644 --- a/crates/kernel/tests/engine_tests.rs +++ b/crates/kernel/tests/engine_tests.rs @@ -1112,3 +1112,35 @@ fn ddl_rollback_discards_catalog_changes_and_epoch_validation_detects_stale_snap Err(Error::SchemaChanged) ); } + +#[test] +fn row_lock_is_released_on_drop_rollback_and_commit() { + let (_temp, engine) = test_engine(); + engine.set_busy_timeout(Duration::from_millis(200)); + + let mut holder = engine.begin(Isolation::Snapshot).unwrap(); + engine.reserve_begin_lock(&mut holder).unwrap(); + drop(holder); + let mut next = engine.begin(Isolation::Snapshot).unwrap(); + engine + .reserve_begin_lock(&mut next) + .expect("drop left the begin lock held"); + engine.rollback(next).unwrap(); + + let mut holder = engine.begin(Isolation::Snapshot).unwrap(); + engine.reserve_begin_lock(&mut holder).unwrap(); + engine.rollback(holder).unwrap(); + let mut next = engine.begin(Isolation::Snapshot).unwrap(); + engine + .reserve_begin_lock(&mut next) + .expect("rollback left the begin lock held"); + engine.rollback(next).unwrap(); + + let mut holder = engine.begin(Isolation::Snapshot).unwrap(); + engine.reserve_begin_lock(&mut holder).unwrap(); + engine.commit(holder).unwrap(); + let mut next = engine.begin(Isolation::Snapshot).unwrap(); + engine + .reserve_begin_lock(&mut next) + .expect("commit left the begin lock held"); +}