fix: replace mutex with single flight pattern (#1193)

Signed-off-by: ferhat elmas <elmas.ferhat@gmail.com>

Author: ferhatelmas

package-lock.json

@@ -80,7 +80,6 @@
     "@platformatic/globals": "^3.59.0",
     "@platformatic/node": "^3.59.0",
     "@platformatic/wattpm-pprof-capture": "^3.59.0",
-    "@shopify/semaphore": "^3.0.2",
     "@smithy/node-http-handler": "^2.3.1",
     "@tus/file-store": "2.1.0",
     "@tus/s3-store": "2.0.3",

package.json

@@ -1,10 +1,15 @@
 import Fastify, { type FastifyRequest } from 'fastify'
 import { vi } from 'vitest'
+import { MultitenantMigrationStrategy } from '../../config'
 
 async function loadDbPlugins({
   databaseEnableQueryCancellation = false,
+  dbMigrationStrategy = MultitenantMigrationStrategy.PROGRESSIVE,
+  isMultitenant = false,
 }: {
   databaseEnableQueryCancellation?: boolean
+  dbMigrationStrategy?: MultitenantMigrationStrategy
+  isMultitenant?: boolean
 } = {}) {
   vi.resetModules()
 
@@ -19,43 +24,54 @@ async function loadDbPlugins({
       role: 'service_role',
     },
   })
+  const getTenantConfig = vi.fn()
+  const areMigrationsUpToDate = vi.fn()
+  const runMigrationsOnTenant = vi.fn()
+  const updateTenantMigrationsState = vi.fn()
+  const progressiveMigrations = {
+    addTenant: vi.fn(),
+  }
 
   vi.doMock('@internal/database', () => {
     return {
       getPostgresConnection,
       getServiceKeyUser,
-      getTenantConfig: vi.fn(),
+      getTenantConfig,
       PgTenantConnection: class {},
     }
   })
 
   vi.doMock('@internal/database/migrations', () => {
     return {
-      areMigrationsUpToDate: vi.fn(),
+      areMigrationsUpToDate,
       DBMigration: {},
       lastLocalMigrationName: vi.fn().mockResolvedValue('initialmigration'),
-      progressiveMigrations: {
-        addTenant: vi.fn(),
-      },
-      runMigrationsOnTenant: vi.fn(),
-      updateTenantMigrationsState: vi.fn(),
+      progressiveMigrations,
+      runMigrationsOnTenant,
+      updateTenantMigrationsState,
     }
   })
 
   const configModule = await import('../../config')
   configModule.getConfig({ reload: true })
   configModule.mergeConfig({
     databaseEnableQueryCancellation,
-    isMultitenant: false,
+    dbMigrationStrategy,
+    isMultitenant,
   })
 
   const { db, dbSuperUser } = await import('./db')
 
   return {
     db,
     dbSuperUser,
+    areMigrationsUpToDate,
     getPostgresConnection,
+    getTenantConfig,
+    progressiveMigrations,
     requestDb,
+    runMigrationsOnTenant,
+    updateTenantMigrationsState,
   }
 }
 
@@ -99,6 +115,285 @@ describe('dbSuperUser plugin', () => {
   })
 })
 
