feat: new queing system - #1292
Conversation
There was a problem hiding this comment.
Pull request overview
This PR migrates the storage event/worker queueing layer from the legacy in-repo pg-boss v10 fork + custom polling implementation to the Wave-based queueing system, standardizing topics, handlers, and middleware while updating producers, workers, and tests to the new message/handler APIs.
Changes:
- Replaces the legacy
@internal/queueworker implementation with Wave (@supabase-labs/wave-core) and pg-boss v12 adapters, including new middleware for tenant-disable gating and sync fallback. - Refactors storage events into Wave message classes +
TopicHandlerworkers, with a centralizedTOPICSregistry and storage-queue helpers (startStorageQueue,getStorageQueue). - Updates storage producers and test suites to use
getStorageQueue().produce()/invoke()and new handler contexts.
Reviewed changes
Copilot reviewed 82 out of 83 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| src/test/webhooks.test.ts | Updates webhook job assertions/dispatch to Wave handler context. |
| src/test/uploader.test.ts | Switches queue assertions to Wave test seam (mockQueue). |
| src/test/tenant-pg-store-runtime.test.ts | Updates migration-topic naming/schema references for v2 queue. |
| src/test/sharding.test.ts | Repoints import to consolidated events export surface. |
| src/test/s3-protocol.test.ts | Aligns ObjectRemovedEvent type import with new lifecycle-events module. |
| src/test/queue-mocks.test.ts | Updates webhook handler module path and Wave handler invocation contract. |
| src/test/object.test.ts | Switches ObjectAdminDelete send spying to Wave test seam. |
| src/test/iceberg.test.ts | Refactors DeleteIcebergResources invocation to Wave handler + base createStorage seam. |
| src/test/common.ts | Replaces legacy Queue singleton mocking with setWaveForTesting seam. |
| src/test/admin-migrations.test.ts | Updates admin migration queue checks to Wave stats seam and TOPICS naming. |
| src/storage/uploader.ts | Migrates ObjectAdminDelete scheduling to getStorageQueue().produce(new ...). |
| src/storage/storage.ts | Migrates bucket/object admin event scheduling to Wave produce/invoke APIs. |
| src/storage/scanner/scanner.ts | Migrates backup scheduling from batchSend to Wave produce of message instances. |
| src/storage/scanner/scanner-pagination.test.ts | Adjusts mocks for new @storage/events exports and getStorageQueue(). |
| src/storage/object.ts | Migrates ObjectAdminDelete scheduling sites to Wave produce(new ...). |
| src/storage/events/workers.ts | Replaces v1 registration with buildHandlers() returning Wave TopicHandler instances. |
| src/storage/events/webhooks/webhook.ts | Adds Wave-based webhook topic/message + handler implementation. |
| src/storage/events/webhooks/lifecycle-events.ts | Introduces webhook lifecycle event wrappers that preserve v1 call shape. |
| src/storage/events/upgrades/sync-catalog-ids.ts | Migrates fleet upgrade event to Wave message + handler and run-once helper. |
| src/storage/events/upgrades/base.ts | Adds runUpgradeOnce helper replacing v1 upgrade base event abstraction. |
| src/storage/events/upgrades/base.test.ts | Updates tests to target runUpgradeOnce helper. |
| src/storage/events/upgrades/base-event.ts | Removes legacy UpgradeBaseEvent implementation. |
| src/storage/events/topics.ts | Adds TOPICS constants and Wave topic registry with retry/DLQ policies. |
| src/storage/events/queue.ts | Adds startStorageQueue and typed getStorageQueue helpers. |
| src/storage/events/pgboss/pg-branches.test.ts | Removes v1 pg-boss maintenance test. |
| src/storage/events/pgboss/move-jobs.ts | Removes v1 MoveJobs maintenance task. |
| src/storage/events/pgboss/move-jobs.test.ts | Removes v1 MoveJobs unit tests. |
| src/storage/events/objects/object-admin-delete.ts | Migrates ObjectAdminDelete to Wave message + handler. |
| src/storage/events/objects/object-admin-delete-all-before.ts | Migrates ObjectAdminDeleteAllBefore to Wave message + handler and recursive produce. |
| src/storage/events/objects/backup-object.ts | Migrates BackupObjectEvent to Wave message + handler. |
| src/storage/events/migrations/run-migrations.ts | Migrates tenant migrations worker to Wave handler and new retry semantics. |
| src/storage/events/migrations/run-migrations.test.ts | Updates run-migrations tests to Wave handler context and stubs. |
| src/storage/events/migrations/reset-migrations.ts | Migrates reset-migrations worker and follow-up job enqueue to Wave. |
| src/storage/events/migrations/reset-migrations.test.ts | Updates reset-migrations tests to Wave handler context and produce assertions. |
| src/storage/events/lifecycle/webhook.ts | Removes legacy webhook queue worker implementation. |
| src/storage/events/lifecycle/object-updated.ts | Removes legacy object-updated event class. |
| src/storage/events/lifecycle/object-removed.ts | Removes legacy object-removed event class. |
| src/storage/events/lifecycle/object-created.ts | Removes legacy object-created event classes. |
| src/storage/events/lifecycle/bucket-deleted.ts | Migrates bucket-deleted worker to Wave handler and transactional enqueue support. |
| src/storage/events/lifecycle/bucket-created.ts | Migrates bucket-created worker to Wave handler; adds real worker registration. |
| src/storage/events/jwks/jwks-roll-url-signing-key.ts | Migrates JWKS roll worker to Wave message + handler. |
| src/storage/events/jwks/jwks-roll-url-signing-key.test.ts | Updates JWKS roll tests to Wave handler context. |
| src/storage/events/jwks/jwks-create-signing-secret.ts | Migrates JWKS create worker to Wave message + handler. |
| src/storage/events/index.ts | Re-exports new Wave event modules/topics/queue helpers; drops legacy exports. |
| src/storage/events/iceberg/reconcile-catalog.ts | Migrates reconcile-catalog worker to Wave message + handler. |
| src/storage/events/iceberg/index.ts | Removes legacy iceberg barrel export. |
| src/storage/events/iceberg/delete-iceberg-resources.ts | Migrates delete-iceberg-resources worker to Wave message + handler. |
| src/storage/events/iceberg/delete-iceberg-resources.test.ts | Updates delete-iceberg-resources tests to Wave handler context and new base seam. |
| src/storage/events/base.ts | Adds Wave message factory (storageEvent) and shared storage backend + createStorage seam. |
| src/storage/events/base.test.ts | Adds unit tests for storageEvent versioning/idempotency/middleware config surface. |
| src/storage/events/base-event.ts | Removes legacy BaseEvent implementation. |
| src/start/worker.ts | Switches worker startup to startStorageQueue. |
| src/start/server.ts | Starts storage queue in all modes; migrates SyncCatalogIds trigger to Wave invoke. |
| src/internal/queue/queue.ts | Removes legacy polling-based Queue implementation. |
| src/internal/queue/queue.test.ts | Removes tests for legacy Queue implementation. |
| src/internal/queue/middleware.ts | Adds Wave middleware for tenant disabling, metrics, sync fallback, and sync-mode guard. |
| src/internal/queue/middleware.test.ts | Adds unit tests for scheduling metrics middleware. |
| src/internal/queue/instance.ts | Adds Wave singleton start/get/stop and test seam (setWaveForTesting) + queueSize helper. |
| src/internal/queue/instance.test.ts | Adds stopQueue timeout/idempotency unit tests. |
| src/internal/queue/index.ts | Re-exports new queue modules (instance, middleware, events, database). |
| src/internal/queue/events.ts | Introduces BasePayload/WirePayload and queue-event option surface + runtime guard. |
| src/internal/queue/event.ts | Removes legacy Event base class. |
| src/internal/queue/event.test.ts | Removes legacy Event tests. |
| src/internal/queue/database.ts | Replaces legacy QueueDB/PgQueueDB with transactional produce ctx adapter. |
| src/internal/queue/database.test.ts | Updates db adapter tests for pg-boss v12 insert plan and tx ctx wrapper. |
| src/internal/queue/constants.ts | Adjusts system-tenant constants to non-empty ref and object form. |
| src/internal/queue/boss.ts | Adds pg-boss v12 boss factory + queue default policy mapping. |
| src/internal/monitoring/otel-class-instrumentations.ts | Removes legacy Event.send/batchSend instrumentation note for follow-up middleware tracing. |
| src/internal/database/migrations/vector-store-migrations.test.ts | Updates migration tests to mock Wave queue surface. |
| src/internal/database/migrations/progressive.ts | Migrates progressive migration enqueueing to Wave produce + delayMs pacing. |
| src/internal/database/migrations/progressive.test.ts | Updates progressive migration tests to mock Wave produce surface. |
| src/internal/database/migrations/migrate.ts | Migrates tenant migration/reset batch enqueueing to Wave produce. |
| src/internal/database/migrations/migrate.test.ts | Updates migration tests to validate Wave produce routing via fake message classes. |
| src/internal/auth/jwks/generator.ts | Migrates JWKS create batch enqueueing to Wave produce. |
| src/http/routes/admin/tenants.ts | Updates migration admin store schema/topic naming to v2 schema + TOPICS. |
| src/http/routes/admin/queue.ts | Disables MoveJobs endpoint (501) with explanatory comment. |
| src/http/routes/admin/queue.test.ts | Updates tests to assert MoveJobs endpoint is unavailable. |
| src/http/routes/admin/migrations.ts | Replaces legacy Queue.getQueueSize with queueSize() helper + TOPICS. |
| src/http/routes/admin/jwks.ts | Migrates JWKS roll enqueueing to Wave produce. |
| src/http/routes/admin/iceberg-admin.ts | Migrates iceberg cleanup batch enqueueing to Wave produce. |
| src/config.ts | Adds pgQueueSchemaV2 config setting (env PG_QUEUE_V2_SCHEMA). |
| package.json | Adds Wave packages and updates pg-boss dependency to v12. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
![claude[bot]](https://avatars.githubusercontent.com/in/1236702?s=60&v=4){kind=small}
| let instance: AnyWave | undefined | ||
| let stopping: Promise<void> | undefined | ||
|
|
||
| const boss = createQueueBoss({ enableWorkers: pgQueueEnableWorkers ?? true }) |
There was a problem hiding this comment.
🟡 In src/internal/queue/instance.ts:38, createQueueBoss() runs unconditionally at module scope (import time of @internal/queue), instead of being constructed lazily inside startQueue gated on pgQueueEnable as v1 did inside Queue.start(). For a multitenant deployment that sets only DB_MULTITENANT_DATABASE_POOL_URL (no direct DB_MULTITENANT_DATABASE_URL, no PG_QUEUE_CONNECTION_URL), this now throws at process boot even if PG_QUEUE_ENABLE=false. In practice this trigger config is unlikely to be viable anyway since multitenant pubsub also requires the direct URL, so this is more a robustness/design improvement than a blocking fix.
Extended reasoning...
src/internal/queue/instance.ts:38 calls const boss = createQueueBoss({ enableWorkers: pgQueueEnableWorkers ?? true }) at MODULE SCOPE -- it runs as a side effect of simply importing @internal/queue, which is imported very widely (e.g. via storage/events, and server.ts now calls startStorageQueue in every mode). This is unconditional: it does not check pgQueueEnable (which defaults to false, per config.ts). In v1, the pg-boss instance was only constructed inside Queue.start(), and server.ts only called Queue.start() when if (pgQueueEnable). So construction moved from a runtime call gated on pgQueueEnable to an unconditional import-time call.
createQueueBoss (src/internal/queue/boss.ts:27-30) throws synchronously when isMultitenant && !pgQueueConnectionURL && !multitenantDatabaseUrl -- note this checks only the direct multitenantDatabaseUrl, not multitenantDatabasePoolUrl. Combined with the eager construction, a multitenant deployment configured with only DB_MULTITENANT_DATABASE_POOL_URL (no direct URL, no queue URL) will now crash at process startup with 'running storage in multi-tenant but DB_MULTITENANT_DATABASE_URL is not set' -- merely from importing the queue module -- even when PG_QUEUE_ENABLE=false and the queue is never used (sync-wave mode).
Step-by-step to reproduce: (1) Set DB_MULTITENANT=true, DB_MULTITENANT_DATABASE_POOL_URL=postgres://pooler/db, leave DB_MULTITENANT_DATABASE_URL unset, leave PG_QUEUE_CONNECTION_URL unset, and PG_QUEUE_ENABLE=false. (2) Start the server. (3) Any module that transitively imports @internal/queue (e.g. storage/events, storage/object) triggers evaluation of src/internal/queue/instance.ts, which calls createQueueBoss() at module scope. (4) createQueueBoss sees isMultitenant=true, pgQueueConnectionURL unset, multitenantDatabaseUrl unset, and throws synchronously. (5) The throw propagates up through the import chain and crashes the process before the app ever binds to a port -- regardless of the queue being disabled.
However, this crash's practical blast radius is narrow, which is why I'm marking this a nit rather than blocking. First, the exact trigger config (multitenant + pool-URL-only + no direct URL + no queue URL) is very likely not a viable deployment shape in the first place: multitenantDatabaseUrl (the direct, non-pooled connection) is also required by pubsub.ts's tenant-update LISTEN, which needs a session-level (non-transaction-pooled) connection -- so a multitenant deployment lacking the direct URL was already degraded/broken before this PR regardless of the queue. Second, v1's own Queue.createPgBoss (used on the transactional-send path) carried the identical single-field !multitenantDatabaseUrl check, so v1 also threw for 'pool-only' whenever the queue was actually used with pgQueueEnable=true -- the narrowing introduced here is specific to the pgQueueEnable=false case, and even that case likely wasn't a supported combination to begin with. Third, in every properly-configured deployment (single-tenant, or multitenant with the direct URL set), createQueueBoss does not throw and merely constructs an unstarted, connection-less PgBoss instance, so the eager construction itself is otherwise inert.
The fix is straightforward and worth doing regardless of the narrow blast radius: move createQueueBoss() out of module scope and into startQueue (or otherwise gate it behind pgQueueEnable), so the queue module can be imported freely without any side effect, and so any related config validation only fires when the queue is actually going to be used. This also better matches v1's behavior where importing queue-adjacent code carried no side effects.
This comment has been minimized.
This comment has been minimized.
Coverage Report for CI Build 30835492250Coverage increased (+0.5%) to 81.002%Details
Uncovered Changes
Coverage Regressions23 previously-covered lines in 7 files lost coverage.
Coverage Stats
💛 - Coveralls |
- createIcebergBucket: invoke the BucketCreatedEvent handler inline first (v1 invokeOrSend order), enqueue only as fallback; StorageBackendError surfaces to the caller again - wire PG_QUEUE_READ_WRITE_TIMEOUT as statement_timeout on the pg-boss pool, bounding queue produce/fetch/maintenance statements as in v1 - skip tenant disable-events gating for system-tenant messages, which have no tenant config row - fix jobId typos (jodId) and BackupObject error log event mislabel
topics.ts reads every event class at module top level, while event modules import it back only for TOPICS/retry postures they read lazily. Loading topics.ts first lets every event module finish initializing before the registry reads their classes; alphabetical order makes the first event file the cycle entry and its class is still in TDZ when the registry reads it (Cannot access before initialization). A biome-ignore-all suppression keeps lint:fix from re-sorting the file.
Production starts the wave in start/server.ts before serving; tests build the app straight from src/app.ts, so produce paths threw 'queue is not started'. vitest-setup now installs the in-process sync wave (v1 parity: produce runs handlers inline) via a new startSyncWaveForTesting seam — unconditionally, so files that set PG_QUEUE_ENABLE=true to exercise queue-enabled app branches still never construct a real pg-boss. Files that stub the wave keep overriding the instance as before. bucket.test.ts CDN purge tests switch from mockQueue() (whose stub replaced the wave for the rest of the file) to restorable spies on the live wave's produce.
v1 spies received send arguments ({ name, data, options }); the v2
wave's produce receives the message instance, where only .data
exists — queue name and options live in the topic registry. Drop the
name/options wrapper from webhooks assertions, the name field from
tenant-jwks, and nest the object move-rollback expectation under
data.
What kind of change does this PR introduce?