diff --git a/src/dialects/mysql.ts b/src/dialects/mysql.ts index a629cf580..7fccc825d 100644 --- a/src/dialects/mysql.ts +++ b/src/dialects/mysql.ts @@ -265,18 +265,30 @@ export class MysqlDialect implements DialectContract { * Attempts to add advisory lock to the database and * returns it's status. */ - async getAdvisoryLock(key: string, timeout: number = 0): Promise { - const response = await this.client.rawQuery( - `SELECT GET_LOCK('${key}', ${timeout}) as lock_status;` - ) + async getAdvisoryLock(key: string, timeout: number = 0, connection?: any): Promise { + const query = this.client + .getWriteClient() + .raw(`SELECT GET_LOCK('${key}', ${timeout}) as lock_status;`) + + if (connection) { + query.connection(connection) + } + + const response = await query return response[0] && response[0][0] && response[0][0].lock_status === 1 } /** * Releases the advisory lock */ - async releaseAdvisoryLock(key: string): Promise { - const response = await this.client.rawQuery(`SELECT RELEASE_LOCK('${key}') as lock_status;`) + async releaseAdvisoryLock(key: string, connection?: any): Promise { + const query = this.client.getWriteClient().raw(`SELECT RELEASE_LOCK('${key}') as lock_status;`) + + if (connection) { + query.connection(connection) + } + + const response = await query return response[0] && response[0][0] && response[0][0].lock_status === 1 } } diff --git a/src/dialects/pg.ts b/src/dialects/pg.ts index 7159ae48b..492ce8939 100644 --- a/src/dialects/pg.ts +++ b/src/dialects/pg.ts @@ -264,20 +264,32 @@ export class PgDialect implements DialectContract { * Attempts to add advisory lock to the database and * returns it's status. */ - async getAdvisoryLock(key: string): Promise { - const response = await this.client.rawQuery( - `SELECT PG_TRY_ADVISORY_LOCK('${key}') as lock_status;` - ) + async getAdvisoryLock(key: string, _timeout?: number, connection?: any): Promise { + const query = this.client + .getWriteClient() + .raw(`SELECT PG_TRY_ADVISORY_LOCK('${key}') as lock_status;`) + + if (connection) { + query.connection(connection) + } + + const response = await query return response.rows[0] && response.rows[0].lock_status === true } /** * Releases the advisory lock */ - async releaseAdvisoryLock(key: string): Promise { - const response = await this.client.rawQuery( - `SELECT PG_ADVISORY_UNLOCK('${key}') as lock_status;` - ) + async releaseAdvisoryLock(key: string, connection?: any): Promise { + const query = this.client + .getWriteClient() + .raw(`SELECT PG_ADVISORY_UNLOCK('${key}') as lock_status;`) + + if (connection) { + query.connection(connection) + } + + const response = await query return response.rows[0] && response.rows[0].lock_status === true } } diff --git a/src/migration/runner.ts b/src/migration/runner.ts index a66caa616..a2636357c 100644 --- a/src/migration/runner.ts +++ b/src/migration/runner.ts @@ -90,6 +90,22 @@ export class MigrationRunner extends EventEmitter { */ disableLocks: boolean + /** + * A dedicated connection pinned for the advisory lock lifecycle. + * When the pool has more than one connection, we hold this connection + * between acquire and release to ensure both operations run on the + * same database session. + */ + private lockConnection: any = null + + /** + * Tracks whether we have actually acquired the advisory lock. We only + * attempt to release the lock when this is true, otherwise a failed + * acquisition would trigger a bogus release (releasing a lock we never + * held), masking the original "unable to acquire lock" error. + */ + private lockAcquired: boolean = false + /** * An array of files we have successfully migrated. The files are * collected regardless of `up` or `down` methods @@ -269,21 +285,38 @@ export class MigrationRunner extends EventEmitter { * Acquires a lock to disallow concurrent transactions. Only works with * `Mysql`, `PostgresSQL` and `MariaDb` for now. * - * Make sure we are acquiring lock outside the transactions, since we want - * to block other processes from acquiring the same lock. - * - * Locks are always acquired in dry run too, since we want to stay close - * to the real execution cycle + * Advisory locks are session-scoped, meaning both acquire and release + * must happen on the same database connection. When the pool has more + * than one connection, we pin a dedicated connection to guarantee this. + * With a single-connection pool, pinning is unnecessary (and would + * starve other queries). */ private async acquireLock() { if (!this.client.dialect.supportsAdvisoryLocks || this.disableLocks) { return } - const acquired = await this.client.dialect.getAdvisoryLock(1) - if (!acquired) { - throw new errors.E_UNABLE_ACQUIRE_LOCK() + const knexClient = this.client.getWriteClient() + if (knexClient.client.pool.max > 1) { + this.lockConnection = await knexClient.client.acquireConnection() } + + try { + const acquired = await this.client.dialect.getAdvisoryLock(1, undefined, this.lockConnection) + if (!acquired) { + throw new errors.E_UNABLE_ACQUIRE_LOCK() + } + this.lockAcquired = true + } catch (error) { + this.releaseLockConnection() + throw error + } + + /** + * Emitted outside the try/catch above. The lock is already acquired at + * this point, so a throwing listener must not trigger the catch (which + * would release the pinned connection while leaving the lock held on it). + */ this.emit('acquire:lock') } @@ -296,11 +329,35 @@ export class MigrationRunner extends EventEmitter { return } - const released = await this.client.dialect.releaseAdvisoryLock(1) - if (!released) { - throw new errors.E_UNABLE_RELEASE_LOCK() + /** + * Never attempt to release a lock we did not acquire. Doing so would + * throw "E_UNABLE_RELEASE_LOCK" and mask the real error (for example + * "E_UNABLE_ACQUIRE_LOCK" raised when another process holds the lock). + */ + if (!this.lockAcquired) { + return + } + + try { + const released = await this.client.dialect.releaseAdvisoryLock(1, this.lockConnection) + if (!released) { + throw new errors.E_UNABLE_RELEASE_LOCK() + } + this.emit('release:lock') + } finally { + this.lockAcquired = false + this.releaseLockConnection() + } + } + + /** + * Release the pinned lock connection back to the pool + */ + private releaseLockConnection() { + if (this.lockConnection) { + this.client.getWriteClient().client.releaseConnection(this.lockConnection) + this.lockConnection = null } - this.emit('release:lock') } /** diff --git a/src/types/database.ts b/src/types/database.ts index 58ce062e9..27cdb62f7 100644 --- a/src/types/database.ts +++ b/src/types/database.ts @@ -83,8 +83,8 @@ export interface DialectContract { getPrimaryKeys(tableName: string): Promise - getAdvisoryLock(key: string | number, timeout?: number): Promise - releaseAdvisoryLock(key: string | number): Promise + getAdvisoryLock(key: string | number, timeout?: number, connection?: any): Promise + releaseAdvisoryLock(key: string | number, connection?: any): Promise } /** diff --git a/test/database/query_client.spec.ts b/test/database/query_client.spec.ts index f006eb4c5..951c54b4c 100644 --- a/test/database/query_client.spec.ts +++ b/test/database/query_client.spec.ts @@ -14,7 +14,7 @@ import { QueryClient } from '../../src/query_client/index.js' import { logger, setup, - cleanup, + cleanup as dbCleanup, getConfig, resetTables, createEmitter, @@ -26,7 +26,7 @@ test.group('Query client', (group) => { }) group.teardown(async () => { - await cleanup() + await dbCleanup() }) group.each.teardown(async () => { @@ -277,7 +277,7 @@ test.group('Query client | dual mode', (group) => { }) group.teardown(async () => { - await cleanup() + await dbCleanup() }) group.each.teardown(async () => { @@ -356,7 +356,7 @@ test.group('Query client | read mode', (group) => { }) group.teardown(async () => { - await cleanup() + await dbCleanup() }) group.each.teardown(async () => { @@ -423,7 +423,7 @@ test.group('Query client | write mode', (group) => { }) group.teardown(async () => { - await cleanup() + await dbCleanup() }) group.each.teardown(async () => { @@ -503,7 +503,7 @@ if (!['sqlite', 'mssql', 'better_sqlite', 'libsql'].includes(process.env.DB!)) { }) group.teardown(async () => { - await cleanup() + await dbCleanup() }) group.each.teardown(async () => { @@ -533,6 +533,100 @@ if (!['sqlite', 'mssql', 'better_sqlite', 'libsql'].includes(process.env.DB!)) { await connection.disconnect() }) + + test('release advisory lock with pool of 1') + .disableTimeout() + .run(async ({ assert, cleanup }) => { + const config = getConfig() + config.pool = { min: 1, max: 1, acquireTimeoutMillis: 2000 } + + const connection = new Connection('primary', config, logger) + connection.connect() + + const client = new QueryClient('dual', connection, createEmitter()) + const knexClient = client.getWriteClient() + + /** + * Simulate what would happen if we pinned a connection for + * the lock: acquire a connection, hold it, then try to run + * a normal query. With pool max=1 the query should hang + * because there are no free connections. + */ + const heldConnection = await knexClient.client.acquireConnection() + cleanup(async () => { + knexClient.client.releaseConnection(heldConnection) + await connection.disconnect() + }) + + /** + * This simulates running a migration query while the lock + * connection is held. It should timeout since the only + * pool connection is occupied. + */ + await assert.rejects(() => client.rawQuery('SELECT 1')) + }) + + test('release advisory lock fails with pool of 2 without pinned connection', async ({ + assert, + cleanup, + }) => { + const config = getConfig() + config.pool = { min: 2, max: 2 } + + const connection = new Connection('primary', config, logger) + connection.connect() + cleanup(() => connection.disconnect()) + + const client = new QueryClient('dual', connection, createEmitter()) + const knexClient = client.getWriteClient() + + const acquired = await client.dialect.getAdvisoryLock(1) + assert.isTrue(acquired) + + /** + * Hold one connection so the release call is forced onto + * the other one. With min:2/max:2 both connections exist, + * and holding one guarantees releaseAdvisoryLock gets a + * different connection than the one that acquired the lock. + */ + const heldConnection = await knexClient.client.acquireConnection() + + const released = await client.dialect.releaseAdvisoryLock(1) + + knexClient.client.releaseConnection(heldConnection) + + assert.isFalse(released) + }) + + test('release advisory lock succeeds with pool of 2 with pinned connection', async ({ + assert, + cleanup, + }) => { + const config = getConfig() + config.pool = { min: 2, max: 2 } + + const connection = new Connection('primary', config, logger) + connection.connect() + + const client = new QueryClient('dual', connection, createEmitter()) + const knexClient = client.getWriteClient() + + /** + * Pin a connection and pass it to both acquire and release. + * This ensures both operations run on the same database session. + */ + const pinnedConnection = await knexClient.client.acquireConnection() + cleanup(async () => { + knexClient.client.releaseConnection(pinnedConnection) + await connection.disconnect() + }) + + const acquired = await client.dialect.getAdvisoryLock(1, undefined, pinnedConnection) + assert.isTrue(acquired) + + const released = await client.dialect.releaseAdvisoryLock(1, pinnedConnection) + assert.isTrue(released) + }) }) } @@ -542,7 +636,7 @@ test.group('Query client | get tables', (group) => { }) group.teardown(async () => { - await cleanup() + await dbCleanup() }) group.each.teardown(async () => { @@ -598,7 +692,7 @@ test.group('Query client | get primary keys', (group) => { }) group.teardown(async () => { - await cleanup() + await dbCleanup() }) group.each.teardown(async () => { @@ -662,7 +756,7 @@ if (process.env.DB === 'pg') { }) group.teardown(async () => { - await cleanup() + await dbCleanup() }) test('get tables with schema info for non-public schemas', async ({ diff --git a/test/migrations/migrator.spec.ts b/test/migrations/migrator.spec.ts index b981b2f98..13704c491 100644 --- a/test/migrations/migrator.spec.ts +++ b/test/migrations/migrator.spec.ts @@ -14,6 +14,7 @@ import { AppFactory } from '@adonisjs/core/factories/app' import { setup, getDb, + getConfig, resetTables, getMigrator, cleanup as cleanupTables, @@ -1526,4 +1527,261 @@ test.group('Migrator', (group) => { assert.deepEqual(migratedFiles, []) assert.equal(migrator.status, 'skipped') }) + + test('migrate with pool of 4 acquires and releases advisory lock correctly', async ({ + fs, + assert, + cleanup, + }) => { + const app = new AppFactory().create(fs.baseUrl, () => {}) + await app.init() + + const config = getConfig() + config.pool = { min: 4, max: 4 } + const db = getDb(undefined, { + connection: 'primary', + connections: { primary: config, secondary: config }, + }) + cleanup(() => db.manager.closeAll()) + + await fs.create( + 'database/migrations/users_pool4.ts', + ` + import { BaseSchema as Schema } from '../../../../src/schema/main.js' + export default class User extends Schema { + public async up () { + this.schema.createTable('schema_users', (table) => { + table.increments() + }) + } + } + ` + ) + + const migrator = getMigrator(db, app, { + direction: 'up', + connectionName: 'primary', + }) + + await migrator.run() + + assert.isNull(migrator.error) + assert.equal(migrator.status, 'completed') + assert.isTrue(await db.connection().schema.hasTable('schema_users')) + }) + + test('migrate with pool of 2 acquires and releases advisory lock correctly', async ({ + fs, + assert, + cleanup, + }) => { + const app = new AppFactory().create(fs.baseUrl, () => {}) + await app.init() + + /** + * Pool of 2 is the tightest case for connection pinning: one + * connection is held for the advisory lock lifecycle and the + * other is left to run the migration queries. If pinning starved + * the pool, this migration would deadlock. + */ + const config = getConfig() + config.pool = { min: 2, max: 2 } + const db = getDb(undefined, { + connection: 'primary', + connections: { primary: config, secondary: config }, + }) + cleanup(() => db.manager.closeAll()) + + await fs.create( + 'database/migrations/users_pool2.ts', + ` + import { BaseSchema as Schema } from '../../../../src/schema/main.js' + export default class User extends Schema { + public async up () { + this.schema.createTable('schema_users', (table) => { + table.increments() + }) + } + } + ` + ) + + const migrator = getMigrator(db, app, { + direction: 'up', + connectionName: 'primary', + }) + + await migrator.run() + + assert.isNull(migrator.error) + assert.equal(migrator.status, 'completed') + assert.isTrue(await db.connection().schema.hasTable('schema_users')) + }) + + test('migrate with pool of 1 acquires and releases advisory lock correctly', async ({ + fs, + assert, + cleanup, + }) => { + const app = new AppFactory().create(fs.baseUrl, () => {}) + await app.init() + + const config = getConfig() + config.pool = { min: 1, max: 1 } + const db = getDb(undefined, { + connection: 'primary', + connections: { primary: config, secondary: config }, + }) + cleanup(() => db.manager.closeAll()) + + await fs.create( + 'database/migrations/users_pool1.ts', + ` + import { BaseSchema as Schema } from '../../../../src/schema/main.js' + export default class User extends Schema { + public async up () { + this.schema.createTable('schema_users', (table) => { + table.increments() + }) + } + } + ` + ) + + const migrator = getMigrator(db, app, { + direction: 'up', + connectionName: 'primary', + }) + + await migrator.run() + + assert.isNull(migrator.error) + assert.equal(migrator.status, 'completed') + assert.isTrue(await db.connection().schema.hasTable('schema_users')) + }) + + test('rollback with a multi connection pool acquires and releases advisory lock correctly', async ({ + fs, + assert, + cleanup, + }) => { + const app = new AppFactory().create(fs.baseUrl, () => {}) + await app.init() + + /** + * The "up" pool tests already exercise pinning during acquire. This + * test covers the "down" direction, ensuring the pinned connection + * is acquired and released correctly across a rollback as well. + */ + const config = getConfig() + config.pool = { min: 2, max: 2 } + const db = getDb(undefined, { + connection: 'primary', + connections: { primary: config, secondary: config }, + }) + cleanup(() => db.manager.closeAll()) + + await fs.create( + 'database/migrations/users_pool_rollback.ts', + ` + import { BaseSchema as Schema } from '../../../../src/schema/main.js' + export default class User extends Schema { + public async up () { + this.schema.createTable('schema_users', (table) => { + table.increments() + }) + } + + public async down () { + this.schema.dropTable('schema_users') + } + } + ` + ) + + const migrator = getMigrator(db, app, { + direction: 'up', + connectionName: 'primary', + }) + await migrator.run() + + assert.isNull(migrator.error) + assert.equal(migrator.status, 'completed') + assert.isTrue(await db.connection().schema.hasTable('schema_users')) + + const rollbackMigrator = getMigrator(db, app, { + direction: 'down', + connectionName: 'primary', + }) + await rollbackMigrator.run() + + assert.isNull(rollbackMigrator.error) + assert.equal(rollbackMigrator.status, 'completed') + assert.isFalse(await db.connection().schema.hasTable('schema_users')) + }) + + test('surface the acquire lock error without masking it as a release error', async ({ + fs, + assert, + cleanup, + }) => { + const app = new AppFactory().create(fs.baseUrl, () => {}) + await app.init() + + const config = getConfig() + config.pool = { min: 2, max: 4 } + const db = getDb(undefined, { + connection: 'primary', + connections: { primary: config, secondary: config }, + }) + cleanup(() => db.manager.closeAll()) + + /** + * Simulate another process holding the advisory lock on its own + * database session, so the migrator is unable to acquire it. + */ + const client = db.connection('primary') + const knexClient = client.getWriteClient() + const heldConnection = await knexClient.client.acquireConnection() + const locked = await client.dialect.getAdvisoryLock(1, undefined, heldConnection) + assert.isTrue(locked) + + cleanup(async () => { + await client.dialect.releaseAdvisoryLock(1, heldConnection) + knexClient.client.releaseConnection(heldConnection) + }) + + await fs.create( + 'database/migrations/users_contended.ts', + ` + import { BaseSchema as Schema } from '../../../../src/schema/main.js' + export default class User extends Schema { + public async up () { + this.schema.createTable('schema_users', (table) => { + table.increments() + }) + } + } + ` + ) + + const migrator = getMigrator(db, app, { + direction: 'up', + connectionName: 'primary', + }) + + /** + * run() must resolve (not throw). Previously the failed acquisition + * triggered a bogus release that threw E_UNABLE_RELEASE_LOCK and + * masked the real "unable to acquire lock" error. + */ + await migrator.run() + + assert.equal(migrator.status, 'error') + assert.equal((migrator.error as any)?.code, 'E_UNABLE_ACQUIRE_LOCK') + assert.isFalse(await db.connection().schema.hasTable('schema_users')) + }).skip( + ['sqlite', 'better_sqlite', 'libsql', 'mssql'].includes(process.env.DB!), + 'Advisory locks are only supported on PostgreSQL and MySQL' + ) })