fix: ADR 0050 callback-error precedence, and the honest limits of transaction control

Narrow contract and implementation adjustments from review. No SDK, manifest,
packaging, or Signage work.

Callback-error precedence:
- A transaction or savepoint callback that rejects keeps its own error, even
  while sibling work it started is still settling (one failed half of a
  Promise.all). Core drains that work, rolls back, and never commits or
  releases.
- A callback that resolves with work outstanding (a forgotten await) still
  fails with MODULE_DATABASE_SCOPE_WORK_OUTSTANDING and rolls back.

Transaction-control SQL:
- ADR 0050 forbids it through query() as a conformance rule for trusted code,
  not a security guarantee. Core does not parse SQL, so PostgreSQL has run
  the statement before Core sees it. Core detects the violation from the
  command tag or the transaction status, and claims nothing it cannot do: a
  module's own COMMIT has committed.
- After a violation, the transaction's scopes refuse every later statement
  without sending it (it would otherwise autocommit). Core never commits the
  transaction itself, and the connection is destroyed rather than reused,
  for top-level violations too.
- The commit and rollback guarantees apply to conforming use.

Duplicate object-mode columns: documented as found only after execution, so a
mutation can have taken effect; atomicity needs an ABI transaction.

Tests prove each of these against PostgreSQL: Promise.all failures in a
transaction and a savepoint keep 23505; a forgotten await still rolls back; a
module COMMIT's prior work is committed while nothing after it runs; a backend
PID changes after every violation; duplicate RETURNING columns persist at the
top level and roll back in a transaction.

ADR 0050 stays Accepted: the executor conforms to the revised text.

Co-Authored-By: Claude Opus 5.5 <[email protected]>

core/tests/module-database-abi.integration.ts

@@ -228,6 +228,28 @@ describe("query", () => {
     );
   });
 
