fix(core): Bound database pool teardown so connection recovery always completes (#36013)

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<void>; recoverDataSource: () => Promise<void> };
+type MonitorInternals = {
+	ping: () => Promise<void>;
+	recoverDataSource: () => Promise<void>;
+	forceClosePostgresPool: () => void;
+};
 
 type PgPoolClient = {
 	query: (query: string | { text: string }) => Promise<unknown>;
 	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<DatabaseConfig['postgresdb']>({ destroyTimeoutMs }),
+				}),
+				mock<Logger>(),
+				mock<ErrorReporter>(),
+				mock<DbConnectionMetrics>(),
+			);
+			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) => {

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<DatabaseConfig>({
-				pingTimeoutMs: 5_000,
-				pingMaxFailuresBeforeRecovery: 3,
-				minRecoveryBackoffMs: 1_000,
-				maxRecoveryBackoffMs: 30_000,
-				connectionAcquisitionTimeoutMs: 30_000,
-				postgresdb: mock<DatabaseConfig['postgresdb']>({ destroyTimeoutMs }),
-			});
+		const buildPgMonitor = (destroyTimeoutMs = 10_000) =>
+			new DbConnectionMonitor(
+				dataSource,
+				onConnectedChange,
+				mock<DatabaseConfig>({
+					pingTimeoutMs: 5_000,
+					pingMaxFailuresBeforeRecovery: 3,
+					minRecoveryBackoffMs: 1_000,
+					maxRecoveryBackoffMs: 30_000,
+					connectionAcquisitionTimeoutMs: 30_000,
+					postgresdb: mock<DatabaseConfig['postgresdb']>({ 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<void>((resolve) => (resolveDestroy = resolve));
+			let resolveDrain!: () => void;
+			const drainPromise = new Promise<void>((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<void>(() => {}));
@@ -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<void>(() => {}));
+			// @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<void>((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', () => {

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<T>(work: Promise<T>): Promise<T> {
+	private async raceTimeout<T>(
+		work: Promise<T>,
+		timeoutMs = this.databaseConfig.pingTimeoutMs,
+	): Promise<T> {
 		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<void>;
+			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<unknown>, timeoutMs: number): Promise<boolean> {
+		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)',