fix: use exactly once queue semantics for specific events (#755)

Author: fenos

package-lock.json

@@ -76,7 +76,7 @@
     "multistream": "^4.1.0",
     "object-sizeof": "^2.6.4",
     "pg": "^8.12.0",
-    "pg-boss": "^10.2.0",
+    "pg-boss": "github:supabase/pg-boss#feat/exactly_once",
     "pg-listen": "^1.7.0",
     "pino": "^9.7.0",
     "pino-logflare": "^0.5.2",

package.json

@@ -1,10 +1,34 @@
-import { FastifyInstance } from 'fastify'
+import { FastifyInstance, RequestGenericInterface } from 'fastify'
 import apiKey from '../../plugins/apikey'
 import { getConfig } from '../../../config'
-import { UpgradePgBossV10 } from '@storage/events'
+import { MoveJobs, UpgradePgBossV10 } from '@storage/events'
+import { FromSchema } from 'json-schema-to-ts'
 
 const { pgQueueEnable } = getConfig()
 
+const moveJobsSchema = {
+  body: {
+    type: 'object',
+    properties: {
+      fromQueue: {
+        type: 'string',
+      },
+      toQueue: {
+        type: 'string',
+      },
+      deleteJobsFromOriginalQueue: {
+        type: 'boolean',
+        default: false,
+      },
+    },
+    required: ['fromQueue', 'toQueue'],
+  },
+} as const
+
+interface MoveJobsRequestInterface extends RequestGenericInterface {
+  Body: FromSchema<typeof moveJobsSchema.body>
+}
+
 export default async function routes(fastify: FastifyInstance) {
   fastify.register(apiKey)
 
@@ -17,4 +41,22 @@ export default async function routes(fastify: FastifyInstance) {
 
     return reply.send({ message: 'Migration scheduled' })
   })
+
+  fastify.post<MoveJobsRequestInterface>('/move', async (req, reply) => {
+    if (!pgQueueEnable) {
+      return reply.status(400).send({ message: 'Queue is not enabled' })
+    }
+
+    const fromQueue = req.body.fromQueue
+    const toQueue = req.body.toQueue
+    const deleteJobsFromOriginalQueue = req.body.deleteJobsFromOriginalQueue || false
+
+    await MoveJobs.send({
+      fromQueue,
+      toQueue,
+      deleteJobsFromOriginalQueue,
+    })
+
+    return reply.send({ message: 'Move jobs scheduled' })
+  })
 }

src/http/routes/admin/queue.ts

