From 10876e839699a921ac4d7ffa832824dc1f776852 Mon Sep 17 00:00:00 2001 From: Tim Jones Date: Fri, 3 Jul 2026 10:19:13 -0500 Subject: [PATCH] feat: upgrade pg-boss from custom fork to upstream 12.25.0 The supabase/pg-boss fork existed to add the exactly_once queue policy on top of pg-boss 10.3.2. That feature was merged upstream in 11.1.0 and is now called the "exclusive" policy, so the fork is no longer needed. Because the fork's schema version 25 collides with upstream's v11 schema rewrite (also version 25), the queue moves to a fresh pgboss_v12 schema, mirroring the previous pgboss -> pgboss_v10 upgrade. A new UpgradePgBossV12 event (triggered via POST /queue/migrate/pgboss-v12) copies pending jobs from pgboss_v10, mapping exactly_once -> exclusive and expire_in intervals -> expire_seconds. Breaking API changes handled: - exactly_once policy renamed to exclusive - constructor-level retention/expiration/retry options moved to per-queue options in seconds (deleteAfterDays/Hours -> deleteAfterSeconds, retentionDays -> retentionSeconds, expireInHours -> expireInSeconds); archiveCompletedAfterSeconds removed (v11+ has no archive table) - send options expireInHours/expireInMinutes/singletonHours replaced with expireInSeconds/singletonSeconds - insert() is now per-queue: insert(name, jobs) - getQueueSize() replaced by getQueueStats() (returns snapshots with queuedCount; queue existence checked via getQueue()) - stop() no longer takes wait (it always waits) - PgBoss is a named export; Db/QueueOptions/JobInsert are named types - Job requires heartbeatSeconds and signal (sync-invoke jobs updated) - externally provided Db connections must be opened by the caller (the isOurs auto-open hack no longer works) - job table column expire_in (interval) is now expire_seconds (int) - deep requires move from pg-boss/src/* to pg-boss/dist/*, and the schema version lives in package.json pgboss.schema Co-Authored-By: Claude Fable 5 --- .env.sample | 1 - package-lock.json | 217 +++++++++++------- package.json | 2 +- src/config.ts | 5 - src/http/routes/admin/migrations.ts | 11 +- src/http/routes/admin/queue.test.ts | 50 +++- src/http/routes/admin/queue.ts | 15 +- .../database/migrations/progressive.test.ts | 4 +- src/internal/queue/database.test.ts | 205 +++++++++++------ src/internal/queue/event.test.ts | 3 +- src/internal/queue/event.ts | 57 ++--- src/internal/queue/queue.test.ts | 1 - src/internal/queue/queue.ts | 65 ++++-- .../iceberg/delete-iceberg-resources.ts | 7 +- .../events/iceberg/reconcile-catalog.ts | 6 +- src/storage/events/index.ts | 1 + .../events/jwks/jwks-create-signing-secret.ts | 4 +- .../events/jwks/jwks-roll-url-signing-key.ts | 4 +- .../events/migrations/reset-migrations.ts | 4 +- .../events/migrations/run-migrations.ts | 6 +- src/storage/events/pgboss/move-jobs.test.ts | 2 +- src/storage/events/pgboss/move-jobs.ts | 10 +- src/storage/events/pgboss/pg-branches.test.ts | 51 +++- src/storage/events/pgboss/upgrade-v12.ts | 149 ++++++++++++ src/storage/events/upgrades/base-event.ts | 6 +- src/storage/events/workers.ts | 2 + src/test/admin-migrations.test.ts | 8 +- src/test/tenant-jwks.test.ts | 5 +- 28 files changed, 651 insertions(+), 250 deletions(-) create mode 100644 src/storage/events/pgboss/upgrade-v12.ts diff --git a/.env.sample b/.env.sample index d31f4dcac..e930d2297 100644 --- a/.env.sample +++ b/.env.sample @@ -145,7 +145,6 @@ RATE_LIMITER_REDIS_COMMAND_TIMEOUT=2 ####################################### PG_QUEUE_ENABLE=false PG_QUEUE_DELETE_AFTER_DAYS=2 -PG_QUEUE_ARCHIVE_COMPLETED_AFTER_SECONDS=7200 PG_QUEUE_RETENTION_DAYS=2 PG_QUEUE_CONNECTION_URL= diff --git a/package-lock.json b/package-lock.json index 81b548069..160857224 100644 --- a/package-lock.json +++ b/package-lock.json @@ -59,7 +59,7 @@ "json-bigint": "^1.0.0", "lru-cache": "^11.2.0", "pg": "^8.12.0", - "pg-boss": "github:supabase/pg-boss#feat/exactly_once", + "pg-boss": "12.25.0", "pg-listen": "^1.7.0", "pino": "^10.3.1", "pino-logflare": "^0.5.2", @@ -12547,14 +12547,15 @@ } }, "node_modules/cron-parser": { - "version": "4.9.0", - "resolved": "https://registry.npmjs.org/cron-parser/-/cron-parser-4.9.0.tgz", - "integrity": "sha512-p0SaNjrHOnQeR8/VnfGbmg9te2kfyYSQ7Sc/j/6DtPL3JQvKxmjO9TSjNFpujqV3vEYYBvNNvXSxzyksBWAx1Q==", + "version": "5.6.1", + "resolved": "https://registry.npmjs.org/cron-parser/-/cron-parser-5.6.1.tgz", + "integrity": "sha512-QBm4o1PwZiuY7KFbVvW7FLC8bozy7YWzv+Fz6KRS7sQghzcbDZCGxr/Bc5b6TQreAoSwuWVP491dIcK0THCX6A==", + "license": "MIT", "dependencies": { - "luxon": "^3.2.1" + "luxon": "^3.7.2" }, "engines": { - "node": ">=12.0.0" + "node": ">=18" } }, "node_modules/cross-spawn": { @@ -14674,6 +14675,18 @@ "node-gyp-build-test": "build-test.js" } }, + "node_modules/non-error": { + "version": "0.1.0", + "resolved": "https://registry.npmjs.org/non-error/-/non-error-0.1.0.tgz", + "integrity": "sha512-TMB1uHiGsHRGv1uYclfhivcnf0/PdFp2pNqRxXjncaAsjYMoisaQJI+SSZCqRq+VliwRTC8tsMQfmrWjDMhkPQ==", + "license": "MIT", + "engines": { + "node": ">=20" + }, + "funding": { + "url": "https://github.com/sponsors/sindresorhus" + } + }, "node_modules/npm-run-path": { "version": "6.0.0", "resolved": "https://registry.npmjs.org/npm-run-path/-/npm-run-path-6.0.0.tgz", @@ -15026,13 +15039,14 @@ "license": "MIT" }, "node_modules/pg": { - "version": "8.16.3", - "resolved": "https://registry.npmjs.org/pg/-/pg-8.16.3.tgz", - "integrity": "sha512-enxc1h0jA/aq5oSDMvqyW3q89ra6XIIDZgCX9vkMrnz5DFTw/Ny3Li2lFQ+pt3L6MCgm/5o2o8HW9hiJji+xvw==", + "version": "8.22.0", + "resolved": "https://registry.npmjs.org/pg/-/pg-8.22.0.tgz", + "integrity": "sha512-8wih1vVIBMxoUM2oB4soJsD9tDnDpLv4OXBJ+EJzFsvycD+lfyIreC2gGHq78f8jbLLt+bvlPTFdFZfJkOuzAA==", + "license": "MIT", "dependencies": { - "pg-connection-string": "^2.9.1", - "pg-pool": "^3.10.1", - "pg-protocol": "^1.10.3", + "pg-connection-string": "^2.14.0", + "pg-pool": "^3.14.0", + "pg-protocol": "^1.15.0", "pg-types": "2.2.0", "pgpass": "1.0.5" }, @@ -15040,7 +15054,7 @@ "node": ">= 16.0.0" }, "optionalDependencies": { - "pg-cloudflare": "^1.2.7" + "pg-cloudflare": "^1.4.0" }, "peerDependencies": { "pg-native": ">=3.0.1" @@ -15052,24 +15066,35 @@ } }, "node_modules/pg-boss": { - "version": "10.3.2", - "resolved": "git+ssh://git@github.com/supabase/pg-boss.git#89b675b057ff7cec3ca2c434e4d3005c35a76bd9", + "version": "12.25.0", + "resolved": "https://registry.npmjs.org/pg-boss/-/pg-boss-12.25.0.tgz", + "integrity": "sha512-m0kpRsrj5j3hGtzslQzLo/5EfSDtQlNf2N6uvCft3tPHrbSXqdxJRzjb675gf3ZnWcTsu6IIYjCN5zezkZQ+WQ==", "license": "MIT", "dependencies": { - "cron-parser": "^4.9.0", - "pg": "^8.16.0", - "serialize-error": "^8.1.0" + "cron-parser": "^5.6.1", + "pg": "^8.22.0", + "serialize-error": "^13.0.1" + }, + "bin": { + "pg-boss": "dist/cli.js" }, "engines": { - "node": ">=20" + "node": ">=22.12.0" } }, "node_modules/pg-cloudflare": { - "version": "1.2.7", - "resolved": "https://registry.npmjs.org/pg-cloudflare/-/pg-cloudflare-1.2.7.tgz", - "integrity": "sha512-YgCtzMH0ptvZJslLM1ffsY4EuGaU0cx4XSdXLRFae8bPP4dS5xL1tNB3k2o/N64cHJpwU7dxKli/nZ2lUa5fLg==", + "version": "1.4.0", + "resolved": "https://registry.npmjs.org/pg-cloudflare/-/pg-cloudflare-1.4.0.tgz", + "integrity": "sha512-Vo7z/6rrQYxpNRylp4Tlob2elzbh+N/MOQbxFVWCxS7oEx6jF53GTJFxK2WWpKuBRkmiin4Mt+xofFDjx09R0A==", + "license": "MIT", "optional": true }, + "node_modules/pg-connection-string": { + "version": "2.14.0", + "resolved": "https://registry.npmjs.org/pg-connection-string/-/pg-connection-string-2.14.0.tgz", + "integrity": "sha512-XwWDGcLRGCXAR8F/AM5bG7Q+A3Wm2s6QeEjlOKZLlH3UYcguiqCWKyWXVag5TLTIjR7oOJUY8kcADaZgWPyLeg==", + "license": "MIT" + }, "node_modules/pg-format": { "version": "1.0.4", "resolved": "https://registry.npmjs.org/pg-format/-/pg-format-1.0.4.tgz", @@ -15100,17 +15125,19 @@ } }, "node_modules/pg-pool": { - "version": "3.10.1", - "resolved": "https://registry.npmjs.org/pg-pool/-/pg-pool-3.10.1.tgz", - "integrity": "sha512-Tu8jMlcX+9d8+QVzKIvM/uJtp07PKr82IUOYEphaWcoBhIYkoHpLXN3qO59nAI11ripznDsEzEv8nUxBVWajGg==", + "version": "3.14.0", + "resolved": "https://registry.npmjs.org/pg-pool/-/pg-pool-3.14.0.tgz", + "integrity": "sha512-gKtPkFdQPU3DksooVLi9LsjZxrsBUZIpa+7aVx+LV5pNh0KzP4Zleud2po+ConrxbuXGBJ6Hfer6hdgpIBpBaw==", + "license": "MIT", "peerDependencies": { "pg": ">=8.0" } }, "node_modules/pg-protocol": { - "version": "1.10.3", - "resolved": "https://registry.npmjs.org/pg-protocol/-/pg-protocol-1.10.3.tgz", - "integrity": "sha512-6DIBgBQaTKDJyxnXaLiLR8wBpQQcGWuAESkRBX/t6OwA8YsqP+iVSiond2EDy6Y/dsGk8rh/jtax3js5NeV7JQ==" + "version": "1.15.0", + "resolved": "https://registry.npmjs.org/pg-protocol/-/pg-protocol-1.15.0.tgz", + "integrity": "sha512-cq9sECI5s0+uPUXjbz8ioyPJni6RzsRib0US67i5IoTZKw8fNeYlVE7u8F4dG7vEJJtc5wdD1K189lCCUwqWTQ==", + "license": "MIT" }, "node_modules/pg-types": { "version": "2.2.0", @@ -15127,11 +15154,6 @@ "node": ">=4" } }, - "node_modules/pg/node_modules/pg-connection-string": { - "version": "2.9.1", - "resolved": "https://registry.npmjs.org/pg-connection-string/-/pg-connection-string-2.9.1.tgz", - "integrity": "sha512-nkc6NpDcvPVpZXxrreI/FOtX3XemeLl8E0qFr6F2Lrm/I8WOnaWNhIPK2Z7OHpw7gh5XJThi6j6ppgNoaT1w4w==" - }, "node_modules/pgpass": { "version": "1.0.5", "resolved": "https://registry.npmjs.org/pgpass/-/pgpass-1.0.5.tgz", @@ -15962,14 +15984,16 @@ } }, "node_modules/serialize-error": { - "version": "8.1.0", - "resolved": "https://registry.npmjs.org/serialize-error/-/serialize-error-8.1.0.tgz", - "integrity": "sha512-3NnuWfM6vBYoy5gZFvHiYsVbafvI9vZv/+jlIigFn4oP4zjNPK3LhcY0xSCgeb1a5L8jO71Mit9LlNoi2UfDDQ==", + "version": "13.0.1", + "resolved": "https://registry.npmjs.org/serialize-error/-/serialize-error-13.0.1.tgz", + "integrity": "sha512-bBZaRwLH9PN5HbLCjPId4dP5bNGEtumcErgOX952IsvOhVPrm3/AeK1y0UHA/QaPG701eg0yEnOKsCOC6X/kaA==", + "license": "MIT", "dependencies": { - "type-fest": "^0.20.2" + "non-error": "^0.1.0", + "type-fest": "^5.4.1" }, "engines": { - "node": ">=10" + "node": ">=20" }, "funding": { "url": "https://github.com/sponsors/sindresorhus" @@ -16329,6 +16353,18 @@ "node": ">=10.0.0" } }, + "node_modules/tagged-tag": { + "version": "1.0.0", + "resolved": "https://registry.npmjs.org/tagged-tag/-/tagged-tag-1.0.0.tgz", + "integrity": "sha512-yEFYrVhod+hdNyx7g5Bnkkb0G6si8HJurOoOEgC8B/O0uXLHlaey/65KRv6cuWBNhBgHKAROVpc7QyYqE5gFng==", + "license": "MIT", + "engines": { + "node": ">=20" + }, + "funding": { + "url": "https://github.com/sponsors/sindresorhus" + } + }, "node_modules/tar-fs": { "version": "3.1.1", "resolved": "https://registry.npmjs.org/tar-fs/-/tar-fs-3.1.1.tgz", @@ -17046,11 +17082,15 @@ } }, "node_modules/type-fest": { - "version": "0.20.2", - "resolved": "https://registry.npmjs.org/type-fest/-/type-fest-0.20.2.tgz", - "integrity": "sha512-Ne+eE4r0/iWnpAxD852z3A+N0Bt5RN//NjJwRd2VFHEmrywxf5vsZlh4R6lixl6B+wz/8d+maTSAkN1FIkI3LQ==", + "version": "5.7.0", + "resolved": "https://registry.npmjs.org/type-fest/-/type-fest-5.7.0.tgz", + "integrity": "sha512-1URUxUqfHFM1c+zfSPsa3gnkO7Aq21qyH75SIduNYz4SzY964rn1X2vCMQaHSHhktiw+0kPa2iyb6PUpXqB6Vg==", + "license": "(MIT OR CC0-1.0)", + "dependencies": { + "tagged-tag": "^1.0.0" + }, "engines": { - "node": ">=10" + "node": ">=20" }, "funding": { "url": "https://github.com/sponsors/sindresorhus" @@ -26159,11 +26199,11 @@ } }, "cron-parser": { - "version": "4.9.0", - "resolved": "https://registry.npmjs.org/cron-parser/-/cron-parser-4.9.0.tgz", - "integrity": "sha512-p0SaNjrHOnQeR8/VnfGbmg9te2kfyYSQ7Sc/j/6DtPL3JQvKxmjO9TSjNFpujqV3vEYYBvNNvXSxzyksBWAx1Q==", + "version": "5.6.1", + "resolved": "https://registry.npmjs.org/cron-parser/-/cron-parser-5.6.1.tgz", + "integrity": "sha512-QBm4o1PwZiuY7KFbVvW7FLC8bozy7YWzv+Fz6KRS7sQghzcbDZCGxr/Bc5b6TQreAoSwuWVP491dIcK0THCX6A==", "requires": { - "luxon": "^3.2.1" + "luxon": "^3.7.2" } }, "cross-spawn": { @@ -27470,6 +27510,11 @@ "resolved": "https://registry.npmjs.org/node-gyp-build/-/node-gyp-build-3.9.0.tgz", "integrity": "sha512-zLcTg6P4AbcHPq465ZMFNXx7XpKKJh+7kkN699NiQWisR2uWYOWNWqRHAmbnmKiL4e9aLSlmy5U7rEMUXV59+A==" }, + "non-error": { + "version": "0.1.0", + "resolved": "https://registry.npmjs.org/non-error/-/non-error-0.1.0.tgz", + "integrity": "sha512-TMB1uHiGsHRGv1uYclfhivcnf0/PdFp2pNqRxXjncaAsjYMoisaQJI+SSZCqRq+VliwRTC8tsMQfmrWjDMhkPQ==" + }, "npm-run-path": { "version": "6.0.0", "resolved": "https://registry.npmjs.org/npm-run-path/-/npm-run-path-6.0.0.tgz", @@ -27694,40 +27739,39 @@ "dev": true }, "pg": { - "version": "8.16.3", - "resolved": "https://registry.npmjs.org/pg/-/pg-8.16.3.tgz", - "integrity": "sha512-enxc1h0jA/aq5oSDMvqyW3q89ra6XIIDZgCX9vkMrnz5DFTw/Ny3Li2lFQ+pt3L6MCgm/5o2o8HW9hiJji+xvw==", - "requires": { - "pg-cloudflare": "^1.2.7", - "pg-connection-string": "^2.9.1", - "pg-pool": "^3.10.1", - "pg-protocol": "^1.10.3", + "version": "8.22.0", + "resolved": "https://registry.npmjs.org/pg/-/pg-8.22.0.tgz", + "integrity": "sha512-8wih1vVIBMxoUM2oB4soJsD9tDnDpLv4OXBJ+EJzFsvycD+lfyIreC2gGHq78f8jbLLt+bvlPTFdFZfJkOuzAA==", + "requires": { + "pg-cloudflare": "^1.4.0", + "pg-connection-string": "^2.14.0", + "pg-pool": "^3.14.0", + "pg-protocol": "^1.15.0", "pg-types": "2.2.0", "pgpass": "1.0.5" - }, - "dependencies": { - "pg-connection-string": { - "version": "2.9.1", - "resolved": "https://registry.npmjs.org/pg-connection-string/-/pg-connection-string-2.9.1.tgz", - "integrity": "sha512-nkc6NpDcvPVpZXxrreI/FOtX3XemeLl8E0qFr6F2Lrm/I8WOnaWNhIPK2Z7OHpw7gh5XJThi6j6ppgNoaT1w4w==" - } } }, "pg-boss": { - "version": "git+ssh://git@github.com/supabase/pg-boss.git#89b675b057ff7cec3ca2c434e4d3005c35a76bd9", - "from": "pg-boss@github:supabase/pg-boss#feat/exactly_once", + "version": "12.25.0", + "resolved": "https://registry.npmjs.org/pg-boss/-/pg-boss-12.25.0.tgz", + "integrity": "sha512-m0kpRsrj5j3hGtzslQzLo/5EfSDtQlNf2N6uvCft3tPHrbSXqdxJRzjb675gf3ZnWcTsu6IIYjCN5zezkZQ+WQ==", "requires": { - "cron-parser": "^4.9.0", - "pg": "^8.16.0", - "serialize-error": "^8.1.0" + "cron-parser": "^5.6.1", + "pg": "^8.22.0", + "serialize-error": "^13.0.1" } }, "pg-cloudflare": { - "version": "1.2.7", - "resolved": "https://registry.npmjs.org/pg-cloudflare/-/pg-cloudflare-1.2.7.tgz", - "integrity": "sha512-YgCtzMH0ptvZJslLM1ffsY4EuGaU0cx4XSdXLRFae8bPP4dS5xL1tNB3k2o/N64cHJpwU7dxKli/nZ2lUa5fLg==", + "version": "1.4.0", + "resolved": "https://registry.npmjs.org/pg-cloudflare/-/pg-cloudflare-1.4.0.tgz", + "integrity": "sha512-Vo7z/6rrQYxpNRylp4Tlob2elzbh+N/MOQbxFVWCxS7oEx6jF53GTJFxK2WWpKuBRkmiin4Mt+xofFDjx09R0A==", "optional": true }, + "pg-connection-string": { + "version": "2.14.0", + "resolved": "https://registry.npmjs.org/pg-connection-string/-/pg-connection-string-2.14.0.tgz", + "integrity": "sha512-XwWDGcLRGCXAR8F/AM5bG7Q+A3Wm2s6QeEjlOKZLlH3UYcguiqCWKyWXVag5TLTIjR7oOJUY8kcADaZgWPyLeg==" + }, "pg-format": { "version": "1.0.4", "resolved": "https://registry.npmjs.org/pg-format/-/pg-format-1.0.4.tgz", @@ -27749,15 +27793,15 @@ } }, "pg-pool": { - "version": "3.10.1", - "resolved": "https://registry.npmjs.org/pg-pool/-/pg-pool-3.10.1.tgz", - "integrity": "sha512-Tu8jMlcX+9d8+QVzKIvM/uJtp07PKr82IUOYEphaWcoBhIYkoHpLXN3qO59nAI11ripznDsEzEv8nUxBVWajGg==", + "version": "3.14.0", + "resolved": "https://registry.npmjs.org/pg-pool/-/pg-pool-3.14.0.tgz", + "integrity": "sha512-gKtPkFdQPU3DksooVLi9LsjZxrsBUZIpa+7aVx+LV5pNh0KzP4Zleud2po+ConrxbuXGBJ6Hfer6hdgpIBpBaw==", "requires": {} }, "pg-protocol": { - "version": "1.10.3", - "resolved": "https://registry.npmjs.org/pg-protocol/-/pg-protocol-1.10.3.tgz", - "integrity": "sha512-6DIBgBQaTKDJyxnXaLiLR8wBpQQcGWuAESkRBX/t6OwA8YsqP+iVSiond2EDy6Y/dsGk8rh/jtax3js5NeV7JQ==" + "version": "1.15.0", + "resolved": "https://registry.npmjs.org/pg-protocol/-/pg-protocol-1.15.0.tgz", + "integrity": "sha512-cq9sECI5s0+uPUXjbz8ioyPJni6RzsRib0US67i5IoTZKw8fNeYlVE7u8F4dG7vEJJtc5wdD1K189lCCUwqWTQ==" }, "pg-types": { "version": "2.2.0", @@ -28326,11 +28370,12 @@ "integrity": "sha512-vFKC2IEtQnVhpT78h1Yp8wzwrf8CM+MzKMHGJZfBtzhZNycRFnXsHk6E5TxIkkMsgNS7mdX3AGB7x2QM2di4lA==" }, "serialize-error": { - "version": "8.1.0", - "resolved": "https://registry.npmjs.org/serialize-error/-/serialize-error-8.1.0.tgz", - "integrity": "sha512-3NnuWfM6vBYoy5gZFvHiYsVbafvI9vZv/+jlIigFn4oP4zjNPK3LhcY0xSCgeb1a5L8jO71Mit9LlNoi2UfDDQ==", + "version": "13.0.1", + "resolved": "https://registry.npmjs.org/serialize-error/-/serialize-error-13.0.1.tgz", + "integrity": "sha512-bBZaRwLH9PN5HbLCjPId4dP5bNGEtumcErgOX952IsvOhVPrm3/AeK1y0UHA/QaPG701eg0yEnOKsCOC6X/kaA==", "requires": { - "type-fest": "^0.20.2" + "non-error": "^0.1.0", + "type-fest": "^5.4.1" } }, "set-cookie-parser": { @@ -28572,6 +28617,11 @@ "strip-ansi": "^6.0.1" } }, + "tagged-tag": { + "version": "1.0.0", + "resolved": "https://registry.npmjs.org/tagged-tag/-/tagged-tag-1.0.0.tgz", + "integrity": "sha512-yEFYrVhod+hdNyx7g5Bnkkb0G6si8HJurOoOEgC8B/O0uXLHlaey/65KRv6cuWBNhBgHKAROVpc7QyYqE5gFng==" + }, "tar-fs": { "version": "3.1.1", "resolved": "https://registry.npmjs.org/tar-fs/-/tar-fs-3.1.1.tgz", @@ -28957,9 +29007,12 @@ } }, "type-fest": { - "version": "0.20.2", - "resolved": "https://registry.npmjs.org/type-fest/-/type-fest-0.20.2.tgz", - "integrity": "sha512-Ne+eE4r0/iWnpAxD852z3A+N0Bt5RN//NjJwRd2VFHEmrywxf5vsZlh4R6lixl6B+wz/8d+maTSAkN1FIkI3LQ==" + "version": "5.7.0", + "resolved": "https://registry.npmjs.org/type-fest/-/type-fest-5.7.0.tgz", + "integrity": "sha512-1URUxUqfHFM1c+zfSPsa3gnkO7Aq21qyH75SIduNYz4SzY964rn1X2vCMQaHSHhktiw+0kPa2iyb6PUpXqB6Vg==", + "requires": { + "tagged-tag": "^1.0.0" + } }, "typed-emitter": { "version": "0.1.0", diff --git a/package.json b/package.json index 832a0f28e..f0f1f3d9e 100644 --- a/package.json +++ b/package.json @@ -101,7 +101,7 @@ "json-bigint": "^1.0.0", "lru-cache": "^11.2.0", "pg": "^8.12.0", - "pg-boss": "github:supabase/pg-boss#feat/exactly_once", + "pg-boss": "12.25.0", "pg-listen": "^1.7.0", "pino": "^10.3.1", "pino-logflare": "^0.5.2", diff --git a/src/config.ts b/src/config.ts index 5147d9d4d..93439bac5 100644 --- a/src/config.ts +++ b/src/config.ts @@ -138,7 +138,6 @@ type StorageConfigType = { pgQueueConnectionURL?: string pgQueueDeleteAfterHours?: number pgQueueDeleteAfterDays?: number - pgQueueArchiveCompletedAfterSeconds?: number pgQueueRetentionDays?: number pgQueueConcurrentTasksPerQueue: number webhookURL?: string @@ -563,10 +562,6 @@ export function getConfig(options?: { reload?: boolean }): StorageConfigType { ), pgQueueDeleteAfterHours: envNumber(getOptionalConfigFromEnv('PG_QUEUE_DELETE_AFTER_HOURS')) || undefined, - pgQueueArchiveCompletedAfterSeconds: parseInt( - getOptionalConfigFromEnv('PG_QUEUE_ARCHIVE_COMPLETED_AFTER_SECONDS') || '7200', - 10 - ), pgQueueRetentionDays: parseInt(getOptionalConfigFromEnv('PG_QUEUE_RETENTION_DAYS') || '2', 10), pgQueueConcurrentTasksPerQueue: parseInt( getOptionalConfigFromEnv('PG_QUEUE_CONCURRENT_TASKS_PER_QUEUE') || '50', diff --git a/src/http/routes/admin/migrations.ts b/src/http/routes/admin/migrations.ts index dd9184d8f..e11c29770 100644 --- a/src/http/routes/admin/migrations.ts +++ b/src/http/routes/admin/migrations.ts @@ -98,8 +98,15 @@ export default async function routes(fastify: FastifyInstance) { if (!pgQueueEnable) { return reply.code(400).send({ message: 'Queue is not enabled' }) } - const queueSize = await Queue.getInstance().getQueueSize(RunMigrationsOnTenants.getQueueName()) - return { remaining: queueSize } + const queueName = RunMigrationsOnTenants.getQueueName() + const queue = await Queue.getInstance().getQueue(queueName) + + if (!queue) { + return { remaining: 0 } + } + + const stats = await Queue.getInstance().getQueueStats(queueName) + return { remaining: stats.at(-1)?.queuedCount ?? 0 } }) fastify.get( diff --git a/src/http/routes/admin/queue.test.ts b/src/http/routes/admin/queue.test.ts index 78c51491a..14e3e6a9c 100644 --- a/src/http/routes/admin/queue.test.ts +++ b/src/http/routes/admin/queue.test.ts @@ -1,14 +1,18 @@ import { SYSTEM_TENANT } from '@internal/queue' import { vi } from 'vitest' -const { mockMoveJobsSend } = vi.hoisted(() => ({ +const { mockMoveJobsSend, mockUpgradeSend } = vi.hoisted(() => ({ mockMoveJobsSend: vi.fn(), + mockUpgradeSend: vi.fn(), })) vi.mock('@storage/events', () => ({ MoveJobs: { send: mockMoveJobsSend, }, + UpgradePgBossV12: { + send: mockUpgradeSend, + }, })) describe('admin queue routes', () => { @@ -68,6 +72,50 @@ describe('admin queue routes', () => { } }) + it('schedules the pgboss v12 migration', async () => { + vi.resetModules() + + const { mergeConfig } = await import('../../../config') + mergeConfig({ + pgQueueEnable: true, + adminApiKeys: 'test-admin-key', + }) + + const fastify = (await import('fastify')).default + const { default: routes } = await import('./queue') + + const app = fastify() + app.decorateRequest('sbReqId', undefined) + app.addHook('onRequest', (request, _reply, done) => { + request.sbReqId = + typeof request.headers['sb-request-id'] === 'string' + ? request.headers['sb-request-id'] + : undefined + done() + }) + app.register(routes, { prefix: '/queue' }) + + try { + const response = await app.inject({ + method: 'POST', + url: '/queue/migrate/pgboss-v12', + headers: { + apikey: 'test-admin-key', + 'sb-request-id': 'sb-req-456', + }, + }) + + expect(response.statusCode).toBe(200) + expect(response.json()).toEqual({ message: 'Migration scheduled' }) + expect(mockUpgradeSend).toHaveBeenCalledWith({ + sbReqId: 'sb-req-456', + tenant: SYSTEM_TENANT, + }) + } finally { + await app.close() + } + }) + it('rejects move jobs requests without queue names', async () => { vi.resetModules() diff --git a/src/http/routes/admin/queue.ts b/src/http/routes/admin/queue.ts index bd8548020..bacb82e21 100644 --- a/src/http/routes/admin/queue.ts +++ b/src/http/routes/admin/queue.ts @@ -1,5 +1,5 @@ import { SYSTEM_TENANT } from '@internal/queue/constants' -import { MoveJobs } from '@storage/events' +import { MoveJobs, UpgradePgBossV12 } from '@storage/events' import { FastifyInstance, RequestGenericInterface } from 'fastify' import { FromSchema } from 'json-schema-to-ts' import { getConfig } from '../../../config' @@ -56,4 +56,17 @@ export default async function routes(fastify: FastifyInstance) { return reply.send({ message: 'Move jobs scheduled' }) } ) + + fastify.post('/migrate/pgboss-v12', { schema: { tags: ['queue'] } }, async (req, reply) => { + if (!pgQueueEnable) { + return reply.status(400).send({ message: 'Queue is not enabled' }) + } + + await UpgradePgBossV12.send({ + sbReqId: req.sbReqId, + tenant: SYSTEM_TENANT, + }) + + return reply.send({ message: 'Migration scheduled' }) + }) } diff --git a/src/internal/database/migrations/progressive.test.ts b/src/internal/database/migrations/progressive.test.ts index a2d2330d0..16ca81af6 100644 --- a/src/internal/database/migrations/progressive.test.ts +++ b/src/internal/database/migrations/progressive.test.ts @@ -218,12 +218,12 @@ describe('ProgressiveMigrations', () => { }) mockRunMigrationsBatchSend - .mockImplementationOnce(async () => { + .mockImplementationOnce((async () => { queueMicrotask(() => { migrations.addTenant('tenant-b') void migrations.drain() }) - }) + }) as never) .mockResolvedValueOnce(undefined as never) migrations.seed('tenant-a') diff --git a/src/internal/queue/database.test.ts b/src/internal/queue/database.test.ts index 18cd469b1..dceeee8ba 100644 --- a/src/internal/queue/database.test.ts +++ b/src/internal/queue/database.test.ts @@ -10,10 +10,22 @@ type PgBossStatic = { getRollbackPlans(schema: string, version: number): string } -type FetchNextJobOptions = { +type SqlQuery = { + text: string + values: unknown[] +} + +type FetchJobOptions = { + schema: string + table: string + name: string + policy: string | undefined + limit: number includeMetadata?: boolean priority?: boolean + orderByCreatedOn?: boolean ignoreStartAfter?: boolean + ignoreSingletons: string[] | null } type PgBossMigration = { @@ -23,46 +35,63 @@ type PgBossMigration = { type PgBossPlans = { JOB_STATES: Record - create(schema: string, version: number): string + create(schema: string, version: number, options?: { createSchema?: boolean }): string insertVersion(schema: string, version: number): string getVersion(schema: string): string setVersion(schema: string, version: number): string - versionTableExists(schema: string): string - fetchNextJob(schema: string): (options?: FetchNextJobOptions) => string - completeJobs(schema: string): string - cancelJobs(schema: string): string - resumeJobs(schema: string): string - deleteJobs(schema: string): string - retryJobs(schema: string): string - failJobsById(schema: string): string - failJobsByTimeout(schema: string): string - insertJob(schema: string): string - insertJobs(schema: string): string + fetchNextJob(options: FetchJobOptions): SqlQuery + completeJobs(schema: string, table: string, includeQueued?: boolean): string + cancelJobs(schema: string, table: string): string + resumeJobs(schema: string, table: string): string + restoreJobs(schema: string, table: string): string + retryJobs(schema: string, table: string): string + deleteJobsById(schema: string, table: string): string + deleteQueuedJobs(schema: string, table: string): string + deleteStoredJobs(schema: string, table: string): string + deleteAllJobs(schema: string, table: string): string + truncateTable(schema: string, table: string): string + failJobsById(schema: string, table: string): string + failJobsByTimeout(schema: string, table: string, queues: string[]): string + failJobsByHeartbeat(schema: string, table: string, queues: string[]): string + touchJobs(schema: string, table: string): string + insertJobs(schema: string, options: { table: string; name: string; returnId?: boolean }): string getTime(): string getSchedules(schema: string): string + getSchedulesByQueue(schema: string): string schedule(schema: string): string unschedule(schema: string): string subscribe(schema: string): string unsubscribe(schema: string): string getQueuesForEvent(schema: string): string - archive(schema: string, completedInterval: string, failedInterval?: string): string - drop(schema: string, interval: string): string - countStates(schema: string): string - updateQueue(schema: string): string - createQueue(schema: string): string - deleteQueue(schema: string): string - getQueues(schema: string): string - getQueueByName(schema: string): string - getQueueSize(schema: string, options?: { before?: string }): string - purgeQueue(schema: string): string - clearStorage(schema: string): string - trySetMaintenanceTime(schema: string): string - trySetMonitorTime(schema: string): string - trySetCronTime(schema: string): string - locked(schema: string, query: string | string[]): string + deletion(schema: string, table: string, queues: string[]): string + getQueueStats(schema: string, table: string, queues: string[]): SqlQuery + cacheQueueStats(schema: string, table: string, queues: string[]): string + createQueue(schema: string, name: string, options: unknown): string + updateQueue(schema: string, options?: { deadLetter?: string }): string + deleteQueue(schema: string, name: string): string + getQueues(schema: string, names?: string[]): SqlQuery + getJobById(schema: string, table: string): string + findJobs( + schema: string, + table: string, + options: { queued: boolean; byKey: boolean; byData: boolean; byId: boolean } + ): string + getBlockedKeys(schema: string, table: string): string + trySetQueueMonitorTime(schema: string, queues: string[], seconds: number): SqlQuery + trySetQueueDeletionTime(schema: string, queues: string[], seconds: number): SqlQuery + trySetCronTime(schema: string, seconds: number): string + trySetBamTime(schema: string, seconds: number): string + insertWarning(schema: string): string + getWarnings(schema: string): string + getWarningsCount(schema: string): string + deleteOldWarnings(schema: string, days: number): string + getBamStatus(schema: string): string + getBamEntries(schema: string): string + getNextBamCommand(schema: string): string + setBamCompleted(schema: string, id: string): string + setBamFailed(schema: string, id: string, error: string): string + locked(schema: string, query: string | string[], key?: string): string assertMigration(schema: string, version: number): string - getArchivedJobById(schema: string): string - getJobById(schema: string): string } type MigrationStore = { @@ -70,72 +99,114 @@ type MigrationStore = { } const PgBoss = loadCjs('pg-boss') as PgBossStatic -const plans = loadCjs('pg-boss/src/plans') as PgBossPlans -const migrationStore = loadCjs('pg-boss/src/migrationStore') as MigrationStore -const schemaVersion = (loadCjs('pg-boss/version.json') as { schema: number }).schema +const plans = loadCjs('pg-boss/dist/plans') as PgBossPlans +const migrationStore = loadCjs('pg-boss/dist/migrationStore') as MigrationStore +const schemaVersion = (loadCjs('pg-boss/package.json') as { pgboss: { schema: number } }).pgboss + .schema function normalizeSql(sql: string): string { return sql.replace(/\s+/g, ' ').trim() } function stripSqlStrings(sql: string): string { - return sql.replace(/\$[A-Za-z_]*\$[\s\S]*?\$[A-Za-z_]*\$/g, '').replace(/'(?:''|[^'])*'/g, '') + return ( + sql + .replace(/\$[A-Za-z_]*\$[\s\S]*?\$[A-Za-z_]*\$/g, '') + // Drop the jsonb key-exists operator (e.g. "o.data ? 'retryDelayMax'") so it + // is not mistaken for a question-mark placeholder. + .replace(/\?(?=\s*')/g, ' ') + .replace(/'(?:''|[^'])*'/g, '') + ) } function collectPgBossSqlSurface(): Array<[string, string]> { - const schema = 'pgboss_v10' + const schema = 'pgboss_v12' + const table = 'job_common' + const queues = ['queue-name'] const migrations = migrationStore.getAll(schema) as PgBossMigration[] const statements: Array<[string, string]> = [ ['create', plans.create(schema, schemaVersion)], ['insertVersion', plans.insertVersion(schema, schemaVersion)], ['getVersion', plans.getVersion(schema)], ['setVersion', plans.setVersion(schema, schemaVersion)], - ['versionTableExists', plans.versionTableExists(schema)], - ['fetchNextJob default', plans.fetchNextJob(schema)()], + [ + 'fetchNextJob default', + plans.fetchNextJob({ + schema, + table, + name: 'queue-name', + policy: 'standard', + limit: 5, + ignoreSingletons: null, + }).text, + ], [ 'fetchNextJob metadata no-priority ignore-start-after', - plans.fetchNextJob(schema)({ + plans.fetchNextJob({ + schema, + table, + name: 'queue-name', + policy: 'standard', + limit: 5, includeMetadata: true, priority: false, ignoreStartAfter: true, - }), + ignoreSingletons: ['singleton-key'], + }).text, ], - ['completeJobs', plans.completeJobs(schema)], - ['cancelJobs', plans.cancelJobs(schema)], - ['resumeJobs', plans.resumeJobs(schema)], - ['deleteJobs', plans.deleteJobs(schema)], - ['retryJobs', plans.retryJobs(schema)], - ['failJobsById', plans.failJobsById(schema)], - ['failJobsByTimeout', plans.failJobsByTimeout(schema)], - ['insertJob', plans.insertJob(schema)], - ['insertJobs', plans.insertJobs(schema)], + ['completeJobs', plans.completeJobs(schema, table)], + ['completeJobs queued', plans.completeJobs(schema, table, true)], + ['cancelJobs', plans.cancelJobs(schema, table)], + ['resumeJobs', plans.resumeJobs(schema, table)], + ['restoreJobs', plans.restoreJobs(schema, table)], + ['retryJobs', plans.retryJobs(schema, table)], + ['deleteJobsById', plans.deleteJobsById(schema, table)], + ['deleteQueuedJobs', plans.deleteQueuedJobs(schema, table)], + ['deleteStoredJobs', plans.deleteStoredJobs(schema, table)], + ['deleteAllJobs', plans.deleteAllJobs(schema, table)], + ['truncateTable', plans.truncateTable(schema, table)], + ['failJobsById', plans.failJobsById(schema, table)], + ['failJobsByTimeout', plans.failJobsByTimeout(schema, table, queues)], + ['failJobsByHeartbeat', plans.failJobsByHeartbeat(schema, table, queues)], + ['touchJobs', plans.touchJobs(schema, table)], + ['insertJobs', plans.insertJobs(schema, { table, name: 'queue-name' })], ['getTime', plans.getTime()], ['getSchedules', plans.getSchedules(schema)], + ['getSchedulesByQueue', plans.getSchedulesByQueue(schema)], ['schedule', plans.schedule(schema)], ['unschedule', plans.unschedule(schema)], ['subscribe', plans.subscribe(schema)], ['unsubscribe', plans.unsubscribe(schema)], ['getQueuesForEvent', plans.getQueuesForEvent(schema)], - ['archive', plans.archive(schema, '1 day', '2 days')], - ['drop', plans.drop(schema, '1 day')], - ['countStates', plans.countStates(schema)], - ['updateQueue', plans.updateQueue(schema)], - ['createQueue', plans.createQueue(schema)], - ['deleteQueue', plans.deleteQueue(schema)], - ['getQueues', plans.getQueues(schema)], - ['getQueueByName', plans.getQueueByName(schema)], - ['getQueueSize default', plans.getQueueSize(schema)], - ['getQueueSize failed', plans.getQueueSize(schema, { before: plans.JOB_STATES.failed })], - ['purgeQueue', plans.purgeQueue(schema)], - ['clearStorage', plans.clearStorage(schema)], - ['trySetMaintenanceTime', plans.trySetMaintenanceTime(schema)], - ['trySetMonitorTime', plans.trySetMonitorTime(schema)], - ['trySetCronTime', plans.trySetCronTime(schema)], + ['deletion', plans.deletion(schema, table, queues)], + ['getQueueStats', plans.getQueueStats(schema, table, queues).text], + ['cacheQueueStats', plans.cacheQueueStats(schema, table, queues)], + ['createQueue', plans.createQueue(schema, 'queue-name', { policy: 'standard' })], + ['updateQueue', plans.updateQueue(schema, { deadLetter: 'dead-letter-queue' })], + ['deleteQueue', plans.deleteQueue(schema, 'queue-name')], + ['getQueues', plans.getQueues(schema, queues).text], + ['getJobById', plans.getJobById(schema, table)], + [ + 'findJobs', + plans.findJobs(schema, table, { queued: true, byKey: true, byData: true, byId: true }), + ], + ['getBlockedKeys', plans.getBlockedKeys(schema, table)], + ['trySetQueueMonitorTime', plans.trySetQueueMonitorTime(schema, queues, 60).text], + ['trySetQueueDeletionTime', plans.trySetQueueDeletionTime(schema, queues, 60).text], + ['trySetCronTime', plans.trySetCronTime(schema, 60)], + ['trySetBamTime', plans.trySetBamTime(schema, 60)], + ['insertWarning', plans.insertWarning(schema)], + ['getWarnings', plans.getWarnings(schema)], + ['getWarningsCount', plans.getWarningsCount(schema)], + ['deleteOldWarnings', plans.deleteOldWarnings(schema, 7)], + ['getBamStatus', plans.getBamStatus(schema)], + ['getBamEntries', plans.getBamEntries(schema)], + ['getNextBamCommand', plans.getNextBamCommand(schema)], + ['setBamCompleted', plans.setBamCompleted(schema, 'bam-id')], + ['setBamFailed', plans.setBamFailed(schema, 'bam-id', 'bam-error')], ['locked single query', plans.locked(schema, plans.getTime())], - ['locked query array', plans.locked(schema, [plans.getTime(), plans.countStates(schema)])], + ['locked query array', plans.locked(schema, [plans.getTime(), plans.getVersion(schema)])], ['assertMigration', plans.assertMigration(schema, schemaVersion)], - ['getArchivedJobById', plans.getArchivedJobById(schema)], - ['getJobById', plans.getJobById(schema)], ['constructionPlans', PgBoss.getConstructionPlans(schema)], ] @@ -172,7 +243,7 @@ describe('PgQueueDB', () => { } satisfies PgStatement) }) - it('covers pg-boss v10 generated SQL with only pg-compatible placeholders', () => { + it('covers pg-boss v12 generated SQL with only pg-compatible placeholders', () => { const invalidStatements = collectPgBossSqlSurface().flatMap(([name, sql]) => { const stripped = stripSqlStrings(sql) const hasQuestionPlaceholder = stripped.includes('?') diff --git a/src/internal/queue/event.test.ts b/src/internal/queue/event.test.ts index c2bca6daf..e0bc3944b 100644 --- a/src/internal/queue/event.test.ts +++ b/src/internal/queue/event.test.ts @@ -125,9 +125,8 @@ describe('Event payload versioning', () => { expect(payload).toEqual(createPayload({ scheduleAt: new Date('2026-04-07T10:00:00.000Z') })) expect(payload).not.toHaveProperty('$version') - expect(insert).toHaveBeenCalledWith([ + expect(insert).toHaveBeenCalledWith('test-event', [ expect.objectContaining({ - name: 'test-event', deadLetter: 'test-event-dead-letter', data: expect.objectContaining({ $version: 'v-test', diff --git a/src/internal/queue/event.ts b/src/internal/queue/event.ts index c11e1bdba..62003e3fd 100644 --- a/src/internal/queue/event.ts +++ b/src/internal/queue/event.ts @@ -4,7 +4,7 @@ import { ERRORS } from '@internal/errors' import { logger, logSchema } from '@internal/monitoring' import { queueJobScheduled, queueJobSchedulingTime } from '@internal/monitoring/metrics' import { PgQueueDB } from '@internal/queue/database' -import PgBoss, { Job, Queue as PgBossQueue, SendOptions, WorkOptions } from 'pg-boss' +import { Job, JobInsert, Queue as PgBossQueue, SendOptions, WorkOptions } from 'pg-boss' import { getConfig } from '../../config' import { SYSTEM_TENANT_REF } from './constants' import { PG_BOSS_SCHEMA, Queue } from './queue' @@ -34,6 +34,21 @@ function withPayloadVersion( } } +/** + * Builds an in-memory job for handlers invoked synchronously (without going + * through pg-boss), matching the pg-boss v12 Job shape. + */ +function syncJob(name: string, data: T): Job { + return { + id: '__sync', + name, + data, + expireInSeconds: 0, + heartbeatSeconds: null, + signal: new AbortController().signal, + } +} + export type EventInputPayload = Omit export type QueueEvent = Event export type StaticThis = BaseEventConstructor @@ -112,10 +127,10 @@ export class Event> { } return Queue.getInstance().insert( + eventClass.getQueueName(), messages.map((message) => { const payloadWithVersion = withPayloadVersion(message.payload, eventClass.version) - const sendOptions = - (eventClass.getSendOptions(payloadWithVersion) as PgBoss.JobInsert) || {} + const sendOptions = (eventClass.getSendOptions(payloadWithVersion) as JobInsert) || {} if (payloadWithVersion.scheduleAt) { sendOptions.startAfter = new Date(payloadWithVersion.scheduleAt) @@ -123,7 +138,6 @@ export class Event> { return { ...sendOptions, - name: eventClass.getQueueName(), data: payloadWithVersion, deadLetter: eventClass.deadLetterQueueName(), } @@ -234,16 +248,13 @@ export class Event> { throw ERRORS.InternalError(undefined, 'Cannot send this event synchronously') } - await eventClass.handle({ - id: '__sync', - expireInSeconds: 0, - name: eventClass.getQueueName(), - data: { + await eventClass.handle( + syncJob(eventClass.getQueueName(), { region, ...this.payload, $version: eventClass.version, - }, - }) + }) + ) } async send( @@ -259,16 +270,13 @@ export class Event> { if (!pgQueueEnable) { if (eventClass.allowSync) { - return eventClass.handle({ - id: '__sync', - expireInSeconds: 0, - name: eventClass.getQueueName(), - data: { + return eventClass.handle( + syncJob(eventClass.getQueueName(), { region, ...this.payload, $version: eventClass.version, - }, - }) + }) + ) } else { logger.warn( { @@ -312,7 +320,7 @@ export class Event> { }) // pg-boss returns null when the insert was dropped by a queue policy - // (e.g. an exactly_once job with the same singleton key is still queued or active). + // (e.g. an exclusive job with the same singleton key is still queued or active). if (res === null) { logSchema.info(logger, `[Queue Sender] Job not queued, dropped by queue policy`, { type: 'queue', @@ -349,16 +357,13 @@ export class Event> { throw e } - return eventClass.handle({ - id: '__sync', - expireInSeconds: 0, - name: eventClass.getQueueName(), - data: { + return eventClass.handle( + syncJob(eventClass.getQueueName(), { region, ...this.payload, $version: eventClass.version, - }, - }) + }) + ) } finally { const duration = Number(process.hrtime.bigint() - startTime) / 1e9 queueJobSchedulingTime.record(duration, { diff --git a/src/internal/queue/queue.test.ts b/src/internal/queue/queue.test.ts index b48cee3cd..26f90d9a4 100644 --- a/src/internal/queue/queue.test.ts +++ b/src/internal/queue/queue.test.ts @@ -327,7 +327,6 @@ describe('Queue.stop', () => { expect(boss.stop).toHaveBeenCalledWith({ timeout: 20 * 1000, graceful: false, - wait: true, }) expect((Queue as unknown as { pgBoss?: unknown }).pgBoss).toBeUndefined() expect((Queue as unknown as { pgBossDb?: unknown }).pgBossDb).toBeUndefined() diff --git a/src/internal/queue/queue.ts b/src/internal/queue/queue.ts index 940cfbea1..e9bfd0572 100644 --- a/src/internal/queue/queue.ts +++ b/src/internal/queue/queue.ts @@ -1,7 +1,7 @@ import { createConcurrencyLimiter } from '@internal/concurrency' import { ERRORS } from '@internal/errors' import { QueueDB } from '@internal/queue/database' -import PgBoss, { Db, Job, JobWithMetadata } from 'pg-boss' +import { Db, Job, JobWithMetadata, PgBoss, QueueOptions } from 'pg-boss' import { getConfig } from '../../config' import { getSbReqIdFromPayload, logger, logSchema } from '../monitoring' import { @@ -23,13 +23,13 @@ type RegisteredEvent = { name: string } -export const PG_BOSS_SCHEMA = 'pgboss_v10' +export const PG_BOSS_SCHEMA = 'pgboss_v12' const queueStopTimeoutMs = 25_000 export abstract class Queue { protected static events: RegisteredEvent[] = [] private static pgBoss?: PgBoss - private static pgBossDb?: PgBoss.Db + private static pgBossDb?: QueueDB static createPgBoss(opts: { db: Db; enableWorkers: boolean }) { const { @@ -38,10 +38,6 @@ export abstract class Queue { multitenantDatabasePoolUrl, multitenantDatabaseUrl, pgQueueConnectionURL, - pgQueueArchiveCompletedAfterSeconds, - pgQueueDeleteAfterDays, - pgQueueDeleteAfterHours, - pgQueueRetentionDays, } = getConfig() let url = pgQueueConnectionURL ?? databaseURL @@ -65,20 +61,31 @@ export abstract class Queue { migrate, db: opts.db, schema: PG_BOSS_SCHEMA, - ...(pgQueueDeleteAfterHours - ? { deleteAfterHours: pgQueueDeleteAfterHours } - : { deleteAfterDays: pgQueueDeleteAfterDays }), - archiveCompletedAfterSeconds: pgQueueArchiveCompletedAfterSeconds, - retentionDays: pgQueueRetentionDays, - retryBackoff: true, - retryLimit: 20, - expireInHours: 23, maintenanceIntervalSeconds: 60 * 5, schedule: opts.enableWorkers, supervise: opts.enableWorkers, }) } + /** + * Retention, expiration and retry settings moved from the pg-boss constructor + * to per-queue options in pg-boss v11+. These defaults are applied when each + * queue is created and can be overridden per event via getQueueOptions(). + */ + static getDefaultQueueOptions(): QueueOptions { + const { pgQueueDeleteAfterDays, pgQueueDeleteAfterHours, pgQueueRetentionDays } = getConfig() + + return { + deleteAfterSeconds: pgQueueDeleteAfterHours + ? pgQueueDeleteAfterHours * 60 * 60 + : (pgQueueDeleteAfterDays ?? 2) * 24 * 60 * 60, + retentionSeconds: (pgQueueRetentionDays ?? 2) * 24 * 60 * 60, + expireInSeconds: 23 * 60 * 60, + retryBackoff: true, + retryLimit: 20, + } + } + static async start(opts: { signal?: AbortSignal onMessage?: (job: Job) => void @@ -124,6 +131,10 @@ export abstract class Queue { statement_timeout: pgQueueReadWriteTimeout > 0 ? pgQueueReadWriteTimeout : undefined, }) + // pg-boss v11+ only opens/closes the connection pool of its own internal + // db, so an externally provided db must be opened before start. + await Queue.pgBossDb.open() + Queue.pgBoss = this.createPgBoss({ db: Queue.pgBossDb, enableWorkers: pgQueueEnableWorkers !== false, @@ -210,7 +221,6 @@ export abstract class Queue { boss.stop({ timeout: 20 * 1000, graceful: isProduction, - wait: true, }), 'Queue stop' ) @@ -218,6 +228,12 @@ export abstract class Queue { try { await withQueueStopTimeout(this.callClose(), 'Queue close') } finally { + try { + await db?.close() + } catch { + // no-op + } + if (Queue.pgBoss === boss) { Queue.pgBoss = undefined } @@ -267,23 +283,22 @@ export abstract class Queue { const concurrentTaskCount = event.getWorkerOptions().concurrentTaskCount || maxConcurrentTasks try { // Create dead-letter queue and the normal queue - const queueOptions = { - policy: 'standard', + const { name: _name, ...eventQueueOptions } = { + policy: 'standard' as const, + ...Queue.getDefaultQueueOptions(), ...event.getQueueOptions(), - } as const + } // dead-letter await this.pgBoss?.createQueue(deadLetterName, { - ...queueOptions, - name: deadLetterName, - retentionDays: 30, + ...eventQueueOptions, + retentionSeconds: 30 * 24 * 60 * 60, retryBackoff: true, }) - // // normal queue + // normal queue await this.pgBoss?.createQueue(queueName, { - ...queueOptions, - name: queueName, + ...eventQueueOptions, deadLetter: deadLetterName, }) } catch { diff --git a/src/storage/events/iceberg/delete-iceberg-resources.ts b/src/storage/events/iceberg/delete-iceberg-resources.ts index d5a989554..decb404f1 100644 --- a/src/storage/events/iceberg/delete-iceberg-resources.ts +++ b/src/storage/events/iceberg/delete-iceberg-resources.ts @@ -26,7 +26,7 @@ export class DeleteIcebergResources extends BaseEvent { static getQueueOptions(): Queue { return { name: this.queueName, - policy: 'exactly_once', + policy: 'exclusive', } as const } @@ -30,7 +30,7 @@ export class ResetMigrationsOnTenant extends BaseEvent { static getSendOptions(payload: ResetMigrationsPayload): SendOptions { return { - expireInHours: 2, + expireInSeconds: 2 * 60 * 60, singletonKey: payload.tenantId, retryLimit: 3, retryDelay: 5, diff --git a/src/storage/events/migrations/run-migrations.ts b/src/storage/events/migrations/run-migrations.ts index e61360941..64b925e03 100644 --- a/src/storage/events/migrations/run-migrations.ts +++ b/src/storage/events/migrations/run-migrations.ts @@ -23,7 +23,7 @@ export class RunMigrationsOnTenants extends BaseEvent { static getQueueOptions(): Queue { return { name: this.queueName, - policy: 'exactly_once', + policy: 'exclusive', } as const } @@ -36,8 +36,8 @@ export class RunMigrationsOnTenants extends BaseEvent { static getSendOptions(payload: RunMigrationsPayload): SendOptions { return { singletonKey: `migrations_${payload.tenantId}`, - singletonHours: 1, - expireInMinutes: 10, + singletonSeconds: 60 * 60, + expireInSeconds: 10 * 60, retryLimit: 3, retryDelay: 5, priority: 10, diff --git a/src/storage/events/pgboss/move-jobs.test.ts b/src/storage/events/pgboss/move-jobs.test.ts index f58adb190..7a06437a9 100644 --- a/src/storage/events/pgboss/move-jobs.test.ts +++ b/src/storage/events/pgboss/move-jobs.test.ts @@ -102,7 +102,7 @@ describe('MoveJobs.handle', () => { mockLockedTransaction(query) mockGetQueue.mockResolvedValue({ name: 'target-queue', - policy: 'exactly_once', + policy: 'exclusive', }) await expect(MoveJobs.handle(makeJob() as never)).resolves.toBeUndefined() diff --git a/src/storage/events/pgboss/move-jobs.ts b/src/storage/events/pgboss/move-jobs.ts index 60e6884c6..f353df04d 100644 --- a/src/storage/events/pgboss/move-jobs.ts +++ b/src/storage/events/pgboss/move-jobs.ts @@ -16,7 +16,7 @@ export class MoveJobs extends BaseEvent { static getQueueOptions(): PgBossQueue { return { name: this.queueName, - policy: 'exactly_once', + policy: 'exclusive', } as const } @@ -28,9 +28,9 @@ export class MoveJobs extends BaseEvent { static getSendOptions(payload: MoveJobsPayload): SendOptions { return { - expireInHours: 2, + expireInSeconds: 2 * 60 * 60, singletonKey: `move_${payload.fromQueue}_to_${payload.toQueue}`, - singletonHours: 12, + singletonSeconds: 12 * 60 * 60, retryLimit: 3, retryDelay: 5, priority: 10, @@ -82,7 +82,7 @@ export class MoveJobs extends BaseEvent { start_after, singleton_key, singleton_on, - expire_in, + expire_seconds, created_on, keep_until, output, @@ -101,7 +101,7 @@ export class MoveJobs extends BaseEvent { start_after, singleton_key, singleton_on, - expire_in, + expire_seconds, created_on, keep_until, output, diff --git a/src/storage/events/pgboss/pg-branches.test.ts b/src/storage/events/pgboss/pg-branches.test.ts index ee4a619cf..87480c9a2 100644 --- a/src/storage/events/pgboss/pg-branches.test.ts +++ b/src/storage/events/pgboss/pg-branches.test.ts @@ -27,7 +27,7 @@ vi.mock('@internal/monitoring', () => ({ })) vi.mock('@internal/queue', () => ({ - PG_BOSS_SCHEMA: 'pgboss_v10', + PG_BOSS_SCHEMA: 'pgboss_v12', SYSTEM_TENANT_REF: 'system-tenant', Queue: { getInstance: () => ({ @@ -41,6 +41,7 @@ vi.mock('../base-event', () => ({ })) import { MoveJobs } from './move-jobs' +import { UpgradePgBossV12 } from './upgrade-v12' function makeTransaction() { return { @@ -64,6 +65,17 @@ function makeMoveJob() { } } +function makeUpgradeJob() { + return { + data: { + sbReqId: 'sb-req-456', + tenant: { + ref: 'tenant-a', + }, + }, + } +} + describe('pg-boss maintenance pg branches', () => { beforeEach(() => { vi.clearAllMocks() @@ -75,7 +87,7 @@ describe('pg-boss maintenance pg branches', () => { mockBeginTransaction.mockResolvedValue(tx) mockGetQueue.mockResolvedValue({ name: 'target-queue', - policy: 'exactly_once', + policy: 'exclusive', }) await MoveJobs.handle(makeMoveJob() as never) @@ -83,11 +95,42 @@ describe('pg-boss maintenance pg branches', () => { expect(tx.query).toHaveBeenNthCalledWith( 2, expect.objectContaining({ - text: expect.stringContaining('INSERT INTO pgboss_v10.job'), - values: ['target-queue', 'exactly_once', 'source-queue'], + text: expect.stringContaining('INSERT INTO pgboss_v12.job'), + values: ['target-queue', 'exclusive', 'source-queue'], }) ) expect(tx.commit).toHaveBeenCalledTimes(1) expect(tx.rollback).not.toHaveBeenCalled() }) + + it('copies pending jobs from the pgboss_v10 schema through the pg transaction branch', async () => { + const tx = makeTransaction() + tx.query + .mockResolvedValueOnce({ rows: [{ locked: true }] }) + .mockResolvedValueOnce({ rows: [{ exists: true }] }) + .mockResolvedValueOnce({ rows: [] }) + mockBeginTransaction.mockResolvedValue(tx) + + await UpgradePgBossV12.handle(makeUpgradeJob() as never) + + const copySql = tx.query.mock.calls[2][0] as string + expect(copySql).toContain('INSERT INTO pgboss_v12.job') + expect(copySql).toContain('FROM pgboss_v10.job') + expect(tx.commit).toHaveBeenCalledTimes(1) + expect(tx.rollback).not.toHaveBeenCalled() + }) + + it('skips the copy when the pgboss_v10 schema does not exist', async () => { + const tx = makeTransaction() + tx.query + .mockResolvedValueOnce({ rows: [{ locked: true }] }) + .mockResolvedValueOnce({ rows: [{ exists: false }] }) + mockBeginTransaction.mockResolvedValue(tx) + + await UpgradePgBossV12.handle(makeUpgradeJob() as never) + + expect(tx.query).toHaveBeenCalledTimes(2) + expect(tx.commit).toHaveBeenCalledTimes(1) + expect(tx.rollback).not.toHaveBeenCalled() + }) }) diff --git a/src/storage/events/pgboss/upgrade-v12.ts b/src/storage/events/pgboss/upgrade-v12.ts new file mode 100644 index 000000000..5131d82e4 --- /dev/null +++ b/src/storage/events/pgboss/upgrade-v12.ts @@ -0,0 +1,149 @@ +import { multitenantPgExecutor, PgTransaction } from '@internal/database' +import { logger, logSchema } from '@internal/monitoring' +import { BasePayload, PG_BOSS_SCHEMA, SYSTEM_TENANT_REF } from '@internal/queue' +import { Job, Queue as PgBossQueue, SendOptions, WorkOptions } from 'pg-boss' +import { BaseEvent } from '../base-event' + +type UpgradePgBossV12Payload = BasePayload + +const SOURCE_SCHEMA = 'pgboss_v10' + +/** + * Copies pending jobs from the pg-boss v10 fork schema (pgboss_v10) into the + * upstream pg-boss v12 schema. Only jobs whose queue exists in the new schema + * are copied; the queue's policy in the new schema is applied so jobs queued + * under the fork-only `exactly_once` policy continue under `exclusive`. + */ +export class UpgradePgBossV12 extends BaseEvent { + static queueName = 'upgrade-pgboss-v12' + + static getQueueOptions(): PgBossQueue { + return { + name: this.queueName, + policy: 'exclusive', + } as const + } + + static getWorkerOptions(): WorkOptions { + return { + includeMetadata: true, + } + } + + static getSendOptions(): SendOptions { + return { + expireInSeconds: 2 * 60 * 60, + singletonKey: 'pgboss-upgrade-v12', + singletonSeconds: 12 * 60 * 60, + retryLimit: 3, + retryDelay: 5, + priority: 10, + } + } + + static async handle(job: Job) { + const { sbReqId } = job.data + + await withPgTransaction(async (tnx) => { + const resultLock = await tnx.query<{ locked: boolean }>( + `SELECT pg_try_advisory_xact_lock(-5525285245963000612) AS locked` + ) + const lockAcquired = resultLock.rows.shift()?.locked || false + + if (!lockAcquired) { + return + } + + const sourceSchemaExists = await tnx.query<{ exists: boolean }>( + `SELECT EXISTS(SELECT 1 FROM pg_namespace WHERE nspname = $1) AS exists`, + [SOURCE_SCHEMA] + ) + + if (!sourceSchemaExists.rows.shift()?.exists) { + logSchema.info(logger, `[PgBoss] Source schema ${SOURCE_SCHEMA} does not exist, skipping`, { + type: 'pgboss', + project: job.data.tenant?.ref || SYSTEM_TENANT_REF, + sbReqId, + }) + return + } + + try { + await tnx.query(` + INSERT INTO ${PG_BOSS_SCHEMA}.job ( + id, + name, + priority, + data, + retry_limit, + retry_count, + retry_delay, + retry_backoff, + start_after, + singleton_key, + singleton_on, + expire_seconds, + created_on, + keep_until, + output, + policy, + dead_letter, + state + ) + SELECT + source.id, + source.name, + source.priority, + source.data, + source.retry_limit, + source.retry_count, + source.retry_delay, + source.retry_backoff, + source.start_after, + source.singleton_key, + source.singleton_on, + GREATEST(1, EXTRACT(EPOCH FROM source.expire_in))::int, + source.created_on, + source.keep_until, + source.output, + queue.policy, + dlq.name, + 'created' as state + FROM ${SOURCE_SCHEMA}.job AS source + JOIN ${PG_BOSS_SCHEMA}.queue AS queue ON queue.name = source.name + LEFT JOIN ${PG_BOSS_SCHEMA}.queue AS dlq ON dlq.name = source.dead_letter + WHERE source.state IN ('created', 'active', 'retry') + ON CONFLICT DO NOTHING + `) + } catch (error) { + logSchema.error(logger, '[PgBoss] Error while copying jobs from previous schema', { + type: 'pgboss', + error, + project: job.data.tenant?.ref || SYSTEM_TENANT_REF, + sbReqId, + }) + } + }) + } +} + +async function withPgTransaction(fn: (tnx: PgTransaction) => Promise): Promise { + const tnx = await multitenantPgExecutor.beginTransaction() + + try { + const result = await fn(tnx) + await tnx.commit() + return result + } catch (e) { + try { + await tnx.rollback() + } catch (rollbackError) { + logSchema.warning(logger, '[UpgradePgBossV12] Failed to rollback transaction', { + type: 'pgboss', + error: rollbackError, + metadata: JSON.stringify({ originalError: String(e) }), + }) + } + throw e + } +} diff --git a/src/storage/events/upgrades/base-event.ts b/src/storage/events/upgrades/base-event.ts index 7d36b81af..03e7dab9a 100644 --- a/src/storage/events/upgrades/base-event.ts +++ b/src/storage/events/upgrades/base-event.ts @@ -15,15 +15,15 @@ export abstract class UpgradeBaseEvent extend static getQueueOptions(): PgBossQueue { return { name: this.queueName, - policy: 'exactly_once', + policy: 'exclusive', } as const } static getSendOptions(): SendOptions { return { - expireInHours: 2, + expireInSeconds: 2 * 60 * 60, singletonKey: this.getQueueName(), - singletonHours: 12, + singletonSeconds: 12 * 60 * 60, retryBackoff: false, retryLimit: 3, retryDelay: 5, diff --git a/src/storage/events/workers.ts b/src/storage/events/workers.ts index 433126cf7..44bfa9718 100644 --- a/src/storage/events/workers.ts +++ b/src/storage/events/workers.ts @@ -10,6 +10,7 @@ import { BackupObjectEvent } from './objects/backup-object' import { ObjectAdminDelete } from './objects/object-admin-delete' import { ObjectAdminDeleteAllBefore } from './objects/object-admin-delete-all-before' import { MoveJobs } from './pgboss/move-jobs' +import { UpgradePgBossV12 } from './pgboss/upgrade-v12' import { SyncCatalogIds } from './upgrades/sync-catalog-ids' export function registerWorkers() { @@ -22,6 +23,7 @@ export function registerWorkers() { Queue.register(JwksCreateSigningSecret) Queue.register(JwksRollUrlSigningKey) Queue.register(MoveJobs) + Queue.register(UpgradePgBossV12) Queue.register(ReconcileIcebergCatalog) Queue.register(DeleteIcebergResources) Queue.register(SyncCatalogIds) diff --git a/src/test/admin-migrations.test.ts b/src/test/admin-migrations.test.ts index 144043d76..0d8e30f2c 100644 --- a/src/test/admin-migrations.test.ts +++ b/src/test/admin-migrations.test.ts @@ -534,9 +534,11 @@ describe('Admin migrations routes', () => { }) test('returns queue progress for the current migration queue', async () => { - const getQueueSize = vi.fn().mockResolvedValue(7) + const getQueueStats = vi.fn().mockResolvedValue([{ queuedCount: 7 }]) + const getQueue = vi.fn().mockResolvedValue({ name: RunMigrationsOnTenants.getQueueName() }) vi.spyOn(Queue, 'getInstance').mockReturnValue({ - getQueueSize, + getQueue, + getQueueStats, } as never) const response = await adminApp.inject({ @@ -547,7 +549,7 @@ describe('Admin migrations routes', () => { expect(response.statusCode).toBe(200) expect(JSON.parse(response.body)).toEqual({ remaining: 7 }) - expect(getQueueSize).toHaveBeenCalledWith(RunMigrationsOnTenants.getQueueName()) + expect(getQueueStats).toHaveBeenCalledWith(RunMigrationsOnTenants.getQueueName()) }) test('lists failed tenants and paginates by cursor', async () => { diff --git a/src/test/tenant-jwks.test.ts b/src/test/tenant-jwks.test.ts index f2af397a7..3a12173b7 100644 --- a/src/test/tenant-jwks.test.ts +++ b/src/test/tenant-jwks.test.ts @@ -512,9 +512,10 @@ describe('Tenant jwks configs', () => { await queueSpyAwaiter expect(queueInsertSpy).toHaveBeenCalledTimes(1) - const [[callArg]] = queueInsertSpy.mock.calls + const [[queueName, callArg]] = queueInsertSpy.mock.calls + expect(queueName).toBe('tenants-jwks-create-v2') expect(callArg).toHaveLength(1) - expect(callArg[0]).toMatchObject({ data: { tenantId }, name: 'tenants-jwks-create-v2' }) + expect(callArg[0]).toMatchObject({ data: { tenantId } }) } finally { queueInsertSpy.mockRestore() }