fix: recover from panic/cancellation-induced mutex poisoning and leaked transactions

hyperdb-api/CHANGELOG.md

@@ -89,6 +89,30 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
   units depending on which platform produced the row — a 4.86% discrepancy.
   All `MB`-labelled output is now decimal (10^6). Installed-RAM reporting stays
   binary, since RAM is conventionally quoted that way.
+- **The async pool's default recycle probe now discharges a transaction
+  left open by a panicked or cancelled task.** Rust has no async `Drop`, so
+  `AsyncTransaction::drop` cannot issue a `ROLLBACK` when dropped without an
+  explicit `commit()`/`rollback()` — it only warns, leaving the transaction
+  open on the connection. `ConnectionManager::recycle`'s default
+  `RecycleStrategy::SelectOne` probe previously ran `SELECT 1`, which
+  succeeds even inside an open transaction (confirmed against the real
+  engine) and so never detected the leak; the next checkout silently
+  inherited a connection mid-transaction. `SelectOne` now issues an
+  unconditional `ROLLBACK` instead — a no-op when nothing is open, confirmed
+  empirically — at the same one-round-trip cost as the probe it replaces.
+  `RecycleStrategy::Ping`, `::None` and `::Custom` are unchanged and do not
+  discharge a leaked transaction; see their doc comments. The sync pool's
+  `SyncRecycleStrategy::SelectOne` is unaffected — `Transaction`'s `Drop` can
+  and does roll back synchronously, so the sync pool never leaks one of
+  these. Fixes [issue #263](https://github.com/tableau/hyper-api-rust/issues/263).
+- **Corrected the documented behavior of `RecycleStrategy::None`.** Its doc
+  comment claimed the pool "still drops connections that fail the passive
+  `AsyncConnection::is_alive` check". The async connection manager never calls
+  `is_alive` — `None` is a genuine no-op, so a connection is handed out in
+  whatever state the previous borrower left it, and a dead one surfaces on
+  first use. (Only the *sync* pool performs an `is_alive` check.) Behavior is
+  unchanged; the documentation was wrong, and misleadingly reassuring for the
+  caller most exposed to it — one combining `None` with `AsyncTransaction`.
 
 ### Changed
 

hyperdb-api/src/pool.rs

@@ -67,9 +67,14 @@
 //! # Tuning knobs (shared by both pools)
 //!
 //! - **Recycle strategy** ([`PoolConfig::recycle`] / [`SyncPoolConfig::recycle`])
-//!   controls the per-checkout health probe. Defaults to `SelectOne` (a
-//!   `SELECT 1` round-trip). Use `Ping` for the connection's native ping,
-//!   `None` to skip the probe on hot paths, or `Custom(..)` for a bespoke check.
+//!   controls the per-checkout health probe. Defaults to `SelectOne` — on the
+//!   async pool this is an unconditional `ROLLBACK` round-trip, which both
+//!   probes liveness and discharges any transaction a panicked or cancelled
+//!   `AsyncTransaction` guard left open (see [`RecycleStrategy::SelectOne`]);
+//!   on the sync pool (which has no such leak — `Transaction`'s `Drop` rolls
+//!   back synchronously) it stays a plain `SELECT 1`. Use `Ping` for the
+//!   connection's native ping, `None` to skip the probe on hot paths, or
+//!   `Custom(..)` for a bespoke check.
 //! - **`max_lifetime`** caps how long a physical connection may live before it
 //!   is retired at checkout, regardless of health.
 //! - **`idle_timeout`** retires connections that have sat idle too long (down to
@@ -172,18 +177,42 @@ pub type RecycleCheck = Arc<dyn Fn(&AsyncConnection) -> HookFuture<'_> + Send +
 /// evicts the connection and the pool transparently builds a fresh one.
 #[derive(Clone, Default)]
 pub enum RecycleStrategy {
-    /// Run a `SELECT 1` round-trip on every checkout. The default — catches a
-    /// half-dead connection at acquire time at the cost of one round-trip.
+    /// Run an unconditional `ROLLBACK` round-trip on every checkout. The
+    /// default — catches a half-dead connection at acquire time at the cost
+    /// of one round-trip, **and** discharges any transaction left open by a
+    /// panicked or cancelled [`AsyncTransaction`](crate::AsyncTransaction)
+    /// guard (issue #263: Rust has no async `Drop`, so that guard cannot
+    /// roll back itself and can only warn). `ROLLBACK` on a connection with
+    /// no open transaction is a harmless, zero-row no-op — confirmed against
+    /// the real engine — so this costs exactly the same one round-trip that
+    /// the prior `SELECT 1` probe did; it does not add a second round-trip.
     #[default]
     SelectOne,
     /// Call [`AsyncConnection::ping`] on every checkout (equivalent round-trip,
     /// expressed via the connection's own health primitive).
+    ///
+    /// Unlike [`SelectOne`](Self::SelectOne), this does **not** discharge a
+    /// transaction left open by a panicked or cancelled `AsyncTransaction`
+    /// guard — `ping` only reads. Prefer `SelectOne` (the default) if your
+    /// workload uses `AsyncTransaction`.
     Ping,
-    /// Skip the active probe entirely. The pool still drops connections that
-    /// fail the passive [`AsyncConnection::is_alive`] check. Use on hot paths
-    /// where the round-trip cost outweighs detecting a dead connection early.
+    /// Skip connection validation entirely: no round-trip, and no passive
+    /// check either — recycling is a genuine no-op, so a connection is
+    /// handed out in whatever state the previous borrower left it. Use on
+    /// hot paths where the round-trip cost outweighs detecting a dead
+    /// connection early, and expect the failure to surface on first use
+    /// instead.
+    ///
+    /// Does not discharge a transaction left open by a panicked or cancelled
+    /// `AsyncTransaction` guard — there is no round-trip at all to piggyback
+    /// on. Avoid combining with `AsyncTransaction` unless you independently
+    /// guarantee every transaction is committed or rolled back.
     None,
     /// Run a user-supplied async check on every checkout.
+    ///
+    /// Like [`Ping`](Self::Ping), does not itself discharge a transaction
+    /// left open by a panicked or cancelled `AsyncTransaction` guard; add an
+    /// unconditional `ROLLBACK` to your check if your workload needs that.
     Custom(RecycleCheck),
 }
 
@@ -538,7 +567,14 @@ impl Manager for ConnectionManager {
         // Active health probe per the configured strategy.
         match &self.config.recycle {
             RecycleStrategy::SelectOne => {
-                conn.execute_command("SELECT 1")
+                // Issue #263: an unconditional `ROLLBACK` replaces the
+                // former `SELECT 1` probe at the same one-round-trip cost.
+                // It doubles as the liveness check (a dead connection fails
+                // `ROLLBACK` exactly as it would fail `SELECT 1`) while also
+                // discharging any transaction a panicked or cancelled
+                // `AsyncTransaction` guard left open — see the type doc on
+                // `RecycleStrategy::SelectOne` for why this is safe.
+                conn.execute_command("ROLLBACK")
                     .await
                     .map_err(RecycleError::Backend)?;
             }

hyperdb-api/tests/pool_tests.rs

@@ -9,8 +9,8 @@
 
 mod common;
 
-use std::sync::Arc;
 use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use common::{test_hyper_params, test_result_path};
@@ -217,6 +217,198 @@ async fn async_pool_idle_timeout_retires_connection() {
     );
 }
 
+/// Regression test for [issue #263](https://github.com/tableau/hyper-api-rust/issues/263):
+/// a connection returned to the pool with an open transaction (left by a
+/// panicked or cancelled task holding an `AsyncTransaction`) must not be
+/// handed to the next borrower still mid-transaction.
+///
+/// `AsyncTransaction::drop` cannot issue an async `ROLLBACK` (Rust has no
+/// async `Drop`), so it only warns when dropped without an explicit
+/// `commit()`/`rollback()` — exactly what happens when the task holding the
+/// guard panics, mirroring the real call sites in `ingest.rs` /
+/// `ingest_arrow.rs`, which already catch the `tokio::spawn` join error and
+/// continue rather than propagating a panic.
+///
+/// `max_size(1)` caps the pool at one physical connection at a time, but it
+/// does **not** by itself prove the fix: deadpool evicts and builds a
+/// replacement on any `RecycleError`, and a fresh session cannot see another
+/// session's uncommitted row, so the `count == 0` assertion below would pass
+/// even if the connection had merely been replaced. The
+/// [`async_session_id`] comparison is what pins the causal mechanism —
+/// identical session ids mean the *same physical connection* was recycled,
+/// so `ConnectionManager::recycle` is what discharged the transaction.
+///
+/// That comparison also covers the cost premise of the fix, which nothing
+/// else in this file does: it is the only assertion that `ROLLBACK` *itself
+/// succeeds* on a recycled connection. (`async_pool_recycle_ping_strategy_works`
+/// uses `Ping`, `async_pool_custom_recycle_failure_replaces_connection` uses
+/// `Custom`, and the `max_lifetime` / `idle_timeout` tests return early from
+/// `recycle` before the probe runs.) If the probe ever began erroring on an
+/// idle connection, the pool would silently rebuild on *every* checkout — a
+/// throughput cliff — and `assert_eq!` on the session id is what fails.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn async_pool_recycle_discharges_transaction_left_open_by_panicked_task() {
+    let (_hyper, endpoint) = fresh_server("pool_async_txn_leak").unwrap();
+    let config = PoolConfig::new(&endpoint, db_path("pool_async_txn_leak"))
+        .create_mode(CreateMode::CreateAndReplace)
+        .max_size(1);
+    let pool = create_pool(config).unwrap();
+
+    {
+        let conn = pool.get().await.expect("setup checkout");
+        conn.execute_command("CREATE TABLE leaked (v INT)")
+            .await
+            .unwrap();
+    }
+
+    // Simulate the defect: BEGIN, write a row, then panic before
+    // commit/rollback. The `AsyncTransaction` guard's `Drop` only warns; the
+    // transaction is still open on the physical connection when the pooled
+    // guard (`conn`) is itself dropped during the same unwind and returns
+    // that connection to deadpool's idle set.
+    //
+    // The `leaked_sid` slot publishes the panicking task's session id so the
+    // assertion below can prove the next checkout got that same physical
+    // connection back.
+    let leaked_sid = Arc::new(Mutex::new(None));
+    let pool_for_task = pool.clone();
+    let sid_slot = Arc::clone(&leaked_sid);
+    let join_result = tokio::spawn(async move {
+        let mut conn = pool_for_task.get().await.expect("panic-task checkout");
+        let sid = async_session_id(&conn).await;
+        *sid_slot.lock().expect("session-id slot") = Some(sid);
+        let txn = conn.transaction().await.expect("begin txn");
+        txn.execute_command("INSERT INTO leaked VALUES (999)")
+            .await
+            .expect("insert inside txn");
+        panic!("simulated tool handler bug mid-transaction");
+    })
+    .await;
+    assert!(
+        join_result.is_err(),
+        "spawned task should have panicked, mirroring the real join-error \
+         handling in load_files"
+    );
+    let leaked_sid = leaked_sid
+        .lock()
+        .expect("session-id slot")
+        .clone()
+        .expect("panicking task must have recorded its session id before the panic");
+
+    // This checkout runs the leaked connection through
+    // `ConnectionManager::recycle`. Without the fix, the old `SELECT 1` probe
+    // succeeds despite the open transaction and this borrower silently
+    // inherits it; with the fix, `recycle` issues an unconditional
+    // `ROLLBACK` first.
+    let conn2 = pool.get().await.expect("checkout after panicking task");
+
+    // The load-bearing assertion: same session id means `recycle` succeeded
+    // and returned this exact connection to service. A `ROLLBACK` that
+    // errored would instead have evicted it, and deadpool would have handed
+    // us a replacement with a different id — which would still satisfy the
+    // `count == 0` check below, silently hiding both a broken probe and the
+    // per-checkout rebuild cliff it would cause.
+    assert_eq!(
+        leaked_sid,
+        async_session_id(&conn2).await,
+        "recycle must ROLLBACK and reuse the same physical connection, not \
+         evict it and build a replacement"
+    );
+
+    // Prove the leaked row did NOT survive — i.e. `recycle` genuinely rolled
+    // it back, rather than merely succeeding without touching the
+    // transaction.
+    let count: i64 = conn2
+        .query_count("SELECT COUNT(*) FROM leaked WHERE v = 999")
+        .await
+        .expect("query after recycle");
+    assert_eq!(
+        count, 0,
+        "recycle must roll back a transaction leaked by a panicked task \
+         before handing the connection to the next borrower"
+    );
+
+    // And the connection must be immediately usable for ordinary work, not
+    // stuck mid-transaction from the next borrower's perspective.
+    conn2
+        .execute_command("INSERT INTO leaked VALUES (1)")
+        .await
+        .expect("connection must be usable, not wedged mid-transaction");
+    let total: i64 = conn2
+        .query_count("SELECT COUNT(*) FROM leaked")
+        .await
+        .unwrap();
+    assert_eq!(total, 1, "only the post-recycle insert should be present");
+}
+
+/// The cancellation half of [issue #263](https://github.com/tableau/hyper-api-rust/issues/263),
+/// which names "panic and cancellation" — the sibling test above covers only
+/// the panic.
+///
+/// The fix covers cancellation by construction: dropping a pending future
+/// drops `AsyncTransaction` and then `PooledConnection` in the same order an
+/// unwind does, so it reaches the same `ConnectionManager::recycle`. This
+/// test exists so that equivalence is asserted rather than assumed, and so
+/// the changelog's "panicked or cancelled" claim is test-backed.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn async_pool_recycle_discharges_transaction_left_open_by_cancelled_task() {
+    let (_hyper, endpoint) = fresh_server("pool_async_txn_cancel").unwrap();
+    let config = PoolConfig::new(&endpoint, db_path("pool_async_txn_cancel"))
+        .create_mode(CreateMode::CreateAndReplace)
+        .max_size(1);
+    let pool = create_pool(config).unwrap();
+
+    {
+        let conn = pool.get().await.expect("setup checkout");
+        conn.execute_command("CREATE TABLE leaked (v INT)")
+            .await
+            .unwrap();
+    }
+
+    // Cancel the task by timing it out mid-transaction: `tokio::time::timeout`
+    // drops the inner future once the deadline passes, which drops the
+    // `AsyncTransaction` (BEGIN still open) and then the pooled connection.
+    let leaked_sid = Arc::new(Mutex::new(None));
+    let sid_slot = Arc::clone(&leaked_sid);
+    let cancelled = tokio::time::timeout(Duration::from_millis(150), async {
+        let mut conn = pool.get().await.expect("cancel-task checkout");
+        let sid = async_session_id(&conn).await;
+        *sid_slot.lock().expect("session-id slot") = Some(sid);
+        let txn = conn.transaction().await.expect("begin txn");
+        txn.execute_command("INSERT INTO leaked VALUES (999)")
+            .await
+            .expect("insert inside txn");
+        // Outlive the deadline so the future is dropped right here, with the
+        // transaction open and never committed or rolled back.
+        tokio::time::sleep(Duration::from_secs(30)).await;
+        unreachable!("the timeout must fire first");
+    })
+    .await;
+    assert!(cancelled.is_err(), "the task must have been cancelled");
+    let leaked_sid = leaked_sid
+        .lock()
+        .expect("session-id slot")
+        .clone()
+        .expect("cancelled task must have recorded its session id");
+
+    let conn2 = pool.get().await.expect("checkout after cancelled task");
+    assert_eq!(
+        leaked_sid,
+        async_session_id(&conn2).await,
+        "recycle must ROLLBACK and reuse the same physical connection the \
+         cancelled task left mid-transaction"
+    );
+    let count: i64 = conn2
+        .query_count("SELECT COUNT(*) FROM leaked WHERE v = 999")
+        .await
+        .expect("query after recycle");
+    assert_eq!(
+        count, 0,
+        "recycle must roll back a transaction leaked by a cancelled task, \
+         exactly as it does for a panicked one"
+    );
+}
+
 // ---------------------------------------------------------------------------
 // Sync pool
 // ---------------------------------------------------------------------------

