From 6cbaaf97f00db1b18f398c758bbaa42608fc65e9 Mon Sep 17 00:00:00 2001 From: alfredorubin96 Date: Mon, 27 Jul 2026 23:44:54 +0200 Subject: [PATCH 1/3] fix(connection): drain PostgreSQL writes through a cursor instead of buffering MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The write branch called client.query(), which materialises the entire result set before the row limit is applied: const result = await client.query(query, paramValues); fetchedRows = result.rows; // whole set, then sliced later So rowLimit bounded what was displayed, not what was resident. One Form submit against a large table could exhaust the heap shared by every tenant on the process. The obvious fix is the dangerous one. readBoundedCursor performs ONE bounded read and closes the cursor in its finally — right for a SELECT, but PostgreSQL applies an UPDATE ... RETURNING incrementally, so rows never pulled are never modified and closing the portal abandons them. Reusing it here would silently turn "update 1,000 rows" into "update 11" and still report success. New drainBoundedCursor reads to exhaustion — every row produced, every side effect run — while retaining only maxRows. Peak memory becomes the batch size plus maxRows instead of the whole result. One subtlety the integration test caught: the completing batch of a RETURNING statement reports rowCount 0, which overwrote the real count and made the module report NO_DATA for an UPDATE that changed a thousand rows. affectedRowCount is now reported only when the statement returned no rows — exactly the INSERT-without-RETURNING case it exists for. Three integration tests against a real PostgreSQL, because a stubbed client cannot prove any of this: - UPDATE ... RETURNING over 1000 rows with rowLimit 10 returns <= 10 rows AND leaves all 1000 updated. This is the regression guard: it fails on any implementation that stops reading early. - INSERT without RETURNING still reports its affected-row count, not NO_DATA. - a write exceeding the limit still flags COMPLETE_TRUNCATED. postgres-error-handling.test.ts adapted: its failing-INSERT simulation drove client.query, which the write path no longer uses. The failure now originates in drainBoundedCursor; the test's subject — a failed query whose ROLLBACK also fails must still surface the ORIGINAL error — is unchanged. Verified: 8 pg suites, 72 tests. Closes #1326 Refs #1298 Co-Authored-By: Claude Opus 5 --- .../postgres-error-handling.test.ts | 18 +++ .../postgresql/postgres-write-drain.ts | 140 ++++++++++++++++++ .../postgresql/PostgresConnectionModule.ts | 20 ++- connection/src/postgresql/cursor-read.ts | 99 +++++++++++++ 4 files changed, 272 insertions(+), 5 deletions(-) create mode 100644 connection/__tests__/postgresql/postgres-write-drain.ts diff --git a/connection/__tests__/postgresql/postgres-error-handling.test.ts b/connection/__tests__/postgresql/postgres-error-handling.test.ts index 17d2ac69..00a4d559 100644 --- a/connection/__tests__/postgresql/postgres-error-handling.test.ts +++ b/connection/__tests__/postgresql/postgres-error-handling.test.ts @@ -17,6 +17,15 @@ import { // (which can't back a real pg-cursor) still exercises the transaction logic. jest.mock("../../src/postgresql/cursor-read", () => ({ readBoundedCursor: jest.fn().mockResolvedValue({ rows: [], fields: [] }), + // Writes drain rather than stopping early (#1298); the write path calls + // this one, so the double has to provide it or every write test fails on + // "drainBoundedCursor is not a function" rather than on its own assertion. + drainBoundedCursor: jest.fn().mockResolvedValue({ + rows: [], + fields: [], + affectedRowCount: 0, + truncated: false, + }), })); function makeModule(): PostgresConnectionModule { @@ -199,6 +208,15 @@ describe("PostgresConnectionModule — error-path routing", () => { jest.spyOn(mod.authModule, "getPool").mockReturnValue(pool as any); const errSpy = jest.spyOn(console, "error").mockImplementation(() => {}); + // Writes now stream through drainBoundedCursor rather than client.query + // (#1298), so that is where the statement failure originates. The test's + // subject is unchanged: a failing query whose ROLLBACK also fails must + // still surface the ORIGINAL error through onFail. + const { drainBoundedCursor } = require("../../src/postgresql/cursor-read"); + (drainBoundedCursor as jest.Mock).mockRejectedValueOnce( + new Error("insert exploded"), + ); + const onFail = jest.fn(); await mod.runQuery( { query: "INSERT INTO t VALUES (1)", params: {} }, diff --git a/connection/__tests__/postgresql/postgres-write-drain.ts b/connection/__tests__/postgresql/postgres-write-drain.ts new file mode 100644 index 00000000..8a9378e9 --- /dev/null +++ b/connection/__tests__/postgresql/postgres-write-drain.ts @@ -0,0 +1,140 @@ +import { PostgresConnectionModule } from "../../src/postgresql/PostgresConnectionModule"; +import { + DEFAULT_CONNECTION_CONFIG, + QueryStatus, + AuthType, + ConnectionTypes, +} from "@neoboard/connector-sdk"; +import { PostgreSqlContainer } from "@testcontainers/postgresql"; + +/** + * Write-path row limiting against a REAL PostgreSQL (#1298 / #1326). + * + * The write branch buffered the entire result set and sliced afterwards, so + * `rowLimit` bounded what was displayed, not what was resident — one Form + * submit against a large table could exhaust the heap shared by every tenant. + * + * The obvious fix is the dangerous one. `readBoundedCursor` performs ONE + * bounded `cursor.read()` and closes the cursor in its `finally`. PostgreSQL + * executes a portal incrementally, so an `UPDATE … RETURNING` suspended after + * `rowLimit + 1` rows has NOT applied the rest, and closing the portal + * abandons that work — turning "update 1,000 rows" into "update 11" while + * still reporting success. + * + * These tests exist to make that failure impossible to ship. A stubbed client + * cannot prove any of it; only a real database can. + */ +describe("PostgreSQL write path — row limit must not truncate side effects", () => { + let container: PostgreSqlContainer; + let connectionModule: PostgresConnectionModule; + + const ROW_COUNT = 1000; + const ROW_LIMIT = 10; + + beforeAll(async () => { + container = await new PostgreSqlContainer("postgres:16-alpine").start(); + + connectionModule = new PostgresConnectionModule({ + username: container.getUsername(), + password: container.getPassword(), + authType: AuthType.NATIVE, + uri: `postgresql://${container.getHost()}:${container.getPort()}/${container.getDatabase()}`, + }); + + expect(await connectionModule.authModule.verifyAuthentication()).toBe(true); + + const client = await connectionModule.getPool()!.connect(); + try { + await client.query( + `CREATE TABLE counters (id SERIAL PRIMARY KEY, n INT NOT NULL)`, + ); + await client.query( + `INSERT INTO counters (n) SELECT 0 FROM generate_series(1, $1)`, + [ROW_COUNT], + ); + } finally { + client.release(); + } + }, 120_000); + + afterAll(async () => { + await connectionModule.close(); + await container?.stop(); + }, 60_000); + + /** Run a write query through the module and collect what the caller sees. */ + function runWrite(query: string) { + return new Promise<{ rows: unknown[]; statuses: QueryStatus[] }>( + (resolve, reject) => { + const statuses: QueryStatus[] = []; + connectionModule.runQuery( + { query, parameters: {} }, + { + onSuccess: (rows: unknown) => + resolve({ rows: rows as unknown[], statuses }), + onFail: reject, + setStatus: (s: QueryStatus) => statuses.push(s), + setFields: () => {}, + setSchema: () => {}, + }, + { + ...DEFAULT_CONNECTION_CONFIG, + type: ConnectionTypes.POSTGRESQL, + rowLimit: ROW_LIMIT, + accessMode: "WRITE", + }, + ); + }, + ); + } + + it("applies the UPDATE to EVERY row while returning at most rowLimit", async () => { + const { rows } = await runWrite( + `UPDATE counters SET n = n + 1 RETURNING *`, + ); + + // The caller sees only the capped page... + expect(rows.length).toBeLessThanOrEqual(ROW_LIMIT); + + // ...but every row must have been updated. This is the assertion that + // fails on any implementation which stops reading the portal early. + const client = await connectionModule.getPool()!.connect(); + try { + const { rows: check } = await client.query( + `SELECT count(*)::int AS updated FROM counters WHERE n = 1`, + ); + expect(check[0].updated).toBe(ROW_COUNT); + } finally { + client.release(); + } + }, 60_000); + + it("still reports the affected-row count for an INSERT without RETURNING", async () => { + // The buffered path was kept originally because result.rowCount is what + // makes a non-returning write report COMPLETE rather than NO_DATA. Any + // cursor-based rewrite has to preserve that. + const { statuses } = await runWrite( + `INSERT INTO counters (n) SELECT 99 FROM generate_series(1, 5)`, + ); + + expect(statuses).not.toContain(QueryStatus.NO_DATA); + + const client = await connectionModule.getPool()!.connect(); + try { + const { rows: check } = await client.query( + `SELECT count(*)::int AS inserted FROM counters WHERE n = 99`, + ); + expect(check[0].inserted).toBe(5); + } finally { + client.release(); + } + }, 60_000); + + it("flags truncation when a write returns more rows than the limit", async () => { + const { statuses } = await runWrite( + `UPDATE counters SET n = n WHERE n <> 99 RETURNING *`, + ); + + expect(statuses).toContain(QueryStatus.COMPLETE_TRUNCATED); + }, 60_000); +}); diff --git a/connection/src/postgresql/PostgresConnectionModule.ts b/connection/src/postgresql/PostgresConnectionModule.ts index a2d65693..edd809b7 100644 --- a/connection/src/postgresql/PostgresConnectionModule.ts +++ b/connection/src/postgresql/PostgresConnectionModule.ts @@ -11,7 +11,7 @@ import { } from "@neoboard/connector-sdk"; import { PostgresRecordParser } from "./PostgresRecordParser"; import { Pool, PoolClient, FieldDef } from "pg"; -import { readBoundedCursor } from "./cursor-read"; +import { readBoundedCursor, drainBoundedCursor } from "./cursor-read"; import { extractTableSchemaFromFields, isAuthenticationError } from "./utils"; import { determineQueryStatus } from "@neoboard/connector-sdk"; import { wrapError, ConnectorErrorType } from "@neoboard/connector-sdk"; @@ -170,10 +170,20 @@ export class PostgresConnectionModule extends ConnectionModule { fetchedRows = batch.rows; fields = batch.fields; } else { - const result = await client.query(query, paramValues); - fetchedRows = result.rows; - fields = result.fields; - affectedRowCount = result.rowCount ?? undefined; + // Writes stream too, but they must be DRAINED rather than stopped + // early: PostgreSQL applies an UPDATE ... RETURNING incrementally, so + // rows never pulled are never modified. readBoundedCursor stops after + // one bounded read and closes the portal — correct for a SELECT, + // silently partially-applied for a write (#1298, #1326). + const batch = await drainBoundedCursor( + client, + query, + paramValues, + config.rowLimit + 1, + ); + fetchedRows = batch.rows; + fields = batch.fields; + affectedRowCount = batch.affectedRowCount; } // Commit transaction diff --git a/connection/src/postgresql/cursor-read.ts b/connection/src/postgresql/cursor-read.ts index 7290f581..1123462f 100644 --- a/connection/src/postgresql/cursor-read.ts +++ b/connection/src/postgresql/cursor-read.ts @@ -77,3 +77,102 @@ function closeCursorSafely(cursor: Cursor): Promise { } }); } + +/** A drained cursor: retained rows, field descriptors, and the true row count. */ +export interface DrainedCursor extends CursorBatch { + /** + * Rows the statement actually produced or affected — not the retained + * count. Feeds the COMPLETE / NO_DATA decision for writes that return + * nothing, e.g. an INSERT without RETURNING. + */ + affectedRowCount: number | undefined; + /** True when the statement produced more rows than `maxRows`. */ + truncated: boolean; +} + +/** Rows pulled per round-trip while draining. Bounds memory, not correctness. */ +const DRAIN_BATCH_SIZE = 500; + +/** + * Executes `query` through a server-side cursor and reads it **to exhaustion**, + * retaining at most `maxRows` rows. + * + * This is the WRITE-path counterpart to `readBoundedCursor`, and the difference + * is correctness rather than performance (#1298, #1326). + * + * PostgreSQL executes a portal incrementally. `readBoundedCursor` does one + * bounded read and then closes the cursor, which is exactly right for a SELECT + * — but on an `UPDATE … RETURNING` the rows never pulled are never modified, + * and closing the portal abandons them. Reusing it for writes would silently + * turn "update 1,000,000 rows" into "update 26" and still report success. + * + * So this keeps reading until a batch comes back empty — every row is produced, + * every side effect runs — while retaining only `maxRows`. Peak memory becomes + * the batch size plus `maxRows` instead of the whole result set. + * + * The user's query text is passed unmodified; parameters stay positional. + */ +export async function drainBoundedCursor( + client: PoolClient, + query: string, + values: unknown[], + maxRows: number, +): Promise { + const cursor = client.query(new Cursor(query, values)); + const rows: Record[] = []; + let fields: FieldDef[] = []; + let affectedRowCount: number | undefined; + let total = 0; + + try { + for (;;) { + const batch = await new Promise<{ + rows: Record[]; + fields: FieldDef[]; + rowCount: number | undefined; + }>((resolve, reject) => { + cursor.read(DRAIN_BATCH_SIZE, (err, batchRows, result) => { + if (err) { + reject(err); + return; + } + resolve({ + rows: batchRows as Record[], + fields: result?.fields ?? [], + rowCount: result?.rowCount ?? undefined, + }); + }); + }); + + if (batch.fields.length > 0 && fields.length === 0) { + fields = batch.fields; + } + // A non-returning statement (INSERT without RETURNING) yields no rows, + // so its affected count only ever arrives on the completing batch. + if (batch.rowCount !== undefined && batch.rowCount !== null) { + affectedRowCount = batch.rowCount; + } + + if (batch.rows.length === 0) break; + + total += batch.rows.length; + for (const row of batch.rows) { + if (rows.length < maxRows) rows.push(row); + } + } + } finally { + await closeCursorSafely(cursor); + } + + return { + rows, + fields, + // Only meaningful for statements that RETURN nothing. A returning + // statement's completing batch reports rowCount 0, which would otherwise + // overwrite the real count and make the caller report NO_DATA for an + // UPDATE that changed a thousand rows. When rows were produced, let the + // caller derive the count from them. + affectedRowCount: total > 0 ? undefined : affectedRowCount, + truncated: total > maxRows, + }; +} From 324edc06f30239f3cd51e732c1370883f651344c Mon Sep 17 00:00:00 2001 From: alfredorubin96 Date: Tue, 28 Jul 2026 00:07:34 +0200 Subject: [PATCH 2/3] fix(connection): address review on the PostgreSQL write drain MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit on #1328, four findings — all correct: - the test passed `parameters: {}` where the module destructures `params`, and `type:` where the config declares `connectionType`. The second is the one that mattered: DEFAULT_CONNECTION_CONFIG.connectionType defaults to NEO4J, so the suite was configuring a Neo4j type against the Postgres module. Behaviour was still right, but the test would have started lying the moment anything branched on it. - `expect(rows.length).toBeLessThanOrEqual(ROW_LIMIT)` also passes when the drain returns ZERO rows — exactly the regression this suite exists to catch. Now asserts the exact bound. - afterAll called connectionModule.close() unguarded, so a failure in beforeAll before construction raised a TypeError that masked the real setup error. `container?.stop()` was already guarded. - the block comment above the branch still claimed writes "keep the direct path", the opposite of what it now does. Rewritten to say what actually differs: both stream, they differ in how they STOP. DrainedCursor.truncated is dropped rather than wired up. The caller derives truncation uniformly from the retained row count, so the field was dead on arrival; shipping an unused one invites a future reader to trust it. Verified: 8 pg suites, 72 tests. Refs #1326 Co-Authored-By: Claude Opus 5 --- .../postgresql/postgres-error-handling.test.ts | 1 - .../__tests__/postgresql/postgres-write-drain.ts | 14 +++++++++----- .../src/postgresql/PostgresConnectionModule.ts | 15 +++++++++------ connection/src/postgresql/cursor-read.ts | 3 --- 4 files changed, 18 insertions(+), 15 deletions(-) diff --git a/connection/__tests__/postgresql/postgres-error-handling.test.ts b/connection/__tests__/postgresql/postgres-error-handling.test.ts index 00a4d559..6ca52b9a 100644 --- a/connection/__tests__/postgresql/postgres-error-handling.test.ts +++ b/connection/__tests__/postgresql/postgres-error-handling.test.ts @@ -24,7 +24,6 @@ jest.mock("../../src/postgresql/cursor-read", () => ({ rows: [], fields: [], affectedRowCount: 0, - truncated: false, }), })); diff --git a/connection/__tests__/postgresql/postgres-write-drain.ts b/connection/__tests__/postgresql/postgres-write-drain.ts index 8a9378e9..754489b7 100644 --- a/connection/__tests__/postgresql/postgres-write-drain.ts +++ b/connection/__tests__/postgresql/postgres-write-drain.ts @@ -58,7 +58,9 @@ describe("PostgreSQL write path — row limit must not truncate side effects", ( }, 120_000); afterAll(async () => { - await connectionModule.close(); + // Guarded: if beforeAll throws before construction, an unguarded close() + // raises a TypeError that masks the real setup failure. + await connectionModule?.close(); await container?.stop(); }, 60_000); @@ -68,7 +70,7 @@ describe("PostgreSQL write path — row limit must not truncate side effects", ( (resolve, reject) => { const statuses: QueryStatus[] = []; connectionModule.runQuery( - { query, parameters: {} }, + { query, params: {} }, { onSuccess: (rows: unknown) => resolve({ rows: rows as unknown[], statuses }), @@ -79,7 +81,7 @@ describe("PostgreSQL write path — row limit must not truncate side effects", ( }, { ...DEFAULT_CONNECTION_CONFIG, - type: ConnectionTypes.POSTGRESQL, + connectionType: ConnectionTypes.POSTGRESQL, rowLimit: ROW_LIMIT, accessMode: "WRITE", }, @@ -93,8 +95,10 @@ describe("PostgreSQL write path — row limit must not truncate side effects", ( `UPDATE counters SET n = n + 1 RETURNING *`, ); - // The caller sees only the capped page... - expect(rows.length).toBeLessThanOrEqual(ROW_LIMIT); + // Exactly the cap, not merely "no more than" — `<=` would also pass if + // the drain returned nothing, which is the regression this suite exists + // to catch. + expect(rows).toHaveLength(ROW_LIMIT); // ...but every row must have been updated. This is the assertion that // fails on any implementation which stops reading the portal early. diff --git a/connection/src/postgresql/PostgresConnectionModule.ts b/connection/src/postgresql/PostgresConnectionModule.ts index edd809b7..76b82f4a 100644 --- a/connection/src/postgresql/PostgresConnectionModule.ts +++ b/connection/src/postgresql/PostgresConnectionModule.ts @@ -150,12 +150,15 @@ export class PostgresConnectionModule extends ConnectionModule { .map((k) => params[k]) : []; - // Fetch rows. READ queries stream through a server-side cursor so a - // huge result set never buffers in memory — we pull at most rowLimit + 1 - // rows (the MAX_ROWS+1 truncation probe). WRITE queries (Form widgets) - // keep the direct path: their result sets are small and we need the - // driver's affected-row count so an INSERT without RETURNING still - // reports COMPLETE rather than NO_DATA. + // Both paths stream through a server-side cursor so a huge result set + // never buffers in memory; each pulls at most rowLimit + 1 rows for the + // MAX_ROWS+1 truncation probe. They differ in how they STOP: + // READ stops as soon as truncation is known, releasing the portal. + // WRITE drains to exhaustion, because PostgreSQL applies an + // UPDATE ... RETURNING incrementally — rows never pulled are + // never modified — and then reports the driver's affected-row + // count so an INSERT without RETURNING still reads as COMPLETE + // rather than NO_DATA (#1298, #1326). let fetchedRows: Record[]; let fields: FieldDef[] | undefined; let affectedRowCount: number | undefined; diff --git a/connection/src/postgresql/cursor-read.ts b/connection/src/postgresql/cursor-read.ts index 1123462f..afad8a00 100644 --- a/connection/src/postgresql/cursor-read.ts +++ b/connection/src/postgresql/cursor-read.ts @@ -86,8 +86,6 @@ export interface DrainedCursor extends CursorBatch { * nothing, e.g. an INSERT without RETURNING. */ affectedRowCount: number | undefined; - /** True when the statement produced more rows than `maxRows`. */ - truncated: boolean; } /** Rows pulled per round-trip while draining. Bounds memory, not correctness. */ @@ -173,6 +171,5 @@ export async function drainBoundedCursor( // UPDATE that changed a thousand rows. When rows were produced, let the // caller derive the count from them. affectedRowCount: total > 0 ? undefined : affectedRowCount, - truncated: total > maxRows, }; } From d0891b8ff93067f7ee3e4e759c5535429b95fef8 Mon Sep 17 00:00:00 2001 From: alfredorubin96 Date: Tue, 28 Jul 2026 01:04:31 +0200 Subject: [PATCH 3/3] test(connection): type the container as StartedPostgreSqlContainer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CodeRabbit on #1328: start() resolves to StartedPostgreSqlContainer, so getUsername/getPort/stop were being called on the wrong type. postgres-query.ts has the identical mistake and is left alone here — an unrelated file in a P1 PR. Flagged on the PR instead. Refs #1326 Co-Authored-By: Claude Opus 5 --- connection/__tests__/postgresql/postgres-write-drain.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/connection/__tests__/postgresql/postgres-write-drain.ts b/connection/__tests__/postgresql/postgres-write-drain.ts index 754489b7..fa73df3b 100644 --- a/connection/__tests__/postgresql/postgres-write-drain.ts +++ b/connection/__tests__/postgresql/postgres-write-drain.ts @@ -5,7 +5,10 @@ import { AuthType, ConnectionTypes, } from "@neoboard/connector-sdk"; -import { PostgreSqlContainer } from "@testcontainers/postgresql"; +import { + PostgreSqlContainer, + type StartedPostgreSqlContainer, +} from "@testcontainers/postgresql"; /** * Write-path row limiting against a REAL PostgreSQL (#1298 / #1326). @@ -25,7 +28,7 @@ import { PostgreSqlContainer } from "@testcontainers/postgresql"; * cannot prove any of it; only a real database can. */ describe("PostgreSQL write path — row limit must not truncate side effects", () => { - let container: PostgreSqlContainer; + let container: StartedPostgreSqlContainer; let connectionModule: PostgresConnectionModule; const ROW_COUNT = 1000;