diff --git a/packages/@n8n/db/src/connection/__tests__/db-connection.test.ts b/packages/@n8n/db/src/connection/__tests__/db-connection.test.ts index 3f0372230b6..bdea6066ed4 100644 --- a/packages/@n8n/db/src/connection/__tests__/db-connection.test.ts +++ b/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('@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(); + const queryRunner = mock(); + const tx = mock({ queryRunner }); + const migrationExecutor = mock(); + // 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', () => { diff --git a/packages/@n8n/db/src/connection/db-connection.ts b/packages/@n8n/db/src/connection/db-connection.ts index 53b89731842..d92746b976b 100644 --- a/packages/@n8n/db/src/connection/db-connection.ts +++ b/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; diff --git a/packages/@n8n/db/src/services/__tests__/db-lock.service.test.ts b/packages/@n8n/db/src/services/__tests__/db-lock.service.test.ts index d6abb0e3cfb..222a48b69be 100644 --- a/packages/@n8n/db/src/services/__tests__/db-lock.service.test.ts +++ b/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((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((r) => { diff --git a/packages/@n8n/db/src/services/db-lock.service.ts b/packages/@n8n/db/src/services/db-lock.service.ts index 6dd5edf336c..30ee6090061 100644 --- a/packages/@n8n/db/src/services/db-lock.service.ts +++ b/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(); + private readonly locks = new Map(); 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( lockId: DbLock, fn: (tx: EntityManager) => Promise, - options?: { timeoutMs?: number }, + options?: WithLockOptions, ): Promise { 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 { - const lockState = this.getOrCreateLockState(lockId); + private async acquireLock( + lockId: number, + timeoutMs?: number, + subKey?: number, + ): Promise { + 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; } diff --git a/packages/cli/test/integration/database/services/db-lock.service.test.ts b/packages/cli/test/integration/database/services/db-lock.service.test.ts index e77de41c0d8..c6c04348544 100644 --- a/packages/cli/test/integration/database/services/db-lock.service.test.ts +++ b/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((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((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']); + }); + }); });