fix: pin connection for advisory lock lifecycle

commands/make_factory.ts

@@ -35,8 +35,15 @@ export default class MakeFactory extends BaseCommand {
   @flags.string({ description: 'Use the contents of the given file as the generated output' })
   declare contentsFrom: string
 
+  /**
+   * Forcefully overwrite existing files
+   */
+  @flags.boolean({ description: 'Forcefully overwrite existing files', alias: 'f' })
+  declare force: boolean
+
   async run() {
     const codemods = await this.createCodemods()
+    codemods.overwriteExisting = this.force === true
     await codemods.makeUsingStub(
       stubsRoot,
       'make/factory/main.stub',

commands/make_migration.ts

@@ -63,6 +63,12 @@ export default class MakeMigration extends BaseCommand {
   @flags.string({ description: 'Use the contents of the given file as the generated output' })
   declare contentsFrom: string
 
+  /**
+   * Forcefully overwrite existing files
+   */
+  @flags.boolean({ description: 'Forcefully overwrite existing files', alias: 'f' })
+  declare force: boolean
+
   /**
    * Not a valid connection
    */
@@ -137,6 +143,7 @@ export default class MakeMigration extends BaseCommand {
     const fileName = `${prefix}_${action}_${tableName}_table.ts`
 
     const codemods = await this.createCodemods()
+    codemods.overwriteExisting = this.force === true
     await codemods.makeUsingStub(
       stubsRoot,
       `make/migration/${action}.stub`,

commands/make_model.ts

@@ -72,6 +72,12 @@ export default class MakeModel extends BaseCommand {
   @flags.string({ description: 'Use the contents of the given file as the generated output' })
   declare contentsFrom: string
 
+  /**
+   * Forcefully overwrite existing files
+   */
+  @flags.boolean({ description: 'Forcefully overwrite existing files' })
+  declare force: boolean
+
   /**
    * Run migrations
    */
@@ -80,7 +86,8 @@ export default class MakeModel extends BaseCommand {
       return
     }
 
-    const makeMigration = await this.kernel.exec('make:migration', [this.name])
+    const migrationArgs = this.force ? [this.name, '--force'] : [this.name]
+    const makeMigration = await this.kernel.exec('make:migration', migrationArgs)
     this.exitCode = makeMigration.exitCode
     this.error = makeMigration.error
   }
@@ -93,7 +100,8 @@ export default class MakeModel extends BaseCommand {
       return
     }
 
-    const makeController = await this.kernel.exec('make:controller', [this.name])
+    const controllerArgs = this.force ? [this.name, '--force'] : [this.name]
+    const makeController = await this.kernel.exec('make:controller', controllerArgs)
     this.exitCode = makeController.exitCode
     this.error = makeController.error
   }
@@ -106,7 +114,8 @@ export default class MakeModel extends BaseCommand {
       return
     }
 
-    const makeTransformer = await this.kernel.exec('make:transformer', [this.name])
+    const transformerArgs = this.force ? [this.name, '--force'] : [this.name]
+    const makeTransformer = await this.kernel.exec('make:transformer', transformerArgs)
     this.exitCode = makeTransformer.exitCode
     this.error = makeTransformer.error
   }
@@ -119,7 +128,8 @@ export default class MakeModel extends BaseCommand {
       return
     }
 
-    const makeFactory = await this.kernel.exec('make:factory', [this.name])
+    const factoryArgs = this.force ? [this.name, '--force'] : [this.name]
+    const makeFactory = await this.kernel.exec('make:factory', factoryArgs)
     this.exitCode = makeFactory.exitCode
     this.error = makeFactory.error
   }
@@ -129,6 +139,7 @@ export default class MakeModel extends BaseCommand {
    */
   async run(): Promise<void> {
     const codemods = await this.createCodemods()
+    codemods.overwriteExisting = this.force === true
     await codemods.makeUsingStub(
       stubsRoot,
       'make/model/main.stub',

commands/make_seeder.ts

@@ -32,11 +32,18 @@ export default class MakeSeeder extends BaseCommand {
   @flags.string({ description: 'Use the contents of the given file as the generated output' })
   declare contentsFrom: string
 
+  /**
+   * Forcefully overwrite existing files
+   */
+  @flags.boolean({ description: 'Forcefully overwrite existing files', alias: 'f' })
+  declare force: boolean
+
   /**
    * Execute command
    */
   async run(): Promise<void> {
     const codemods = await this.createCodemods()
+    codemods.overwriteExisting = this.force === true
     await codemods.makeUsingStub(
       stubsRoot,
       'make/seeder/main.stub',

src/database/main.ts

@@ -227,8 +227,7 @@ export class Database extends Macroable {
   transaction(options?: { isolationLevel?: IsolationLevels }): Promise<TransactionClientContract>
   transaction<T>(
     callbackOrOptions?:
-      | ((trx: TransactionClientContract) => Promise<T>)
-      | { isolationLevel?: IsolationLevels },
+      ((trx: TransactionClientContract) => Promise<T>) | { isolationLevel?: IsolationLevels },
     options?: { isolationLevel?: IsolationLevels }
   ): Promise<TransactionClientContract | T> {
     const client = this.connection()

src/dialects/mysql.ts

@@ -249,18 +249,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

@@ -252,20 +252,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/migration/schema_dump/schema_states/sqlite.ts

@@ -59,8 +59,7 @@ export class SqliteSchemaState extends BaseSchemaState {
    */
   private getDatabasePath() {
     const connection = this.connectionConfig.connection as
-      | SqliteConfig['connection']
-      | LibSQLConfig['connection']
+      SqliteConfig['connection'] | LibSQLConfig['connection']
     const filename = connection?.filename
 
     if (!filename) {

src/query_client/index.ts

@@ -235,8 +235,7 @@ export class QueryClient implements QueryClientContract {
   transaction(options?: { isolationLevel?: IsolationLevels }): Promise<TransactionClientContract>
   async transaction<T>(
     callback?:
-      | { isolationLevel?: IsolationLevels }
-      | ((trx: TransactionClientContract) => Promise<any>),
+      { isolationLevel?: IsolationLevels } | ((trx: TransactionClientContract) => Promise<any>),
     options?: { isolationLevel?: IsolationLevels }
   ): Promise<TransactionClientContract | T> {
     const trx = await this.getWriteClient().transaction(options)

src/transaction_client/index.ts

@@ -258,8 +258,7 @@ export class TransactionClient extends EventEmitter implements TransactionClient
    */
   async transaction(
     callback?:
-      | { isolationLevel?: IsolationLevels }
-      | ((trx: TransactionClientContract) => Promise<any>),
+      { isolationLevel?: IsolationLevels } | ((trx: TransactionClientContract) => Promise<any>),
     options?: { isolationLevel?: IsolationLevels }
   ): Promise<any> {
     const trx = await this.knexClient.transaction(options)

src/types/database.ts

@@ -29,11 +29,7 @@ import { type OrmSchemaGeneratorConfig } from './schema_generator.ts'
  * type
  */
 export type IsolationLevels =
-  | 'read uncommitted'
-  | 'read committed'
-  | 'snapshot'
-  | 'repeatable read'
-  | 'serializable'
+  'read uncommitted' | 'read committed' | 'snapshot' | 'repeatable read' | 'serializable'
 
 export type ColumnInfo = Knex.ColumnInfo
 
@@ -87,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>
 }
 
 /**
@@ -568,11 +564,7 @@ type MssqlConnectionNode = {
     instanceName?: string
     enableArithAbort?: boolean
     isolationLevel?:
-      | 'READ_UNCOMMITTED'
-      | 'READ_COMMITTED'
-      | 'REPEATABLE_READ'
-      | 'SERIALIZABLE'
-      | 'SNAPSHOT'
+      'READ_UNCOMMITTED' | 'READ_COMMITTED' | 'REPEATABLE_READ' | 'SERIALIZABLE' | 'SNAPSHOT'
     maxRetriesOnTransientErrors?: number
     multiSubnetFailover?: boolean
     packetSize?: number

src/types/model.ts

@@ -97,13 +97,15 @@ export type OptionalTypedDecorator<PropType> = <
  * property.
  */
 export type ModelAttributes<Model extends LucidRow> = {
-  [Filtered in {
-    [P in keyof Model]: P extends keyof LucidRow | 'serializeExtras'
-      ? never
-      : Model[P] extends Function | ModelRelationTypes
+  [
+    Filtered in {
+      [P in keyof Model]: P extends keyof LucidRow | 'serializeExtras'
         ? never
-        : P
-  }[keyof Model]]: Model[Filtered]
+        : Model[P] extends Function | ModelRelationTypes
+          ? never
+          : P
+    }[keyof Model]
+  ]: Model[Filtered]
 }
 
 /**