fix: pin connection for advisory lock lifecycle

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<boolean> {
-    const response = await this.client.rawQuery(
-      `SELECT GET_LOCK('${key}', ${timeout}) as lock_status;`
-    )
+  async getAdvisoryLock(key: string, timeout: number = 0, connection?: any): Promise<boolean> {
+    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<boolean> {
-    const response = await this.client.rawQuery(`SELECT RELEASE_LOCK('${key}') as lock_status;`)
+  async releaseAdvisoryLock(key: string, connection?: any): Promise<boolean> {
+    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
   }
 }

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<boolean> {
-    const response = await this.client.rawQuery(
-      `SELECT PG_TRY_ADVISORY_LOCK('${key}') as lock_status;`
-    )
+  async getAdvisoryLock(key: string, _timeout?: number, connection?: any): Promise<boolean> {
+    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<boolean> {
-    const response = await this.client.rawQuery(
-      `SELECT PG_ADVISORY_UNLOCK('${key}') as lock_status;`
-    )
+  async releaseAdvisoryLock(key: string, connection?: any): Promise<boolean> {
+    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
   }
 }

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,22 +285,33 @@ 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
+      this.emit('acquire:lock')
+    } catch (error) {
+      this.releaseLockConnection()
+      throw error
     }
-    this.emit('acquire:lock')
   }
 
   /**
@@ -296,11 +323,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')
   }
 
   /**

src/types/database.ts

@@ -83,8 +83,8 @@ export interface DialectContract {
 
   getPrimaryKeys(tableName: string): Promise<string[]>
 
-  getAdvisoryLock(key: string | number, timeout?: number): Promise<boolean>
-  releaseAdvisoryLock(key: string | number): Promise<boolean>
+  getAdvisoryLock(key: string | number, timeout?: number, connection?: any): Promise<boolean>
+  releaseAdvisoryLock(key: string | number, connection?: any): Promise<boolean>
 }
 
 /**

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 ({