Release row locks when an open transaction is dropped.

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.

crates/kernel/src/engine/mod.rs

@@ -191,7 +191,7 @@ pub struct Engine {
     heap: PageBackedHeap,
     catalog: CatalogManager,
     catalog_store: CatalogStore,
-    locks: RowLockManager,
+    locks: Arc<RowLockManager>,
     wal: Arc<WalCoordinator>,
     /// Phase 11 Wave 0: engine-level aggregator for the new
     /// telemetry counters. Lives next to `wal` because it is the

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));

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<()> {

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<RowKey>,
     /// Membership for `has_row_lock`. The vec alone is O(n) per probe.
     row_lock_set: HashSet<RowKey>,
+    /// Present for engine transactions so `Drop` can release locks that
+    /// commit and rollback did not already drain.
+    row_lock_manager: Option<Arc<RowLockManager>>,
     open: bool,
     lifecycle: Option<Arc<TxnLifecycle>>,
     /// 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<RowLockManager>) {
+        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::*;

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");
+}