fix(objectql): ScopedContext.transaction 补齐 ADR-0067 D2 的 ambient join (#6168)

.changeset/scoped-context-transaction-ambient-join.md

@@ -0,0 +1,54 @@
+---
+"@objectstack/objectql": patch
+---
+
+fix(objectql): `ctx.api.transaction()` joins an open ambient transaction instead of opening a second one (#6168)
+
+`ObjectQL.transaction()` has always started with the ADR-0067 D2 join: if an
+ambient transaction is already open, it runs the callback inside that one and
+reports `owned: false` rather than beginning a nested driver transaction.
+`ScopedContext.transaction()` — the second implementation of the same
+primitive, reached as `ctx.api.transaction(fn)` from hook and action bodies —
+did not. It went straight to the default driver and called `beginTransaction()`
+unconditionally.
+
+Its own TSDoc called it "a second implementation of the same thing" and lined
+up against ADR-0119 D1's caveats one by one; the join was the single point that
+never got aligned. That is now fixed, with the same branch, in the same
+position — before the driver lookup and before `opts.require`, because an
+ambient transaction *is* a transaction and a caller who declared they cannot run
+without one is served by joining it.
+
+**Behaviour change — a nested sandbox/hook transaction now rolls back with the
+outer one.** Previously a hook fired from inside an `engine.transaction()`
+whose body called `ctx.api.transaction(fn)` got a **separate** driver
+transaction. That transaction committed itself, so its writes **survived a
+rollback of the outer one**: the caller was told the unit of work had been
+undone while some of its rows were still there, with nothing failing and
+nothing logged. It also took a second connection for the duration — the
+deadlock ADR-0067 D2 exists to avoid on a single-connection pool such as
+SQLite's. After this change the inner call joins, writes on the outer handle,
+and is undone by the outer rollback.
+
+If you have a hook or action body that used `ctx.api.transaction()` inside a
+larger transaction *specifically* to get an independently-committing unit —
+an audit trail that must outlive a rollback, say — it no longer does. The
+supported way to have a write survive a rollback is the ADR-0057 §3.6 system
+ledger carve-out (`lifecycle.class` of `audit` / `telemetry` / `event`), which
+routes the row to its own datasource and executes it outside the transaction by
+decision rather than by accident.
+
+The callback's `owned` signal (#5696) now reports `false` on this path, as it
+already did on the engine surface. It was never wrong before — this surface
+really did always open its own transaction — but what it honestly described was
+the defect.
+
+Two limits stay as they are, and are now stated in the method's TSDoc. The join
+reads the engine's ambient `AsyncLocalStorage` store only, so the discrete
+`beginTransaction`/`commit`/`rollback` trio — which deliberately never
+populates that store, because its handle is threaded explicitly across
+`setImmediate` boundaries — is invisible to it and is not joined. That is what
+keeps the branch from mistaking an explicitly-threaded handle for an ambient
+one. The QuickJS sandbox drives its VM-side `ctx.api.transaction(fn)` through
+that trio rather than through this method, so a VM-side body is outside this
+join; unattributable handles are tracked separately in #6167.

packages/objectql/src/engine-ambient-transaction.test.ts

@@ -7,7 +7,7 @@
 // and deadlock on the single-connection SQLite pool.
 
 import { describe, it, expect, beforeEach } from 'vitest';
-import { ObjectQL } from './engine.js';
+import { ObjectQL, ScopedContext } from './engine.js';
 
 function makeRecordingDriver() {
   const stores = new Map<string, Map<string, any>>();
@@ -133,3 +133,273 @@ describe('engine ambient transaction (ADR-0034)', () => {
     expect(seen.commit.length).toBe(0);
   });
 });
+
+// ---------------------------------------------------------------------------
+// #6168 — the SECOND implementation of the same primitive joins too
+// ---------------------------------------------------------------------------
+//
+// `ScopedContext.transaction` — what a hook or action body reaches as
+// `ctx.api.transaction(fn)` — had no ADR-0067 D2 join branch. Inside an
+// `engine.transaction()` it opened a SECOND driver transaction: a second
+// connection (the deadlock D2 exists to avoid on a single-connection pool),
+// committed independently, and therefore NOT covered by the outer rollback.
+//
+// The driver below is deliberately ROLLBACK-HONEST where the one above is only
+// rollback-RECORDING: writes carrying a handle are staged per handle, `commit`
+// flushes them into the committed store and `rollback` discards them. That is
+// the difference between pinning "rollback was called" and pinning the fact
+// this issue is actually about — whether the inner row is still there
+// afterwards. With the join removed, the inner transaction commits its own
+// stage and the row SURVIVES the outer rollback; the assertion that fails is
+// the one on committed state, in the shape of leftover residue.
+
+function makeRollbackHonestDriver() {
+  const committed = new Map<string, Map<string, any>>();
+  const staged = new Map<unknown, Array<{ object: string; row: any }>>();
+  const seen = {
+    begins: [] as unknown[],
+    commits: [] as unknown[],
+    rollbacks: [] as unknown[],
+    creates: [] as Array<{ object: string; transaction: unknown }>,
+  };
+  let nextId = 0;
+  let nextTrx = 0;
+  const committedFor = (o: string) => {
+    let s = committed.get(o);
+    if (!s) { s = new Map(); committed.set(o, s); }
+    return s;
+  };
+  const driver: any = {
+    name: 'memory',
+    version: '0.0.0',
+    supports: {},
+    async connect() {},
+    async disconnect() {},
+    async checkHealth() { return true; },
+    async execute() { return null; },
+    // Reads see COMMITTED state only. Nothing in these cases reads back its own
+    // uncommitted write, and keeping it simple keeps the durability assertion
+    // unambiguous: what `find` returns at the end is what actually landed.
+    async find(object: string) { return Array.from(committedFor(object).values()); },
+    async findOne(object: string) {
+      for (const r of committedFor(object).values()) return r;
+      return null;
+    },
+    async create(object: string, data: Record<string, unknown>, options: any) {
+      const trx = options?.transaction;
+      seen.creates.push({ object, transaction: trx });
+      nextId += 1;
+      const id = (data.id as string) ?? `r_${nextId}`;
+      const row = { ...data, id };
+      if (trx === undefined) committedFor(object).set(id, row);
+      else staged.set(trx, [...(staged.get(trx) ?? []), { object, row }]);
+      return row;
+    },
+    async update(object: string, id: string, data: Record<string, unknown>) {
+      const s = committedFor(object);
+      const row = { ...s.get(id), ...data, id };
+      s.set(id, row);
+      return row;
+    },
+    async delete(object: string, id: string) { return committedFor(object).delete(id); },
+    async count() { return 0; },
+    async bulkCreate(object: string, rows: Record<string, unknown>[]) {
+      return Promise.all(rows.map((r) => this.create(object, r, undefined)));
+    },
+    async bulkUpdate() { return []; },
+    async bulkDelete() {},
+    async syncSchema() {},
+    async beginTransaction() {
+      nextTrx += 1;
+      const handle = { __trx: nextTrx };
+      seen.begins.push(handle);
+      staged.set(handle, []);
+      return handle;
+    },
+    async commit(trx: unknown) {
+      seen.commits.push(trx);
+      for (const { object, row } of staged.get(trx) ?? []) committedFor(object).set(row.id, row);
+      staged.delete(trx);
+    },
+    async rollback(trx: unknown) {
+      seen.rollbacks.push(trx);
+      staged.delete(trx);
+    },
+  };
+  const committedNames = (object: string) =>
+    Array.from(committedFor(object).values()).map((r) => r.name).sort();
+  return { driver, seen, committedNames };
+}
+
+describe('ScopedContext.transaction joins the ambient transaction (ADR-0067 D2, #6168)', () => {
+  let engine: ObjectQL;
+  let seen: ReturnType<typeof makeRollbackHonestDriver>['seen'];
+  let committedNames: ReturnType<typeof makeRollbackHonestDriver>['committedNames'];
+
+  beforeEach(async () => {
+    engine = new ObjectQL();
+    const d = makeRollbackHonestDriver();
+    seen = d.seen;
+    committedNames = d.committedNames;
+    engine.registerDriver(d.driver, true);
+    await engine.init();
+    engine.registry.registerObject({ name: 'thing', fields: { name: { type: 'text' } } } as any);
+  });
+
+  /**
+   * A hook body's `ctx.api.transaction(fn)`, registered on `thing` and fired by
+   * a write made inside `engine.transaction()`. This is the real reachability
+   * path — `HookContext.api` IS a `ScopedContext` (`ObjectQL.buildHookApi`) —
+   * not a `createContext` stand-in for one.
+   */
+  function hookThatOpensATransaction(
+    body: (trxCtx: any, info: any) => Promise<void>,
+    onlyFor = 'outer',
+  ) {
+    (engine as any).registerHook(
+      'afterInsert',
+      async (ctx: any) => {
+        // The inner write fires this hook again; without the guard the join
+        // would be measured against a recursion, not against the outer call.
+        if (ctx.result?.name !== onlyFor) return;
+        await ctx.api.transaction(body);
+      },
+      { object: 'thing' },
+    );
+  }
+
+  it('joins the outer transaction — same handle, owned: false, no second begin', async () => {
+    let innerHandle: unknown;
+    let innerOwned: boolean | undefined;
+    let outerHandle: unknown;
+
+    hookThatOpensATransaction(async (trxCtx: any, info: any) => {
+      innerHandle = trxCtx.transactionHandle;
+      innerOwned = info.owned;
+    });
+
+    await engine.transaction(async (ctx: any) => {
+      outerHandle = ctx.transaction;
+      await engine.insert('thing', { name: 'outer' });
+    });
+
+    expect(innerOwned).toBe(false);
+    expect(innerHandle).toBe(outerHandle);
+    // The whole point of D2: ONE begin, so ONE connection.
+    expect(seen.begins).toHaveLength(1);
+    expect(seen.commits).toHaveLength(1);
+  });
+
+  it('the joined write is UNDONE by the outer rollback — the durability pin', async () => {
+    hookThatOpensATransaction(async (trxCtx: any) => {
+      await trxCtx.object('thing').insert({ name: 'inner' });
+    });
+
+    await expect(
+      engine.transaction(async () => {
+        await engine.insert('thing', { name: 'outer' });
+        throw new Error('outer boom');
+      }),
+    ).rejects.toThrow('outer boom');
+
+    // Before #6168 the inner transaction committed itself, so 'inner' was still
+    // here after the outer rollback: a write the caller was told had been undone
+    // and had not been. Nothing failed, nothing was logged.
+    expect(committedNames('thing')).toEqual([]);
+    expect(seen.rollbacks).toHaveLength(1);
+    expect(seen.commits).toHaveLength(0);
+    // Both writes rode the ONE handle the outer call owns.
+    expect(seen.creates).toHaveLength(2);
+    expect(seen.creates[1].transaction).toBe(seen.creates[0].transaction);
+  });
+
+  it('the joined write commits with the outer one when the outer succeeds', async () => {
+    hookThatOpensATransaction(async (trxCtx: any) => {
+      await trxCtx.object('thing').insert({ name: 'inner' });
+    });
+
+    await engine.transaction(async () => {
+      await engine.insert('thing', { name: 'outer' });
+    });
+
+    expect(committedNames('thing')).toEqual(['inner', 'outer']);
+    expect(seen.begins).toHaveLength(1);
+  });
+
+  it('with NO ambient transaction it still OPENS one — owned: true, unchanged', async () => {
+    const scoped = (engine as any).createContext({ userId: 'u1' }) as ScopedContext;
+    let owned: boolean | undefined;
+
+    await scoped.transaction(async (trxCtx: any, info: any) => {
+      owned = info.owned;
+      await trxCtx.object('thing').insert({ name: 'standalone' });
+    });
+
+    expect(owned).toBe(true);
+    expect(seen.begins).toHaveLength(1);
+    expect(seen.commits).toHaveLength(1);
+    expect(committedNames('thing')).toEqual(['standalone']);
+  });
+
+  /**
+   * The OTHER half of ADR-0067 D2's rationale: the second `beginTransaction`
+   * asks the pool for a second connection, and on a single-connection pool
+   * (the knex/SQLite one D2 names) there is no second connection to give.
+   *
+   * The real pool BLOCKS there — that is the deadlock — and a test that
+   * modelled the block faithfully would fail only by hitting vitest's timeout:
+   * slow, and flaky under parallel load. So this double models a pool of size 1
+   * WITH an acquire timeout, which refuses the second checkout instead of
+   * queueing for it. The refusal is a stand-in for the hang; what is measured
+   * honestly either way is the thing that causes both — whether a second
+   * connection is asked for at all.
+   */
+  it('never asks for a second connection — a single-connection pool survives the nested call', async () => {
+    let checkedOut = false;
+    const checkouts: number[] = [];
+    const d = makeRollbackHonestDriver();
+    d.driver.beginTransaction = async () => {
+      if (checkedOut) {
+        // Where a real single-connection pool would wait forever.
+        throw new Error('pool exhausted: no connection available (max=1)');
+      }
+      checkedOut = true;
+      checkouts.push(checkouts.length + 1);
+      return { __trx: 'only' };
+    };
+    d.driver.commit = async () => { checkedOut = false; };
+    d.driver.rollback = async () => { checkedOut = false; };
+
+    const oneConn = new ObjectQL();
+    oneConn.registerDriver(d.driver, true);
+    await oneConn.init();
+    oneConn.registry.registerObject({ name: 'thing', fields: { name: { type: 'text' } } } as any);
+    (oneConn as any).registerHook(
+      'afterInsert',
+      async (ctx: any) => {
+        if (ctx.result?.name !== 'outer') return;
+        await ctx.api.transaction(async () => { /* joins — asks for nothing */ });
+      },
+      { object: 'thing' },
+    );
+
+    await expect(
+      oneConn.transaction(async () => { await oneConn.insert('thing', { name: 'outer' }); }),
+    ).resolves.toBeUndefined();
+
+    expect(checkouts).toHaveLength(1);
+  });
+
+  it('does not join a transaction that has already closed — the store does not leak', async () => {
+    await engine.transaction(async () => {
+      await engine.insert('thing', { name: 'covered' });
+    });
+
+    const scoped = (engine as any).createContext({ userId: 'u1' }) as ScopedContext;
+    let owned: boolean | undefined;
+    await scoped.transaction(async (_ctx: any, info: any) => { owned = info.owned; });
+
+    expect(owned).toBe(true);
+    expect(seen.begins).toHaveLength(2); // the outer one, then a fresh one
+  });
+});

packages/objectql/src/engine-transaction-contract.test.ts

@@ -309,7 +309,7 @@ describe('ScopedContext.transaction (ctx.api.transaction) carries the same two p
     await expect(scopedOf(engine).transaction(async () => 'ran')).resolves.toBe('ran');
   });
 