@@ -65,7 +65,7 @@ export abstract class Queue {
 
     Queue.pgBoss = new PgBoss({
       connectionString: url,
-
+      migrate: true,
       db: Queue.pgBossDb,
       schema: 'pgboss_v10',
       application_name: 'storage-pgboss',

src/internal/queue/queue.ts

@@ -12,4 +12,5 @@ export * from './migrations/run-migrations'
 export * from './migrations/reset-migrations'
 export * from './jwks/jwks-create-signing-secret'
 export * from './pgboss/upgrade-v10'
+export * from './pgboss/move-jobs'
 export * from './workers'

src/storage/events/index.ts

@@ -9,12 +9,12 @@ interface JwksCreateSigningSecretPayload extends BasePayload {
 }
 
 export class JwksCreateSigningSecret extends BaseEvent<JwksCreateSigningSecretPayload> {
-  static queueName = 'tenants-jwks-create'
+  static queueName = 'tenants-jwks-create-v2'
 
   static getQueueOptions(): Queue {
     return {
       name: this.queueName,
-      policy: 'stately',
+      policy: 'exactly_once',
     } as const
   }
 

src/storage/events/jwks/jwks-create-signing-secret.ts

@@ -13,12 +13,12 @@ interface ResetMigrationsPayload extends BasePayload {
 }
 
 export class ResetMigrationsOnTenant extends BaseEvent<ResetMigrationsPayload> {
-  static queueName = 'tenants-migrations-reset'
+  static queueName = 'tenants-migrations-reset-v2'
 
   static getQueueOptions(): Queue {
     return {
       name: this.queueName,
-      policy: 'stately',
+      policy: 'exactly_once',
     } as const
   }
 

src/storage/events/migrations/reset-migrations.ts

@@ -17,13 +17,13 @@ interface RunMigrationsPayload extends BasePayload {
 }
 
 export class RunMigrationsOnTenants extends BaseEvent<RunMigrationsPayload> {
-  static queueName = 'tenants-migrations'
+  static queueName = 'tenants-migrations-v2'
   static allowSync = false
 
   static getQueueOptions(): Queue {
     return {
       name: this.queueName,
-      policy: 'stately',
+      policy: 'exactly_once',
     } as const
   }
 

src/storage/events/migrations/run-migrations.ts

@@ -0,0 +1,123 @@
+import { BaseEvent } from '../base-event'
+import { Job, Queue as PgBossQueue, SendOptions, WorkOptions } from 'pg-boss'
+import { BasePayload, Queue } from '@internal/queue'
+import { multitenantKnex } from '@internal/database'
+import { logger, logSchema } from '@internal/monitoring'
+
+interface MoveJobsPayload extends BasePayload {
+  fromQueue: string
+  toQueue: string
+  deleteJobsFromOriginalQueue?: boolean
+}
+
+export class MoveJobs extends BaseEvent<MoveJobsPayload> {
+  static queueName = 'move-jobs'
+
+  static getQueueOptions(): PgBossQueue {
+    return {
+      name: this.queueName,
+      policy: 'exactly_once',
+    } as const
+  }
+
+  static getWorkerOptions(): WorkOptions {
+    return {
+      includeMetadata: true,
+    }
+  }
+
+  static getSendOptions(payload: MoveJobsPayload): SendOptions {
+    return {
+      expireInHours: 2,
+      singletonKey: `move_${payload.fromQueue}_to_${payload.toQueue}`,
+      singletonHours: 12,
+      retryLimit: 3,
+      retryDelay: 5,
+      priority: 10,
+    }
+  }
+
+  static async handle(job: Job<MoveJobsPayload>) {
+    await multitenantKnex.transaction(async (tnx) => {
+      const resultLock = await tnx.raw('SELECT pg_try_advisory_xact_lock(-5525285245963000611)')
+      const lockAcquired = resultLock.rows.shift()?.pg_try_advisory_xact_lock || false
+
+      if (!lockAcquired) {
+        return
+      }
+
+      const schema = 'pgboss_v10'
+      const fromQueueName = job.data.fromQueue
+      const toQueue = await Queue.getInstance().getQueue(job.data.toQueue)
+
+      if (!toQueue) {
+        logSchema.error(logger, `[PgBoss] Target queue ${job.data.toQueue} does not exist`, {
+          type: 'pgboss',
+        })
+        return
+      }
+
+      try {
+        const sql = `
+            INSERT INTO ${schema}.job (
+                id,
+                name,
+                priority,
+                data,
+                retry_limit,
+                retry_count,
+                retry_delay,
+                retry_backoff,
+                start_after,
+                singleton_key,
+                singleton_on,
+                expire_in,
+                created_on,
+                keep_until,
+                output,
+                policy,
+                state
+            )
+            SELECT 
+                id,
+                '${toQueue.name}' as name,
+                priority,
+                data,
+                retry_limit,
+                retry_count,
+                retry_delay,
+                retry_backoff,
+                start_after,
+                singleton_key,
+                singleton_on,
+                expire_in,
+                created_on,
+                keep_until,
+                output,
+                '${toQueue.policy}' as policy,
+                'created' as state
+            FROM ${schema}.job
+            WHERE name = '${fromQueueName}'
+                AND state IN ('created', 'active', 'retry')
+            ON CONFLICT DO NOTHING
+        `
+
+        await tnx.raw(sql)
+
+        if (job.data.deleteJobsFromOriginalQueue) {
+          const deleteSql = `
+                DELETE FROM ${schema}.job
+                WHERE name = '${fromQueueName}'
+                    AND state IN ('created', 'active', 'retry')
+            `
+          await tnx.raw(deleteSql)
+        }
+      } catch (error) {
+        logSchema.error(logger, '[PgBoss] Error while copying jobs', {
+          type: 'pgboss',
+          error,
+        })
+      }
+    })
+  }
+}

src/storage/events/pgboss/move-jobs.ts

@@ -12,7 +12,7 @@ export class UpgradePgBossV10 extends BaseEvent<UpgradePgBossV10Payload> {
   static getQueueOptions(): PgBossQueue {
     return {
       name: this.queueName,
-      policy: 'stately',
+      policy: 'exactly_once',
     } as const
   }