fix(distributed): safe-pinning fail-closed + deterministic 2PC phase timeouts (issues 01, 02)

Combines the two prior-session 2PC corrections that share
distributed_txn.h.

Issue 01 (P0) — distributed 2PC must require safe session pinning

Pooled executors that cannot pin a physical connection across the
DML and PREPARE/COMMIT phases would silently corrupt 2PC:
phase 1 PREPARE could land on one backend session, phase 2 COMMIT
on another, and the participant would never see the commit. Old
code fell back to the unpinned path whenever checkout_session()
returned nullptr -- which is exactly what every pooled executor
returned, with no warning.

Fix:

- New virtual bool RemoteExecutor::allows_unpinned_distributed_2pc()
  defaults to false. Executors must explicitly opt in to the
  unpinned path.
- MySQLRemoteExecutor and PgSQLRemoteExecutor opt in (they keep one
  durable physical connection per backend for the executor's
  lifetime).
- MultiRemoteExecutor opts in (it composes the two single-conn
  executors).
- ThreadSafeMultiRemoteExecutor (the pooled, production path) does
  NOT opt in and so DistributedTransactionManager fails closed when
  asked to enlist a backend without a pinned session.

Issue 02 (P1) — make 2PC phase timeouts deterministic

The previous code tried to set max_execution_time on MySQL XA
control statements, which silently does nothing (MySQL ignores it
for XA). PostgreSQL got a separate SET statement_timeout that ran
some indeterminate time before PREPARE / COMMIT PREPARED.

Fix:

- PostgreSQL: SET statement_timeout = N is issued on the same
  pinned session immediately before each phase command, so the
  timeout governs that exact statement.
- MySQL: dropped the misleading per-phase SQL timeout. XA control
  statements are bounded only by connection-level read/write
  timeouts (which the connection pool already configures). Doctrine
  is now stated explicitly in the comments instead of pretending
  there's a per-statement bound.
- distributed_txn.h is ~111 lines smaller thanks to the cleanup.

Tests:

- tests/test_distributed_txn.cpp updated to:
  * exercise the fail-closed path via a mock executor that does not
    opt in to unpinned 2PC,
  * verify the per-phase PostgreSQL timeout SET is issued on the
    pinned session,
  * keep coverage for happy paths (XA + PREPARE TRANSACTION) and
    in-doubt WAL recovery.
- tests/test_session.cpp gets a small update for the new flag.

Verification:

- make test: 1208 passed, 37 skipped (live-backend integration), 0
  failed.

Plan:
- docs/superpowers/plans/2026-04-15-distributed-2pc-safe-pinning.md
  is the implementation plan that drove issue 01. Committed
  alongside the change.

Author: renecannao

docs/superpowers/plans/2026-04-15-distributed-2pc-safe-pinning.md

