diff --git a/packages/@n8n/db/src/connection/__tests__/db-connection-monitor.recovery.postgres.test.ts b/packages/@n8n/db/src/connection/__tests__/db-connection-monitor.recovery.postgres.test.ts index 392d0a22950..2da0d648462 100644 --- a/packages/@n8n/db/src/connection/__tests__/db-connection-monitor.recovery.postgres.test.ts +++ b/packages/@n8n/db/src/connection/__tests__/db-connection-monitor.recovery.postgres.test.ts @@ -10,7 +10,7 @@ import { setTimeout as setTimeoutP } from 'timers/promises'; import { mock } from 'vitest-mock-extended'; import type { DbConnectionMetrics } from '../db-connection-metrics'; -import { DbConnectionMonitor } from '../db-connection-monitor'; +import { DbConnectionMonitor, FORCE_CLOSE_GRACE_MS } from '../db-connection-monitor'; // Minimal pg pool view: check out a client the pool can never drain on its own. type PgDriver = { master: PgPool }; @@ -108,11 +108,16 @@ const externalHost = process.env.DB_POSTGRESDB_HOST ?? ''; const useExternalPostgres = externalHost.length > 0; // Private monitor methods exposed for integration tests without loosening prod typing. -type MonitorInternals = { ping: () => Promise; recoverDataSource: () => Promise }; +type MonitorInternals = { + ping: () => Promise; + recoverDataSource: () => Promise; + forceClosePostgresPool: () => void; +}; type PgPoolClient = { query: (query: string | { text: string }) => Promise; release: (error?: Error) => void; + on: (event: 'error', listener: (error: Error) => void) => void; }; type PgPool = { @@ -364,6 +369,71 @@ describe('DbConnectionMonitor recovery against real Postgres', () => { TEST_TIMEOUT_MS, ); + it( + 'abandons a teardown that force-closing cannot unblock', + async (ctx) => { + if (!connection) { + ctx.skip(); + return; + } + + const proxy = new FreezableProxy(connection.host, connection.port); + const proxyPort = await proxy.listen(); + const dataSource = newDataSource( + { ...connection, host: '127.0.0.1', port: proxyPort }, + { poolSize: 2 }, + ); + await dataSource.initialize(); + + const destroyTimeoutMs = 2_000; + const monitor = new DbConnectionMonitor( + dataSource, + () => {}, + buildDatabaseConfig({ + postgresdb: mock({ destroyTimeoutMs }), + }), + mock(), + mock(), + mock(), + ); + monitor.start(); + + // Neuter force-close so the drain is genuinely unblockable. `pool.end()` can wedge + // deeper than the sockets we can reach: after the last `_remove`, pg-pool only + // re-enters `_pulseQueue` from a `client.end()` callback a frozen socket never fires. + (monitor as unknown as MonitorInternals).forceClosePostgresPool = () => {}; + + const pool = (dataSource.driver as unknown as PgDriver).master; + const leaked = await pool.connect(); + expect(leaked).toBeDefined(); + // This client is deliberately never released, so nothing else listens for its + // error when `proxy.close()` kills the socket in the teardown below. In production + // a checked-out client is held by a query runner, which attaches its own listener. + leaked.on('error', () => {}); + proxy.freezeExisting(); + + try { + const start = Date.now(); + // Before the fix this never resolves: the post-force-close await was unbounded. + await (monitor as unknown as MonitorInternals).recoverDataSource(); + const elapsed = Date.now() - start; + + // Both bounds were traversed, then the teardown was abandoned and the pool rebuilt. + expect(elapsed).toBeGreaterThanOrEqual(destroyTimeoutMs + FORCE_CLOSE_GRACE_MS); + expect(elapsed).toBeLessThan(destroyTimeoutMs + FORCE_CLOSE_GRACE_MS + 15_000); + expect(dataSource.isInitialized).toBe(true); + expect(await dataSource.query('SELECT 1 AS ok')).toEqual([{ ok: 1 }]); + } finally { + await monitor.stop(); + if (dataSource.isInitialized) { + await dataSource.destroy(); + } + await proxy.close(); + } + }, + TEST_TIMEOUT_MS, + ); + it( 'destroys a timed-out ping client so the pg pool slot is reusable', async (ctx) => { diff --git a/packages/@n8n/db/src/connection/__tests__/db-connection-monitor.test.ts b/packages/@n8n/db/src/connection/__tests__/db-connection-monitor.test.ts index 82448869283..b0c4eb49b6b 100644 --- a/packages/@n8n/db/src/connection/__tests__/db-connection-monitor.test.ts +++ b/packages/@n8n/db/src/connection/__tests__/db-connection-monitor.test.ts @@ -929,30 +929,42 @@ describe('DbConnectionMonitor', () => { on: Mock; }; type DriverShape = { - master: PoolShape; + master: PoolShape | undefined; obtainMasterConnection: Mock; + disconnect: Mock; }; - const buildPgConfig = (destroyTimeoutMs: number) => - mock({ - pingTimeoutMs: 5_000, - pingMaxFailuresBeforeRecovery: 3, - minRecoveryBackoffMs: 1_000, - maxRecoveryBackoffMs: 30_000, - connectionAcquisitionTimeoutMs: 30_000, - postgresdb: mock({ destroyTimeoutMs }), - }); + const buildPgMonitor = (destroyTimeoutMs = 10_000) => + new DbConnectionMonitor( + dataSource, + onConnectedChange, + mock({ + pingTimeoutMs: 5_000, + pingMaxFailuresBeforeRecovery: 3, + minRecoveryBackoffMs: 1_000, + maxRecoveryBackoffMs: 30_000, + connectionAcquisitionTimeoutMs: 30_000, + postgresdb: mock({ destroyTimeoutMs }), + }), + logger, + errorReporter, + dbConnectionMetrics, + ); - // One un-drained client whose `destroy()` resolves only once it is both released - // with an error and has its socket destroyed, mirroring pg-pool where neither step - // alone unblocks `pool.end()`. So the test goes red if either step is dropped. + // One un-drained client whose drain resolves only once it is both released with an + // error and has its socket destroyed, mirroring pg-pool where neither step alone + // unblocks `pool.end()`. So the test goes red if either step is dropped. + // + // `destroy()` delegates to `driver.disconnect()` and only clears `isInitialized` + // once that resolves, exactly like TypeORM's `DataSource.destroy()`. The monitor + // bounds the disconnect, so a fixture that skipped it would not exercise the fix. const setupFrozenPool = () => { - let resolveDestroy!: () => void; - const destroyPromise = new Promise((resolve) => (resolveDestroy = resolve)); + let resolveDrain!: () => void; + const drainPromise = new Promise((resolve) => (resolveDrain = resolve)); let releasedWithError = false; let socketDestroyed = false; const settleIfForceClosed = () => { - if (releasedWithError && socketDestroyed) resolveDestroy(); + if (releasedWithError && socketDestroyed) resolveDrain(); }; const stream: StreamShape = { destroy: vi.fn(() => { @@ -970,24 +982,30 @@ describe('DbConnectionMonitor', () => { const driver: DriverShape = { master: { _clients: [client], on: vi.fn() }, obtainMasterConnection: vi.fn().mockResolvedValue(undefined), + disconnect: vi.fn(async () => { + await drainPromise; + driver.master = undefined; + }), }; (dataSource as unknown as { driver: DriverShape }).driver = driver; // @ts-expect-error readonly property dataSource.isInitialized = true; - dataSource.destroy.mockReturnValue(destroyPromise); - dataSource.initialize.mockResolvedValue(dataSource); - return { stream, client }; + dataSource.destroy.mockImplementation(async () => { + // Reads the property at call time, so it picks up the monitor's wrapper. + await driver.disconnect(); + // @ts-expect-error readonly property + dataSource.isInitialized = false; + }); + dataSource.initialize.mockImplementation(async () => { + // @ts-expect-error readonly property + dataSource.isInitialized = true; + return dataSource; + }); + return { stream, client, driver, resolveDrain }; }; it('should force-close the pool and continue recovery when destroy() exceeds the timeout', async () => { - const pgMonitor = new DbConnectionMonitor( - dataSource, - onConnectedChange, - buildPgConfig(10_000), - logger, - errorReporter, - dbConnectionMetrics, - ); + const pgMonitor = buildPgMonitor(10_000); const { stream, client } = setupFrozenPool(); // @ts-expect-error private property pgMonitor.connected = false; @@ -1010,17 +1028,10 @@ describe('DbConnectionMonitor', () => { }); it('should not force-close the pool when destroy() resolves before the timeout', async () => { - const pgMonitor = new DbConnectionMonitor( - dataSource, - onConnectedChange, - buildPgConfig(10_000), - logger, - errorReporter, - dbConnectionMetrics, - ); - const { stream, client } = setupFrozenPool(); - // `destroy()` resolves on its own here, before the never-resolving timeout. - dataSource.destroy.mockResolvedValue(); + const pgMonitor = buildPgMonitor(10_000); + const { stream, client, resolveDrain } = setupFrozenPool(); + // The drain completes on its own here, before the never-resolving timeout. + resolveDrain(); // @ts-expect-error private property pgMonitor.connected = false; @@ -1034,14 +1045,7 @@ describe('DbConnectionMonitor', () => { }); it('should wait indefinitely for destroy() when the destroy timeout is 0', async () => { - const pgMonitor = new DbConnectionMonitor( - dataSource, - onConnectedChange, - buildPgConfig(0), - logger, - errorReporter, - dbConnectionMetrics, - ); + const pgMonitor = buildPgMonitor(0); const { stream, client } = setupFrozenPool(); // `destroy()` never resolves and the timeout is disabled, so recovery stays parked. dataSource.destroy.mockReturnValue(new Promise(() => {})); @@ -1058,6 +1062,81 @@ describe('DbConnectionMonitor', () => { void pgMonitor.stop(); void recovery; }); + + it('should abandon a teardown that force-closing cannot unblock, and still recover', async () => { + const pgMonitor = buildPgMonitor(10_000); + const { client, driver } = setupFrozenPool(); + // The drain stays wedged even after force-close, so both bounds have to fire. + driver.disconnect.mockReturnValue(new Promise(() => {})); + // @ts-expect-error private property + pgMonitor.connected = false; + mockedSetTimeoutP.mockResolvedValueOnce(undefined).mockResolvedValueOnce(undefined); + + // @ts-expect-error private property + await pgMonitor.recoverDataSource(); + + // Force-close was still attempted, then the teardown was abandoned so the pool + // could be rebuilt instead of pinning recovery at attempt 1 forever. + expect(client.release).toHaveBeenCalledWith(expect.any(Error)); + expect(logger.error).toHaveBeenCalledWith(expect.stringContaining('abandoning the pool')); + expect(dataSource.initialize).toHaveBeenCalledTimes(1); + expect(onConnectedChange).toHaveBeenLastCalledWith(true); + // @ts-expect-error private property + expect(pgMonitor.recovering).toBe(false); + }); + + it('should leave the rebuilt connection alone when an abandoned teardown settles later', async () => { + const pgMonitor = buildPgMonitor(10_000); + const { driver } = setupFrozenPool(); + let settleAbandonedDrain!: () => void; + driver.disconnect.mockReturnValue( + new Promise((resolve) => (settleAbandonedDrain = resolve)), + ); + mockedSetTimeoutP.mockResolvedValueOnce(undefined).mockResolvedValueOnce(undefined); + + // @ts-expect-error private property + await pgMonitor.recoverDataSource(); + expect(dataSource.isInitialized).toBe(true); + + // The drain finally completes minutes later. It was bound at the driver, so it + // cannot reach `destroy()`'s `isInitialized = false` on the rebuilt connection. + settleAbandonedDrain(); + await flushMicrotasks(); + + expect(dataSource.isInitialized).toBe(true); + expect(dataSource.initialize).toHaveBeenCalledTimes(1); + expect(dataSource.destroy).toHaveBeenCalledTimes(1); + }); + + it('should recover rather than retry forever when the teardown rejects', async () => { + const pgMonitor = buildPgMonitor(10_000); + const { driver } = setupFrozenPool(); + // e.g. TypeORM's ConnectionIsNotSetError once `driver.master` is already gone. + driver.disconnect.mockRejectedValue(new Error('Connection is not established')); + // @ts-expect-error private property + pgMonitor.connected = false; + + // @ts-expect-error private property + await pgMonitor.recoverDataSource(); + + // Previously the rejection propagated, `isInitialized` stayed set, and every + // retry re-ran the same failing teardown without ever reaching initialize(). + expect(dataSource.destroy).toHaveBeenCalledTimes(1); + expect(dataSource.initialize).toHaveBeenCalledTimes(1); + expect(onConnectedChange).toHaveBeenLastCalledWith(true); + }); + + it('should restore the original disconnect after tearing down', async () => { + const pgMonitor = buildPgMonitor(10_000); + const { driver, resolveDrain } = setupFrozenPool(); + const originalDisconnect = driver.disconnect; + resolveDrain(); + + // @ts-expect-error private property + await pgMonitor.recoverDataSource(); + + expect(driver.disconnect).toBe(originalDisconnect); + }); }); describe('start', () => { diff --git a/packages/@n8n/db/src/connection/db-connection-monitor.ts b/packages/@n8n/db/src/connection/db-connection-monitor.ts index 597fd905f5a..1e98204703f 100644 --- a/packages/@n8n/db/src/connection/db-connection-monitor.ts +++ b/packages/@n8n/db/src/connection/db-connection-monitor.ts @@ -41,6 +41,13 @@ interface PgPoolInternals { _clients?: PgForceCloseClient[]; } +/** + * Grace period after force-closing the pool, before the teardown is abandoned. Force-close + * is synchronous and pg-pool's end callback only needs an I/O turn, so this bounds the + * pathological case without adding latency to a teardown that is going to finish anyway. + */ +export const FORCE_CLOSE_GRACE_MS = 1_000; + /** * Watches a DataSource and recovers it when the connection goes bad. * - Pings on `databaseConfig.pingIntervalSeconds`, races against `databaseConfig.pingTimeoutMs`. @@ -254,16 +261,20 @@ export class DbConnectionMonitor { } /** - * Races `work` against `pingTimeoutMs`. Throws OperationalError on timeout so - * the "don't report timeouts to Sentry" rule in `ping()` applies. The timer is - * always cancelled in `finally` so it never leaks when `work` wins. + * Races `work` against `timeoutMs`, defaulting to `pingTimeoutMs`. Throws + * OperationalError on timeout so the "don't report timeouts to Sentry" rule in + * `ping()` applies. The timer is always cancelled in `finally` so it never leaks + * when `work` wins. */ - private async raceTimeout(work: Promise): Promise { + private async raceTimeout( + work: Promise, + timeoutMs = this.databaseConfig.pingTimeoutMs, + ): Promise { const abortController = new AbortController(); try { return await Promise.race([ work, - setTimeoutP(this.databaseConfig.pingTimeoutMs, undefined, { + setTimeoutP(timeoutMs, undefined, { signal: abortController.signal, }).then(() => { throw new OperationalError('Database connection timed out'); @@ -349,49 +360,72 @@ export class DbConnectionMonitor { } /** - * Tear down the DataSource, bounding the Postgres drain so recovery can't hang on - * it. `pool.end()` only resolves once every pooled connection drains, so one frozen - * against an unreachable backend blocks `destroy()` forever, pinning recovery at - * attempt 1. Race the drain against `destroyTimeoutMs` and force-close on timeout so - * the original `destroy()` resolves and `initialize()` can run. SQLite uses its own - * driver `destroyTimeout`; `destroyTimeoutMs <= 0` disables the bound. + * Tear down the DataSource, bounding the Postgres drain so recovery can't hang on it. + * `pool.end()` only resolves once every pooled connection drains, so one frozen against + * an unreachable backend blocks the drain forever, pinning recovery at attempt 1. + * + * The bound goes on `driver.disconnect()` rather than on `destroy()` itself, because + * `destroy()` clears `isInitialized` only after `disconnect()` resolves, and + * `initialize()` refuses to run while that flag is set. Returning from the wrapper lets + * `destroy()` complete its own bookkeeping, so recovery can always rebuild the pool, and + * a drain we walked away from settling later touches only this discarded driver instance + * — `initialize()` builds a fresh one. + * + * SQLite uses its own driver `destroyTimeout`; `destroyTimeoutMs <= 0` disables the bound. */ private async destroyDataSource() { - const destroyPromise = this.dataSource.destroy(); - const timeoutMs = Number(this.databaseConfig.postgresdb?.destroyTimeoutMs); if (!this.isPostgres || !Number.isFinite(timeoutMs) || timeoutMs <= 0) { - await destroyPromise; + await this.dataSource.destroy(); return; } - // Capture the pool before `destroy()` nulls `driver.master`. - const pool = this.postgresDriver.master; - // Timeout may win the race; swallow a late rejection from the abandoned `destroy()`. - destroyPromise.catch(() => {}); + const driver = this.postgresDriver; + // Capture the pool before `disconnect()` nulls `driver.master`. + const pool = driver.master; + const originalDisconnect = driver.disconnect; + + driver.disconnect = async () => { + const drain = originalDisconnect.call(driver) as Promise; + if (await this.settlesWithin(drain, timeoutMs)) return; - const abortController = new AbortController(); - let timedOut = false; - try { - await Promise.race([ - destroyPromise, - setTimeoutP(timeoutMs, undefined, { signal: abortController.signal }).then(() => { - timedOut = true; - throw new OperationalError(`Database pool teardown timed out after ${timeoutMs}ms`); - }), - ]); - } catch (error) { - if (!timedOut) { - throw error; // a genuine `destroy()` failure, not the timeout - } this.logger.warn( `Database pool teardown exceeded ${timeoutMs}ms; force-closing connection sockets to continue recovery`, ); this.forceClosePostgresPool(pool); - // Force-close lets the original `destroy()` resolve and clear `isInitialized`. - await destroyPromise; + if (await this.settlesWithin(drain, FORCE_CLOSE_GRACE_MS)) return; + + this.logger.error( + `Database pool teardown did not complete ${FORCE_CLOSE_GRACE_MS}ms after force-closing sockets; abandoning the pool to continue recovery`, + ); + }; + + try { + await this.dataSource.destroy(); } finally { - abortController.abort(); + // `destroy()` continues past `disconnect()`, so a throw further along leaves this + // driver installed; without the restore the next attempt would wrap the wrapper + // and stack another set of bounds on top. + driver.disconnect = originalDisconnect; + } + } + + /** + * Resolves `true` if `work` settles within `timeoutMs`, `false` if it is still pending. + * A rejection counts as settled: the pool is unusable either way, and propagating it + * would leave `isInitialized` set, wedging recovery in a retry loop that re-runs the + * same failing teardown forever. Swallowing it up front also leaves the timeout as the + * only thing `raceTimeout` can throw, and keeps a drain we walk away from handled. + */ + private async settlesWithin(work: Promise, timeoutMs: number): Promise { + try { + await this.raceTimeout( + work.catch(() => {}), + timeoutMs, + ); + return true; + } catch { + return false; } } @@ -403,7 +437,13 @@ export class DbConnectionMonitor { * recovery integration test stops unblocking and fails, which is the guard. */ private forceClosePostgresPool(pool: PostgresDriver['master']) { - const clients = (pool as unknown as PgPoolInternals | undefined)?._clients; + if (!pool) { + // Normal, not evidence that pg-pool internals changed, so this stays below a warning. + this.logger.debug('Skipping Postgres pool force-close: the pool is already gone'); + return; + } + + const clients = (pool as unknown as PgPoolInternals)._clients; if (!Array.isArray(clients)) { this.logger.warn( 'Cannot force-close Postgres pool: pool._clients is unavailable (pg-pool internals may have changed)',