From da78dece011fd5758ce45a758d9722c49ac473c8 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 9 Sep 2026 18:35:10 +0000 Subject: [PATCH] fix(storage): decide the join pushdown on enlistment, not the instance flag MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `planSqlJoin` sent a join to the hash fallback when the right-hand storage sat inside a connection transaction the left side was not enlisted in, by testing `right.inTransaction`. That flag is written only by `setConnectionTransactionActive`, which only `runSingleSessionConnectionTransaction` calls — the SQLite / DuckDB / PGlite arm. `PostgresTabularStorage`'s pooled branch checks out its own client and deliberately flags no participant, so on a real `pg.Pool` the guard was dead: the pushdown ran, read the right-hand table off a different pooled client, and missed the transaction's uncommitted rows. An inner join returned zero rows where the same call on SQLite, or through the hash fallback, returned the row — a wrong answer with no error, differing between backends and between the two strategies for identical inputs. The question is now enlistment, unioned with the flag: `isEnlistedInConnectionTransaction()` defaults to `false` for the arm that sets the flag on every participant, and `PostgresTabularStorage` overrides it with the context check its pooled write path and `connectionTxQuery` already use. The `inTransaction` half stays for the `tx`-proxy case. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01T8DU24G9GWuRUJncc5TJFZ --- .../src/tabular/BaseSqlTabularStorage.ts | 35 ++++- .../PostgresPooledJoinTransaction.test.ts | 142 ++++++++++++++++++ .../src/storage/PostgresTabularStorage.ts | 11 ++ 3 files changed, 185 insertions(+), 3 deletions(-) create mode 100644 packages/test/src/test/storage-tabular/PostgresPooledJoinTransaction.test.ts diff --git a/packages/storage/src/tabular/BaseSqlTabularStorage.ts b/packages/storage/src/tabular/BaseSqlTabularStorage.ts index d554febaf..d66fcaeb2 100644 --- a/packages/storage/src/tabular/BaseSqlTabularStorage.ts +++ b/packages/storage/src/tabular/BaseSqlTabularStorage.ts @@ -185,6 +185,30 @@ export abstract class BaseSqlTabularStorage< this.inTransaction = active; } + /** + * Whether a connection-scoped transaction this instance is enlisted in is + * open in the current async context. + * + * {@link inTransaction} does not answer this on every backend. It is an + * instance flag, and a pooled backend deliberately sets it on nobody: its + * transaction runs on one checked-out client while the same instances keep + * serving unrelated callers off the pool, so enlistment is a property of the + * async context rather than of the instance. Such a backend overrides this + * with the context question; the default is `false`, which is right for the + * single-session arm, where every participant carries the flag instead. + * + * The two answers are unioned wherever the question is "is this instance + * inside an open transaction", so neither arm has to be special-cased twice. + */ + protected isEnlistedInConnectionTransaction(): boolean { + return false; + } + + /** Either way an instance can be inside an open transaction. */ + private insideOpenTransaction(): boolean { + return this.inTransaction || this.isEnlistedInConnectionTransaction(); + } + /** * Per-instance promise-chain mutex. On a backend that runs every statement * on one shared connection, all public read/write methods queue behind the @@ -747,9 +771,14 @@ export abstract class BaseSqlTabularStorage< // the right side is enlisted on and miss its uncommitted rows entirely. // The hash fallback goes through `right.query()`, which does serialize on // the right's own lock and does resolve its enlisted client, so hand this - // case to it. `this.inTransaction` means the join is already inside that - // transaction, where reading its own uncommitted rows is correct. - if (right.inTransaction && !this.inTransaction) return null; + // case to it. The left side being inside that same transaction is the + // exception: reading its own uncommitted rows there is correct. + // + // Asked through {@link insideOpenTransaction} rather than the + // `inTransaction` flag alone, because the pooled arm — the one whose + // uncommitted rows sit on another client entirely — sets that flag on + // nobody. + if (right.insideOpenTransaction() && !this.insideOpenTransaction()) return null; return { right, dialect: mine.dialect }; } diff --git a/packages/test/src/test/storage-tabular/PostgresPooledJoinTransaction.test.ts b/packages/test/src/test/storage-tabular/PostgresPooledJoinTransaction.test.ts new file mode 100644 index 000000000..5dc735993 --- /dev/null +++ b/packages/test/src/test/storage-tabular/PostgresPooledJoinTransaction.test.ts @@ -0,0 +1,142 @@ +/** + * @license + * Copyright 2026 Steven Roussey + * SPDX-License-Identifier: Apache-2.0 + */ + +import { PGlite } from "@electric-sql/pglite"; +import { PostgresTabularStorage } from "@workglow/postgres/storage"; +import { withConnectionTransaction } from "@workglow/storage"; +import type { Pool } from "pg"; +import { describe, expect, it } from "vitest"; + +const RowSchema = { + type: "object", + properties: { + name: { type: "string" }, + tag: { type: "string" }, + }, + required: ["name", "tag"], + additionalProperties: false, +} as const; + +const RowPrimaryKeyNames = ["name"] as const; + +type RowStorage = PostgresTabularStorage; + +/** + * A `pg.Pool`-shaped facade over one PGlite session. + * + * `PostgresTabularStorage` decides which arm it is on by whether the db it was + * given has `connect()`, so this puts it on the POOLED arm — the one whose + * connection transaction checks out its own client and therefore flags no + * participant with `inTransaction`. Every statement still runs on the single + * PGlite session underneath, which is what makes the arm reachable without a + * live Postgres: what is under test is which join strategy gets chosen from + * that state, not the isolation a real pool would then impose. + */ +function pooledFacade(db: PGlite): Pool { + const query = db.query.bind(db); + return { + query, + connect: async () => ({ query, release: (): void => {} }), + } as unknown as Pool; +} + +/** + * Counts calls to `query` on one storage without a spy, so this file needs + * nothing from Vitest's module registry and runs unchanged under either runner. + */ +function countQueries(storage: RowStorage): () => number { + let calls = 0; + const original = storage.query.bind(storage); + (storage as { query: RowStorage["query"] }).query = ((...args: Parameters) => { + calls += 1; + return original(...args); + }) as RowStorage["query"]; + return () => calls; +} + +describe("join pushdown against a pooled connection transaction", () => { + it("hands the join to the hash fallback when only the right side is enlisted", async () => { + // The pushdown reads the right table while holding only the left's lock, so + // on a real pool it runs on a different client from the one the right side + // is enlisted on and cannot see that transaction's uncommitted rows — an + // inner join returning zero rows where the hash fallback returns the row. + // The fallback goes through `right.query()`, which resolves the enlisted + // client, so the strategy chosen is the whole question here. Only the + // right side is a participant; a join is a read, so nothing else refuses it. + const db = new PGlite(); + const pool = pooledFacade(db); + const left: RowStorage = new PostgresTabularStorage( + pool, + "pooled_join_left", + RowSchema, + RowPrimaryKeyNames + ); + const right: RowStorage = new PostgresTabularStorage( + pool, + "pooled_join_right", + RowSchema, + RowPrimaryKeyNames + ); + await left.setupDatabase(); + await right.setupDatabase(); + await left.put({ name: "a", tag: "left-row" }); + + const rightQueries = countQueries(right); + + await withConnectionTransaction([right], async () => { + await right.put({ name: "a", tag: "uncommitted" }); + + const rows = await left.join( + { + type: "inner", + on: [{ left: "name", right: "name" }], + orderBy: [{ side: "left", column: "name", direction: "ASC" }], + }, + right + ); + + expect(rows.map((row) => `${row.left.tag}:${row.right.tag}`)).toEqual([ + "left-row:uncommitted", + ]); + }); + + expect(rightQueries()).toBeGreaterThan(0); + }); + + it("still pushes down when neither side is in a transaction", async () => { + const db = new PGlite(); + const pool = pooledFacade(db); + const left: RowStorage = new PostgresTabularStorage( + pool, + "pooled_plain_left", + RowSchema, + RowPrimaryKeyNames + ); + const right: RowStorage = new PostgresTabularStorage( + pool, + "pooled_plain_right", + RowSchema, + RowPrimaryKeyNames + ); + await left.setupDatabase(); + await right.setupDatabase(); + await left.put({ name: "a", tag: "left-row" }); + await right.put({ name: "a", tag: "right-row" }); + + const rightQueries = countQueries(right); + const rows = await left.join( + { + type: "inner", + on: [{ left: "name", right: "name" }], + orderBy: [{ side: "left", column: "name", direction: "ASC" }], + }, + right + ); + + expect(rows.map((row) => `${row.left.tag}:${row.right.tag}`)).toEqual(["left-row:right-row"]); + expect(rightQueries()).toBe(0); + }); +}); diff --git a/providers/postgres/src/storage/PostgresTabularStorage.ts b/providers/postgres/src/storage/PostgresTabularStorage.ts index 78b15edfb..8c30a23fa 100644 --- a/providers/postgres/src/storage/PostgresTabularStorage.ts +++ b/providers/postgres/src/storage/PostgresTabularStorage.ts @@ -1045,6 +1045,17 @@ export class PostgresTabularStorage< }); } + /** + * The real-pool arm sets `inTransaction` on nobody — its transaction runs on + * a checked-out client while these instances keep serving other callers off + * the pool — so the async context is the only thing that can say whether this + * instance is enlisted. The same question `_putBulkInternal` and + * `withTransaction` already ask on this path. + */ + protected override isEnlistedInConnectionTransaction(): boolean { + return isEnlistedInConnectionTx(this); + } + /** * A real `pg.Pool` isolates per checked-out client, so serializing through * the base's per-instance chain would turn the pool's main benefit into a