diff --git a/AGENT_CHAT.md b/AGENT_CHAT.md index f287c0b0..9c3562b1 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 d8b2ea67..084e6aee 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 376e2c5d..08f190a9 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 21481cc5..1c6c83fa 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 fcc2dedd..9d41e6e0 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 62ccf8ab..25c37df5 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"); +}