+  it("finds duplicate object-mode columns only after the statement ran, so atomicity needs a transaction", async () => {
+    await reset();
+    // At the top level, the mutation has happened by the time the result's shape is known.
+    await expectAbiError(
+      database.query({
+        text: "INSERT INTO example_records (id, value) VALUES (1, 'a') RETURNING id, id",
+      }),
+      "MODULE_DATABASE_REQUEST_INVALID",
+    );
+    expect(await storedIds()).toEqual([1]);
+    // Inside a transaction, the same error rolls the mutation back with everything else.
+    await expectAbiError(
+      database.transaction((tx) =>
+        tx.query({
+          text: "INSERT INTO example_records (id, value) VALUES (2, 'b') RETURNING id, id",
+        }),
+      ),
+      "MODULE_DATABASE_REQUEST_INVALID",
+    );
+    expect(await storedIds()).toEqual([1]);
+  });
+
   it("keeps a column named __proto__ a column", async () => {
     const result = await database.query({ text: 'SELECT 1 AS "__proto__"' });
     const row = result.rows[0] as Record<string, unknown>;
@@ -838,22 +860,25 @@ describe("transactions", () => {
     expect(called).toBe(false);
   });
 
-  it("refuses transaction control sent through query, and never commits the transaction", async () => {
-    await reset();
-    for (const statements of [
-      ["SAVEPOINT mine"],
-      ["SAVEPOINT mine", "RELEASE SAVEPOINT mine"],
-      ["SAVEPOINT mine", "ROLLBACK TO SAVEPOINT mine"],
-      ["BEGIN"],
-      ["START TRANSACTION"],
-    ]) {
+  it("detects transaction control sent through query once PostgreSQL has run it, then sends nothing more and commits nothing", async () => {
+    // Forbidden by ADR 0050, and not refusable beforehand without parsing SQL. Core detects each
+    // from PostgreSQL's command tag or the protocol's transaction status.
+    for (const statement of ["SAVEPOINT mine", "BEGIN", "START TRANSACTION", "ROLLBACK", "ABORT"]) {
+      await reset();
       await expectAbiError(
         database.transaction(async (tx) => {
           await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
-          for (const statement of statements) {
-            // PostgreSQL has run each statement by the time it is refused.
-            await expectAbiError(tx.query({ text: statement }), "MODULE_DATABASE_REQUEST_INVALID");
-          }
+          await expectAbiError(tx.query({ text: statement }), "MODULE_DATABASE_REQUEST_INVALID");
+          // Anything after the violation is refused without being sent: it would otherwise run
+          // outside any transaction.
+          await expectAbiError(
+            tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" }),
+            "MODULE_DATABASE_REQUEST_INVALID",
+          );
+          await expectAbiError(
+            tx.savepoint(async () => undefined),
+            "MODULE_DATABASE_REQUEST_INVALID",
+          );
           return "carried on";
         }),
         "MODULE_DATABASE_REQUEST_INVALID",
@@ -862,15 +887,25 @@ describe("transactions", () => {
     }
   });
 
-  it("refuses a COMMIT sent through query and fails the transaction", async () => {
-    await reset();
-    await expectAbiError(
-      database.transaction(async (tx) => {
-        await expectAbiError(tx.query({ text: "COMMIT" }), "MODULE_DATABASE_REQUEST_INVALID");
-        return "carried on";
-      }),
-      "MODULE_DATABASE_REQUEST_INVALID",
-    );
+  it("cannot undo a COMMIT sent through query: PostgreSQL has committed, and Core reports the violation and commits nothing more", async () => {
+    for (const statement of ["COMMIT", "END"]) {
+      await reset();
+      await expectAbiError(
+        database.transaction(async (tx) => {
+          await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
+          await expectAbiError(tx.query({ text: statement }), "MODULE_DATABASE_REQUEST_INVALID");
+          await expectAbiError(
+            tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" }),
+            "MODULE_DATABASE_REQUEST_INVALID",
+          );
+          return "carried on";
+        }),
+        "MODULE_DATABASE_REQUEST_INVALID",
+      );
+      // The documented limit: the work before the module's own COMMIT was committed by PostgreSQL
+      // when it ran that statement. Nothing after it ran.
+      expect(await storedIds()).toEqual([1]);
+    }
   });
 
   it("applies Core's statement bound inside a transaction, and transaction-local settings work", async () => {
@@ -917,6 +952,65 @@ describe("outstanding work", () => {
     expect(await storedIds()).toEqual([]);
   });
 
+  it("keeps the failing query's own error when one half of a Promise.all fails while the other is still running", async () => {
+    await reset();
+    await database.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
+    let sibling: Promise<unknown> | undefined;
+    await expectAbiError(
+      database.transaction(async (tx) => {
+        const conflicting = tx.query({
+          text: "INSERT INTO example_records (id, value) VALUES (1, 'dup')",
+        });
+        sibling = tx.query({
+          text: "INSERT INTO example_records (id, value) SELECT 2, 'b' FROM pg_sleep(0.1)",
+        });
+        sibling.catch(() => undefined);
+        await Promise.all([conflicting, sibling]);
+      }),
+      // The PostgreSQL failure, not MODULE_DATABASE_SCOPE_WORK_OUTSTANDING.
+      "MODULE_DATABASE_QUERY_FAILED",
+      "23505",
+    );
+    // The sibling was still outstanding when the callback rejected: it settled afterwards (failing,
+    // because the transaction was already aborted), before Core rolled back.
+    await expect(sibling).rejects.toMatchObject({ sqlState: "25P02" });
+    expect(await storedIds()).toEqual([1]);
+  });
+
+  it("keeps a savepoint's failing query error from a Promise.all, and the transaction carries on", async () => {
+    await reset();
+    await database.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" });
+    await database.transaction(async (tx) => {
+      await expectAbiError(
+        tx.savepoint(async (sp) => {
+          const sibling = sp.query({
+            text: "INSERT INTO example_records (id, value) SELECT 2, 'b' FROM pg_sleep(0.1)",
+          });
+          sibling.catch(() => undefined);
+          await Promise.all([
+            sp.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'dup')" }),
+            sibling,
+          ]);
+        }),
+        "MODULE_DATABASE_QUERY_FAILED",
+        "23505",
+      );
+      await tx.query({ text: "INSERT INTO example_records (id, value) VALUES (3, 'c')" });
+    });
+    expect(await storedIds()).toEqual([1, 3]);
+  });
+
+  it("commits a Promise.all whose queries all succeed", async () => {
+    await reset();
+    await database.transaction(async (tx) => {
+      await Promise.all([
+        tx.query({ text: "INSERT INTO example_records (id, value) VALUES (1, 'a')" }),
+        tx.query({ text: "INSERT INTO example_records (id, value) VALUES (2, 'b')" }),
+      ]);
+    });
+    expect(await storedIds()).toEqual([1, 2]);
+  });
+
   it("treats an unawaited savepoint as outstanding work", async () => {
     await reset();
     let forgotten: Promise<unknown> | undefined;
@@ -1259,6 +1353,34 @@ describe("pool safety", () => {
     });
   }
 
+  it("destroys, never reuses, a connection after transaction-control misuse", async () => {
+    const pid = async (executor: ModuleDatabaseExecutor | ModuleDatabaseTransaction) =>
+      (
+        (await executor.query({ text: "SELECT pg_backend_pid() AS pid" })).rows[0] as {
+          pid: number;
+        }
+      ).pid;
+    for (const statement of ["COMMIT", "ROLLBACK", "SAVEPOINT mine", "BEGIN"]) {
+      let used: number | undefined;
+      await rejection(
+        single.transaction(async (tx) => {
+          used = await pid(tx);
+          await rejection(tx.query({ text: statement }));
+        }),
+      );
+      expect(used).toBeTypeOf("number");
+      expect(await pid(single)).not.toBe(used);
+      expect(await nextBorrower(onePool)).toEqual(clean);
+    }
+    for (const statement of ["BEGIN", "COMMIT"]) {
+      const before = await pid(single);
+      // With one connection, the next top-level query would reuse it unless it was destroyed.
+      await rejection(single.query({ text: statement }));
+      expect(await pid(single)).not.toBe(before);
+      expect(await nextBorrower(onePool)).toEqual(clean);
+    }
+  });
+
   it("reports a terminated connection without a SQLSTATE of its own making", async () => {
     const error = await rejection(
       single.transaction(async (tx) => {

core/tests/module-database-abi.test.ts

@@ -654,20 +654,25 @@ describe("transaction scopes", () => {
     expect(texts(script)).not.toContain("COMMIT");
   });
 
-  it("reports outstanding work even when the callback itself rejected", async () => {
+  it("keeps a rejected callback's own error when sibling work is still settling, and still never commits", async () => {
     const script = scriptedPool();
     const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" });
+    const own = new Error("module failure");
     let releaseHeld: (() => void) | undefined;
 
     const outcome = executor.transaction(async (tx) => {
       releaseHeld = script.holdNext();
       void tx.query({ text: "INSERT INTO example_records VALUES (1)" });
-      throw new Error("module failure");
+      throw own;
     });
     await vi.waitFor(() => expect(releaseHeld).toBeDefined());
+    // The sibling statement is drained before Core rolls back.
+    await new Promise((resolve) => setTimeout(resolve, 10));
+    expect(texts(script)).not.toContain("ROLLBACK");
     releaseHeld?.();
 
-    await expectAbiError(outcome, "MODULE_DATABASE_SCOPE_WORK_OUTSTANDING");
+    expect(await rejection(outcome)).toBe(own);
+    expect(texts(script).slice(-2)).toEqual(["ROLLBACK", "DISCARD ALL"]);
     expect(texts(script)).not.toContain("COMMIT");
   });
 
@@ -688,6 +693,9 @@ describe("transaction scopes", () => {
     expect(error).toMatchObject({ code: "MODULE_DATABASE_REQUEST_INVALID" });
     expect(texts(script)).not.toContain("COMMIT");
     expect(texts(script)).toContain("ROLLBACK");
+    // The connection is destroyed, never reset and reused.
+    expect(texts(script)).not.toContain("DISCARD ALL");
+    expect(script.released).toEqual([true]);
   });
 
   it("reports a COMMIT that PostgreSQL answered with ROLLBACK as a failed transaction", async () => {
@@ -794,6 +802,32 @@ describe("savepoint scopes", () => {
     ]);
   });
 
+  it("keeps a rejected savepoint callback's own error, rolling back and releasing after its sibling work settles", async () => {
+    const script = scriptedPool();
+    const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" });
+    const own = new Error("savepoint failure");
+
+    await executor.transaction(async (tx) => {
+      const error = await rejection(
+        tx.savepoint(async (sp) => {
+          const release = script.holdNext();
+          void sp.query({ text: "INSERT INTO example_records VALUES (1)" });
+          setTimeout(release, 10);
+          throw own;
+        }),
+      );
+      expect(error).toBe(own);
+    });
+
+    expect(texts(script).slice(2, -2)).toEqual([
+      "SAVEPOINT ekklesia_savepoint_1",
+      "INSERT INTO example_records VALUES (1)",
+      "ROLLBACK TO SAVEPOINT ekklesia_savepoint_1",
+      "RELEASE SAVEPOINT ekklesia_savepoint_1",
+    ]);
+    expect(texts(script)).toContain("COMMIT");
+  });
+
   it("counts an unawaited savepoint as the transaction's outstanding work", async () => {
     const script = scriptedPool();
     const executor = createModuleDatabaseExecutor(script.pool, { moduleId: "example" });