fix(db): recover stuck sqlx transaction-depth counter in begin_immediate

crates/gitforge-db/src/connection.rs

@@ -34,21 +34,35 @@ const BEGIN_RECOVERY_PASSES: usize = 3;
 /// Begin a `BEGIN IMMEDIATE` transaction, recovering a poisoned connection.
 ///
 /// sqlx tracks transaction depth per connection, but a write that loses a
-/// race against task cancellation or a write freeze can leave the SQLite
-/// handle inside a transaction while the depth counter says zero (the
-/// rollback paths that would resync the pair only run when the original
-/// command both succeeded and was acknowledged). The poisoned connection
-/// returns to the pool, and because the pool hands out the most recently
-/// used connection first, every subsequent `begin_with` fails with
-/// `(code: 1) cannot start a transaction within a transaction`. That is
-/// not a transient lock: `persist_with_retry` alone cannot clear it,
-/// because every retry is handed the same poisoned connection.
+/// race against task cancellation or a write freeze can desync that counter
+/// from the SQLite handle in either direction:
 ///
-/// Recovery is a bare `ROLLBACK` on a fresh pooled connection — a no-op
-/// error on a healthy one — then the begin is retried. Observed live
-/// 2026-10-08: five `persist_with_retry` attempts at job completion all
-/// failed against the same poisoned connection while the runner's work had
-/// succeeded; the lease-dead fence then graded the run failed.
+/// 1. *Handle inside a transaction, counter at zero* — the next
+///    `begin_with` fails with `(code: 1) cannot start a transaction within
+///    a transaction`. Recovery is a bare `ROLLBACK` on a fresh pooled
+///    connection — a no-op error on a healthy one — then the begin is
+///    retried. Observed live 2026-10-08: five `persist_with_retry` attempts
+///    at job completion all failed against the same poisoned connection
+///    while the runner's work had succeeded; the lease-dead fence then
+///    graded the run failed.
+/// 2. *Counter above zero with no live `Transaction` owner* — the next
+///    `begin_with` carrying a custom statement fails with `attempted to
+///    call begin_with at non-zero transaction depth` (sqlx rejects a custom
+///    BEGIN inside what it believes is an open transaction) before any
+///    statement reaches SQLite. A raw ROLLBACK cannot fix this direction:
+///    it clears the SQLite side but the counter only moves through sqlx's
+///    managed commit/rollback commands. Recovery opens plain managed
+///    transactions (`pool.begin()`, which issues a SAVEPOINT at depth > 0)
+///    and rolls them back, decrementing the counter one level per pass.
+///    Observed live 2026-11-02: every `pipeline activation` trigger failed
+///    five times with the non-zero-depth error while healthy runs kept
+///    serving from the same pool, so commit `c844a4b` (OpenCodifier) sat
+///    un-run for hours.
+///
+/// Neither failure is a transient lock: `persist_with_retry` alone cannot
+/// clear either, because every retry is handed the same poisoned
+/// connection (the pool hands out the most recently used connection
+/// first).
 pub(crate) async fn begin_immediate(
     pool: &SqlitePool,
     context: &str,
@@ -62,11 +76,61 @@ pub(crate) async fn begin_immediate(
             Ok(tx)
         }
         Err(begin_error) => {
+            let counter_stuck = begin_error
+                .to_string()
+                .contains("non-zero transaction depth");
             let mut last = begin_error;
             for _ in 0..BEGIN_RECOVERY_PASSES {
                 let Ok(mut conn) = pool.acquire().await else {
                     break;
                 };
+                if counter_stuck {
+                    // Direction 2: unwind sqlx's own counter. A managed
+                    // begin at depth > 0 issues a SAVEPOINT (no custom
+                    // statement, so sqlx allows it); the managed rollback
+                    // decrements the counter one level. Keep the
+                    // connection out of the pool across the unwind —
+                    // dropping it mid-desync would hand the next caller
+                    // the same stuck counter.
+                    drop(conn);
+                    let Ok(mut unwind_conn) = pool.acquire().await else {
+                        break;
+                    };
+                    let unwind =
+                        match sqlx::Transaction::<sqlx::Sqlite>::begin(&mut *unwind_conn, None)
+                            .await
+                        {
+                            Ok(tx) => tx.rollback().await,
+                            Err(error) => Err(error),
+                        };
+                    match unwind {
+                        Ok(()) => {
+                            tracing::warn!(
+                                context = %context,
+                                "unwound a stuck sqlx transaction-depth counter with a managed SAVEPOINT rollback"
+                            );
+                        }
+                        Err(error) => {
+                            // The counter cannot be unwound on this
+                            // connection, and the pool keeps handing it
+                            // back: further passes would repeat the same
+                            // failure. Surface the unwind error instead.
+                            last = error;
+                            break;
+                        }
+                    }
+                    drop(unwind_conn);
+                    match pool.begin_with("BEGIN IMMEDIATE").await {
+                        Ok(tx) => {
+                            if let Some(m) = metrics() {
+                                m.record_db_write(context, write_start.elapsed().as_secs_f64());
+                            }
+                            return Ok(tx);
+                        }
+                        Err(error) => last = error,
+                    }
+                    continue;
+                }
                 // A bare ROLLBACK clears the desynced SQLite-side
                 // transaction. On a healthy connection it fails with
                 // "no transaction is active", which carries no signal.

crates/gitforge-db/src/queries.rs

@@ -4162,6 +4162,68 @@ mod tests {
         pool.close().await;
     }
 
+    /// Regression for the 2026-11-02 pipeline-activation failures (commit
+    /// `c844a4b` on OpenCodifier sat un-run for hours): a pooled connection
+    /// whose sqlx depth counter is stuck ABOVE zero with no live
+    /// `Transaction` owner rejects every `begin_with` carrying a custom
+    /// statement with `attempted to call begin_with at non-zero transaction
+    /// depth` — before any statement reaches SQLite, so the raw-ROLLBACK
+    /// recovery for direction 1 is useless here. `begin_immediate` must
+    /// unwind the counter with managed SAVEPOINT rollbacks and yield a
+    /// usable transaction.
+    ///
+    /// The pool is capped at one connection so every borrower
+    /// deterministically gets the stuck one; the poison is injected with
+    /// `pool.begin()` followed by leaking the transaction (simulating the
+    /// lost-ack path that leaves the counter raised without an owner).
+    #[tokio::test]
+    async fn test_begin_immediate_unwinds_stuck_depth_counter() {
+        let pool = sqlx::sqlite::SqlitePoolOptions::new()
+            .max_connections(1)
+            .connect("sqlite::memory:")
+            .await
+            .unwrap();
+
+        // Raise the counter without a live owner: begin managed (depth 1),
+        // then drop the connection without commit/rollback (Transaction's
+        // Drop tries an async rollback, but detaching the connection
+        // prevents it — instead we simulate the stuck state by leaking the
+        // Transaction's handle via std::mem::forget, so no rollback fires).
+        let mut conn = pool.acquire().await.unwrap();
+        let tx = sqlx::Transaction::<sqlx::Sqlite>::begin(&mut *conn, None)
+            .await
+            .unwrap();
+        std::mem::forget(tx);
+        drop(conn);
+
+        // Sanity: the custom-statement begin now fails with the live error
+        // signature.
+        let stuck = pool.begin_with("BEGIN IMMEDIATE").await;
+        let message = format!("{}", stuck.unwrap_err());
+        assert!(
+            message.contains("non-zero transaction depth"),
+            "expected the stuck-counter begin to fail with the depth error, \
+             got: {message}"
+        );
+
+        // Recovery: the begin must succeed and the transaction must work.
+        let mut tx = begin_immediate(&pool, "pipeline activation").await.unwrap();
+        sqlx::query("CREATE TABLE depth_probe (id INTEGER)")
+            .execute(&mut *tx)
+            .await
+            .unwrap();
+        tx.commit().await.unwrap();
+
+        // And the pool keeps serving healthy transactions afterwards.
+        let mut tx = begin_immediate(&pool, "pipeline activation").await.unwrap();
+        sqlx::query("INSERT INTO depth_probe (id) VALUES (1)")
+            .execute(&mut *tx)
+            .await
+            .unwrap();
+        tx.commit().await.unwrap();
+        pool.close().await;
+    }
+
     /// Regression for the swallowed-write flavor of connection poisoning
     /// (run `99c463db`, 2026-10-08): a single-statement write routed in
     /// autocommit to a connection with an orphaned SQLite transaction