@@ -0,0 +1,161 @@
+# Distributed 2PC Safe Pinning Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Prevent distributed 2PC from silently using unsafe unpinned fallback paths unless the executor explicitly declares that fallback safe.
+
+**Architecture:** Extend the `RemoteExecutor` contract with an explicit capability for legacy-safe unpinned 2PC fallback. `DistributedTransactionManager` will keep using pinned sessions when available, but it will reject `nullptr` session fallback unless the executor opts in. Single-connection executors and test mocks can opt in; pooled executors will remain pinned-session-only.
+
+**Tech Stack:** C++17, GoogleTest, existing `RemoteExecutor`/`DistributedTransactionManager` interfaces
+
+---
+
+### Task 1: Add the executor capability flag
+
+**Files:**
+- Modify: `include/sql_engine/remote_executor.h`
+- Modify: `include/sql_engine/mysql_remote_executor.h`
+- Modify: `include/sql_engine/pgsql_remote_executor.h`
+- Modify: `include/sql_engine/multi_remote_executor.h`
+- Modify: `src/sql_engine/multi_remote_executor.cpp`
+
+- [ ] **Step 1: Add the failing tests first**
+
+Add tests that assert distributed 2PC fallback is rejected for an executor that returns `nullptr` from `checkout_session()` and does not explicitly opt in.
+
+- [ ] **Step 2: Run targeted tests to verify failure**
+
+Run: `./run_tests --gtest_filter="DistributedTxnOverrides.*:DistributedTxn.*" --gtest_brief=1`
+
+Expected: at least one new test fails because the current code still routes through the legacy fallback.
+
+- [ ] **Step 3: Add the API hook**
+
+Update `RemoteExecutor` with a default capability method:
+
+```cpp
+virtual bool allows_unpinned_distributed_2pc() const { return false; }
+```
+
+Override it in single-connection executors and wrappers that are safe without pinning:
+
+```cpp
+bool allows_unpinned_distributed_2pc() const override { return true; }
+```
+
+- [ ] **Step 4: Run compile-targeted tests**
+
+Run: `make test`
+
+Expected: build succeeds even if the new regression tests still fail.
+
+### Task 2: Harden distributed transaction fallback
+
+**Files:**
+- Modify: `include/sql_engine/distributed_txn.h`
+- Test: `tests/test_distributed_txn.cpp`
+- Test: `tests/test_session.cpp`
+
+- [ ] **Step 1: Update the fallback contract**
+
+Change the `nullptr` session path from implicit fallback to guarded fallback:
+
+```cpp
+if (!session) {
+    if (!executor_.allows_unpinned_distributed_2pc()) {
+        return false;
+    }
+    // legacy-safe single-connection fallback
+}
+```
+
+Apply the same rule in `execute_participant_dml()` so `route_dml()` cannot bypass the safety check.
+
+- [ ] **Step 2: Add or update regression tests**
+
+Cover:
+
+- executor with no sessions and no opt-in => enlist fails
+- executor with no sessions and explicit opt-in => legacy fallback still works
+- pinned-session executor path remains valid
+
+- [ ] **Step 3: Run targeted tests**
+
+Run: `./run_tests --gtest_filter="DistributedTxn.*:DistributedTxnOverrides.*" --gtest_brief=1`
+
+Expected: all targeted distributed transaction tests pass.
+
+### Task 3: Align comments and route expectations
+
+**Files:**
+- Modify: `include/sql_engine/remote_executor.h`
+- Modify: `include/sql_engine/distributed_txn.h`
+- Modify: `tests/test_session.cpp`
+
+- [ ] **Step 1: Update comments to match the new contract**
+
+Replace comments that describe `nullptr` as an acceptable generic fallback with language that distinguishes:
+
+- pinned-session path
+- explicit legacy-safe fallback
+- rejected unsafe fallback
+
+- [ ] **Step 2: Update tests that currently assume every `nullptr` path is allowed**
+
+For example, tests using `TrackingRemoteExecutor` should explicitly opt in if they intend to validate the legacy-safe fallback.
+
+- [ ] **Step 3: Run focused session tests**
+
+Run: `./run_tests --gtest_filter="DistributedTxnOverrides.*:Session*" --gtest_brief=1`
+
+Expected: updated transaction and session routing tests pass.
+
+### Task 4: Final verification
+
+**Files:**
+- Modify: `include/sql_engine/remote_executor.h`
+- Modify: `include/sql_engine/distributed_txn.h`
+- Modify: `include/sql_engine/mysql_remote_executor.h`
+- Modify: `include/sql_engine/pgsql_remote_executor.h`
+- Modify: `include/sql_engine/multi_remote_executor.h`
+- Modify: `src/sql_engine/multi_remote_executor.cpp`
+- Modify: `tests/test_distributed_txn.cpp`
+- Modify: `tests/test_session.cpp`
+
+- [ ] **Step 1: Run the full targeted verification set**
+
+Run:
+
+```bash
+./run_tests --gtest_filter="DistributedTxn.*:DistributedTxnOverrides.*:Session*" --gtest_brief=1
+```
+
+Expected: all matching tests pass.
+
+- [ ] **Step 2: Run the full suite**
+
+Run:
+
+```bash
+./run_tests --gtest_brief=1
+```
+
+Expected: no regressions; backend-dependent tests may still skip if local services are unavailable.
+
+- [ ] **Step 3: Commit**
+
+```bash
+git add include/sql_engine/remote_executor.h \
+        include/sql_engine/distributed_txn.h \
+        include/sql_engine/mysql_remote_executor.h \
+        include/sql_engine/pgsql_remote_executor.h \
+        include/sql_engine/multi_remote_executor.h \
+        src/sql_engine/multi_remote_executor.cpp \
+        tests/test_distributed_txn.cpp \
+        tests/test_session.cpp \
+        docs/issues/README.md \
+        docs/issues/01-distributed-2pc-safe-session-pinning.md \
+        docs/superpowers/specs/2026-04-15-implementation-gap-backlog-design.md \
+        docs/superpowers/plans/2026-04-15-distributed-2pc-safe-pinning.md
+git commit -m "fix: fail closed on unsafe distributed 2pc fallback"
+```

