From fc8105e13b49d481afb8c51ec3f0579cec83d1a8 Mon Sep 17 00:00:00 2001 From: ferhat elmas Date: Mon, 15 Jun 2026 11:21:45 +0200 Subject: [PATCH 01/12] fix: fold timeout into scope query cache serialization of headers/jwt payload per request Signed-off-by: ferhat elmas --- src/internal/database/pg-connection.test.ts | 231 ++++++++++++++++++++ src/internal/database/pg-connection.ts | 119 +++++++--- src/test/storage-pg-db.test.ts | 126 ++++++++--- 3 files changed, 415 insertions(+), 61 deletions(-) diff --git a/src/internal/database/pg-connection.test.ts b/src/internal/database/pg-connection.test.ts index 2abf9c9af..d0251218d 100644 --- a/src/internal/database/pg-connection.test.ts +++ b/src/internal/database/pg-connection.test.ts @@ -7,6 +7,7 @@ import PgConnection from 'pg/lib/connection' import { vi } from 'vitest' import { getPgCancelConnectionTarget, + type PgExecutor, PgPoolExecutor, PgPoolManager, PgPoolStrategy, @@ -553,6 +554,84 @@ describe('PgPoolExecutor', () => { }) describe('PgTransaction', () => { + it('applies a pending statement timeout before the first direct query', async () => { + const client = { + query: vi.fn().mockResolvedValue({ rows: [] }), + release: vi.fn(), + } as unknown as PoolClient + const transaction = new PgTransaction(client) + + transaction.setPendingStatementTimeout(4321) + + await transaction.query('SELECT 1') + await transaction.query('SELECT 2') + + expect(client.query).toHaveBeenCalledTimes(3) + expect(client.query).toHaveBeenNthCalledWith( + 1, + "SELECT set_config('statement_timeout', $1, true)", + ['4321ms'] + ) + expect(client.query).toHaveBeenNthCalledWith(2, 'SELECT 1', undefined) + expect(client.query).toHaveBeenNthCalledWith(3, 'SELECT 2', undefined) + }) + + it('rejects a pre-aborted direct query before applying a pending statement timeout', async () => { + const client = { + query: vi.fn().mockResolvedValue({ rows: [] }), + release: vi.fn(), + } as unknown as PoolClient + const transaction = new PgTransaction(client) + + transaction.setPendingStatementTimeout(4321) + + await expect( + transaction.query('SELECT 1', { signal: AbortSignal.abort() }) + ).rejects.toMatchObject({ + name: 'AbortError', + code: 'ABORT_ERR', + }) + expect(client.query).not.toHaveBeenCalled() + expect(client.release).toHaveBeenCalledWith(expect.objectContaining({ name: 'AbortError' })) + + await expect(transaction.query('SELECT 2')).rejects.toThrow( + 'Cannot query a completed transaction' + ) + expect(client.query).not.toHaveBeenCalled() + }) + + it('honors abort signals while applying a pending statement timeout', async () => { + const controller = new AbortController() + const client = Object.assign(new EventEmitter(), { + query: vi.fn(() => { + controller.abort() + return new Promise(() => undefined) + }), + release: vi.fn(), + }) as unknown as PoolClient & EventEmitter + const transaction = new PgTransaction(client) + + transaction.setPendingStatementTimeout(4321) + + await expect( + transaction.query('SELECT 1', { signal: controller.signal }) + ).rejects.toMatchObject({ + name: 'AbortError', + code: 'ABORT_ERR', + }) + + expect(client.query).toHaveBeenCalledTimes(1) + expect(client.query).toHaveBeenNthCalledWith( + 1, + "SELECT set_config('statement_timeout', $1, true)", + ['4321ms'] + ) + expect(client.release).toHaveBeenCalledWith(expect.objectContaining({ name: 'AbortError' })) + + await transaction.rollback() + expect(client.query).toHaveBeenCalledTimes(1) + }) + it('rejects queries after commit releases the client', async () => { const client = { query: vi.fn().mockResolvedValue({ rows: [] }), @@ -708,6 +787,158 @@ describe('PgTenantConnection', () => { }) }) + it('defers statement_timeout setup until the first scope application', async () => { + const query = vi.fn().mockResolvedValue({ rows: [] }) + const client = { + query, + release: vi.fn(), + } as unknown as PoolClient + const transaction = new PgTransaction(client) + const pool = { + acquire: vi.fn().mockReturnValue({ + beginTransaction: vi.fn().mockResolvedValue(transaction), + }), + } as unknown as PgPoolStrategy + const connection = new PgTenantConnection( + pool, + createPoolStrategySettings({ + isExternalPool: false, + }) + ) + + await expect(connection.transaction({ timeout: 4321 })).resolves.toBe(transaction) + expect(query).not.toHaveBeenCalled() + + await connection.setScope(transaction) + + expect(query).toHaveBeenCalledTimes(1) + const [statement, values] = query.mock.calls[0] + expect(statement).toContain("set_config('statement_timeout'") + expect(values).toEqual([ + '4321ms', + 'authenticated', + 'authenticated', + 'jwt', + '', + JSON.stringify({ role: 'authenticated' }), + '{}', + '', + '', + '', + ]) + }) + + it('keeps external-pool search_path setup before deferring statement_timeout', async () => { + const query = vi.fn().mockResolvedValue({ rows: [] }) + const client = { + query, + release: vi.fn(), + } as unknown as PoolClient + const transaction = new PgTransaction(client) + const pool = { + acquire: vi.fn().mockReturnValue({ + beginTransaction: vi.fn().mockResolvedValue(transaction), + }), + } as unknown as PgPoolStrategy + const connection = new PgTenantConnection( + pool, + createPoolStrategySettings({ + isExternalPool: true, + }) + ) + + await expect(connection.transaction({ timeout: 4321 })).resolves.toBe(transaction) + + expect(query).toHaveBeenCalledTimes(1) + expect(query).toHaveBeenNthCalledWith( + 1, + "SELECT set_config('search_path', $1, true)", + expect.any(Array) + ) + + await connection.setScope(transaction) + + expect(query).toHaveBeenCalledTimes(2) + const [scopeStatement, scopeValues] = query.mock.calls[1] + expect(scopeStatement).toContain("set_config('statement_timeout'") + expect(scopeValues).toEqual([ + '4321ms', + 'authenticated', + 'authenticated', + 'jwt', + '', + JSON.stringify({ role: 'authenticated' }), + '{}', + '', + '', + '', + ]) + }) + + it('does not re-apply statement_timeout after setScope consumes it', async () => { + const query = vi.fn().mockResolvedValue({ rows: [] }) + const client = { + query, + release: vi.fn(), + } as unknown as PoolClient + const transaction = new PgTransaction(client) + const pool = { + acquire: vi.fn().mockReturnValue({ + beginTransaction: vi.fn().mockResolvedValue(transaction), + }), + } as unknown as PgPoolStrategy + const connection = new PgTenantConnection( + pool, + createPoolStrategySettings({ + isExternalPool: false, + }) + ) + + await connection.transaction({ timeout: 4321 }) + await connection.setScope(transaction) + await transaction.query('SELECT 1') + + expect(query).toHaveBeenCalledTimes(2) + expect(query).toHaveBeenNthCalledWith(2, 'SELECT 1', undefined) + expect( + query.mock.calls.filter(([statement]) => String(statement).includes('statement_timeout')) + ).toHaveLength(1) + }) + + it('reuses precomputed scope JSON payloads across repeated scope applications', async () => { + const pool = { + acquire: vi.fn(), + } as unknown as PgPoolStrategy + const connection = new PgTenantConnection( + pool, + createPoolStrategySettings({ + headers: { + 'x-test-header': 'test-value', + }, + user: { + jwt: 'jwt', + payload: { + role: 'authenticated', + sub: 'user-id', + }, + }, + }) + ) + const executor = { + query: vi.fn().mockResolvedValue({ rows: [] }), + } as unknown as PgExecutor + const stringifySpy = vi.spyOn(JSON, 'stringify') + + try { + await connection.setScope(executor) + await connection.setScope(executor) + + expect(stringifySpy).not.toHaveBeenCalled() + } finally { + stringifySpy.mockRestore() + } + }) + it('preserves setup errors when external-pool rollback fails', async () => { const setupError = new Error('search_path setup failed') const rollbackError = new Error('rollback failed') diff --git a/src/internal/database/pg-connection.ts b/src/internal/database/pg-connection.ts index 643f6a6b4..37c660d82 100644 --- a/src/internal/database/pg-connection.ts +++ b/src/internal/database/pg-connection.ts @@ -45,6 +45,24 @@ export interface PgQueryOptions { type PgQueryArgument = PgQueryOptions | unknown[] +const scopeConfigNames = [ + 'role', + 'request.jwt.claim.role', + 'request.jwt', + 'request.jwt.claim.sub', + 'request.jwt.claims', + 'request.headers', + 'request.method', + 'request.path', + 'storage.operation', +] as const + +const scopeConfigStatement = buildScopeConfigStatement(scopeConfigNames) +const scopeConfigStatementWithTimeout = buildScopeConfigStatement([ + 'statement_timeout', + ...scopeConfigNames, +]) + export interface PgExecutor { query( statement: string | PgStatement, @@ -303,6 +321,18 @@ function pulsePgPoolQueue(pool: Pool): void { ;(pool as Pool & { _pulseQueue?: () => void })._pulseQueue?.() } +function buildScopeConfigStatement(configNames: readonly string[]): string { + const configStatements = configNames.map((name, index) => { + return `set_config('${name}', $${index + 1}, true)` + }) + configStatements.push(`set_config('storage.allow_delete_query', 'true', true)`) + + return ` + SELECT + ${configStatements.join(',\n ')}; + ` +} + export class PgPoolManager extends PoolManager { protected newPool(settings: PoolStrategySettings): PgPoolStrategy { return new PgPoolStrategy(settings) @@ -432,6 +462,7 @@ export class PgPoolExecutor implements PgTransactionalExecutor { export class PgTransaction implements PgExecutor { private completed = false + private pendingStatementTimeoutMs?: number constructor( private readonly client: PoolClient, @@ -442,6 +473,16 @@ export class PgTransaction implements PgExecutor { return this.completed } + setPendingStatementTimeout(timeoutMs: number | undefined): void { + this.pendingStatementTimeoutMs = timeoutMs && timeoutMs > 0 ? timeoutMs : undefined + } + + takePendingStatementTimeout(): number | undefined { + const timeoutMs = this.pendingStatementTimeoutMs + this.pendingStatementTimeoutMs = undefined + return timeoutMs + } + async query( statement: string | PgStatement, options?: PgQueryArgument @@ -450,7 +491,16 @@ export class PgTransaction implements PgExecutor { throw new Error('Cannot query a completed transaction') } + const signal = getQuerySignal(options) + try { + assertValidSignal(signal) + + const pendingStatementTimeout = this.takePendingStatementTimeoutStatement() + if (pendingStatementTimeout) { + await runPgQuery(this.client, pendingStatementTimeout, { signal }) + } + const result = await runPgQuery(this.client, statement, options) this.clientErrorTracker?.throwIfErrored() return result @@ -515,6 +565,19 @@ export class PgTransaction implements PgExecutor { this.clientErrorTracker?.detach() } } + + private takePendingStatementTimeoutStatement(): PgStatement | undefined { + const timeoutMs = this.takePendingStatementTimeout() + + if (!timeoutMs) { + return undefined + } + + return { + text: `SELECT set_config('statement_timeout', $1, true)`, + values: [`${timeoutMs}ms`], + } + } } export class PgTenantConnection { @@ -522,12 +585,16 @@ export class PgTenantConnection { public readonly role: string private abortSignal?: AbortSignal private disposed = false + private readonly headersPayload: string + private readonly userPayload: string constructor( public readonly pool: PgPoolStrategy, protected readonly options: TenantConnectionOptions ) { this.role = options.user.payload.role || 'anon' + this.headersPayload = JSON.stringify(options.headers || {}) + this.userPayload = JSON.stringify(options.user.payload) } static stop() { @@ -624,17 +691,7 @@ export class PgTenantConnection { } const statementTimeout = opts?.timeout ?? databaseStatementTimeout - if (statementTimeout > 0) { - try { - await transaction.query({ - text: `SELECT set_config('statement_timeout', $1, true)`, - values: [`${statementTimeout}ms`], - }) - } catch (e) { - await this.rollbackTransactionSafely(transaction, e, 'statement_timeout setup') - throw e - } - } + transaction.setPendingStatementTimeout(statementTimeout) return transaction } catch (e) { @@ -674,32 +731,22 @@ export class PgTenantConnection { } async setScope(tnx: PgExecutor) { - const headers = JSON.stringify(this.options.headers || {}) + const statementTimeout = tnx instanceof PgTransaction ? tnx.takePendingStatementTimeout() : 0 + const scopeValues = [ + this.role, + this.role, + this.options.user.jwt || '', + this.options.user.payload.sub || '', + this.userPayload, + this.headersPayload, + this.options.method || '', + this.options.path || '', + this.options.operation?.() || '', + ] + await tnx.query({ - text: ` - SELECT - set_config('role', $1, true), - set_config('request.jwt.claim.role', $2, true), - set_config('request.jwt', $3, true), - set_config('request.jwt.claim.sub', $4, true), - set_config('request.jwt.claims', $5, true), - set_config('request.headers', $6, true), - set_config('request.method', $7, true), - set_config('request.path', $8, true), - set_config('storage.operation', $9, true), - set_config('storage.allow_delete_query', 'true', true); - `, - values: [ - this.role, - this.role, - this.options.user.jwt || '', - this.options.user.payload.sub || '', - JSON.stringify(this.options.user.payload), - headers, - this.options.method || '', - this.options.path || '', - this.options.operation?.() || '', - ], + text: statementTimeout ? scopeConfigStatementWithTimeout : scopeConfigStatement, + values: statementTimeout ? [`${statementTimeout}ms`, ...scopeValues] : scopeValues, }) } } diff --git a/src/test/storage-pg-db.test.ts b/src/test/storage-pg-db.test.ts index 90f8591df..c5020dc73 100644 --- a/src/test/storage-pg-db.test.ts +++ b/src/test/storage-pg-db.test.ts @@ -352,28 +352,7 @@ describe('StoragePgDB bucket metadata', () => { }) it('restores parent transaction scope after failed super-user queries', async () => { - const authenticatedUser = { - jwt: 'storage-pg-db-authenticated-jwt', - payload: { - role: 'authenticated', - sub: randomUUID(), - }, - } - const authenticatedSettings = { - ...connectionSettings, - user: authenticatedUser, - superUser, - } - const authenticatedPool = new PgPoolStrategy(authenticatedSettings) - const authenticatedDb = new StoragePgDB( - new PgTenantConnection(authenticatedPool, authenticatedSettings), - { - tenantId, - host: 'localhost', - } - ) - - try { + await withAuthenticatedDb('storage-pg-db-authenticated-jwt', async (authenticatedDb) => { await authenticatedDb.withTransaction(async (tx) => { await expect(readCurrentRole(tx)).resolves.toBe('authenticated') @@ -409,9 +388,62 @@ describe('StoragePgDB bucket metadata', () => { ) expect(result.rows[0].n).toBe(1) }) - } finally { - await authenticatedPool.destroy() - } + }) + }) + + it('preserves custom statement timeout after nested super-user scope restores', async () => { + await withAuthenticatedDb('storage-pg-db-timeout-jwt', async (authenticatedDb) => { + await authenticatedDb.withTransaction( + async (tx) => { + await expect(readCurrentRole(tx)).resolves.toBe('authenticated') + await expect(readCurrentStatementTimeout(tx)).resolves.toBe('4321ms') + + await expect( + runStorageQuery( + tx.asSuperUser() as StoragePgDB, + 'ReadSuperUserStatementTimeout', + async (pg) => { + await expect(readCurrentRoleFromExecutor(pg)).resolves.toBe(superUser.payload.role) + await expect(readCurrentStatementTimeoutFromExecutor(pg)).resolves.toBe('4321ms') + } + ) + ).resolves.toBeUndefined() + + await expect(readCurrentRole(tx)).resolves.toBe('authenticated') + await expect(readCurrentStatementTimeout(tx)).resolves.toBe('4321ms') + }, + { timeout: 4321 } + ) + }) + }) + + it('preserves custom statement timeout after failed nested super-user scope rolls back', async () => { + await withAuthenticatedDb('storage-pg-db-timeout-rollback-jwt', async (authenticatedDb) => { + await authenticatedDb.withTransaction( + async (tx) => { + await expect(readCurrentRole(tx)).resolves.toBe('authenticated') + await expect(readCurrentStatementTimeout(tx)).resolves.toBe('4321ms') + + await expect( + runStorageQuery( + tx.asSuperUser() as StoragePgDB, + 'FailSuperUserStatementTimeout', + async (pg) => { + await expect(readCurrentRoleFromExecutor(pg)).resolves.toBe(superUser.payload.role) + await expect(readCurrentStatementTimeoutFromExecutor(pg)).resolves.toBe('4321ms') + await pg.query("SELECT set_config('statement_timeout', '30s', true)") + await expect(readCurrentStatementTimeoutFromExecutor(pg)).resolves.toBe('30s') + throw new Error('failed nested super-user timeout query') + } + ) + ).rejects.toThrow('failed nested super-user timeout query') + + await expect(readCurrentRole(tx)).resolves.toBe('authenticated') + await expect(readCurrentStatementTimeout(tx)).resolves.toBe('4321ms') + }, + { timeout: 4321 } + ) + }) }) it('preserves original errors when best-effort parent scope restoration fails', async () => { @@ -1748,6 +1780,50 @@ describe('StoragePgDB bucket metadata', () => { return result.rows[0].role } + function readCurrentStatementTimeout(storage: StoragePgDB): Promise { + return runStorageQuery(storage, 'ReadCurrentStatementTimeout', (pg) => + readCurrentStatementTimeoutFromExecutor(pg) + ) + } + + async function readCurrentStatementTimeoutFromExecutor(pg: PgExecutor): Promise { + const result = await pg.query<{ statement_timeout: string }>('SHOW statement_timeout') + + return result.rows[0].statement_timeout + } + + async function withAuthenticatedDb( + jwt: string, + fn: (authenticatedDb: StoragePgDB) => Promise + ): Promise { + const authenticatedUser = { + jwt, + payload: { + role: 'authenticated', + sub: randomUUID(), + }, + } + const authenticatedSettings = { + ...connectionSettings, + user: authenticatedUser, + superUser, + } + const authenticatedPool = new PgPoolStrategy(authenticatedSettings) + const authenticatedDb = new StoragePgDB( + new PgTenantConnection(authenticatedPool, authenticatedSettings), + { + tenantId, + host: 'localhost', + } + ) + + try { + await fn(authenticatedDb) + } finally { + await authenticatedPool.destroy() + } + } + async function tableExists(tableName: string): Promise { const result = await pool.acquire().query<{ exists: boolean }>({ text: `SELECT to_regclass($1) IS NOT NULL AS "exists"`, From 0c7e313ac4626a8f1d67282baf8c43376c3719a2 Mon Sep 17 00:00:00 2001 From: ferhat elmas Date: Mon, 15 Jun 2026 12:25:20 +0200 Subject: [PATCH 02/12] fix: make multigres workaround for set_config Signed-off-by: ferhat elmas --- src/internal/database/client.ts | 4 +- src/internal/database/pg-connection.test.ts | 49 +++++++++++++++++ src/internal/database/pg-connection.ts | 60 ++++++++++++++++++++- src/internal/database/pool.ts | 3 +- src/test/pg-connection.test.ts | 1 + src/test/storage-pg-db.test.ts | 3 +- 6 files changed, 115 insertions(+), 5 deletions(-) diff --git a/src/internal/database/client.ts b/src/internal/database/client.ts index 3c1377424..65df7e982 100644 --- a/src/internal/database/client.ts +++ b/src/internal/database/client.ts @@ -41,7 +41,8 @@ async function getDbSettings( host: string | undefined, options?: { disableHostCheck?: boolean } ) { - const { isMultitenant, databasePoolURL, databaseURL, databaseMaxConnections } = getConfig() + const { isMultitenant, databasePoolURL, databaseURL, databaseEngine, databaseMaxConnections } = + getConfig() let dbUrl = databasePoolURL || databaseURL let maxConnections = databaseMaxConnections @@ -75,6 +76,7 @@ async function getDbSettings( return { dbUrl, + databaseEngine, isExternalPool, maxConnections, } diff --git a/src/internal/database/pg-connection.test.ts b/src/internal/database/pg-connection.test.ts index d0251218d..4814242ea 100644 --- a/src/internal/database/pg-connection.test.ts +++ b/src/internal/database/pg-connection.test.ts @@ -875,6 +875,55 @@ describe('PgTenantConnection', () => { ]) }) + it('uses a standalone statement_timeout setup for Multigres transactions', async () => { + const query = vi.fn().mockResolvedValue({ rows: [] }) + const client = { + query, + release: vi.fn(), + } as unknown as PoolClient + const transaction = new PgTransaction(client) + const pool = { + acquire: vi.fn().mockReturnValue({ + beginTransaction: vi.fn().mockResolvedValue(transaction), + }), + } as unknown as PgPoolStrategy + const connection = new PgTenantConnection( + pool, + createPoolStrategySettings({ + isExternalPool: true, + databaseEngine: 'multigres', + }) + ) + + await expect(connection.transaction({ timeout: 4321 })).resolves.toBe(transaction) + + expect(query).toHaveBeenCalledTimes(2) + expect(query).toHaveBeenNthCalledWith( + 1, + "SELECT set_config('search_path', $1, true)", + expect.any(Array) + ) + expect(query).toHaveBeenNthCalledWith(2, "SET LOCAL statement_timeout = '4321ms'", undefined) + + await connection.setScope(transaction) + + expect(query).toHaveBeenCalledTimes(4) + const [scopeStatement, scopeValues] = query.mock.calls[2] + expect(scopeStatement).not.toContain("set_config('statement_timeout'") + expect(scopeValues).toEqual([ + 'authenticated', + 'authenticated', + 'jwt', + '', + JSON.stringify({ role: 'authenticated' }), + '{}', + '', + '', + '', + ]) + expect(query).toHaveBeenNthCalledWith(4, "SET LOCAL statement_timeout = '4321ms'", undefined) + }) + it('does not re-apply statement_timeout after setScope consumes it', async () => { const query = vi.fn().mockResolvedValue({ rows: [] }) const client = { diff --git a/src/internal/database/pg-connection.ts b/src/internal/database/pg-connection.ts index 37c660d82..3ec4c87cf 100644 --- a/src/internal/database/pg-connection.ts +++ b/src/internal/database/pg-connection.ts @@ -28,6 +28,7 @@ const { databaseMaxConnections, databasePoolDrainTimeout, databaseSSLRootCert, + databaseEngine, databaseStatementTimeout, databaseTlsSessionResumption, } = getConfig() @@ -463,6 +464,7 @@ export class PgPoolExecutor implements PgTransactionalExecutor { export class PgTransaction implements PgExecutor { private completed = false private pendingStatementTimeoutMs?: number + private appliedStatementTimeoutMs?: number constructor( private readonly client: PoolClient, @@ -483,6 +485,14 @@ export class PgTransaction implements PgExecutor { return timeoutMs } + setAppliedStatementTimeout(timeoutMs: number | undefined): void { + this.appliedStatementTimeoutMs = timeoutMs && timeoutMs > 0 ? timeoutMs : undefined + } + + getAppliedStatementTimeout(): number | undefined { + return this.appliedStatementTimeoutMs + } + async query( statement: string | PgStatement, options?: PgQueryArgument @@ -691,7 +701,11 @@ export class PgTenantConnection { } const statementTimeout = opts?.timeout ?? databaseStatementTimeout - transaction.setPendingStatementTimeout(statementTimeout) + if (this.isMultigresDatabase()) { + await this.applyStatementTimeoutImmediately(transaction, statementTimeout) + } else { + transaction.setPendingStatementTimeout(statementTimeout) + } return transaction } catch (e) { @@ -703,6 +717,27 @@ export class PgTenantConnection { } } + private isMultigresDatabase(): boolean { + return (this.options.databaseEngine ?? databaseEngine) === 'multigres' + } + + private async applyStatementTimeoutImmediately( + transaction: PgTransaction, + statementTimeout: number | undefined + ): Promise { + if (!statementTimeout || statementTimeout <= 0) { + return + } + + try { + await transaction.query(buildSetLocalStatementTimeoutStatement(statementTimeout)) + transaction.setAppliedStatementTimeout(statementTimeout) + } catch (e) { + await this.rollbackTransactionSafely(transaction, e, 'statement_timeout setup') + throw e + } + } + private assertNotDisposed(): void { if (this.disposed) { throw createDisposedTenantConnectionError() @@ -731,7 +766,10 @@ export class PgTenantConnection { } async setScope(tnx: PgExecutor) { - const statementTimeout = tnx instanceof PgTransaction ? tnx.takePendingStatementTimeout() : 0 + const transaction = tnx instanceof PgTransaction ? tnx : undefined + const pendingStatementTimeout = transaction ? transaction.takePendingStatementTimeout() : 0 + const isMultigres = this.isMultigresDatabase() + const statementTimeout = isMultigres ? 0 : pendingStatementTimeout const scopeValues = [ this.role, this.role, @@ -748,7 +786,25 @@ export class PgTenantConnection { text: statementTimeout ? scopeConfigStatementWithTimeout : scopeConfigStatement, values: statementTimeout ? [`${statementTimeout}ms`, ...scopeValues] : scopeValues, }) + + if (isMultigres && transaction) { + // Multigres requires SET LOCAL for statement_timeout, and role scope setup resets it. + const timeoutToReapply = transaction.getAppliedStatementTimeout() + + if (timeoutToReapply && timeoutToReapply > 0) { + await transaction.query(buildSetLocalStatementTimeoutStatement(timeoutToReapply)) + transaction.setAppliedStatementTimeout(timeoutToReapply) + } + } + } +} + +function buildSetLocalStatementTimeoutStatement(statementTimeout: number): string { + if (!Number.isFinite(statementTimeout) || statementTimeout <= 0) { + throw new Error(`Invalid statement timeout: ${statementTimeout}`) } + + return `SET LOCAL statement_timeout = '${statementTimeout}ms'` } function createDisposedTenantConnectionError(): Error { diff --git a/src/internal/database/pool.ts b/src/internal/database/pool.ts index 8fa82dcff..56782af6c 100644 --- a/src/internal/database/pool.ts +++ b/src/internal/database/pool.ts @@ -10,7 +10,7 @@ import { recordCacheRequest, } from '@internal/monitoring/metrics' import { JWTPayload } from 'jose' -import { getConfig } from '../../config' +import { type DatabaseEngine, getConfig } from '../../config' const { isMultitenant, @@ -30,6 +30,7 @@ export interface TenantConnectionOptions { idleTimeoutMillis?: number reapIntervalMillis?: number maxConnections: number + databaseEngine?: DatabaseEngine clusterSize?: number numWorkers?: number user: User diff --git a/src/test/pg-connection.test.ts b/src/test/pg-connection.test.ts index 988b878f7..dc8dd5796 100644 --- a/src/test/pg-connection.test.ts +++ b/src/test/pg-connection.test.ts @@ -18,6 +18,7 @@ describe('Pg database foundation', () => { tenantId, isExternalPool: true, maxConnections: 2, + databaseEngine: getConfig().databaseEngine, dbUrl: databasePoolURL || databaseURL, user: superUser, superUser, diff --git a/src/test/storage-pg-db.test.ts b/src/test/storage-pg-db.test.ts index c5020dc73..20ea60590 100644 --- a/src/test/storage-pg-db.test.ts +++ b/src/test/storage-pg-db.test.ts @@ -36,6 +36,7 @@ describe('StoragePgDB bucket metadata', () => { connectionSettings = { tenantId, dbUrl: databaseURL!, + databaseEngine: getConfig().databaseEngine, isExternalPool: false, maxConnections: 2, user: superUser, @@ -431,7 +432,7 @@ describe('StoragePgDB bucket metadata', () => { async (pg) => { await expect(readCurrentRoleFromExecutor(pg)).resolves.toBe(superUser.payload.role) await expect(readCurrentStatementTimeoutFromExecutor(pg)).resolves.toBe('4321ms') - await pg.query("SELECT set_config('statement_timeout', '30s', true)") + await pg.query("SET LOCAL statement_timeout = '30s'") await expect(readCurrentStatementTimeoutFromExecutor(pg)).resolves.toBe('30s') throw new Error('failed nested super-user timeout query') } From 4f40c178a4db4c66dab9b2b389dc0c32948b29b3 Mon Sep 17 00:00:00 2001 From: ferhat elmas Date: Wed, 24 Jun 2026 16:29:05 +0200 Subject: [PATCH 03/12] fix: make scope more readable Signed-off-by: ferhat elmas --- src/internal/database/pg-connection.test.ts | 10 ++- src/internal/database/pg-connection.ts | 88 +++++++++++---------- 2 files changed, 52 insertions(+), 46 deletions(-) diff --git a/src/internal/database/pg-connection.test.ts b/src/internal/database/pg-connection.test.ts index 4814242ea..91f9dec90 100644 --- a/src/internal/database/pg-connection.test.ts +++ b/src/internal/database/pg-connection.test.ts @@ -813,9 +813,9 @@ describe('PgTenantConnection', () => { expect(query).toHaveBeenCalledTimes(1) const [statement, values] = query.mock.calls[0] - expect(statement).toContain("set_config('statement_timeout'") + expect(statement).toContain("set_config('role', $1, true)") + expect(statement).toContain("set_config('statement_timeout', $10, true)") expect(values).toEqual([ - '4321ms', 'authenticated', 'authenticated', 'jwt', @@ -825,6 +825,7 @@ describe('PgTenantConnection', () => { '', '', '', + '4321ms', ]) }) @@ -860,9 +861,9 @@ describe('PgTenantConnection', () => { expect(query).toHaveBeenCalledTimes(2) const [scopeStatement, scopeValues] = query.mock.calls[1] - expect(scopeStatement).toContain("set_config('statement_timeout'") + expect(scopeStatement).toContain("set_config('role', $1, true)") + expect(scopeStatement).toContain("set_config('statement_timeout', $10, true)") expect(scopeValues).toEqual([ - '4321ms', 'authenticated', 'authenticated', 'jwt', @@ -872,6 +873,7 @@ describe('PgTenantConnection', () => { '', '', '', + '4321ms', ]) }) diff --git a/src/internal/database/pg-connection.ts b/src/internal/database/pg-connection.ts index 3ec4c87cf..41be8efd2 100644 --- a/src/internal/database/pg-connection.ts +++ b/src/internal/database/pg-connection.ts @@ -46,23 +46,36 @@ export interface PgQueryOptions { type PgQueryArgument = PgQueryOptions | unknown[] -const scopeConfigNames = [ - 'role', - 'request.jwt.claim.role', - 'request.jwt', - 'request.jwt.claim.sub', - 'request.jwt.claims', - 'request.headers', - 'request.method', - 'request.path', - 'storage.operation', -] as const - -const scopeConfigStatement = buildScopeConfigStatement(scopeConfigNames) -const scopeConfigStatementWithTimeout = buildScopeConfigStatement([ - 'statement_timeout', - ...scopeConfigNames, -]) +// The first nine placeholders must stay in the same order as getScopeValues(). +// The timeout variant appends statement_timeout as $10. +const scopeConfigStatement = ` + SELECT + set_config('role', $1, true), + set_config('request.jwt.claim.role', $2, true), + set_config('request.jwt', $3, true), + set_config('request.jwt.claim.sub', $4, true), + set_config('request.jwt.claims', $5, true), + set_config('request.headers', $6, true), + set_config('request.method', $7, true), + set_config('request.path', $8, true), + set_config('storage.operation', $9, true), + set_config('storage.allow_delete_query', 'true', true); + ` + +const scopeConfigStatementWithTimeout = ` + SELECT + set_config('role', $1, true), + set_config('request.jwt.claim.role', $2, true), + set_config('request.jwt', $3, true), + set_config('request.jwt.claim.sub', $4, true), + set_config('request.jwt.claims', $5, true), + set_config('request.headers', $6, true), + set_config('request.method', $7, true), + set_config('request.path', $8, true), + set_config('storage.operation', $9, true), + set_config('storage.allow_delete_query', 'true', true), + set_config('statement_timeout', $10, true); + ` export interface PgExecutor { query( @@ -322,18 +335,6 @@ function pulsePgPoolQueue(pool: Pool): void { ;(pool as Pool & { _pulseQueue?: () => void })._pulseQueue?.() } -function buildScopeConfigStatement(configNames: readonly string[]): string { - const configStatements = configNames.map((name, index) => { - return `set_config('${name}', $${index + 1}, true)` - }) - configStatements.push(`set_config('storage.allow_delete_query', 'true', true)`) - - return ` - SELECT - ${configStatements.join(',\n ')}; - ` -} - export class PgPoolManager extends PoolManager { protected newPool(settings: PoolStrategySettings): PgPoolStrategy { return new PgPoolStrategy(settings) @@ -770,21 +771,11 @@ export class PgTenantConnection { const pendingStatementTimeout = transaction ? transaction.takePendingStatementTimeout() : 0 const isMultigres = this.isMultigresDatabase() const statementTimeout = isMultigres ? 0 : pendingStatementTimeout - const scopeValues = [ - this.role, - this.role, - this.options.user.jwt || '', - this.options.user.payload.sub || '', - this.userPayload, - this.headersPayload, - this.options.method || '', - this.options.path || '', - this.options.operation?.() || '', - ] + const scopeValues = this.getScopeValues() await tnx.query({ text: statementTimeout ? scopeConfigStatementWithTimeout : scopeConfigStatement, - values: statementTimeout ? [`${statementTimeout}ms`, ...scopeValues] : scopeValues, + values: statementTimeout ? [...scopeValues, `${statementTimeout}ms`] : scopeValues, }) if (isMultigres && transaction) { @@ -793,10 +784,23 @@ export class PgTenantConnection { if (timeoutToReapply && timeoutToReapply > 0) { await transaction.query(buildSetLocalStatementTimeoutStatement(timeoutToReapply)) - transaction.setAppliedStatementTimeout(timeoutToReapply) } } } + + private getScopeValues(): unknown[] { + return [ + this.role, + this.role, + this.options.user.jwt || '', + this.options.user.payload.sub || '', + this.userPayload, + this.headersPayload, + this.options.method || '', + this.options.path || '', + this.options.operation?.() || '', + ] + } } function buildSetLocalStatementTimeoutStatement(statementTimeout: number): string { From 468ca09fd7527108243cf54731b964d8ada7f452 Mon Sep 17 00:00:00 2001 From: ferhat elmas Date: Mon, 29 Jun 2026 17:21:00 +0200 Subject: [PATCH 04/12] fix: improve Signed-off-by: ferhat elmas --- src/internal/database/pg-connection.test.ts | 96 +++++-- src/internal/database/pg-connection.ts | 271 +++++++++++++------- 2 files changed, 254 insertions(+), 113 deletions(-) diff --git a/src/internal/database/pg-connection.test.ts b/src/internal/database/pg-connection.test.ts index 91f9dec90..ac3a8e034 100644 --- a/src/internal/database/pg-connection.test.ts +++ b/src/internal/database/pg-connection.test.ts @@ -559,9 +559,7 @@ describe('PgTransaction', () => { query: vi.fn().mockResolvedValue({ rows: [] }), release: vi.fn(), } as unknown as PoolClient - const transaction = new PgTransaction(client) - - transaction.setPendingStatementTimeout(4321) + const transaction = new PgTransaction(client, undefined, { statementTimeoutMs: 4321 }) await transaction.query('SELECT 1') await transaction.query('SELECT 2') @@ -581,9 +579,7 @@ describe('PgTransaction', () => { query: vi.fn().mockResolvedValue({ rows: [] }), release: vi.fn(), } as unknown as PoolClient - const transaction = new PgTransaction(client) - - transaction.setPendingStatementTimeout(4321) + const transaction = new PgTransaction(client, undefined, { statementTimeoutMs: 4321 }) await expect( transaction.query('SELECT 1', { signal: AbortSignal.abort() }) @@ -609,9 +605,7 @@ describe('PgTransaction', () => { }), release: vi.fn(), }) as unknown as PoolClient & EventEmitter - const transaction = new PgTransaction(client) - - transaction.setPendingStatementTimeout(4321) + const transaction = new PgTransaction(client, undefined, { statementTimeoutMs: 4321 }) await expect( transaction.query('SELECT 1', { signal: controller.signal }) @@ -793,10 +787,18 @@ describe('PgTenantConnection', () => { query, release: vi.fn(), } as unknown as PoolClient - const transaction = new PgTransaction(client) + let transaction: PgTransaction + const beginTransaction = vi.fn( + async (options?: { statementTimeoutMs?: number }): Promise => { + transaction = new PgTransaction(client, undefined, { + statementTimeoutMs: options?.statementTimeoutMs, + }) + return transaction + } + ) const pool = { acquire: vi.fn().mockReturnValue({ - beginTransaction: vi.fn().mockResolvedValue(transaction), + beginTransaction, }), } as unknown as PgPoolStrategy const connection = new PgTenantConnection( @@ -806,10 +808,14 @@ describe('PgTenantConnection', () => { }) ) - await expect(connection.transaction({ timeout: 4321 })).resolves.toBe(transaction) + await expect(connection.transaction({ timeout: 4321 })).resolves.toBe(transaction!) + expect(beginTransaction).toHaveBeenCalledWith({ + timeout: 4321, + statementTimeoutMs: 4321, + }) expect(query).not.toHaveBeenCalled() - await connection.setScope(transaction) + await connection.setScope(transaction!) expect(query).toHaveBeenCalledTimes(1) const [statement, values] = query.mock.calls[0] @@ -835,10 +841,18 @@ describe('PgTenantConnection', () => { query, release: vi.fn(), } as unknown as PoolClient - const transaction = new PgTransaction(client) + let transaction: PgTransaction + const beginTransaction = vi.fn( + async (options?: { statementTimeoutMs?: number }): Promise => { + transaction = new PgTransaction(client, undefined, { + statementTimeoutMs: options?.statementTimeoutMs, + }) + return transaction + } + ) const pool = { acquire: vi.fn().mockReturnValue({ - beginTransaction: vi.fn().mockResolvedValue(transaction), + beginTransaction, }), } as unknown as PgPoolStrategy const connection = new PgTenantConnection( @@ -848,7 +862,11 @@ describe('PgTenantConnection', () => { }) ) - await expect(connection.transaction({ timeout: 4321 })).resolves.toBe(transaction) + await expect(connection.transaction({ timeout: 4321 })).resolves.toBe(transaction!) + expect(beginTransaction).toHaveBeenCalledWith({ + timeout: 4321, + statementTimeoutMs: 4321, + }) expect(query).toHaveBeenCalledTimes(1) expect(query).toHaveBeenNthCalledWith( @@ -857,7 +875,7 @@ describe('PgTenantConnection', () => { expect.any(Array) ) - await connection.setScope(transaction) + await connection.setScope(transaction!) expect(query).toHaveBeenCalledTimes(2) const [scopeStatement, scopeValues] = query.mock.calls[1] @@ -883,10 +901,22 @@ describe('PgTenantConnection', () => { query, release: vi.fn(), } as unknown as PoolClient - const transaction = new PgTransaction(client) + let transaction: PgTransaction + const beginTransaction = vi.fn( + async (options?: { + statementTimeoutMs?: number + statementTimeoutMode?: 'set-config' | 'set-local' + }): Promise => { + transaction = new PgTransaction(client, undefined, { + statementTimeoutMs: options?.statementTimeoutMs, + statementTimeoutMode: options?.statementTimeoutMode, + }) + return transaction + } + ) const pool = { acquire: vi.fn().mockReturnValue({ - beginTransaction: vi.fn().mockResolvedValue(transaction), + beginTransaction, }), } as unknown as PgPoolStrategy const connection = new PgTenantConnection( @@ -897,7 +927,12 @@ describe('PgTenantConnection', () => { }) ) - await expect(connection.transaction({ timeout: 4321 })).resolves.toBe(transaction) + await expect(connection.transaction({ timeout: 4321 })).resolves.toBe(transaction!) + expect(beginTransaction).toHaveBeenCalledWith({ + timeout: 4321, + statementTimeoutMs: 4321, + statementTimeoutMode: 'set-local', + }) expect(query).toHaveBeenCalledTimes(2) expect(query).toHaveBeenNthCalledWith( @@ -907,7 +942,7 @@ describe('PgTenantConnection', () => { ) expect(query).toHaveBeenNthCalledWith(2, "SET LOCAL statement_timeout = '4321ms'", undefined) - await connection.setScope(transaction) + await connection.setScope(transaction!) expect(query).toHaveBeenCalledTimes(4) const [scopeStatement, scopeValues] = query.mock.calls[2] @@ -932,10 +967,18 @@ describe('PgTenantConnection', () => { query, release: vi.fn(), } as unknown as PoolClient - const transaction = new PgTransaction(client) + let transaction: PgTransaction + const beginTransaction = vi.fn( + async (options?: { statementTimeoutMs?: number }): Promise => { + transaction = new PgTransaction(client, undefined, { + statementTimeoutMs: options?.statementTimeoutMs, + }) + return transaction + } + ) const pool = { acquire: vi.fn().mockReturnValue({ - beginTransaction: vi.fn().mockResolvedValue(transaction), + beginTransaction, }), } as unknown as PgPoolStrategy const connection = new PgTenantConnection( @@ -946,8 +989,8 @@ describe('PgTenantConnection', () => { ) await connection.transaction({ timeout: 4321 }) - await connection.setScope(transaction) - await transaction.query('SELECT 1') + await connection.setScope(transaction!) + await transaction!.query('SELECT 1') expect(query).toHaveBeenCalledTimes(2) expect(query).toHaveBeenNthCalledWith(2, 'SELECT 1', undefined) @@ -994,7 +1037,8 @@ describe('PgTenantConnection', () => { const setupError = new Error('search_path setup failed') const rollbackError = new Error('rollback failed') const transaction = { - query: vi.fn().mockRejectedValue(setupError), + query: vi.fn(), + runSetupQuery: vi.fn().mockRejectedValue(setupError), rollback: vi.fn().mockRejectedValue(rollbackError), } as unknown as PgTransaction const pool = { diff --git a/src/internal/database/pg-connection.ts b/src/internal/database/pg-connection.ts index 41be8efd2..78c33397b 100644 --- a/src/internal/database/pg-connection.ts +++ b/src/internal/database/pg-connection.ts @@ -46,36 +46,41 @@ export interface PgQueryOptions { type PgQueryArgument = PgQueryOptions | unknown[] -// The first nine placeholders must stay in the same order as getScopeValues(). -// The timeout variant appends statement_timeout as $10. -const scopeConfigStatement = ` - SELECT - set_config('role', $1, true), - set_config('request.jwt.claim.role', $2, true), - set_config('request.jwt', $3, true), - set_config('request.jwt.claim.sub', $4, true), - set_config('request.jwt.claims', $5, true), - set_config('request.headers', $6, true), - set_config('request.method', $7, true), - set_config('request.path', $8, true), - set_config('storage.operation', $9, true), - set_config('storage.allow_delete_query', 'true', true); - ` +type PgStatementTimeoutMode = 'set-config' | 'set-local' -const scopeConfigStatementWithTimeout = ` +interface PgTransactionOptions { + statementTimeoutMs?: number + statementTimeoutMode?: PgStatementTimeoutMode +} + +type PgBeginTransactionOptions = TransactionOptions & PgTransactionOptions + +const scopeConfigEntries: ReadonlyArray = [ + ['role', '$1'], + ['request.jwt.claim.role', '$2'], + ['request.jwt', '$3'], + ['request.jwt.claim.sub', '$4'], + ['request.jwt.claims', '$5'], + ['request.headers', '$6'], + ['request.method', '$7'], + ['request.path', '$8'], + ['storage.operation', '$9'], + ['storage.allow_delete_query', "'true'"], +] + +const scopeConfigStatement = buildScopeConfigSql(false) +const scopeConfigStatementWithTimeout = buildScopeConfigSql(true) + +function buildScopeConfigSql(includeStatementTimeout: boolean): string { + const entries = includeStatementTimeout + ? [...scopeConfigEntries, ['statement_timeout', '$10'] as const] + : scopeConfigEntries + + return ` SELECT - set_config('role', $1, true), - set_config('request.jwt.claim.role', $2, true), - set_config('request.jwt', $3, true), - set_config('request.jwt.claim.sub', $4, true), - set_config('request.jwt.claims', $5, true), - set_config('request.headers', $6, true), - set_config('request.method', $7, true), - set_config('request.path', $8, true), - set_config('storage.operation', $9, true), - set_config('storage.allow_delete_query', 'true', true), - set_config('statement_timeout', $10, true); + ${entries.map(([name, value]) => `set_config('${name}', ${value}, true)`).join(',\n ')}; ` +} export interface PgExecutor { query( @@ -85,7 +90,7 @@ export interface PgExecutor { } export interface PgTransactionalExecutor extends PgExecutor { - beginTransaction(options?: TransactionOptions): Promise + beginTransaction(options?: PgBeginTransactionOptions): Promise } interface PgPoolErrorContext { @@ -430,7 +435,7 @@ export class PgPoolExecutor implements PgTransactionalExecutor { } } - async beginTransaction(options?: TransactionOptions): Promise { + async beginTransaction(options?: PgBeginTransactionOptions): Promise { let client: PoolClient try { client = await this.pool.connect() @@ -443,10 +448,13 @@ export class PgPoolExecutor implements PgTransactionalExecutor { } const clientErrorTracker = new PgClientErrorTracker(client) - const transaction = new PgTransaction(client, clientErrorTracker) + const transaction = new PgTransaction(client, clientErrorTracker, { + statementTimeoutMs: options?.statementTimeoutMs, + statementTimeoutMode: options?.statementTimeoutMode, + }) try { - await transaction.query(buildBeginStatement(options)) + await transaction.runSetupQuery(buildBeginStatement(options)) return transaction } catch (e) { if (!transaction.isCompleted()) { @@ -464,39 +472,60 @@ export class PgPoolExecutor implements PgTransactionalExecutor { export class PgTransaction implements PgExecutor { private completed = false - private pendingStatementTimeoutMs?: number - private appliedStatementTimeoutMs?: number + private statementTimeoutMs?: number + private readonly statementTimeoutMode: PgStatementTimeoutMode constructor( private readonly client: PoolClient, - private readonly clientErrorTracker?: PgClientErrorTracker - ) {} + private readonly clientErrorTracker?: PgClientErrorTracker, + options: PgTransactionOptions = {} + ) { + this.statementTimeoutMs = normalizeStatementTimeoutMs(options.statementTimeoutMs) + this.statementTimeoutMode = options.statementTimeoutMode ?? 'set-config' + } isCompleted(): boolean { return this.completed } - setPendingStatementTimeout(timeoutMs: number | undefined): void { - this.pendingStatementTimeoutMs = timeoutMs && timeoutMs > 0 ? timeoutMs : undefined + async query( + statement: string | PgStatement, + options?: PgQueryArgument + ): Promise> { + return this.runQuery(statement, options, true) } - takePendingStatementTimeout(): number | undefined { - const timeoutMs = this.pendingStatementTimeoutMs - this.pendingStatementTimeoutMs = undefined - return timeoutMs + async runSetupQuery( + statement: string | PgStatement, + options?: PgQueryArgument + ): Promise> { + // Setup statements must not consume a deferred timeout before scope can fold it in. + return this.runQuery(statement, options, false) } - setAppliedStatementTimeout(timeoutMs: number | undefined): void { - this.appliedStatementTimeoutMs = timeoutMs && timeoutMs > 0 ? timeoutMs : undefined + async runScopeQuery(scopeValues: unknown[]): Promise { + const statementTimeout = this.consumeStatementTimeoutForSetConfig() + + await this.runQuery(buildScopeConfigQuery(scopeValues, statementTimeout), undefined, false) + await this.reapplyLocalStatementTimeoutAfterScope() } - getAppliedStatementTimeout(): number | undefined { - return this.appliedStatementTimeoutMs + async applyLocalStatementTimeout(): Promise { + if (this.statementTimeoutMode !== 'set-local' || !this.statementTimeoutMs) { + return + } + + await this.runQuery( + buildSetLocalStatementTimeoutStatement(this.statementTimeoutMs), + undefined, + false + ) } - async query( + private async runQuery( statement: string | PgStatement, - options?: PgQueryArgument + options: PgQueryArgument | undefined, + applyPendingStatementTimeout: boolean ): Promise> { if (this.completed) { throw new Error('Cannot query a completed transaction') @@ -507,9 +536,8 @@ export class PgTransaction implements PgExecutor { try { assertValidSignal(signal) - const pendingStatementTimeout = this.takePendingStatementTimeoutStatement() - if (pendingStatementTimeout) { - await runPgQuery(this.client, pendingStatementTimeout, { signal }) + if (applyPendingStatementTimeout) { + await this.applyPendingStatementTimeoutBeforeQuery(signal) } const result = await runPgQuery(this.client, statement, options) @@ -577,17 +605,43 @@ export class PgTransaction implements PgExecutor { } } - private takePendingStatementTimeoutStatement(): PgStatement | undefined { - const timeoutMs = this.takePendingStatementTimeout() - + private async applyPendingStatementTimeoutBeforeQuery(signal?: AbortSignal): Promise { + const timeoutMs = this.consumeStatementTimeoutForSetConfig() if (!timeoutMs) { + return + } + + await runPgQuery( + this.client, + { + text: `SELECT set_config('statement_timeout', $1, true)`, + values: [`${timeoutMs}ms`], + }, + { signal } + ) + } + + private consumeStatementTimeoutForSetConfig(): number | undefined { + if (this.statementTimeoutMode !== 'set-config') { return undefined } - return { - text: `SELECT set_config('statement_timeout', $1, true)`, - values: [`${timeoutMs}ms`], + const timeoutMs = this.statementTimeoutMs + this.statementTimeoutMs = undefined + return timeoutMs + } + + private async reapplyLocalStatementTimeoutAfterScope(): Promise { + if (this.statementTimeoutMode !== 'set-local' || !this.statementTimeoutMs) { + return } + + // Multigres requires SET LOCAL for statement_timeout, and role scope setup resets it. + await this.runQuery( + buildSetLocalStatementTimeoutStatement(this.statementTimeoutMs), + undefined, + false + ) } } @@ -640,7 +694,23 @@ export class PgTenantConnection { async beginTransaction(options?: TransactionOptions): Promise { this.assertNotDisposed() - return this.pool.acquire().beginTransaction(options) + const statementTimeout = options?.timeout + const isMultigres = this.isMultigresDatabase() + const transaction = await this.pool + .acquire() + .beginTransaction( + this.withStatementTimeout( + options, + statementTimeout, + isMultigres ? 'set-local' : 'set-config' + ) + ) + + if (isMultigres) { + await this.applyStatementTimeoutImmediately(transaction) + } + + return transaction } asSuperUser() { @@ -662,6 +732,13 @@ export class PgTenantConnection { this.assertNotDisposed() try { + const statementTimeout = opts?.timeout ?? databaseStatementTimeout + const isMultigres = this.isMultigresDatabase() + const beginOptions = this.withStatementTimeout( + opts, + statementTimeout, + isMultigres ? 'set-local' : 'set-config' + ) const transaction = await retry( async (bail) => { if (this.disposed) { @@ -670,7 +747,7 @@ export class PgTenantConnection { } try { - return await this.pool.acquire().beginTransaction(opts) + return await this.pool.acquire().beginTransaction(beginOptions) } catch (e) { if (isConnectionLimitError(e)) { throw e @@ -691,7 +768,7 @@ export class PgTenantConnection { if (this.options.isExternalPool) { try { - await transaction.query({ + await transaction.runSetupQuery({ text: `SELECT set_config('search_path', $1, true)`, values: [searchPath.join(',')], }) @@ -701,11 +778,8 @@ export class PgTenantConnection { } } - const statementTimeout = opts?.timeout ?? databaseStatementTimeout - if (this.isMultigresDatabase()) { - await this.applyStatementTimeoutImmediately(transaction, statementTimeout) - } else { - transaction.setPendingStatementTimeout(statementTimeout) + if (isMultigres) { + await this.applyStatementTimeoutImmediately(transaction) } return transaction @@ -722,17 +796,9 @@ export class PgTenantConnection { return (this.options.databaseEngine ?? databaseEngine) === 'multigres' } - private async applyStatementTimeoutImmediately( - transaction: PgTransaction, - statementTimeout: number | undefined - ): Promise { - if (!statementTimeout || statementTimeout <= 0) { - return - } - + private async applyStatementTimeoutImmediately(transaction: PgTransaction): Promise { try { - await transaction.query(buildSetLocalStatementTimeoutStatement(statementTimeout)) - transaction.setAppliedStatementTimeout(statementTimeout) + await transaction.applyLocalStatementTimeout() } catch (e) { await this.rollbackTransactionSafely(transaction, e, 'statement_timeout setup') throw e @@ -768,23 +834,31 @@ export class PgTenantConnection { async setScope(tnx: PgExecutor) { const transaction = tnx instanceof PgTransaction ? tnx : undefined - const pendingStatementTimeout = transaction ? transaction.takePendingStatementTimeout() : 0 - const isMultigres = this.isMultigresDatabase() - const statementTimeout = isMultigres ? 0 : pendingStatementTimeout const scopeValues = this.getScopeValues() - await tnx.query({ - text: statementTimeout ? scopeConfigStatementWithTimeout : scopeConfigStatement, - values: statementTimeout ? [...scopeValues, `${statementTimeout}ms`] : scopeValues, - }) + if (transaction) { + await transaction.runScopeQuery(scopeValues) + return + } - if (isMultigres && transaction) { - // Multigres requires SET LOCAL for statement_timeout, and role scope setup resets it. - const timeoutToReapply = transaction.getAppliedStatementTimeout() + await tnx.query(buildScopeConfigQuery(scopeValues)) + } - if (timeoutToReapply && timeoutToReapply > 0) { - await transaction.query(buildSetLocalStatementTimeoutStatement(timeoutToReapply)) - } + private withStatementTimeout( + options: TransactionOptions | undefined, + statementTimeout: number | undefined, + statementTimeoutMode: PgStatementTimeoutMode + ): PgBeginTransactionOptions | undefined { + const statementTimeoutMs = normalizeStatementTimeoutMs(statementTimeout) + + if (!statementTimeoutMs) { + return options + } + + return { + ...options, + statementTimeoutMs, + ...(statementTimeoutMode === 'set-local' ? { statementTimeoutMode } : undefined), } } @@ -803,6 +877,29 @@ export class PgTenantConnection { } } +function buildScopeConfigQuery(scopeValues: unknown[], statementTimeout?: number): PgStatement { + const normalizedStatementTimeout = normalizeStatementTimeoutMs(statementTimeout) + + return { + text: normalizedStatementTimeout ? scopeConfigStatementWithTimeout : scopeConfigStatement, + values: normalizedStatementTimeout + ? [...scopeValues, `${normalizedStatementTimeout}ms`] + : scopeValues, + } +} + +function normalizeStatementTimeoutMs(timeoutMs: number | undefined): number | undefined { + if (timeoutMs === undefined || timeoutMs <= 0) { + return undefined + } + + if (!Number.isFinite(timeoutMs)) { + throw new Error(`Invalid statement timeout: ${timeoutMs}`) + } + + return timeoutMs +} + function buildSetLocalStatementTimeoutStatement(statementTimeout: number): string { if (!Number.isFinite(statementTimeout) || statementTimeout <= 0) { throw new Error(`Invalid statement timeout: ${statementTimeout}`) From 103081badff54aad0a124cd2ecdb65e628c02d5b Mon Sep 17 00:00:00 2001 From: ferhat elmas Date: Tue, 30 Jun 2026 17:36:24 +0200 Subject: [PATCH 05/12] fix: more tests Signed-off-by: ferhat elmas --- src/internal/database/pg-connection.test.ts | 318 ++++++++++++++++++-- src/internal/database/pg-connection.ts | 63 ++-- 2 files changed, 326 insertions(+), 55 deletions(-) diff --git a/src/internal/database/pg-connection.test.ts b/src/internal/database/pg-connection.test.ts index ac3a8e034..27b6cb1dc 100644 --- a/src/internal/database/pg-connection.test.ts +++ b/src/internal/database/pg-connection.test.ts @@ -57,6 +57,64 @@ function createDatabaseError(code: string | undefined, message = 'database error return error } +type TestPgBeginTransactionOptions = { + timeout?: number + statementTimeoutMs?: number + statementTimeoutMode?: 'set-config' | 'set-local' +} + +function normalizeTestStatementTimeoutMs( + options?: TestPgBeginTransactionOptions +): number | undefined { + const timeoutMs = options?.statementTimeoutMs ?? options?.timeout + + if (timeoutMs === undefined || !Number.isFinite(timeoutMs) || timeoutMs <= 0) { + return undefined + } + + return timeoutMs +} + +function createMockTenantConnectionWithTransaction( + overrides: Partial[0]> = {}, + query = vi.fn().mockResolvedValue({ rows: [] }) +) { + const client = { + query, + release: vi.fn(), + } as unknown as PoolClient + let transaction: PgTransaction | undefined + const beginTransaction = vi.fn( + async (options?: TestPgBeginTransactionOptions): Promise => { + transaction = new PgTransaction(client, undefined, { + statementTimeoutMs: normalizeTestStatementTimeoutMs(options), + statementTimeoutMode: options?.statementTimeoutMode, + }) + return transaction + } + ) + const pool = { + acquire: vi.fn().mockReturnValue({ + beginTransaction, + }), + } as unknown as PgPoolStrategy + const connection = new PgTenantConnection(pool, createPoolStrategySettings(overrides)) + + return { + beginTransaction, + client, + connection, + query, + getTransaction() { + if (!transaction) { + throw new Error('Expected test transaction to be created') + } + + return transaction + }, + } +} + async function expectQueryErrorRelease(error: Error): Promise> { const release = vi.fn() const client = Object.assign(new EventEmitter(), { @@ -403,6 +461,42 @@ describe('PgPoolExecutor', () => { expect(client.listenerCount('error')).toBe(0) }) + it('uses transaction timeout with isolation and read-only begin modes', async () => { + const client = Object.assign(new EventEmitter(), { + query: vi.fn().mockResolvedValue({ rows: [] }), + release: vi.fn(), + }) as unknown as PoolClient & EventEmitter + const pool = { + connect: vi.fn().mockResolvedValue(client), + } as unknown as Pool + const executor = new PgPoolExecutor(pool) + + const transaction = await executor.beginTransaction({ + timeout: 4321, + isolation: 'repeatable read', + readOnly: true, + }) + + expect(client.query).toHaveBeenCalledTimes(1) + expect(client.query).toHaveBeenNthCalledWith( + 1, + 'BEGIN ISOLATION LEVEL REPEATABLE READ, READ ONLY', + undefined + ) + + await transaction.query('SELECT 1') + + expect(client.query).toHaveBeenCalledTimes(3) + expect(client.query).toHaveBeenNthCalledWith( + 2, + "SELECT set_config('statement_timeout', $1, true)", + ['4321ms'] + ) + expect(client.query).toHaveBeenNthCalledWith(3, 'SELECT 1', undefined) + + await transaction.rollback() + }) + it('returns clients to the pool after regular SQL errors', async () => { for (const code of ['42P01', '23505', '23503', '42501', '22P02', '42703']) { const error = createDatabaseError(code) @@ -574,6 +668,27 @@ describe('PgTransaction', () => { expect(client.query).toHaveBeenNthCalledWith(3, 'SELECT 2', undefined) }) + it('applies SET LOCAL for set-local direct queries', async () => { + const client = { + query: vi.fn().mockResolvedValue({ rows: [] }), + release: vi.fn(), + } as unknown as PoolClient + const transaction = new PgTransaction(client, undefined, { + statementTimeoutMs: 4321, + statementTimeoutMode: 'set-local', + }) + + await transaction.query('SELECT 1') + + expect(client.query).toHaveBeenCalledTimes(2) + expect(client.query).toHaveBeenNthCalledWith( + 1, + "SET LOCAL statement_timeout = '4321ms'", + undefined + ) + expect(client.query).toHaveBeenNthCalledWith(2, 'SELECT 1', undefined) + }) + it('rejects a pre-aborted direct query before applying a pending statement timeout', async () => { const client = { query: vi.fn().mockResolvedValue({ rows: [] }), @@ -781,6 +896,69 @@ describe('PgTenantConnection', () => { }) }) + it('treats non-finite transaction timeouts as disabled', async () => { + for (const timeout of [Number.NaN, Number.POSITIVE_INFINITY]) { + const { beginTransaction, connection, query, getTransaction } = + createMockTenantConnectionWithTransaction({ + isExternalPool: false, + }) + + const transaction = await connection.transaction({ timeout }) + expect(transaction).toBe(getTransaction()) + expect(beginTransaction).toHaveBeenCalledWith({ timeout }) + expect(query).not.toHaveBeenCalled() + + await connection.setScope(getTransaction()) + + expect(query).toHaveBeenCalledTimes(1) + const [scopeStatement, scopeValues] = query.mock.calls[0] + expect(scopeStatement).toContain("set_config('role', $1, true)") + expect(scopeStatement).not.toContain("set_config('statement_timeout'") + expect(scopeValues).toHaveLength(9) + } + }) + + it('defers statement_timeout for low-level Postgres beginTransaction', async () => { + const { beginTransaction, connection, query, getTransaction } = + createMockTenantConnectionWithTransaction({ + isExternalPool: false, + }) + + await expect( + connection.beginTransaction({ + timeout: 4321, + isolation: 'serializable', + readOnly: true, + }) + ).resolves.toBe(getTransaction()) + expect(beginTransaction).toHaveBeenCalledWith({ + timeout: 4321, + isolation: 'serializable', + readOnly: true, + statementTimeoutMs: 4321, + }) + expect(query).not.toHaveBeenCalled() + + await connection.setScope(getTransaction()) + + expect(query).toHaveBeenCalledTimes(1) + const [scopeStatement, scopeValues] = query.mock.calls[0] + expect(scopeStatement).toContain("set_config('role', $1, true)") + expect(scopeStatement).toContain("set_config('statement_timeout', $10, true)") + expect(scopeValues).toEqual([ + 'authenticated', + 'authenticated', + 'jwt', + '', + JSON.stringify({ role: 'authenticated' }), + '{}', + '', + '', + '', + '4321ms', + ]) + }) + it('defers statement_timeout setup until the first scope application', async () => { const query = vi.fn().mockResolvedValue({ rows: [] }) const client = { @@ -789,9 +967,9 @@ describe('PgTenantConnection', () => { } as unknown as PoolClient let transaction: PgTransaction const beginTransaction = vi.fn( - async (options?: { statementTimeoutMs?: number }): Promise => { + async (options?: TestPgBeginTransactionOptions): Promise => { transaction = new PgTransaction(client, undefined, { - statementTimeoutMs: options?.statementTimeoutMs, + statementTimeoutMs: normalizeTestStatementTimeoutMs(options), }) return transaction } @@ -843,9 +1021,9 @@ describe('PgTenantConnection', () => { } as unknown as PoolClient let transaction: PgTransaction const beginTransaction = vi.fn( - async (options?: { statementTimeoutMs?: number }): Promise => { + async (options?: TestPgBeginTransactionOptions): Promise => { transaction = new PgTransaction(client, undefined, { - statementTimeoutMs: options?.statementTimeoutMs, + statementTimeoutMs: normalizeTestStatementTimeoutMs(options), }) return transaction } @@ -895,7 +1073,7 @@ describe('PgTenantConnection', () => { ]) }) - it('uses a standalone statement_timeout setup for Multigres transactions', async () => { + it('applies Multigres statement_timeout after scope setup', async () => { const query = vi.fn().mockResolvedValue({ rows: [] }) const client = { query, @@ -903,12 +1081,9 @@ describe('PgTenantConnection', () => { } as unknown as PoolClient let transaction: PgTransaction const beginTransaction = vi.fn( - async (options?: { - statementTimeoutMs?: number - statementTimeoutMode?: 'set-config' | 'set-local' - }): Promise => { + async (options?: TestPgBeginTransactionOptions): Promise => { transaction = new PgTransaction(client, undefined, { - statementTimeoutMs: options?.statementTimeoutMs, + statementTimeoutMs: normalizeTestStatementTimeoutMs(options), statementTimeoutMode: options?.statementTimeoutMode, }) return transaction @@ -934,18 +1109,17 @@ describe('PgTenantConnection', () => { statementTimeoutMode: 'set-local', }) - expect(query).toHaveBeenCalledTimes(2) + expect(query).toHaveBeenCalledTimes(1) expect(query).toHaveBeenNthCalledWith( 1, "SELECT set_config('search_path', $1, true)", expect.any(Array) ) - expect(query).toHaveBeenNthCalledWith(2, "SET LOCAL statement_timeout = '4321ms'", undefined) await connection.setScope(transaction!) - expect(query).toHaveBeenCalledTimes(4) - const [scopeStatement, scopeValues] = query.mock.calls[2] + expect(query).toHaveBeenCalledTimes(3) + const [scopeStatement, scopeValues] = query.mock.calls[1] expect(scopeStatement).not.toContain("set_config('statement_timeout'") expect(scopeValues).toEqual([ 'authenticated', @@ -958,7 +1132,117 @@ describe('PgTenantConnection', () => { '', '', ]) - expect(query).toHaveBeenNthCalledWith(4, "SET LOCAL statement_timeout = '4321ms'", undefined) + expect(query).toHaveBeenNthCalledWith(3, "SET LOCAL statement_timeout = '4321ms'", undefined) + }) + + it('applies Multigres statement_timeout before the first direct transaction query', async () => { + const { beginTransaction, connection, query, getTransaction } = + createMockTenantConnectionWithTransaction({ + isExternalPool: false, + databaseEngine: 'multigres', + }) + + await expect(connection.transaction({ timeout: 4321 })).resolves.toBe(getTransaction()) + expect(beginTransaction).toHaveBeenCalledWith({ + timeout: 4321, + statementTimeoutMs: 4321, + statementTimeoutMode: 'set-local', + }) + expect(query).not.toHaveBeenCalled() + + await getTransaction().query('SELECT 1') + + expect(query).toHaveBeenCalledTimes(2) + expect(query).toHaveBeenNthCalledWith(1, "SET LOCAL statement_timeout = '4321ms'", undefined) + expect(query).toHaveBeenNthCalledWith(2, 'SELECT 1', undefined) + }) + + it('applies SET LOCAL once for low-level Multigres beginTransaction before direct queries', async () => { + const { beginTransaction, connection, query, getTransaction } = + createMockTenantConnectionWithTransaction({ + isExternalPool: false, + databaseEngine: 'multigres', + }) + + await expect(connection.beginTransaction({ timeout: 4321 })).resolves.toBe(getTransaction()) + expect(beginTransaction).toHaveBeenCalledWith({ + timeout: 4321, + statementTimeoutMs: 4321, + statementTimeoutMode: 'set-local', + }) + expect(query).toHaveBeenCalledTimes(1) + expect(query).toHaveBeenNthCalledWith(1, "SET LOCAL statement_timeout = '4321ms'", undefined) + + await getTransaction().query('SELECT 1') + + expect(query).toHaveBeenCalledTimes(2) + expect(query).toHaveBeenNthCalledWith(2, 'SELECT 1', undefined) + expect( + query.mock.calls.filter(([statement]) => + String(statement).includes("set_config('statement_timeout'") + ) + ).toHaveLength(0) + expect( + query.mock.calls.filter(([statement]) => + String(statement).includes('SET LOCAL statement_timeout') + ) + ).toHaveLength(1) + }) + + it('omits statement_timeout setup for low-level beginTransaction without a positive timeout', async () => { + const cases: Array<{ + databaseEngine?: 'multigres' + options?: { timeout: number } + }> = [ + {}, + { options: { timeout: 0 } }, + { databaseEngine: 'multigres' }, + { databaseEngine: 'multigres', options: { timeout: 0 } }, + ] + + for (const { databaseEngine, options } of cases) { + const { beginTransaction, connection, query, getTransaction } = + createMockTenantConnectionWithTransaction({ + isExternalPool: false, + databaseEngine, + }) + + await connection.beginTransaction(options) + expect(beginTransaction).toHaveBeenCalledWith(options) + expect(query).not.toHaveBeenCalled() + + await connection.setScope(getTransaction()) + + expect(query).toHaveBeenCalledTimes(1) + const [scopeStatement, scopeValues] = query.mock.calls[0] + expect(scopeStatement).toContain("set_config('role', $1, true)") + expect(scopeStatement).not.toContain("set_config('statement_timeout'") + expect(scopeValues).toHaveLength(9) + expect( + query.mock.calls.filter(([statement]) => + String(statement).includes('SET LOCAL statement_timeout') + ) + ).toHaveLength(0) + } + }) + + it('rolls back and propagates Multigres SET LOCAL setup failures', async () => { + const setupError = new Error('statement_timeout setup failed') + const query = vi.fn().mockRejectedValueOnce(setupError).mockResolvedValue({ rows: [] }) + const { client, connection } = createMockTenantConnectionWithTransaction( + { + isExternalPool: false, + databaseEngine: 'multigres', + }, + query + ) + + await expect(connection.beginTransaction({ timeout: 4321 })).rejects.toBe(setupError) + + expect(query).toHaveBeenCalledTimes(2) + expect(query).toHaveBeenNthCalledWith(1, "SET LOCAL statement_timeout = '4321ms'", undefined) + expect(query).toHaveBeenNthCalledWith(2, 'ROLLBACK', undefined) + expect(client.release).toHaveBeenCalledWith(undefined) }) it('does not re-apply statement_timeout after setScope consumes it', async () => { @@ -969,9 +1253,9 @@ describe('PgTenantConnection', () => { } as unknown as PoolClient let transaction: PgTransaction const beginTransaction = vi.fn( - async (options?: { statementTimeoutMs?: number }): Promise => { + async (options?: TestPgBeginTransactionOptions): Promise => { transaction = new PgTransaction(client, undefined, { - statementTimeoutMs: options?.statementTimeoutMs, + statementTimeoutMs: normalizeTestStatementTimeoutMs(options), }) return transaction } diff --git a/src/internal/database/pg-connection.ts b/src/internal/database/pg-connection.ts index 78c33397b..e7acc79aa 100644 --- a/src/internal/database/pg-connection.ts +++ b/src/internal/database/pg-connection.ts @@ -449,7 +449,8 @@ export class PgPoolExecutor implements PgTransactionalExecutor { const clientErrorTracker = new PgClientErrorTracker(client) const transaction = new PgTransaction(client, clientErrorTracker, { - statementTimeoutMs: options?.statementTimeoutMs, + statementTimeoutMs: + options?.statementTimeoutMs ?? normalizeStatementTimeoutMs(options?.timeout), statementTimeoutMode: options?.statementTimeoutMode, }) @@ -473,6 +474,7 @@ export class PgPoolExecutor implements PgTransactionalExecutor { export class PgTransaction implements PgExecutor { private completed = false private statementTimeoutMs?: number + private localStatementTimeoutApplied = false private readonly statementTimeoutMode: PgStatementTimeoutMode constructor( @@ -480,7 +482,7 @@ export class PgTransaction implements PgExecutor { private readonly clientErrorTracker?: PgClientErrorTracker, options: PgTransactionOptions = {} ) { - this.statementTimeoutMs = normalizeStatementTimeoutMs(options.statementTimeoutMs) + this.statementTimeoutMs = options.statementTimeoutMs this.statementTimeoutMode = options.statementTimeoutMode ?? 'set-config' } @@ -507,19 +509,24 @@ export class PgTransaction implements PgExecutor { const statementTimeout = this.consumeStatementTimeoutForSetConfig() await this.runQuery(buildScopeConfigQuery(scopeValues, statementTimeout), undefined, false) - await this.reapplyLocalStatementTimeoutAfterScope() + this.localStatementTimeoutApplied = false + await this.applyLocalStatementTimeout() } - async applyLocalStatementTimeout(): Promise { - if (this.statementTimeoutMode !== 'set-local' || !this.statementTimeoutMs) { + async applyLocalStatementTimeout(signal?: AbortSignal): Promise { + if ( + this.statementTimeoutMode !== 'set-local' || + this.localStatementTimeoutApplied || + !this.statementTimeoutMs + ) { return } - await this.runQuery( - buildSetLocalStatementTimeoutStatement(this.statementTimeoutMs), - undefined, - false - ) + await runPgQuery(this.client, buildSetLocalStatementTimeoutStatement(this.statementTimeoutMs), { + signal, + }) + this.clientErrorTracker?.throwIfErrored() + this.localStatementTimeoutApplied = true } private async runQuery( @@ -606,6 +613,11 @@ export class PgTransaction implements PgExecutor { } private async applyPendingStatementTimeoutBeforeQuery(signal?: AbortSignal): Promise { + if (this.statementTimeoutMode === 'set-local') { + await this.applyLocalStatementTimeout(signal) + return + } + const timeoutMs = this.consumeStatementTimeoutForSetConfig() if (!timeoutMs) { return @@ -630,19 +642,6 @@ export class PgTransaction implements PgExecutor { this.statementTimeoutMs = undefined return timeoutMs } - - private async reapplyLocalStatementTimeoutAfterScope(): Promise { - if (this.statementTimeoutMode !== 'set-local' || !this.statementTimeoutMs) { - return - } - - // Multigres requires SET LOCAL for statement_timeout, and role scope setup resets it. - await this.runQuery( - buildSetLocalStatementTimeoutStatement(this.statementTimeoutMs), - undefined, - false - ) - } } export class PgTenantConnection { @@ -778,10 +777,6 @@ export class PgTenantConnection { } } - if (isMultigres) { - await this.applyStatementTimeoutImmediately(transaction) - } - return transaction } catch (e) { if (isConnectionTimeoutError(e)) { @@ -878,25 +873,17 @@ export class PgTenantConnection { } function buildScopeConfigQuery(scopeValues: unknown[], statementTimeout?: number): PgStatement { - const normalizedStatementTimeout = normalizeStatementTimeoutMs(statementTimeout) - return { - text: normalizedStatementTimeout ? scopeConfigStatementWithTimeout : scopeConfigStatement, - values: normalizedStatementTimeout - ? [...scopeValues, `${normalizedStatementTimeout}ms`] - : scopeValues, + text: statementTimeout ? scopeConfigStatementWithTimeout : scopeConfigStatement, + values: statementTimeout ? [...scopeValues, `${statementTimeout}ms`] : scopeValues, } } function normalizeStatementTimeoutMs(timeoutMs: number | undefined): number | undefined { - if (timeoutMs === undefined || timeoutMs <= 0) { + if (timeoutMs === undefined || !Number.isFinite(timeoutMs) || timeoutMs <= 0) { return undefined } - if (!Number.isFinite(timeoutMs)) { - throw new Error(`Invalid statement timeout: ${timeoutMs}`) - } - return timeoutMs } From 3d2c1ff60b507c43ee129ef6863f3e2775b1e4e5 Mon Sep 17 00:00:00 2001 From: ferhat elmas Date: Wed, 8 Jul 2026 00:32:53 +0200 Subject: [PATCH 06/12] fix: remove multigres timeout workaround Signed-off-by: ferhat elmas --- src/internal/database/client.ts | 4 +- src/internal/database/pg-connection.test.ts | 136 ++------------------ src/internal/database/pg-connection.ts | 84 +----------- src/internal/database/pool.ts | 3 +- src/test/pg-connection.test.ts | 1 - src/test/storage-pg-db.test.ts | 3 +- 6 files changed, 20 insertions(+), 211 deletions(-) diff --git a/src/internal/database/client.ts b/src/internal/database/client.ts index 65df7e982..3c1377424 100644 --- a/src/internal/database/client.ts +++ b/src/internal/database/client.ts @@ -41,8 +41,7 @@ async function getDbSettings( host: string | undefined, options?: { disableHostCheck?: boolean } ) { - const { isMultitenant, databasePoolURL, databaseURL, databaseEngine, databaseMaxConnections } = - getConfig() + const { isMultitenant, databasePoolURL, databaseURL, databaseMaxConnections } = getConfig() let dbUrl = databasePoolURL || databaseURL let maxConnections = databaseMaxConnections @@ -76,7 +75,6 @@ async function getDbSettings( return { dbUrl, - databaseEngine, isExternalPool, maxConnections, } diff --git a/src/internal/database/pg-connection.test.ts b/src/internal/database/pg-connection.test.ts index 27b6cb1dc..772fd42bb 100644 --- a/src/internal/database/pg-connection.test.ts +++ b/src/internal/database/pg-connection.test.ts @@ -60,7 +60,6 @@ function createDatabaseError(code: string | undefined, message = 'database error type TestPgBeginTransactionOptions = { timeout?: number statementTimeoutMs?: number - statementTimeoutMode?: 'set-config' | 'set-local' } function normalizeTestStatementTimeoutMs( @@ -88,7 +87,6 @@ function createMockTenantConnectionWithTransaction( async (options?: TestPgBeginTransactionOptions): Promise => { transaction = new PgTransaction(client, undefined, { statementTimeoutMs: normalizeTestStatementTimeoutMs(options), - statementTimeoutMode: options?.statementTimeoutMode, }) return transaction } @@ -668,27 +666,6 @@ describe('PgTransaction', () => { expect(client.query).toHaveBeenNthCalledWith(3, 'SELECT 2', undefined) }) - it('applies SET LOCAL for set-local direct queries', async () => { - const client = { - query: vi.fn().mockResolvedValue({ rows: [] }), - release: vi.fn(), - } as unknown as PoolClient - const transaction = new PgTransaction(client, undefined, { - statementTimeoutMs: 4321, - statementTimeoutMode: 'set-local', - }) - - await transaction.query('SELECT 1') - - expect(client.query).toHaveBeenCalledTimes(2) - expect(client.query).toHaveBeenNthCalledWith( - 1, - "SET LOCAL statement_timeout = '4321ms'", - undefined - ) - expect(client.query).toHaveBeenNthCalledWith(2, 'SELECT 1', undefined) - }) - it('rejects a pre-aborted direct query before applying a pending statement timeout', async () => { const client = { query: vi.fn().mockResolvedValue({ rows: [] }), @@ -1073,7 +1050,7 @@ describe('PgTenantConnection', () => { ]) }) - it('applies Multigres statement_timeout after scope setup', async () => { + it('folds Multigres statement_timeout into scope setup', async () => { const query = vi.fn().mockResolvedValue({ rows: [] }) const client = { query, @@ -1084,7 +1061,6 @@ describe('PgTenantConnection', () => { async (options?: TestPgBeginTransactionOptions): Promise => { transaction = new PgTransaction(client, undefined, { statementTimeoutMs: normalizeTestStatementTimeoutMs(options), - statementTimeoutMode: options?.statementTimeoutMode, }) return transaction } @@ -1094,19 +1070,18 @@ describe('PgTenantConnection', () => { beginTransaction, }), } as unknown as PgPoolStrategy - const connection = new PgTenantConnection( - pool, - createPoolStrategySettings({ + const settings = { + ...createPoolStrategySettings({ isExternalPool: true, - databaseEngine: 'multigres', - }) - ) + }), + databaseEngine: 'multigres', + } as TenantConnectionOptions + const connection = new PgTenantConnection(pool, settings) await expect(connection.transaction({ timeout: 4321 })).resolves.toBe(transaction!) expect(beginTransaction).toHaveBeenCalledWith({ timeout: 4321, statementTimeoutMs: 4321, - statementTimeoutMode: 'set-local', }) expect(query).toHaveBeenCalledTimes(1) @@ -1118,9 +1093,9 @@ describe('PgTenantConnection', () => { await connection.setScope(transaction!) - expect(query).toHaveBeenCalledTimes(3) + expect(query).toHaveBeenCalledTimes(2) const [scopeStatement, scopeValues] = query.mock.calls[1] - expect(scopeStatement).not.toContain("set_config('statement_timeout'") + expect(scopeStatement).toContain("set_config('statement_timeout', $10, true)") expect(scopeValues).toEqual([ 'authenticated', 'authenticated', @@ -1131,80 +1106,19 @@ describe('PgTenantConnection', () => { '', '', '', + '4321ms', ]) - expect(query).toHaveBeenNthCalledWith(3, "SET LOCAL statement_timeout = '4321ms'", undefined) - }) - - it('applies Multigres statement_timeout before the first direct transaction query', async () => { - const { beginTransaction, connection, query, getTransaction } = - createMockTenantConnectionWithTransaction({ - isExternalPool: false, - databaseEngine: 'multigres', - }) - - await expect(connection.transaction({ timeout: 4321 })).resolves.toBe(getTransaction()) - expect(beginTransaction).toHaveBeenCalledWith({ - timeout: 4321, - statementTimeoutMs: 4321, - statementTimeoutMode: 'set-local', - }) - expect(query).not.toHaveBeenCalled() - - await getTransaction().query('SELECT 1') - - expect(query).toHaveBeenCalledTimes(2) - expect(query).toHaveBeenNthCalledWith(1, "SET LOCAL statement_timeout = '4321ms'", undefined) - expect(query).toHaveBeenNthCalledWith(2, 'SELECT 1', undefined) - }) - - it('applies SET LOCAL once for low-level Multigres beginTransaction before direct queries', async () => { - const { beginTransaction, connection, query, getTransaction } = - createMockTenantConnectionWithTransaction({ - isExternalPool: false, - databaseEngine: 'multigres', - }) - - await expect(connection.beginTransaction({ timeout: 4321 })).resolves.toBe(getTransaction()) - expect(beginTransaction).toHaveBeenCalledWith({ - timeout: 4321, - statementTimeoutMs: 4321, - statementTimeoutMode: 'set-local', - }) - expect(query).toHaveBeenCalledTimes(1) - expect(query).toHaveBeenNthCalledWith(1, "SET LOCAL statement_timeout = '4321ms'", undefined) - - await getTransaction().query('SELECT 1') - - expect(query).toHaveBeenCalledTimes(2) - expect(query).toHaveBeenNthCalledWith(2, 'SELECT 1', undefined) - expect( - query.mock.calls.filter(([statement]) => - String(statement).includes("set_config('statement_timeout'") - ) - ).toHaveLength(0) - expect( - query.mock.calls.filter(([statement]) => - String(statement).includes('SET LOCAL statement_timeout') - ) - ).toHaveLength(1) }) it('omits statement_timeout setup for low-level beginTransaction without a positive timeout', async () => { const cases: Array<{ - databaseEngine?: 'multigres' options?: { timeout: number } - }> = [ - {}, - { options: { timeout: 0 } }, - { databaseEngine: 'multigres' }, - { databaseEngine: 'multigres', options: { timeout: 0 } }, - ] - - for (const { databaseEngine, options } of cases) { + }> = [{}, { options: { timeout: 0 } }] + + for (const { options } of cases) { const { beginTransaction, connection, query, getTransaction } = createMockTenantConnectionWithTransaction({ isExternalPool: false, - databaseEngine, }) await connection.beginTransaction(options) @@ -1218,33 +1132,9 @@ describe('PgTenantConnection', () => { expect(scopeStatement).toContain("set_config('role', $1, true)") expect(scopeStatement).not.toContain("set_config('statement_timeout'") expect(scopeValues).toHaveLength(9) - expect( - query.mock.calls.filter(([statement]) => - String(statement).includes('SET LOCAL statement_timeout') - ) - ).toHaveLength(0) } }) - it('rolls back and propagates Multigres SET LOCAL setup failures', async () => { - const setupError = new Error('statement_timeout setup failed') - const query = vi.fn().mockRejectedValueOnce(setupError).mockResolvedValue({ rows: [] }) - const { client, connection } = createMockTenantConnectionWithTransaction( - { - isExternalPool: false, - databaseEngine: 'multigres', - }, - query - ) - - await expect(connection.beginTransaction({ timeout: 4321 })).rejects.toBe(setupError) - - expect(query).toHaveBeenCalledTimes(2) - expect(query).toHaveBeenNthCalledWith(1, "SET LOCAL statement_timeout = '4321ms'", undefined) - expect(query).toHaveBeenNthCalledWith(2, 'ROLLBACK', undefined) - expect(client.release).toHaveBeenCalledWith(undefined) - }) - it('does not re-apply statement_timeout after setScope consumes it', async () => { const query = vi.fn().mockResolvedValue({ rows: [] }) const client = { diff --git a/src/internal/database/pg-connection.ts b/src/internal/database/pg-connection.ts index e7acc79aa..b883da505 100644 --- a/src/internal/database/pg-connection.ts +++ b/src/internal/database/pg-connection.ts @@ -28,7 +28,6 @@ const { databaseMaxConnections, databasePoolDrainTimeout, databaseSSLRootCert, - databaseEngine, databaseStatementTimeout, databaseTlsSessionResumption, } = getConfig() @@ -46,11 +45,8 @@ export interface PgQueryOptions { type PgQueryArgument = PgQueryOptions | unknown[] -type PgStatementTimeoutMode = 'set-config' | 'set-local' - interface PgTransactionOptions { statementTimeoutMs?: number - statementTimeoutMode?: PgStatementTimeoutMode } type PgBeginTransactionOptions = TransactionOptions & PgTransactionOptions @@ -451,7 +447,6 @@ export class PgPoolExecutor implements PgTransactionalExecutor { const transaction = new PgTransaction(client, clientErrorTracker, { statementTimeoutMs: options?.statementTimeoutMs ?? normalizeStatementTimeoutMs(options?.timeout), - statementTimeoutMode: options?.statementTimeoutMode, }) try { @@ -474,8 +469,6 @@ export class PgPoolExecutor implements PgTransactionalExecutor { export class PgTransaction implements PgExecutor { private completed = false private statementTimeoutMs?: number - private localStatementTimeoutApplied = false - private readonly statementTimeoutMode: PgStatementTimeoutMode constructor( private readonly client: PoolClient, @@ -483,7 +476,6 @@ export class PgTransaction implements PgExecutor { options: PgTransactionOptions = {} ) { this.statementTimeoutMs = options.statementTimeoutMs - this.statementTimeoutMode = options.statementTimeoutMode ?? 'set-config' } isCompleted(): boolean { @@ -509,24 +501,6 @@ export class PgTransaction implements PgExecutor { const statementTimeout = this.consumeStatementTimeoutForSetConfig() await this.runQuery(buildScopeConfigQuery(scopeValues, statementTimeout), undefined, false) - this.localStatementTimeoutApplied = false - await this.applyLocalStatementTimeout() - } - - async applyLocalStatementTimeout(signal?: AbortSignal): Promise { - if ( - this.statementTimeoutMode !== 'set-local' || - this.localStatementTimeoutApplied || - !this.statementTimeoutMs - ) { - return - } - - await runPgQuery(this.client, buildSetLocalStatementTimeoutStatement(this.statementTimeoutMs), { - signal, - }) - this.clientErrorTracker?.throwIfErrored() - this.localStatementTimeoutApplied = true } private async runQuery( @@ -613,11 +587,6 @@ export class PgTransaction implements PgExecutor { } private async applyPendingStatementTimeoutBeforeQuery(signal?: AbortSignal): Promise { - if (this.statementTimeoutMode === 'set-local') { - await this.applyLocalStatementTimeout(signal) - return - } - const timeoutMs = this.consumeStatementTimeoutForSetConfig() if (!timeoutMs) { return @@ -634,10 +603,6 @@ export class PgTransaction implements PgExecutor { } private consumeStatementTimeoutForSetConfig(): number | undefined { - if (this.statementTimeoutMode !== 'set-config') { - return undefined - } - const timeoutMs = this.statementTimeoutMs this.statementTimeoutMs = undefined return timeoutMs @@ -694,22 +659,9 @@ export class PgTenantConnection { async beginTransaction(options?: TransactionOptions): Promise { this.assertNotDisposed() const statementTimeout = options?.timeout - const isMultigres = this.isMultigresDatabase() - const transaction = await this.pool + return await this.pool .acquire() - .beginTransaction( - this.withStatementTimeout( - options, - statementTimeout, - isMultigres ? 'set-local' : 'set-config' - ) - ) - - if (isMultigres) { - await this.applyStatementTimeoutImmediately(transaction) - } - - return transaction + .beginTransaction(this.withStatementTimeout(options, statementTimeout)) } asSuperUser() { @@ -732,12 +684,7 @@ export class PgTenantConnection { try { const statementTimeout = opts?.timeout ?? databaseStatementTimeout - const isMultigres = this.isMultigresDatabase() - const beginOptions = this.withStatementTimeout( - opts, - statementTimeout, - isMultigres ? 'set-local' : 'set-config' - ) + const beginOptions = this.withStatementTimeout(opts, statementTimeout) const transaction = await retry( async (bail) => { if (this.disposed) { @@ -787,19 +734,6 @@ export class PgTenantConnection { } } - private isMultigresDatabase(): boolean { - return (this.options.databaseEngine ?? databaseEngine) === 'multigres' - } - - private async applyStatementTimeoutImmediately(transaction: PgTransaction): Promise { - try { - await transaction.applyLocalStatementTimeout() - } catch (e) { - await this.rollbackTransactionSafely(transaction, e, 'statement_timeout setup') - throw e - } - } - private assertNotDisposed(): void { if (this.disposed) { throw createDisposedTenantConnectionError() @@ -841,8 +775,7 @@ export class PgTenantConnection { private withStatementTimeout( options: TransactionOptions | undefined, - statementTimeout: number | undefined, - statementTimeoutMode: PgStatementTimeoutMode + statementTimeout: number | undefined ): PgBeginTransactionOptions | undefined { const statementTimeoutMs = normalizeStatementTimeoutMs(statementTimeout) @@ -853,7 +786,6 @@ export class PgTenantConnection { return { ...options, statementTimeoutMs, - ...(statementTimeoutMode === 'set-local' ? { statementTimeoutMode } : undefined), } } @@ -887,14 +819,6 @@ function normalizeStatementTimeoutMs(timeoutMs: number | undefined): number | un return timeoutMs } -function buildSetLocalStatementTimeoutStatement(statementTimeout: number): string { - if (!Number.isFinite(statementTimeout) || statementTimeout <= 0) { - throw new Error(`Invalid statement timeout: ${statementTimeout}`) - } - - return `SET LOCAL statement_timeout = '${statementTimeout}ms'` -} - function createDisposedTenantConnectionError(): Error { return new Error('Cannot use a disposed PgTenantConnection') } diff --git a/src/internal/database/pool.ts b/src/internal/database/pool.ts index 56782af6c..8fa82dcff 100644 --- a/src/internal/database/pool.ts +++ b/src/internal/database/pool.ts @@ -10,7 +10,7 @@ import { recordCacheRequest, } from '@internal/monitoring/metrics' import { JWTPayload } from 'jose' -import { type DatabaseEngine, getConfig } from '../../config' +import { getConfig } from '../../config' const { isMultitenant, @@ -30,7 +30,6 @@ export interface TenantConnectionOptions { idleTimeoutMillis?: number reapIntervalMillis?: number maxConnections: number - databaseEngine?: DatabaseEngine clusterSize?: number numWorkers?: number user: User diff --git a/src/test/pg-connection.test.ts b/src/test/pg-connection.test.ts index dc8dd5796..988b878f7 100644 --- a/src/test/pg-connection.test.ts +++ b/src/test/pg-connection.test.ts @@ -18,7 +18,6 @@ describe('Pg database foundation', () => { tenantId, isExternalPool: true, maxConnections: 2, - databaseEngine: getConfig().databaseEngine, dbUrl: databasePoolURL || databaseURL, user: superUser, superUser, diff --git a/src/test/storage-pg-db.test.ts b/src/test/storage-pg-db.test.ts index 20ea60590..c5020dc73 100644 --- a/src/test/storage-pg-db.test.ts +++ b/src/test/storage-pg-db.test.ts @@ -36,7 +36,6 @@ describe('StoragePgDB bucket metadata', () => { connectionSettings = { tenantId, dbUrl: databaseURL!, - databaseEngine: getConfig().databaseEngine, isExternalPool: false, maxConnections: 2, user: superUser, @@ -432,7 +431,7 @@ describe('StoragePgDB bucket metadata', () => { async (pg) => { await expect(readCurrentRoleFromExecutor(pg)).resolves.toBe(superUser.payload.role) await expect(readCurrentStatementTimeoutFromExecutor(pg)).resolves.toBe('4321ms') - await pg.query("SET LOCAL statement_timeout = '30s'") + await pg.query("SELECT set_config('statement_timeout', '30s', true)") await expect(readCurrentStatementTimeoutFromExecutor(pg)).resolves.toBe('30s') throw new Error('failed nested super-user timeout query') } From 8d175c05ad4dca1bd5cf8ad1d1f4daa425143717 Mon Sep 17 00:00:00 2001 From: ferhat elmas Date: Wed, 8 Jul 2026 01:01:37 +0200 Subject: [PATCH 07/12] fix: normalize pg transaction timeout invariant Signed-off-by: ferhat elmas --- src/internal/database/pg-connection.test.ts | 15 +++++++++++++++ src/internal/database/pg-connection.ts | 2 +- 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/src/internal/database/pg-connection.test.ts b/src/internal/database/pg-connection.test.ts index 772fd42bb..bae22cffc 100644 --- a/src/internal/database/pg-connection.test.ts +++ b/src/internal/database/pg-connection.test.ts @@ -666,6 +666,21 @@ describe('PgTransaction', () => { expect(client.query).toHaveBeenNthCalledWith(3, 'SELECT 2', undefined) }) + it('normalizes invalid constructor statement timeouts as disabled', async () => { + for (const statementTimeoutMs of [-5, Number.NaN, Number.POSITIVE_INFINITY, 0]) { + const client = { + query: vi.fn().mockResolvedValue({ rows: [] }), + release: vi.fn(), + } as unknown as PoolClient + const transaction = new PgTransaction(client, undefined, { statementTimeoutMs }) + + await transaction.query('SELECT 1') + + expect(client.query).toHaveBeenCalledTimes(1) + expect(client.query).toHaveBeenNthCalledWith(1, 'SELECT 1', undefined) + } + }) + it('rejects a pre-aborted direct query before applying a pending statement timeout', async () => { const client = { query: vi.fn().mockResolvedValue({ rows: [] }), diff --git a/src/internal/database/pg-connection.ts b/src/internal/database/pg-connection.ts index b883da505..dfa47c053 100644 --- a/src/internal/database/pg-connection.ts +++ b/src/internal/database/pg-connection.ts @@ -475,7 +475,7 @@ export class PgTransaction implements PgExecutor { private readonly clientErrorTracker?: PgClientErrorTracker, options: PgTransactionOptions = {} ) { - this.statementTimeoutMs = options.statementTimeoutMs + this.statementTimeoutMs = normalizeStatementTimeoutMs(options.statementTimeoutMs) } isCompleted(): boolean { From 3550a094c6c092169ad38adc09d5acd245c2222f Mon Sep 17 00:00:00 2001 From: ferhat elmas Date: Wed, 8 Jul 2026 01:15:33 +0200 Subject: [PATCH 08/12] fix: simplify pg timeout handoff Signed-off-by: ferhat elmas --- src/internal/database/pg-connection.ts | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/internal/database/pg-connection.ts b/src/internal/database/pg-connection.ts index dfa47c053..6156c485c 100644 --- a/src/internal/database/pg-connection.ts +++ b/src/internal/database/pg-connection.ts @@ -445,8 +445,7 @@ export class PgPoolExecutor implements PgTransactionalExecutor { const clientErrorTracker = new PgClientErrorTracker(client) const transaction = new PgTransaction(client, clientErrorTracker, { - statementTimeoutMs: - options?.statementTimeoutMs ?? normalizeStatementTimeoutMs(options?.timeout), + statementTimeoutMs: options?.statementTimeoutMs ?? options?.timeout, }) try { From eaa886d56b4ed51e483d9d0068e20df1c3436253 Mon Sep 17 00:00:00 2001 From: ferhat elmas Date: Wed, 8 Jul 2026 15:13:06 +0200 Subject: [PATCH 09/12] fix: simplify Signed-off-by: ferhat elmas --- src/internal/database/pg-connection.test.ts | 16 +-- src/internal/database/pg-connection.ts | 135 +++++++++----------- 2 files changed, 65 insertions(+), 86 deletions(-) diff --git a/src/internal/database/pg-connection.test.ts b/src/internal/database/pg-connection.test.ts index bae22cffc..e3f6dcbff 100644 --- a/src/internal/database/pg-connection.test.ts +++ b/src/internal/database/pg-connection.test.ts @@ -927,7 +927,6 @@ describe('PgTenantConnection', () => { timeout: 4321, isolation: 'serializable', readOnly: true, - statementTimeoutMs: 4321, }) expect(query).not.toHaveBeenCalled() @@ -979,10 +978,7 @@ describe('PgTenantConnection', () => { ) await expect(connection.transaction({ timeout: 4321 })).resolves.toBe(transaction!) - expect(beginTransaction).toHaveBeenCalledWith({ - timeout: 4321, - statementTimeoutMs: 4321, - }) + expect(beginTransaction).toHaveBeenCalledWith({ timeout: 4321 }) expect(query).not.toHaveBeenCalled() await connection.setScope(transaction!) @@ -1033,10 +1029,7 @@ describe('PgTenantConnection', () => { ) await expect(connection.transaction({ timeout: 4321 })).resolves.toBe(transaction!) - expect(beginTransaction).toHaveBeenCalledWith({ - timeout: 4321, - statementTimeoutMs: 4321, - }) + expect(beginTransaction).toHaveBeenCalledWith({ timeout: 4321 }) expect(query).toHaveBeenCalledTimes(1) expect(query).toHaveBeenNthCalledWith( @@ -1094,10 +1087,7 @@ describe('PgTenantConnection', () => { const connection = new PgTenantConnection(pool, settings) await expect(connection.transaction({ timeout: 4321 })).resolves.toBe(transaction!) - expect(beginTransaction).toHaveBeenCalledWith({ - timeout: 4321, - statementTimeoutMs: 4321, - }) + expect(beginTransaction).toHaveBeenCalledWith({ timeout: 4321 }) expect(query).toHaveBeenCalledTimes(1) expect(query).toHaveBeenNthCalledWith( diff --git a/src/internal/database/pg-connection.ts b/src/internal/database/pg-connection.ts index 6156c485c..331330766 100644 --- a/src/internal/database/pg-connection.ts +++ b/src/internal/database/pg-connection.ts @@ -51,32 +51,35 @@ interface PgTransactionOptions { type PgBeginTransactionOptions = TransactionOptions & PgTransactionOptions -const scopeConfigEntries: ReadonlyArray = [ - ['role', '$1'], - ['request.jwt.claim.role', '$2'], - ['request.jwt', '$3'], - ['request.jwt.claim.sub', '$4'], - ['request.jwt.claims', '$5'], - ['request.headers', '$6'], - ['request.method', '$7'], - ['request.path', '$8'], - ['storage.operation', '$9'], - ['storage.allow_delete_query', "'true'"], -] - -const scopeConfigStatement = buildScopeConfigSql(false) -const scopeConfigStatementWithTimeout = buildScopeConfigSql(true) - -function buildScopeConfigSql(includeStatementTimeout: boolean): string { - const entries = includeStatementTimeout - ? [...scopeConfigEntries, ['statement_timeout', '$10'] as const] - : scopeConfigEntries - - return ` +const defaultStatementTimeoutMs = normalizeStatementTimeoutMs(databaseStatementTimeout) + +// Shared frozen options for the common transaction() call without caller options. +const defaultBeginTransactionOptions: PgBeginTransactionOptions | undefined = + defaultStatementTimeoutMs !== undefined + ? Object.freeze({ statementTimeoutMs: defaultStatementTimeoutMs }) + : undefined + +const scopeConfigSetters = `set_config('role', $1, true), + set_config('request.jwt.claim.role', $2, true), + set_config('request.jwt', $3, true), + set_config('request.jwt.claim.sub', $4, true), + set_config('request.jwt.claims', $5, true), + set_config('request.headers', $6, true), + set_config('request.method', $7, true), + set_config('request.path', $8, true), + set_config('storage.operation', $9, true), + set_config('storage.allow_delete_query', 'true', true)` + +const scopeConfigSql = ` SELECT - ${entries.map(([name, value]) => `set_config('${name}', ${value}, true)`).join(',\n ')}; + ${scopeConfigSetters}; + ` + +const scopeConfigSqlWithStatementTimeout = ` + SELECT + ${scopeConfigSetters}, + set_config('statement_timeout', $10, true); ` -} export interface PgExecutor { query( @@ -496,10 +499,12 @@ export class PgTransaction implements PgExecutor { return this.runQuery(statement, options, false) } - async runScopeQuery(scopeValues: unknown[]): Promise { - const statementTimeout = this.consumeStatementTimeoutForSetConfig() - - await this.runQuery(buildScopeConfigQuery(scopeValues, statementTimeout), undefined, false) + // Hands the deferred timeout to the caller (setScope) so it can be folded + // into its next statement instead of costing a separate set_config round trip. + takePendingStatementTimeoutMs(): number | undefined { + const timeoutMs = this.statementTimeoutMs + this.statementTimeoutMs = undefined + return timeoutMs } private async runQuery( @@ -586,7 +591,7 @@ export class PgTransaction implements PgExecutor { } private async applyPendingStatementTimeoutBeforeQuery(signal?: AbortSignal): Promise { - const timeoutMs = this.consumeStatementTimeoutForSetConfig() + const timeoutMs = this.takePendingStatementTimeoutMs() if (!timeoutMs) { return } @@ -600,12 +605,6 @@ export class PgTransaction implements PgExecutor { { signal } ) } - - private consumeStatementTimeoutForSetConfig(): number | undefined { - const timeoutMs = this.statementTimeoutMs - this.statementTimeoutMs = undefined - return timeoutMs - } } export class PgTenantConnection { @@ -657,10 +656,8 @@ export class PgTenantConnection { async beginTransaction(options?: TransactionOptions): Promise { this.assertNotDisposed() - const statementTimeout = options?.timeout - return await this.pool - .acquire() - .beginTransaction(this.withStatementTimeout(options, statementTimeout)) + // PgPoolExecutor derives the deferred statement_timeout from options.timeout. + return this.pool.acquire().beginTransaction(options) } asSuperUser() { @@ -682,8 +679,7 @@ export class PgTenantConnection { this.assertNotDisposed() try { - const statementTimeout = opts?.timeout ?? databaseStatementTimeout - const beginOptions = this.withStatementTimeout(opts, statementTimeout) + const beginOptions = withDefaultStatementTimeout(opts) const transaction = await retry( async (bail) => { if (this.disposed) { @@ -761,35 +757,10 @@ export class PgTenantConnection { } async setScope(tnx: PgExecutor) { - const transaction = tnx instanceof PgTransaction ? tnx : undefined - const scopeValues = this.getScopeValues() - - if (transaction) { - await transaction.runScopeQuery(scopeValues) - return - } - - await tnx.query(buildScopeConfigQuery(scopeValues)) - } - - private withStatementTimeout( - options: TransactionOptions | undefined, - statementTimeout: number | undefined - ): PgBeginTransactionOptions | undefined { - const statementTimeoutMs = normalizeStatementTimeoutMs(statementTimeout) - - if (!statementTimeoutMs) { - return options - } - - return { - ...options, - statementTimeoutMs, - } - } + const statementTimeoutMs = + tnx instanceof PgTransaction ? tnx.takePendingStatementTimeoutMs() : undefined - private getScopeValues(): unknown[] { - return [ + const values: unknown[] = [ this.role, this.role, this.options.user.jwt || '', @@ -800,14 +771,32 @@ export class PgTenantConnection { this.options.path || '', this.options.operation?.() || '', ] + + if (statementTimeoutMs) { + values.push(`${statementTimeoutMs}ms`) + } + + await tnx.query({ + text: statementTimeoutMs ? scopeConfigSqlWithStatementTimeout : scopeConfigSql, + values, + }) } } -function buildScopeConfigQuery(scopeValues: unknown[], statementTimeout?: number): PgStatement { - return { - text: statementTimeout ? scopeConfigStatementWithTimeout : scopeConfigStatement, - values: statementTimeout ? [...scopeValues, `${statementTimeout}ms`] : scopeValues, +// PgPoolExecutor already derives the deferred statement_timeout from options.timeout, +// so options only need rewriting when the global default has to be injected. +function withDefaultStatementTimeout( + options: TransactionOptions | undefined +): PgBeginTransactionOptions | undefined { + if (!defaultBeginTransactionOptions || options?.timeout !== undefined) { + return options + } + + if (!options) { + return defaultBeginTransactionOptions } + + return { ...options, statementTimeoutMs: defaultStatementTimeoutMs } } function normalizeStatementTimeoutMs(timeoutMs: number | undefined): number | undefined { From f90ade536535a5e4b486948c1da76bf2d31a975d Mon Sep 17 00:00:00 2001 From: ferhat elmas Date: Wed, 8 Jul 2026 17:30:40 +0200 Subject: [PATCH 10/12] chore: comments Signed-off-by: ferhat elmas --- src/internal/database/pg-connection.test.ts | 2 +- src/test/storage-pg-db.test.ts | 28 ++++++++++++--------- 2 files changed, 17 insertions(+), 13 deletions(-) diff --git a/src/internal/database/pg-connection.test.ts b/src/internal/database/pg-connection.test.ts index e3f6dcbff..2924fbd01 100644 --- a/src/internal/database/pg-connection.test.ts +++ b/src/internal/database/pg-connection.test.ts @@ -75,7 +75,7 @@ function normalizeTestStatementTimeoutMs( } function createMockTenantConnectionWithTransaction( - overrides: Partial[0]> = {}, + overrides: Partial = {}, query = vi.fn().mockResolvedValue({ rows: [] }) ) { const client = { diff --git a/src/test/storage-pg-db.test.ts b/src/test/storage-pg-db.test.ts index c5020dc73..4be716da1 100644 --- a/src/test/storage-pg-db.test.ts +++ b/src/test/storage-pg-db.test.ts @@ -396,7 +396,7 @@ describe('StoragePgDB bucket metadata', () => { await authenticatedDb.withTransaction( async (tx) => { await expect(readCurrentRole(tx)).resolves.toBe('authenticated') - await expect(readCurrentStatementTimeout(tx)).resolves.toBe('4321ms') + await expect(readCurrentStatementTimeoutMs(tx)).resolves.toBe(4321) await expect( runStorageQuery( @@ -404,13 +404,13 @@ describe('StoragePgDB bucket metadata', () => { 'ReadSuperUserStatementTimeout', async (pg) => { await expect(readCurrentRoleFromExecutor(pg)).resolves.toBe(superUser.payload.role) - await expect(readCurrentStatementTimeoutFromExecutor(pg)).resolves.toBe('4321ms') + await expect(readCurrentStatementTimeoutMsFromExecutor(pg)).resolves.toBe(4321) } ) ).resolves.toBeUndefined() await expect(readCurrentRole(tx)).resolves.toBe('authenticated') - await expect(readCurrentStatementTimeout(tx)).resolves.toBe('4321ms') + await expect(readCurrentStatementTimeoutMs(tx)).resolves.toBe(4321) }, { timeout: 4321 } ) @@ -422,7 +422,7 @@ describe('StoragePgDB bucket metadata', () => { await authenticatedDb.withTransaction( async (tx) => { await expect(readCurrentRole(tx)).resolves.toBe('authenticated') - await expect(readCurrentStatementTimeout(tx)).resolves.toBe('4321ms') + await expect(readCurrentStatementTimeoutMs(tx)).resolves.toBe(4321) await expect( runStorageQuery( @@ -430,16 +430,16 @@ describe('StoragePgDB bucket metadata', () => { 'FailSuperUserStatementTimeout', async (pg) => { await expect(readCurrentRoleFromExecutor(pg)).resolves.toBe(superUser.payload.role) - await expect(readCurrentStatementTimeoutFromExecutor(pg)).resolves.toBe('4321ms') + await expect(readCurrentStatementTimeoutMsFromExecutor(pg)).resolves.toBe(4321) await pg.query("SELECT set_config('statement_timeout', '30s', true)") - await expect(readCurrentStatementTimeoutFromExecutor(pg)).resolves.toBe('30s') + await expect(readCurrentStatementTimeoutMsFromExecutor(pg)).resolves.toBe(30000) throw new Error('failed nested super-user timeout query') } ) ).rejects.toThrow('failed nested super-user timeout query') await expect(readCurrentRole(tx)).resolves.toBe('authenticated') - await expect(readCurrentStatementTimeout(tx)).resolves.toBe('4321ms') + await expect(readCurrentStatementTimeoutMs(tx)).resolves.toBe(4321) }, { timeout: 4321 } ) @@ -1780,16 +1780,20 @@ describe('StoragePgDB bucket metadata', () => { return result.rows[0].role } - function readCurrentStatementTimeout(storage: StoragePgDB): Promise { + function readCurrentStatementTimeoutMs(storage: StoragePgDB): Promise { return runStorageQuery(storage, 'ReadCurrentStatementTimeout', (pg) => - readCurrentStatementTimeoutFromExecutor(pg) + readCurrentStatementTimeoutMsFromExecutor(pg) ) } - async function readCurrentStatementTimeoutFromExecutor(pg: PgExecutor): Promise { - const result = await pg.query<{ statement_timeout: string }>('SHOW statement_timeout') + async function readCurrentStatementTimeoutMsFromExecutor(pg: PgExecutor): Promise { + const result = await pg.query<{ statement_timeout_ms: number }>(` + SELECT setting::int AS statement_timeout_ms + FROM pg_settings + WHERE name = 'statement_timeout' + `) - return result.rows[0].statement_timeout + return result.rows[0].statement_timeout_ms } async function withAuthenticatedDb( From 3776e9412e0d2fa1607cfb959a6b03b2e493c0e9 Mon Sep 17 00:00:00 2001 From: ferhat elmas Date: Wed, 8 Jul 2026 17:38:18 +0200 Subject: [PATCH 11/12] fix: test helper for multigres Signed-off-by: ferhat elmas --- src/test/storage-pg-db.test.ts | 32 ++++++++++++++++++++++++++------ 1 file changed, 26 insertions(+), 6 deletions(-) diff --git a/src/test/storage-pg-db.test.ts b/src/test/storage-pg-db.test.ts index 4be716da1..e149a630c 100644 --- a/src/test/storage-pg-db.test.ts +++ b/src/test/storage-pg-db.test.ts @@ -1787,13 +1787,33 @@ describe('StoragePgDB bucket metadata', () => { } async function readCurrentStatementTimeoutMsFromExecutor(pg: PgExecutor): Promise { - const result = await pg.query<{ statement_timeout_ms: number }>(` - SELECT setting::int AS statement_timeout_ms - FROM pg_settings - WHERE name = 'statement_timeout' - `) + const result = await pg.query<{ statement_timeout: string }>('SHOW statement_timeout') + + return parseStatementTimeoutMs(result.rows[0].statement_timeout) + } + + function parseStatementTimeoutMs(statementTimeout: string): number { + const value = statementTimeout.trim() + if (value === '0') { + return 0 + } + + const match = /^(\d+(?:\.\d+)?)\s*(ms|s|min|h|d)?$/.exec(value) + if (!match) { + throw new Error(`Unexpected statement_timeout value: ${statementTimeout}`) + } + + const multipliers = { + ms: 1, + s: 1000, + min: 60 * 1000, + h: 60 * 60 * 1000, + d: 24 * 60 * 60 * 1000, + } + const amount = Number(match[1]) + const unit = (match[2] ?? 'ms') as keyof typeof multipliers - return result.rows[0].statement_timeout_ms + return Math.round(amount * multipliers[unit]) } async function withAuthenticatedDb( From 5f4253fe07fd6160165e6c218971586f50dc8256 Mon Sep 17 00:00:00 2001 From: ferhat elmas Date: Wed, 8 Jul 2026 17:45:09 +0200 Subject: [PATCH 12/12] chore: cleanup a test Signed-off-by: ferhat elmas --- src/internal/database/pg-connection.test.ts | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/src/internal/database/pg-connection.test.ts b/src/internal/database/pg-connection.test.ts index 2924fbd01..e1153bcb5 100644 --- a/src/internal/database/pg-connection.test.ts +++ b/src/internal/database/pg-connection.test.ts @@ -1058,7 +1058,7 @@ describe('PgTenantConnection', () => { ]) }) - it('folds Multigres statement_timeout into scope setup', async () => { + it('folds deferred statement_timeout into scope setup', async () => { const query = vi.fn().mockResolvedValue({ rows: [] }) const client = { query, @@ -1078,13 +1078,12 @@ describe('PgTenantConnection', () => { beginTransaction, }), } as unknown as PgPoolStrategy - const settings = { - ...createPoolStrategySettings({ + const connection = new PgTenantConnection( + pool, + createPoolStrategySettings({ isExternalPool: true, - }), - databaseEngine: 'multigres', - } as TenantConnectionOptions - const connection = new PgTenantConnection(pool, settings) + }) + ) await expect(connection.transaction({ timeout: 4321 })).resolves.toBe(transaction!) expect(beginTransaction).toHaveBeenCalledWith({ timeout: 4321 })