hyperdb-mcp/CHANGELOG.md

@@ -300,6 +300,46 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
   of failing fast and reconnecting. This is most visible since the daemon
   became resident-by-default in 0.5.0, which made long-lived idle connections
   the norm. (Fixed in `hyperdb-api-core` for both the sync and async clients.)
+- **A panicking tool call no longer bricks the server for the rest of the
+  process's lifetime.** `with_engine` holds a `std::sync::MutexGuard` across
+  the tool closure it invokes; a panic propagating out of that closure dropped
+  the guard mid-unwind and poisoned the engine mutex, and every subsequent
+  tool call then failed with `InternalError "Lock poisoned"` until the process
+  restarted. The engine lock now recovers from poisoning the same way it
+  already recovers from `ConnectionLost`: it discards the (possibly
+  mid-mutation) `Engine` behind the poisoned guard, clears the poison flag,
+  and rebuilds a fresh engine on the next call — never reusing a value a panic
+  may have left in a broken state.
+
+  Recovery runs at every site that locks the engine, not just the one the
+  panic came through, because the server hands the same handle to background
+  watchers via `engine_handle()`. A watcher's connection-lost pool rebuild
+  previously failed with `InternalError "Engine lock poisoned"` and its
+  `_table_catalog` bookkeeping was silently skipped until some unrelated tool
+  call happened to clear the flag — indefinitely, for an unattended ingest
+  where no tool call may arrive. A watcher now sees the recovered (empty)
+  engine slot and returns its transient, retryable "Engine not initialized"
+  error instead. The engine-construction single-flight lock recovers too, so a
+  panic during `Engine::new`, attachment replay, or the internal
+  `debug_assert!` cannot re-brick the server through a second mutex.
+
+  `status`, which reads the engine with `try_lock` so it never waits behind a
+  slow data-plane call, no longer reports a poisoned mutex as contention. It
+  previously answered `engine_busy: true` — whose description tells the client
+  to retry later — permanently after a panic, with no call in flight and no
+  way for the flag to clear itself. It now recovers and reports the empty
+  engine as degraded, which is the same shape of answer but an honest one.
+
+  **Recovery destroys the session's ephemeral tables.** Discarding the
+  `Engine` runs its `Drop`, which deletes the temp directory holding the
+  ephemeral primary database, so every table loaded without `persist: true` is
+  lost and the rebuilt engine starts empty. Attached databases survive —
+  they are replayed onto the new engine and `persistent` is re-attached. This
+  is a deliberate trade rather than an oversight: an `Engine` a panic caught
+  mid-mutation cannot be trusted, and handing it back out silently is worse
+  than losing scratch tables. Load with `persist: true` if data must survive a
+  panicking tool call. Fixes
+  [issue #266](https://github.com/tableau/hyper-api-rust/issues/266).
 
 ## [0.5.0] - 2026-06-07