+describe('migrations plugin', () => {
+  async function buildMigrationApp(
+    plugins: Awaited<ReturnType<typeof loadDbPlugins>>,
+    getTenantId: (request: FastifyRequest) => string = () => 'tenant-id'
+  ) {
+    const app = Fastify()
+
+    app.decorateRequest('tenantId')
+    app.addHook('onRequest', async (request) => {
+      request.tenantId = getTenantId(request)
+    })
+    await app.register(plugins.dbSuperUser)
+    app.get('/test', async () => ({ ok: true }))
+
+    const injectTenant = (headers: Record<string, string> = {}) =>
+      app.inject({ method: 'GET', url: '/test', headers })
+
+    return { app, injectTenant }
+  }
+
+  function waitForImmediate() {
+    return new Promise((resolve) => setImmediate(resolve))
+  }
+
+  afterEach(() => {
+    vi.doUnmock('@internal/database')
+    vi.doUnmock('@internal/database/migrations')
+    vi.restoreAllMocks()
+    vi.resetModules()
+  })
+
+  it('shares the same on-request migration check across concurrent same-tenant requests', async () => {
+    const plugins = await loadDbPlugins({
+      dbMigrationStrategy: MultitenantMigrationStrategy.ON_REQUEST,
+      isMultitenant: true,
+    })
+    const tenant = {
+      databaseUrl: 'postgres://tenant-db',
+      migrationVersion: 'initialmigration',
+      syncMigrationsDone: false,
+    }
+    const migration = Promise.withResolvers<void>()
+
+    plugins.getTenantConfig.mockResolvedValue(tenant)
+    plugins.areMigrationsUpToDate.mockResolvedValue(false)
+    plugins.runMigrationsOnTenant.mockReturnValue(migration.promise)
+    plugins.updateTenantMigrationsState.mockResolvedValue(undefined)
+
+    const { app, injectTenant } = await buildMigrationApp(plugins)
+
+    try {
+      const first = injectTenant()
+      const second = injectTenant()
+
+      await waitForImmediate()
+
+      expect(plugins.areMigrationsUpToDate).toHaveBeenCalledTimes(1)
+      expect(plugins.runMigrationsOnTenant).toHaveBeenCalledTimes(1)
+
+      migration.resolve()
+
+      const responses = await Promise.all([first, second])
+      expect(responses).toEqual([
+        expect.objectContaining({ statusCode: 200 }),
+        expect.objectContaining({ statusCode: 200 }),
+      ])
+      expect(tenant.syncMigrationsDone).toBe(true)
+
+      const third = await injectTenant()
+
+      expect(third.statusCode).toBe(200)
+      expect(plugins.areMigrationsUpToDate).toHaveBeenCalledTimes(1)
+      expect(plugins.runMigrationsOnTenant).toHaveBeenCalledTimes(1)
+    } finally {
+      await app.close()
+    }
+  })
+
+  it('shares the on-request migration check across route scopes', async () => {
+    const plugins = await loadDbPlugins({
+      dbMigrationStrategy: MultitenantMigrationStrategy.ON_REQUEST,
+      isMultitenant: true,
+    })
+    const tenant = {
+      databaseUrl: 'postgres://tenant-db',
+      migrationVersion: 'initialmigration',
+      syncMigrationsDone: false,
+    }
+    const migration = Promise.withResolvers<void>()
+
+    plugins.getTenantConfig.mockResolvedValue(tenant)
+    plugins.areMigrationsUpToDate.mockResolvedValue(false)
+    plugins.runMigrationsOnTenant.mockReturnValue(migration.promise)
+    plugins.updateTenantMigrationsState.mockResolvedValue(undefined)
+
+    // Two separate Fastify apps stand in for two route scopes, each registering
+    // its own copy of the migrations plugin against the same module state.
+    const firstScope = await buildMigrationApp(plugins)
+    const secondScope = await buildMigrationApp(plugins)
+
+    try {
+      const first = firstScope.injectTenant()
+      const second = secondScope.injectTenant()
+
+      await waitForImmediate()
+
+      expect(plugins.runMigrationsOnTenant).toHaveBeenCalledTimes(1)
+
+      migration.resolve()
+
+      await expect(Promise.all([first, second])).resolves.toEqual([
+        expect.objectContaining({ statusCode: 200 }),
+        expect.objectContaining({ statusCode: 200 }),
+      ])
+    } finally {
+      await firstScope.app.close()
+      await secondScope.app.close()
+    }
+  })
+
+  it('shares migration failures across concurrent same-tenant requests and retries later', async () => {
+    const plugins = await loadDbPlugins({
+      dbMigrationStrategy: MultitenantMigrationStrategy.ON_REQUEST,
+      isMultitenant: true,
+    })
+    const tenant = {
+      databaseUrl: 'postgres://tenant-db',
+      migrationVersion: 'initialmigration',
+      syncMigrationsDone: false,
+    }
+    const migration = Promise.withResolvers<void>()
+
+    plugins.getTenantConfig.mockResolvedValue(tenant)
+    plugins.areMigrationsUpToDate.mockResolvedValue(false)
+    plugins.runMigrationsOnTenant
+      .mockReturnValueOnce(migration.promise)
+      .mockResolvedValueOnce(undefined)
+    plugins.updateTenantMigrationsState.mockResolvedValue(undefined)
+
+    const { app, injectTenant } = await buildMigrationApp(plugins)
+
+    try {
+      const first = injectTenant()
+      const second = injectTenant()
+
+      await waitForImmediate()
+
+      expect(plugins.runMigrationsOnTenant).toHaveBeenCalledTimes(1)
+
+      migration.reject(new Error('migration failed'))
+
+      const failedResponses = await Promise.all([first, second])
+      expect(failedResponses).toEqual([
+        expect.objectContaining({ statusCode: 500 }),
+        expect.objectContaining({ statusCode: 500 }),
+      ])
+      expect(tenant.syncMigrationsDone).toBe(false)
+
+      const retry = await injectTenant()
+
+      expect(retry.statusCode).toBe(200)
+      expect(plugins.areMigrationsUpToDate).toHaveBeenCalledTimes(2)
+      expect(plugins.runMigrationsOnTenant).toHaveBeenCalledTimes(2)
+      expect(tenant.syncMigrationsDone).toBe(true)
+    } finally {
+      await app.close()
+    }
+  })
+
+  it('skips on-request migration checks when the tenant is already marked migrated', async () => {
+    const plugins = await loadDbPlugins({
+      dbMigrationStrategy: MultitenantMigrationStrategy.ON_REQUEST,
+      isMultitenant: true,
+    })
+
+    plugins.getTenantConfig.mockResolvedValue({
+      databaseUrl: 'postgres://tenant-db',
+      migrationVersion: 'initialmigration',
+      syncMigrationsDone: true,
+    })
+
+    const { app, injectTenant } = await buildMigrationApp(plugins)
+
+    try {
+      const response = await injectTenant()
+
+      expect(response.statusCode).toBe(200)
+      expect(plugins.areMigrationsUpToDate).not.toHaveBeenCalled()
+      expect(plugins.runMigrationsOnTenant).not.toHaveBeenCalled()
+    } finally {
+      await app.close()
+    }
+  })
+
+  it('skips migration execution when tenant migrations are already up to date', async () => {
+    const plugins = await loadDbPlugins({
+      dbMigrationStrategy: MultitenantMigrationStrategy.ON_REQUEST,
+      isMultitenant: true,
+    })
+
+    plugins.getTenantConfig.mockResolvedValue({
+      databaseUrl: 'postgres://tenant-db',
+      migrationVersion: 'initialmigration',
+      syncMigrationsDone: false,
+    })
+    plugins.areMigrationsUpToDate.mockResolvedValue(true)
+
+    const { app, injectTenant } = await buildMigrationApp(plugins)
+
+    try {
+      const response = await injectTenant()
+
+      expect(response.statusCode).toBe(200)
+      expect(plugins.areMigrationsUpToDate).toHaveBeenCalledTimes(1)
+      expect(plugins.runMigrationsOnTenant).not.toHaveBeenCalled()
+
+      const secondResponse = await injectTenant()
+
+      expect(secondResponse.statusCode).toBe(200)
+      expect(plugins.areMigrationsUpToDate).toHaveBeenCalledTimes(1)
+      expect(plugins.runMigrationsOnTenant).not.toHaveBeenCalled()
+    } finally {
+      await app.close()
+    }
+  })
+
+  it('does not coalesce on-request migrations for different tenants', async () => {
+    const plugins = await loadDbPlugins({
+      dbMigrationStrategy: MultitenantMigrationStrategy.ON_REQUEST,
+      isMultitenant: true,
+    })
+    const migrations = {
+      'tenant-a': Promise.withResolvers<void>(),
+      'tenant-b': Promise.withResolvers<void>(),
+    }
+
+    plugins.getTenantConfig.mockImplementation(async (tenantId: string) => ({
+      databaseUrl: `postgres://${tenantId}`,
+      migrationVersion: 'initialmigration',
+      syncMigrationsDone: false,
+    }))
+    plugins.areMigrationsUpToDate.mockResolvedValue(false)
+    plugins.runMigrationsOnTenant.mockImplementation(
+      ({ tenantId }: { tenantId: keyof typeof migrations }) => migrations[tenantId].promise
+    )
+    plugins.updateTenantMigrationsState.mockResolvedValue(undefined)
+
+    const { app, injectTenant } = await buildMigrationApp(
+      plugins,
+      (request) => request.headers['x-tenant-id'] as string
+    )
+
+    try {
+      const first = injectTenant({ 'x-tenant-id': 'tenant-a' })
+      const second = injectTenant({ 'x-tenant-id': 'tenant-b' })
+
+      await waitForImmediate()
+
+      expect(plugins.runMigrationsOnTenant).toHaveBeenCalledTimes(2)
+      expect(plugins.runMigrationsOnTenant).toHaveBeenCalledWith(
+        expect.objectContaining({ tenantId: 'tenant-a' })
+      )
+      expect(plugins.runMigrationsOnTenant).toHaveBeenCalledWith(
+        expect.objectContaining({ tenantId: 'tenant-b' })
+      )
+
+      migrations['tenant-a'].resolve()
+      migrations['tenant-b'].resolve()
+
+      await expect(Promise.all([first, second])).resolves.toEqual([
+        expect.objectContaining({ statusCode: 200 }),
+        expect.objectContaining({ statusCode: 200 }),
+      ])
+    } finally {
+      await app.close()
+    }
+  })
+})
+
 describe.each([
   {
     name: 'db plugin',