fix(core): Prevent concurrent instance startups from racing database migrations (#34685)

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

packages/@n8n/db/src/connection/__tests__/db-connection.test.ts

@@ -1,14 +1,23 @@
 /* eslint-disable @typescript-eslint/unbound-method */
 import type { Logger } from '@n8n/backend-common';
 import type { DatabaseConfig } from '@n8n/config';
-import { DataSource, type DataSourceOptions } from '@n8n/typeorm';
+import { Container } from '@n8n/di';
+import {
+	DataSource,
+	MigrationExecutor,
+	type DataSourceOptions,
+	type EntityManager,
+	type QueryRunner,
+} from '@n8n/typeorm';
 import type { ErrorReporter } from 'n8n-core';
 import { DbConnectionTimeoutError } from 'n8n-workflow';
+import { createHash } from 'node:crypto';
 import type { Mock } from 'vitest';
 import { mock, mockDeep } from 'vitest-mock-extended';
 
 import * as migrationHelper from '../../migrations/migration-helpers';
 import type { Migration } from '../../migrations/migration-types';
+import { DbLock, DbLockService } from '../../services/db-lock.service';
 import { DbConnection } from '../db-connection';
 import type { DbConnectionMetrics } from '../db-connection-metrics';
 import { DbConnectionMonitor } from '../db-connection-monitor';
@@ -19,6 +28,8 @@ vi.mock('@n8n/typeorm', async () => ({
 	...(await vi.importActual<typeof import('@n8n/typeorm')>('@n8n/typeorm')),
 	// eslint-disable-next-line @typescript-eslint/naming-convention
 	DataSource: vi.fn(),
+	// eslint-disable-next-line @typescript-eslint/naming-convention
+	MigrationExecutor: vi.fn(),
 }));
 
 vi.mock('../db-connection-monitor');
@@ -151,22 +162,93 @@ describe('DbConnection', () => {
 	});
 
 	describe('migrate', () => {
-		it('should wrap migrations and run them', async () => {
-			dataSource.runMigrations.mockResolvedValue([]);
+		const dbLockService = mock<DbLockService>();
+		const queryRunner = mock<QueryRunner>();
+		const tx = mock<EntityManager>({ queryRunner });
+		const migrationExecutor = mock<MigrationExecutor>();
+		// Mirrors migrate()'s derivation for the test's schema-less, prefix-less options
+		const expectedSubKey = createHash('sha256').update('public:').digest().readInt32BE(0);
 
-			const wrapMigrationSpy = vi
-				.spyOn(migrationHelper, 'wrapMigration')
-				.mockImplementation(() => {});
+		beforeEach(() => {
+			Container.set(DbLockService, dbLockService);
+			dbLockService.withLock.mockImplementation(async (_lockId, fn) => await fn(tx));
+			tx.query.mockResolvedValue([]);
+			vi.mocked(MigrationExecutor).mockImplementation(function () {
+				return migrationExecutor;
+			});
+			vi.spyOn(migrationHelper, 'wrapMigration').mockImplementation(() => {});
+		});
 
-			expect(dataSource.runMigrations).not.toHaveBeenCalled();
+		it('should run migrations under the migration advisory lock on postgres', async () => {
 			expect(dbConnection.connectionState.migrated).toBe(false);
 
 			await dbConnection.migrate();
 
-			expect(wrapMigrationSpy).toHaveBeenCalledTimes(2);
-			expect(dataSource.runMigrations).toHaveBeenCalledWith({ transaction: 'each' });
+			expect(migrationHelper.wrapMigration).toHaveBeenCalledTimes(2);
+			expect(dbLockService.withLock).toHaveBeenCalledWith(DbLock.MIGRATIONS, expect.any(Function), {
+				subKey: expectedSubKey,
+				waitIndefinitely: true,
+			});
+			expect(MigrationExecutor).toHaveBeenCalledWith(dataSource, queryRunner);
+			expect(migrationExecutor.transaction).toBe('none');
+			expect(migrationExecutor.executePendingMigrations).toHaveBeenCalled();
 			expect(dbConnection.connectionState.migrated).toBe(true);
 		});
+
+		it('should not override statement_timeout when none is configured', async () => {
+			await dbConnection.migrate();
+
+			expect(tx.query).not.toHaveBeenCalled();
+		});
+
+		it('should restore the configured statement timeout inside the lock transaction', async () => {
+			connectionOptions.getOptions.mockReturnValue({
+				...postgresOptions,
+				statementTimeout: 300_000,
+			});
+			const connection = new DbConnection(
+				errorReporter,
+				connectionOptions,
+				databaseConfig,
+				logger,
+				dbConnectionMetrics,
+			);
+
+			await connection.migrate();
+
+			expect(tx.query).toHaveBeenCalledWith('SET LOCAL statement_timeout = 300000');
+		});
+
+		it('should propagate migration errors', async () => {
+			migrationExecutor.executePendingMigrations.mockRejectedValue(new Error('migration failed'));
+
+			await expect(dbConnection.migrate()).rejects.toThrow('migration failed');
+
+			expect(dbConnection.connectionState.migrated).toBe(false);
+		});
+
+		it('should run migrations directly without the lock on sqlite', async () => {
+			connectionOptions.getOptions.mockReturnValue({
+				type: 'sqlite',
+				database: ':memory:',
+				migrations,
+			});
+			// options are @Memoized and read in the constructor — re-instantiate
+			const sqliteConnection = new DbConnection(
+				errorReporter,
+				connectionOptions,
+				databaseConfig,
+				logger,
+				dbConnectionMetrics,
+			);
+			dataSource.runMigrations.mockResolvedValue([]);
+
+			await sqliteConnection.migrate();
+
+			expect(dbLockService.withLock).not.toHaveBeenCalled();
+			expect(dataSource.runMigrations).toHaveBeenCalledWith({ transaction: 'each' });
+			expect(sqliteConnection.connectionState.migrated).toBe(true);
+		});
 	});
 
 	describe('close', () => {

packages/@n8n/db/src/connection/db-connection.ts

@@ -2,10 +2,11 @@ import { Logger } from '@n8n/backend-common';
 import { DatabaseConfig } from '@n8n/config';
 import { Memoized } from '@n8n/decorators';
 import { Container, Service } from '@n8n/di';
-import { DataSource } from '@n8n/typeorm';
+import { DataSource, MigrationExecutor } from '@n8n/typeorm';
 import { ensureError } from '@n8n/utils/errors/ensure-error';
 import { ErrorReporter } from 'n8n-core';
 import { DbConnectionTimeoutError } from 'n8n-workflow';
+import { createHash } from 'node:crypto';
 import { setTimeout as setTimeoutP } from 'timers/promises';
 
 import { computeBackoff } from './backoff';
@@ -15,6 +16,7 @@ import { DbConnectionOptions } from './db-connection-options';
 import { readPoolStats, type DbPoolStats } from './db-pool-stats';
 import { wrapMigration } from '../migrations/migration-helpers';
 import type { Migration } from '../migrations/migration-types';
+import { DbLock, DbLockService } from '../services/db-lock.service';
 import { TransactionRunner } from '../services/transaction';
 import { TypeOrmTransactionRunner } from '../services/typeorm-transaction';
 
@@ -84,12 +86,52 @@ export class DbConnection {
 	}
 
 	async migrate() {
-		const { dataSource, connectionState } = this;
+		const { dataSource, connectionState, options } = this;
 		(dataSource.options.migrations as Migration[]).forEach(wrapMigration);
-		await dataSource.runMigrations({ transaction: 'each' });
+		if (options.type === 'postgres') {
+			await this.migrateWithAdvisoryLock(options);
+		} else {
+			// SQLite is single-instance, so concurrent migrations aren't possible.
+			await dataSource.runMigrations({ transaction: 'each' });
+		}
 		connectionState.migrated = true;
 	}
 
+	/**
+	 * Runs all pending migrations while holding a Postgres advisory lock, so
+	 * concurrently starting instances migrate one at a time. The migrations run
+	 * on the lock transaction's own connection, which keeps
+	 * `DB_POSTGRESDB_POOL_SIZE=1` working; the trade-off is that they commit or
+	 * roll back as one unit.
+	 */
+	private async migrateWithAdvisoryLock(options: {
+		schema?: string;
+		entityPrefix?: string;
+		statementTimeout?: number;
+	}) {
+		const { dataSource } = this;
+		// Scoped by schema+prefix so co-hosted tenants in one database don't block each other.
+		const subKey = createHash('sha256')
+			.update(`${options.schema ?? 'public'}:${options.entityPrefix ?? ''}`)
+			.digest()
+			.readInt32BE(0);
+		this.logger.info('Acquiring database migration lock...');
+		await Container.get(DbLockService).withLock(
+			DbLock.MIGRATIONS,
+			async (tx) => {
+				if (options.statementTimeout) {
+					// Restore the driver's per-statement cap for the migration statements
+					await tx.query(`SET LOCAL statement_timeout = ${Number(options.statementTimeout)}`);
+				}
+				const migrationExecutor = new MigrationExecutor(dataSource, tx.queryRunner);
+				// 'none': the executor manages no transactions and runs inside the lock's
+				migrationExecutor.transaction = 'none';
+				await migrationExecutor.executePendingMigrations();
+			},
+			{ subKey, waitIndefinitely: true },
+		);
+	}
+
 	async close() {
 		await this.monitor?.stop();
 		this.monitor = undefined;

packages/@n8n/db/src/services/__tests__/db-lock.service.test.ts

@@ -67,6 +67,70 @@ describe('DbLockService', () => {
 			expect(mockTx.query).toHaveBeenCalledWith('SELECT pg_advisory_xact_lock($1)', [1001]);
 		});
 
+		it('should disable all lock-killing timeouts when waitIndefinitely is set', async () => {
+			databaseConfig.type = 'postgresdb';
+			const fn = vi.fn().mockResolvedValue('result');
+
+			await service.withLock(1001, fn, { waitIndefinitely: true });
+
+			expect(mockTx.query).toHaveBeenCalledWith('SET LOCAL statement_timeout = 0');
+			expect(mockTx.query).toHaveBeenCalledWith('SET LOCAL lock_timeout = 0');
+			expect(mockTx.query).toHaveBeenCalledWith(
+				'SET LOCAL idle_in_transaction_session_timeout = 0',
+			);
+		});
+
+		it('should not touch statement timeouts without waitIndefinitely', async () => {
+			databaseConfig.type = 'postgresdb';
+			const fn = vi.fn().mockResolvedValue('result');
+
+			await service.withLock(1001, fn);
+			await service.withLock(1001, fn, { timeoutMs: 5000 });
+
+			expect(mockTx.query).not.toHaveBeenCalledWith(expect.stringContaining('statement_timeout'));
+			expect(mockTx.query).not.toHaveBeenCalledWith(
+				expect.stringContaining('idle_in_transaction_session_timeout'),
+			);
+		});
+
+		it('should reject combining timeoutMs and waitIndefinitely at compile time', async () => {
+			databaseConfig.type = 'postgresdb';
+			const fn = vi.fn().mockResolvedValue('result');
+
+			// @ts-expect-error timeoutMs and waitIndefinitely are mutually exclusive
+			await service.withLock(1001, fn, { timeoutMs: 5000, waitIndefinitely: true });
+		});
+
+		it('should use the two-key lock form when subKey is provided', async () => {
+			databaseConfig.type = 'postgresdb';
+			const fn = vi.fn().mockResolvedValue('result');
+
+			const result = await service.withLock(1005, fn, { subKey: -12345 });
+
+			expect(result).toBe('result');
+			expect(mockTx.query).toHaveBeenCalledWith(
+				'SELECT pg_advisory_xact_lock($1, $2)',
+				[1005, -12345],
+			);
+			expect(fn).toHaveBeenCalledWith(mockTx);
+		});
+
+		it('should include both keys in the timeout error message when subKey is provided', async () => {
+			databaseConfig.type = 'postgresdb';
+			const fn = vi.fn();
+			const timeoutError = new QueryFailedError(
+				'SELECT pg_advisory_xact_lock($1, $2)',
+				[1005, -12345],
+				new Error('canceling statement due to lock timeout'),
+			);
+
+			mockTx.query.mockResolvedValueOnce(undefined).mockRejectedValueOnce(timeoutError);
+
+			await expect(service.withLock(1005, fn, { timeoutMs: 5000, subKey: -12345 })).rejects.toThrow(
+				/Timed out waiting for DbLock 1005:-12345 after 5000ms/,
+			);
+		});
+
 		it('should throw OperationalError when lock timeout is exceeded', async () => {
 			databaseConfig.type = 'postgresdb';
 			const fn = vi.fn();
@@ -194,6 +258,35 @@ describe('DbLockService', () => {
 			expect(executionOrder).toEqual(['fn1-start', 'fn1-end', 'fn2-start']);
 		});
 
+		it('should scope the in-process mutex by subKey', async () => {
+			let resolveFirst!: () => void;
+			const firstBlocking = new Promise<void>((r) => {
+				resolveFirst = r;
+			});
+
+			const fn1 = vi.fn(async () => {
+				await firstBlocking;
+				return 'first';
+			});
+			const fn2 = vi.fn(async () => 'second');
+			const fn3 = vi.fn(async () => 'third');
+
+			const p1 = service.withLock(1001, fn1, { subKey: 1 });
+			// Different subKey — must not block
+			const p2 = service.withLock(1001, fn2, { subKey: 2 });
+			// Same subKey — must block until fn1 releases
+			const p3 = service.withLock(1001, fn3, { subKey: 1 });
+
+			await new Promise((r) => setImmediate(r));
+			expect(fn1).toHaveBeenCalled();
+			expect(fn2).toHaveBeenCalled();
+			expect(fn3).not.toHaveBeenCalled();
+
+			resolveFirst();
+			const results = await Promise.all([p1, p2, p3]);
+			expect(results).toEqual(['first', 'second', 'third']);
+		});
+
 		it('should not block different lockIds', async () => {
 			let resolveFirst!: () => void;
 			const firstBlocking = new Promise<void>((r) => {

packages/@n8n/db/src/services/db-lock.service.ts

@@ -14,12 +14,22 @@ export const enum DbLock {
 	TRUSTED_KEY_REFRESH = 1002,
 	WORKFLOW_STATISTICS_ROLLUP = 1003,
 	WORKFLOW_REVIEW_REQUEST_CREATE = 1004,
+	MIGRATIONS = 1005,
 	/** Reserved for integration tests — never use in production code */
 	TEST = 9999,
 }
 
 type ReleaseFn = () => void;
 
+/**
+ * `timeoutMs` and `waitIndefinitely` are mutually exclusive — passing both is
+ * a compile-time error.
+ */
+type WithLockOptions = { subKey?: number } & (
+	| { timeoutMs?: number; waitIndefinitely?: never }
+	| { timeoutMs?: never; waitIndefinitely?: boolean }
+);
+
 /**
  * Opaque ownership token created on lock acquisition. The release function
  * captures this token and only releases if it still matches `held`, which
@@ -33,6 +43,10 @@ interface LockState {
 	queue: Array<(token: OwnerToken) => void>;
 }
 
+/** Shared key derivation so SQLite mutexes scope by subKey like Postgres locks do */
+const lockKey = (lockId: number, subKey?: number) =>
+	subKey === undefined ? `${lockId}` : `${lockId}:${subKey}`;
+
 /**
  * Provides serialized execution of critical sections across concurrent
  * callers using database-level advisory locks (Postgres) or an in-process
@@ -81,7 +95,7 @@ interface LockState {
  */
 @Service()
 export class DbLockService {
-	private readonly locks = new Map<number, LockState>();
+	private readonly locks = new Map<string, LockState>();
 
 	constructor(
 		private readonly dataSource: DataSource,
@@ -96,15 +110,23 @@ export class DbLockService {
 	 * or rolls back. Concurrent callers block until the lock is available.
 	 *
 	 * @param options.timeoutMs - Maximum time in milliseconds to wait for the lock.
-	 *   If not provided, waits indefinitely.
+	 *   If not provided, the wait is implicitly capped by the connection's
+	 *   `statement_timeout` on Postgres.
+	 * @param options.waitIndefinitely - Wait for the lock with no upper bound and
+	 *   keep the lock transaction alive however long `fn` takes. Mutually
+	 *   exclusive with `timeoutMs`. On SQLite the in-process mutex already waits
+	 *   indefinitely without `timeoutMs`.
+	 * @param options.subKey - Optional int32 second lock key (Postgres two-int
+	 *   advisory lock form), e.g. to scope a lock ID per tenant schema. Scopes
+	 *   the SQLite in-process mutex the same way.
 	 */
 	async withLock<T>(
 		lockId: DbLock,
 		fn: (tx: EntityManager) => Promise<T>,
-		options?: { timeoutMs?: number },
+		options?: WithLockOptions,
 	): Promise<T> {
 		if (this.databaseConfig.type !== 'postgresdb') {
-			const release = await this.acquireLock(lockId, options?.timeoutMs);
+			const release = await this.acquireLock(lockId, options?.timeoutMs, options?.subKey);
 			try {
 				return await this.dataSource.manager.transaction(async (tx) => await fn(tx));
 			} finally {
@@ -117,13 +139,26 @@ export class DbLockService {
 				// SET doesn't support parameterized queries ($1) in Postgres.
 				// Number() coercion is safe here since timeoutMs is typed as number.
 				await tx.query(`SET LOCAL lock_timeout = '${Number(options.timeoutMs)}'`);
+			} else if (options?.waitIndefinitely) {
+				// A blocked lock acquisition is a single statement, so no statement or
+				// lock timeout may kill the wait, and no idle timeout may reap this
+				// transaction while it idles holding the lock.
+				await tx.query('SET LOCAL statement_timeout = 0');
+				await tx.query('SET LOCAL lock_timeout = 0');
+				await tx.query('SET LOCAL idle_in_transaction_session_timeout = 0');
 			}
+			const lockKeys = options?.subKey === undefined ? [lockId] : [lockId, options.subKey];
 			try {
-				await tx.query('SELECT pg_advisory_xact_lock($1)', [lockId]);
+				await tx.query(
+					lockKeys.length === 1
+						? 'SELECT pg_advisory_xact_lock($1)'
+						: 'SELECT pg_advisory_xact_lock($1, $2)',
+					lockKeys,
+				);
 			} catch (error) {
 				if (error instanceof QueryFailedError && error.message.includes('lock timeout')) {
 					throw new OperationalError(
-						`Timed out waiting for DbLock ${lockId} after ${options?.timeoutMs}ms`,
+						`Timed out waiting for DbLock ${lockKeys.join(':')} after ${options?.timeoutMs}ms`,
 						{ cause: error },
 					);
 				}
@@ -165,8 +200,12 @@ export class DbLockService {
 	 * Acquire the in-process lock, blocking until available.
 	 * Returns a one-shot release function bound to this specific acquisition.
 	 */
-	private async acquireLock(lockId: number, timeoutMs?: number): Promise<ReleaseFn> {
-		const lockState = this.getOrCreateLockState(lockId);
+	private async acquireLock(
+		lockId: number,
+		timeoutMs?: number,
+		subKey?: number,
+	): Promise<ReleaseFn> {
+		const lockState = this.getOrCreateLockState(lockId, subKey);
 
 		if (!lockState.held) {
 			const token: OwnerToken = {} as OwnerToken;
@@ -199,7 +238,9 @@ export class DbLockService {
 						lockState.queue.splice(idx, 1);
 					}
 					reject(
-						new OperationalError(`Timed out waiting for DbLock ${lockId} after ${timeoutMs}ms`),
+						new OperationalError(
+							`Timed out waiting for DbLock ${lockKey(lockId, subKey)} after ${timeoutMs}ms`,
+						),
 					);
 				}, timeoutMs);
 			}
@@ -254,11 +295,12 @@ export class DbLockService {
 		};
 	}
 
-	private getOrCreateLockState(lockId: number): LockState {
-		let lockState = this.locks.get(lockId);
+	private getOrCreateLockState(lockId: number, subKey?: number): LockState {
+		const key = lockKey(lockId, subKey);
+		let lockState = this.locks.get(key);
 		if (!lockState) {
 			lockState = { held: null, queue: [] };
-			this.locks.set(lockId, lockState);
+			this.locks.set(key, lockState);
 		}
 		return lockState;
 	}

packages/cli/test/integration/database/services/db-lock.service.test.ts

@@ -172,4 +172,70 @@ describe('DbLockService', () => {
 			expect(result).toBe('free');
 		});
 	});
+
+	describe('subKey scoping (Postgres)', () => {
+		it('should not block when the same lock ID is held with a different subKey', async () => {
+			if (!isPostgres) return;
+
+			let lockAcquired!: () => void;
+			const lockAcquiredPromise = new Promise<void>((resolve) => {
+				lockAcquired = resolve;
+			});
+
+			// Hold (TEST, 1) on the separate connection
+			const holdLockPromise = holdLockDs.manager.transaction(async (tx) => {
+				await tx.query('SELECT pg_advisory_xact_lock($1, $2)', [DbLock.TEST, 1]);
+				lockAcquired();
+				await sleep(500);
+			});
+
+			await lockAcquiredPromise;
+
+			// (TEST, 2) must be acquirable immediately — timeoutMs turns a wrongly
+			// blocking lock into a test failure instead of a hang
+			const result = await dbLockService.withLock(DbLock.TEST, async () => 'independent', {
+				subKey: 2,
+				timeoutMs: 200,
+			});
+			expect(result).toBe('independent');
+
+			await holdLockPromise;
+		});
+
+		it('should serialize callers on the same lock ID and subKey', async () => {
+			if (!isPostgres) return;
+
+			const executionOrder: string[] = [];
+
+			let lockAcquired!: () => void;
+			const lockAcquiredPromise = new Promise<void>((resolve) => {
+				lockAcquired = resolve;
+			});
+
+			const first = holdLockDs.manager.transaction(async (tx) => {
+				await tx.query('SELECT pg_advisory_xact_lock($1, $2)', [DbLock.TEST, 1]);
+				executionOrder.push('first:start');
+				lockAcquired();
+				await sleep(300);
+				executionOrder.push('first:end');
+				return 'first';
+			});
+
+			await lockAcquiredPromise;
+
+			const second = dbLockService.withLock(
+				DbLock.TEST,
+				async () => {
+					executionOrder.push('second:start');
+					return 'second';
+				},
+				{ subKey: 1, waitIndefinitely: true },
+			);
+
+			const results = await Promise.all([first, second]);
+
+			expect(results).toEqual(['first', 'second']);
+			expect(executionOrder).toEqual(['first:start', 'first:end', 'second:start']);
+		});
+	});
 });