include/sql_engine/distributed_txn.h

@@ -56,24 +56,23 @@ class DistributedTransactionManager : public TransactionManager {
     // continues (caller might prefer availability over durability).
     void set_require_durable_log(bool required) { require_durable_log_ = required; }
 
-    // Set a tight per-phase statement timeout (in milliseconds). When > 0,
-    // the manager issues a SET SESSION max_execution_time (MySQL) or
-    // SET LOCAL statement_timeout (PostgreSQL) on each participant BEFORE
-    // phase 1 (XA PREPARE / PREPARE TRANSACTION) and again before phase 2
-    // (XA COMMIT / COMMIT PREPARED). This is independent of the backend
-    // connection's default read/write timeout -- use it to bound 2PC
-    // specifically without affecting other queries.
+    // Set a tight per-phase statement timeout (in milliseconds).
     //
-    // 0 (default) means "don't override", fall back to whatever the
-    // backend connection's read/write timeout provides.
+    // PostgreSQL:
+    //   Before PREPARE TRANSACTION and COMMIT/ROLLBACK PREPARED, the manager
+    //   issues SET statement_timeout = <ms> on the participant session
+    //   immediately before the phase SQL. This gives a deterministic
+    //   session-level timeout on the same connection used for the phase
+    //   statement.
     //
-    // NOTE: because ThreadSafeMultiRemoteExecutor may hand us a different
-    // pooled connection for each execute_dml call, the SET must be issued
-    // immediately before the statement whose timeout it bounds. We do
-    // that by concatenating them with "; " when multi-statement is
-    // supported, OR by issuing two separate execute_dml calls and
-    // tolerating that the second one may not actually have the timeout
-    // in effect (best-effort). For the MVP we use the two-call approach.
+    // MySQL:
+    //   XA control statements are not bounded by max_execution_time, so we do
+    //   not emit a misleading per-phase SQL timeout. MySQL protection comes
+    //   from the connection read/write timeouts configured on the executor.
+    //
+    // 0 (default) means "don't override". PostgreSQL then falls back to the
+    // backend/session defaults, and MySQL continues to rely on connection
+    // read/write timeouts only.
     void set_phase_statement_timeout_ms(uint32_t ms) {
         phase_statement_timeout_ms_ = ms;
     }
@@ -108,14 +107,8 @@ class DistributedTransactionManager : public TransactionManager {
         if (started_.count(name)) return true;  // already enlisted
 
         // Acquire a pinned session. If the executor doesn't support
-        // pinning (returns nullptr), fall back to the unpinned path:
-        // BEGIN/XA START goes through execute_dml and subsequent phase
-        // calls will also use execute_dml. That mode is only correct
-        // for executors that aren't pool-based (e.g. the mock used in
-        // tests OR a non-thread-safe MySQLRemoteExecutor that holds one
-        // connection per backend). ThreadSafeMultiRemoteExecutor DOES
-        // implement checkout_session so the pinning path is taken for
-        // real workloads.
+        // pinning (returns nullptr), only executors that explicitly
+        // declare unpinned 2PC safe are allowed to use the legacy path.
         std::unique_ptr<RemoteSession> session = executor_.checkout_session(backend_name);
 
         bool ok = false;
@@ -127,9 +120,6 @@ class DistributedTransactionManager : public TransactionManager {
             } else {
                 sql = "BEGIN";
             }
-            if (phase_statement_timeout_ms_ > 0) {
-                session->execute_dml(make_timeout_stringref());
-            }
             DmlResult r = session->execute_dml(
                 sql_parser::StringRef{sql.c_str(),
                     static_cast<uint32_t>(sql.size())});
@@ -138,10 +128,11 @@ class DistributedTransactionManager : public TransactionManager {
                 sessions_[name] = std::move(session);
             }
         } else {
-            // Unpinned fallback: executor doesn't support sessions.
-            // Use the legacy path, which works with mock executors and
-            // single-connection executors but is broken for pooled
-            // real-backend 2PC.
+            // Unpinned fallback is valid only for executors that
+            // explicitly guarantee one durable connection per backend.
+            if (!executor_.allows_unpinned_distributed_2pc()) {
+                return false;
+            }
             if (dialect_ == BackendDialect::MYSQL) {
                 std::string sql = "XA START '" + txn_id_ + "'";
                 ok = send_sql(backend_name, sql);
@@ -183,7 +174,7 @@ class DistributedTransactionManager : public TransactionManager {
         if (it != sessions_.end() && it->second) {
             return it->second->execute_dml(sql);
         }
-        // Fallback: unpinned mode. Use the legacy path.
+        // Legacy-safe single-connection fallback.
         return executor_.execute_dml(backend_name, sql);
     }
 
@@ -289,28 +280,6 @@ class DistributedTransactionManager : public TransactionManager {
     uint32_t auto_compact_threshold_ = 0;
     uint32_t completions_since_compact_ = 0;
 
-    // Best-effort: set a per-session statement timeout on a backend before
-    // issuing a phase-1 or phase-2 SQL. Returns true if the SET succeeded
-    // OR if no timeout is configured; false only if the SET itself fails
-    // and the caller asked for a real timeout (in which case the caller
-    // may want to abort rather than risk an unbounded hang).
-    bool maybe_set_statement_timeout(const char* backend) {
-        if (phase_statement_timeout_ms_ == 0) return true;
-        std::string sql;
-        if (dialect_ == BackendDialect::MYSQL) {
-            // MySQL 5.7.4+: max_execution_time is in milliseconds and
-            // only bounds SELECTs. For DML and XA commands, the client
-            // read_timeout is our real protection. We still set this for
-            // SELECTs that might be issued between phases.
-            sql = "SET SESSION max_execution_time = " +
-                  std::to_string(phase_statement_timeout_ms_);
-        } else {
-            sql = "SET LOCAL statement_timeout = " +
-                  std::to_string(phase_statement_timeout_ms_);
-        }
-        return send_sql(backend, sql);
-    }
-
     // Write the phase-2 decision to the durable log before dispatching.
     // Returns true if the commit/rollback can proceed:
     // - log not configured: true (log-less mode preserves legacy behavior)
@@ -475,36 +444,18 @@ class DistributedTransactionManager : public TransactionManager {
         return send_sql(participant.c_str(), sql);
     }
 
-    // Same as maybe_set_statement_timeout but for a participant -- routes
-    // through the pinned session when available.
+    // Set a participant-scoped phase timeout immediately before the phase SQL.
+    // PostgreSQL gets a real session-level statement_timeout on the same
+    // connection. MySQL returns true without emitting SQL because XA control
+    // statements are bounded by connection read/write timeouts, not by
+    // max_execution_time.
     bool maybe_set_statement_timeout_participant(const std::string& participant) {
         if (phase_statement_timeout_ms_ == 0) return true;
-        std::string sql;
-        if (dialect_ == BackendDialect::MYSQL) {
-            sql = "SET SESSION max_execution_time = " +
-                  std::to_string(phase_statement_timeout_ms_);
-        } else {
-            sql = "SET LOCAL statement_timeout = " +
-                  std::to_string(phase_statement_timeout_ms_);
-        }
+        if (dialect_ == BackendDialect::MYSQL) return true;
+        std::string sql = "SET statement_timeout = " +
+                          std::to_string(phase_statement_timeout_ms_);
         return send_sql_participant(participant, sql);
     }
-
-    // For passing a StringRef to session->execute_dml in enlist_backend.
-    // We only need this temporarily inside enlist_backend, so a thread-
-    // local buffer is fine.
-    sql_parser::StringRef make_timeout_stringref() {
-        static thread_local std::string buf;
-        if (dialect_ == BackendDialect::MYSQL) {
-            buf = "SET SESSION max_execution_time = " +
-                  std::to_string(phase_statement_timeout_ms_);
-        } else {
-            buf = "SET LOCAL statement_timeout = " +
-                  std::to_string(phase_statement_timeout_ms_);
-        }
-        return sql_parser::StringRef{buf.c_str(),
-            static_cast<uint32_t>(buf.size())};
-    }
 };
 
 } // namespace sql_engine

include/sql_engine/multi_remote_executor.h

@@ -20,6 +20,7 @@ class MultiRemoteExecutor : public RemoteExecutor {
     void add_backend(const BackendConfig& config);
     ResultSet execute(const char* backend_name, sql_parser::StringRef sql) override;
     DmlResult execute_dml(const char* backend_name, sql_parser::StringRef sql) override;
+    bool allows_unpinned_distributed_2pc() const override;
     void disconnect_all();
 
 private:

include/sql_engine/mysql_remote_executor.h

@@ -21,6 +21,7 @@ class MySQLRemoteExecutor : public RemoteExecutor {
     void add_backend(const BackendConfig& config);
     ResultSet execute(const char* backend_name, sql_parser::StringRef sql) override;
     DmlResult execute_dml(const char* backend_name, sql_parser::StringRef sql) override;
+    bool allows_unpinned_distributed_2pc() const override { return true; }
     void disconnect_all();
 
 private:

include/sql_engine/pgsql_remote_executor.h

@@ -21,6 +21,7 @@ class PgSQLRemoteExecutor : public RemoteExecutor {
     void add_backend(const BackendConfig& config);
     ResultSet execute(const char* backend_name, sql_parser::StringRef sql) override;
     DmlResult execute_dml(const char* backend_name, sql_parser::StringRef sql) override;
+    bool allows_unpinned_distributed_2pc() const override { return true; }
     void disconnect_all();
 
 private:

include/sql_engine/remote_executor.h

@@ -22,18 +22,23 @@ class RemoteExecutor {
         return r;
     }
 
+    // Returns true only when this executor can safely run distributed 2PC
+    // without pinned sessions. This is valid for executors that keep one
+    // durable physical connection per backend for the full lifetime of the
+    // executor. Pooled executors must return false and rely on
+    // checkout_session() instead.
+    virtual bool allows_unpinned_distributed_2pc() const { return false; }
+
     // Check out a pinned session for `backend_name`. The returned session
     // owns a specific physical connection for its lifetime; every call on
     // it goes to that same connection. On destruction the connection is
     // returned to the pool (or closed if the session was poisoned).
     //
-    // REQUIRED for correct 2PC behavior: MySQL XA and PostgreSQL PREPARE
-    // TRANSACTION both need the transaction's DML and PREPARE to share a
-    // session. Executors that cannot honor pinning (or don't need to --
-    // e.g. a single-connection-per-backend executor) may return nullptr;
-    // DistributedTransactionManager detects that and falls back to the
-    // unpinned path, which works for simple single-backend transactions
-    // but silently corrupts real multi-backend 2PC.
+    // REQUIRED for correct pooled 2PC behavior: MySQL XA and PostgreSQL
+    // PREPARE TRANSACTION both need the transaction's DML and PREPARE to
+    // share a session. Executors may return nullptr only if they also
+    // override allows_unpinned_distributed_2pc() to declare the legacy
+    // single-connection fallback safe for their execution model.
     virtual std::unique_ptr<RemoteSession> checkout_session(const char* backend_name) {
         (void)backend_name;
         return nullptr;

src/sql_engine/multi_remote_executor.cpp

@@ -43,6 +43,10 @@ DmlResult MultiRemoteExecutor::execute_dml(const char* backend_name, sql_parser:
     }
 }
 
+bool MultiRemoteExecutor::allows_unpinned_distributed_2pc() const {
+    return true;
+}
+
 void MultiRemoteExecutor::disconnect_all() {
     mysql_exec_.disconnect_all();
     pgsql_exec_.disconnect_all();