-  it('reports owned: true (this surface always opens) and false when degraded', async () => {
+  it('reports owned: true when it opens one, and false when degraded', async () => {
     const withTx = await engineWith({ transactional: true });
     const withoutTx = await engineWith({ transactional: false });
     let ownedOpen: boolean | undefined;
@@ -321,4 +321,48 @@ describe('ScopedContext.transaction (ctx.api.transaction) carries the same two p
     expect(ownedOpen).toBe(true);
     expect(ownedDegraded).toBe(false);
   });
+
+  // The third point of parity, added by #6168. Until then this was the ONE
+  // place the second implementation still diverged: it had no ADR-0067 D2 join
+  // branch, so `owned` reported `true` here forever — honest about a behaviour
+  // that was not. Durability coverage lives in `engine-ambient-transaction.test.ts`
+  // (the inner write must be undone by the outer rollback); what is pinned here
+  // is the CONTRACT parity — same handle, same signal, same ordering.
+  it('JOINS an ambient transaction rather than opening a second one (ADR-0067 D2, #6168)', async () => {
+    const { engine, driver } = await engineWith({ transactional: true });
+    let outerHandle: unknown;
+    let innerHandle: unknown;
+    let innerOwned: boolean | undefined;
+
+    await engine.transaction(async (ctx: any) => {
+      outerHandle = ctx.transaction;
+      await scopedOf(engine).transaction(async (trxCtx, info) => {
+        innerHandle = (trxCtx as unknown as { transactionHandle: unknown }).transactionHandle;
+        innerOwned = info.owned;
+      });
+    });
+
+    // `beginTransaction` mints a fresh object per call, so identity is what
+    // separates "joined the outer handle" from "opened an identical-looking one".
+    expect(innerOwned).toBe(false);
+    expect(innerHandle).toBe(outerHandle);
+    expect(driver.writes).toHaveLength(0);
+  });
+
+  it('joins under require: true — an ambient transaction satisfies it, nothing is refused', async () => {
+    const { engine } = await engineWith({ transactional: true });
+    let owned: boolean | undefined;
+
+    // The join sits BEFORE the driver/`require` handling on both surfaces, and
+    // it must: when there is an ambient transaction there IS a transaction, so
+    // a caller who declared it cannot run without one is served by joining.
+    await engine.transaction(async () => {
+      await scopedOf(engine).transaction(
+        async (_ctx, info) => { owned = info.owned; },
+        { require: true },
+      );
+    });
+
+    expect(owned).toBe(false);
+  });
 });