fix: configure pgBoss from env

supabase · Oct 11, 2023 · bfa4b99 · bfa4b99
1 parent 9007206
commit bfa4b99
Show file tree

Hide file tree

Showing 5 changed files with 81 additions and 115 deletions.
diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -55,7 +55,7 @@
     "knex": "^2.4.2",
     "md5-file": "^5.0.0",
     "pg": "^8.10.0",
-    "pg-boss": "^8.1.1",
+    "pg-boss": "^9.0.3",
     "pg-listen": "^1.7.0",
     "pino": "^8.15.4",
     "pino-logflare": "^0.4.2",

diff --git a/src/config.ts b/src/config.ts
@@ -40,6 +40,9 @@ type StorageConfigType = {
   logflareSourceToken?: string
   enableQueueEvents: boolean
   pgQueueConnectionURL?: string
+  pgQueueDeleteAfterDays?: number
+  pgQueueArchiveCompletedAfterSeconds?: number
+  pgQueueRetentionDays?: number
   webhookURL?: string
   webhookApiKey?: string
   webhookQueuePullInterval?: number
@@ -147,6 +150,15 @@ export function getConfig(): StorageConfigType {
     logflareSourceToken: getOptionalConfigFromEnv('LOGFLARE_SOURCE_TOKEN'),
     enableQueueEvents: getOptionalConfigFromEnv('ENABLE_QUEUE_EVENTS') === 'true',
     pgQueueConnectionURL: getOptionalConfigFromEnv('PG_QUEUE_CONNECTION_URL'),
+    pgQueueDeleteAfterDays: parseInt(
+      getOptionalConfigFromEnv('PG_QUEUE_DELETE_AFTER_DAYS') || '2',
+      10
+    ),
+    pgQueueArchiveCompletedAfterSeconds: parseInt(
+      getOptionalConfigFromEnv('PG_QUEUE_ARCHIVE_COMPLETED_AFTER_SECONDS') || '7200',
+      10
+    ),
+    pgQueueRetentionDays: parseInt(getOptionalConfigFromEnv('PG_QUEUE_RETENTION_DAYS') || '2', 10),
     webhookURL: getOptionalConfigFromEnv('WEBHOOK_URL'),
     webhookApiKey: getOptionalConfigFromEnv('WEBHOOK_API_KEY'),
     webhookQueuePullInterval: parseInt(

diff --git a/src/http/error-handler.ts b/src/http/error-handler.ts
@@ -32,6 +32,7 @@ export const setErrorHandler = (app: FastifyInstance) => {
         'remaining connection slots are reserved for non-replication superuser connections',
         'no more connections allowed',
         'sorry, too many clients already',
+        'server login has been failing, try again later',
       ].some((msg) => (error as DatabaseError).message.includes(msg))
     ) {
       return reply.status(429).send({

diff --git a/src/queue/queue.ts b/src/queue/queue.ts
@@ -19,7 +19,14 @@ export abstract class Queue {
       return Queue.pgBoss
     }
 
-    const { isMultitenant, multitenantDatabaseUrl, pgQueueConnectionURL } = getConfig()
+    const {
+      isMultitenant,
+      multitenantDatabaseUrl,
+      pgQueueConnectionURL,
+      pgQueueDeleteAfterDays,
+      pgQueueArchiveCompletedAfterSeconds,
+      pgQueueRetentionDays,
+    } = getConfig()
 
     let url = pgQueueConnectionURL ?? process.env.DATABASE_URL
 
@@ -31,13 +38,12 @@ export abstract class Queue {
       connectionString: url,
       max: 4,
       application_name: 'storage-pgboss',
-      deleteAfterDays: 7,
-      archiveCompletedAfterSeconds: 14_400,
-      retentionDays: 7,
+      deleteAfterDays: pgQueueDeleteAfterDays,
+      archiveCompletedAfterSeconds: pgQueueArchiveCompletedAfterSeconds,
+      retentionDays: pgQueueRetentionDays,
       retryBackoff: true,
       retryLimit: 20,
       expireInHours: 48,
-      monitorStateIntervalSeconds: 30,
     })
 
     registerWorkers()