diff --git a/.github/workflows/acceptance.yml b/.github/workflows/acceptance.yml index c7722ad2c..66ff3073e 100644 --- a/.github/workflows/acceptance.yml +++ b/.github/workflows/acceptance.yml @@ -79,7 +79,7 @@ jobs: ACCEPTANCE_INFRA_RESTART_SCRIPT: ${{ matrix.database == 'oriole' && 'infra:restart:ci:oriole:pgvector' || matrix.database == 'multigres' && 'infra:restart:ci:multigres' || 'infra:restart:ci' }} ACCEPTANCE_PROFILE: ${{ inputs.profile || 'full' }} ACCEPTANCE_X_FORWARDED_HOST: ${{ matrix.tenancy == 'multitenant' && 'bjhaohmqunupljrqypxz.local.dev' || '' }} - DATABASE_ENGINE: ${{ matrix.database == 'multigres' && 'multigres' || 'postgres' }} + DATABASE_ENGINE: ${{ matrix.database == 'pg' && 'postgres' || matrix.database }} MULTI_TENANT: ${{ matrix.tenancy == 'multitenant' && 'true' || 'false' }} PG_QUEUE_ENABLE: ${{ matrix.tenancy == 'multitenant' && 'true' || 'false' }} REQUEST_X_FORWARDED_HOST_REGEXP: ${{ matrix.tenancy == 'multitenant' && '^([a-z]{20})[.]local[.](?:com|dev)$' || '' }} diff --git a/package.json b/package.json index 99d4f6c22..48921e911 100644 --- a/package.json +++ b/package.json @@ -12,6 +12,9 @@ "migrations:types": "tsx src/scripts/migrations-types.ts", "docs:export": "tsx ./src/scripts/export-docs.ts", "docs:export-ui": "tsx ./src/scripts/export-docs-ui.ts", + "jobs:list": "tsx src/scripts/jobs-client.ts list", + "jobs:backup": "tsx src/scripts/jobs-client.ts backup", + "jobs:restore": "tsx src/scripts/jobs-client.ts restore", "pprof:capture": "tsx src/scripts/pprof-client.ts", "test:dummy-data": "tsx -r dotenv/config ./src/test/db/import-dummy-data.ts", "test:unit": "vitest run --config vitest.unit.config.ts", @@ -24,12 +27,12 @@ "acceptance:run": "tsx acceptance/scripts/run.ts", "acceptance:typecheck": "tsc -p acceptance/tsconfig.json --noEmit", "test": "npm run test:unit && npm run infra:restart && npm run test:dummy-data && npm run test:integration", - "test:oriole": "npm run infra:restart:oriole && npm run test:dummy-data && npm run test:integration", + "test:oriole": "npm run infra:restart:oriole && DATABASE_ENGINE=oriole npm run test:dummy-data && DATABASE_ENGINE=oriole npm run test:integration", "test:multigres": "npm run infra:restart:multigres && DATABASE_ENGINE=multigres npm run test:dummy-data && DATABASE_ENGINE=multigres npm run test:integration", "test:coverage": "npm run infra:restart && npm run test:dummy-data && npm run test:integration:coverage", "test:coverage:ci": "npm run infra:restart:ci && npm run test:dummy-data && npm run test:integration:coverage", - "test:oriole:ci": "npm run infra:restart:ci:oriole && npm run test:dummy-data && npm run test:integration", - "test:vector:oriole:pgvector": "VECTOR_BUCKET_PROVIDER=pgvector VECTOR_STORE_MIGRATIONS_ENABLED=true VECTOR_DATABASE_URL=postgresql://postgres:postgres@127.0.0.1:5432/postgres npm run infra:restart:oriole:pgvector && npm run test:dummy-data && VECTOR_BUCKET_PROVIDER=pgvector VECTOR_STORE_MIGRATIONS_ENABLED=true VECTOR_DATABASE_URL=postgresql://postgres:postgres@127.0.0.1:5432/postgres npm run test:integration -- src/test/pgvector-adapter.test.ts src/test/vectors-pgvector.test.ts", + "test:oriole:ci": "npm run infra:restart:ci:oriole && DATABASE_ENGINE=oriole npm run test:dummy-data && DATABASE_ENGINE=oriole npm run test:integration", + "test:vector:oriole:pgvector": "VECTOR_BUCKET_PROVIDER=pgvector VECTOR_STORE_MIGRATIONS_ENABLED=true VECTOR_DATABASE_URL=postgresql://postgres:postgres@127.0.0.1:5432/postgres npm run infra:restart:oriole:pgvector && DATABASE_ENGINE=oriole npm run test:dummy-data && DATABASE_ENGINE=oriole VECTOR_BUCKET_PROVIDER=pgvector VECTOR_STORE_MIGRATIONS_ENABLED=true VECTOR_DATABASE_URL=postgresql://postgres:postgres@127.0.0.1:5432/postgres npm run test:integration -- src/test/pgvector-adapter.test.ts src/test/vectors-pgvector.test.ts", "test:vector:multigres:pgvector": "VECTOR_BUCKET_PROVIDER=pgvector VECTOR_DATABASE_CREATE=false VECTOR_STORE_MIGRATIONS_ENABLED=true VECTOR_DATABASE_URL=postgresql://postgres:postgres@127.0.0.1:5432/postgres npm run infra:restart:multigres && npm run test:dummy-data && VECTOR_BUCKET_PROVIDER=pgvector VECTOR_DATABASE_CREATE=false VECTOR_STORE_MIGRATIONS_ENABLED=true VECTOR_DATABASE_URL=postgresql://postgres:postgres@127.0.0.1:5432/postgres npm run test:integration -- src/test/pgvector-adapter.test.ts src/test/vectors-pgvector.test.ts", "test:multigres:ci": "npm run infra:restart:ci:multigres && DATABASE_ENGINE=multigres npm run test:dummy-data && DATABASE_ENGINE=multigres npm run test:integration", "infra:stop": "docker compose --project-directory . --profile monitoring -f ./.docker/docker-compose-infra.yml down --remove-orphans", diff --git a/src/config.test.ts b/src/config.test.ts index 873d67232..d3338e41f 100644 --- a/src/config.test.ts +++ b/src/config.test.ts @@ -7,6 +7,7 @@ const CONFIG_ENV_KEYS = [ 'TENANT_POOL_CACHE_HIT_LOG_SAMPLE_RATE', 'TENANT_POOL_CACHE_MISS_LOG_SAMPLE_RATE', 'DATABASE_POOL_DRAIN_TIMEOUT', + 'DATABASE_ENGINE', 'REQUEST_HARD_LIMITS_ENABLED', 'STORAGE_S3_REQUEST_CHECKSUM_CALCULATION', 'STORAGE_S3_RESPONSE_CHECKSUM_VALIDATION', @@ -62,6 +63,19 @@ describe('tenant pool cache config parsing', () => { expect(config.requestHardLimitsEnabled).toBe(false) }) + test.each([ + 'postgres', + 'oriole', + 'multigres', + ] as const)('parses the %s database engine', async (databaseEngine) => { + setConfigEnv({ DATABASE_ENGINE: databaseEngine }) + + const { getConfig } = await import('./config') + const config = getConfig({ reload: true }) + + expect(config.databaseEngine).toBe(databaseEngine) + }) + test('parses request hard limits as disabled by default', async () => { setConfigEnv({}) diff --git a/src/config.ts b/src/config.ts index 768fc7854..dd8582cd1 100644 --- a/src/config.ts +++ b/src/config.ts @@ -5,7 +5,7 @@ import { SignJWT } from 'jose' export type StorageBackendType = 'file' | 's3' export type VectorBucketProvider = 's3' | 'pgvector' export type IcebergCatalogAuthType = 'sigv4' | 'token' -export type DatabaseEngine = 'postgres' | 'multigres' +export type DatabaseEngine = 'postgres' | 'oriole' | 'multigres' export type StorageS3ChecksumConfig = 'WHEN_SUPPORTED' | 'WHEN_REQUIRED' const DEFAULT_S3_UPLOAD_PART_SIZE = 16 * 1024 * 1024 const MIN_S3_UPLOAD_PART_SIZE = 5 * 1024 * 1024 @@ -256,11 +256,13 @@ export function normalizeDatabaseEngine(engine: string | null | undefined): Data return 'postgres' } - if (engine === 'postgres' || engine === 'multigres') { + if (engine === 'postgres' || engine === 'oriole' || engine === 'multigres') { return engine } - throw new Error(`Invalid database engine "${engine}". Expected "postgres" or "multigres".`) + throw new Error( + `Invalid database engine "${engine}". Expected "postgres", "oriole", or "multigres".` + ) } function normalizeStorageS3ChecksumConfig( diff --git a/src/http/routes/admin/queue.test.ts b/src/http/routes/admin/queue.test.ts index 78c51491a..f0d1c7361 100644 --- a/src/http/routes/admin/queue.test.ts +++ b/src/http/routes/admin/queue.test.ts @@ -19,7 +19,8 @@ describe('admin queue routes', () => { it('passes sbReqId to the move jobs task', async () => { vi.resetModules() - const { mergeConfig } = await import('../../../config') + const { getConfig, mergeConfig } = await import('../../../config') + getConfig() mergeConfig({ pgQueueEnable: true, adminApiKeys: 'test-admin-key', @@ -71,7 +72,8 @@ describe('admin queue routes', () => { it('rejects move jobs requests without queue names', async () => { vi.resetModules() - const { mergeConfig } = await import('../../../config') + const { getConfig, mergeConfig } = await import('../../../config') + getConfig() mergeConfig({ pgQueueEnable: true, adminApiKeys: 'test-admin-key', @@ -100,4 +102,42 @@ describe('admin queue routes', () => { await app.close() } }) + + it('rejects queue overflow restore on OrioleDB', async () => { + vi.resetModules() + + const { getConfig, mergeConfig } = await import('../../../config') + getConfig() + mergeConfig({ + pgQueueEnable: true, + adminApiKeys: 'test-admin-key', + databaseEngine: 'oriole', + }) + + const fastify = (await import('fastify')).default + const { default: routes } = await import('./queue') + + const app = fastify() + app.register(routes, { prefix: '/queue' }) + + try { + const response = await app.inject({ + method: 'POST', + url: '/queue/overflow/restore', + headers: { + apikey: 'test-admin-key', + }, + payload: { + name: 'webhooks', + }, + }) + + expect(response.statusCode).toBe(400) + expect(response.json()).toEqual({ + message: 'Queue overflow restore is not supported on OrioleDB', + }) + } finally { + await app.close() + } + }) }) diff --git a/src/http/routes/admin/queue.ts b/src/http/routes/admin/queue.ts index bd8548020..adfcc2a03 100644 --- a/src/http/routes/admin/queue.ts +++ b/src/http/routes/admin/queue.ts @@ -1,11 +1,41 @@ import { SYSTEM_TENANT } from '@internal/queue/constants' +import { + JOB_OVERFLOW_LIST_LIMIT_DEFAULT, + JOB_OVERFLOW_RESTORE_LIMIT_DEFAULT, + QueueOverflowStorePg, +} from '@internal/queue/overflow' +import { Queue } from '@internal/queue/queue' +import { parseCommaSeparatedList } from '@internal/strings' import { MoveJobs } from '@storage/events' import { FastifyInstance, RequestGenericInterface } from 'fastify' import { FromSchema } from 'json-schema-to-ts' import { getConfig } from '../../../config' import { registerApiKeyAuth } from '../../plugins/apikey' -const { pgQueueEnable } = getConfig() +const { databaseEngine, pgQueueEnable } = getConfig() + +function getQueueOverflowStore() { + return new QueueOverflowStorePg(Queue.getDb()) +} + +const nonBlankStringSchema = { + type: 'string', + minLength: 1, + pattern: '\\S', +} as const + +const stringListSchema = { + type: 'array', + minItems: 1, + maxItems: 1000, + items: nonBlankStringSchema, +} as const + +const positiveSafeIntegerSchema = { + type: 'integer', + minimum: 1, + maximum: Number.MAX_SAFE_INTEGER, +} as const const moveJobsSchema = { body: { @@ -26,10 +56,106 @@ const moveJobsSchema = { }, } as const +const listQueueOverflowSchema = { + description: 'List created pgBoss jobs from the live queue table or overflow backup table.', + querystring: { + type: 'object', + properties: { + source: { + type: 'string', + enum: ['job', 'backup'], + default: 'job', + }, + groupBy: { + type: 'string', + enum: ['summary', 'tenant'], + default: 'summary', + }, + name: nonBlankStringSchema, + eventTypes: { + ...nonBlankStringSchema, + description: 'Comma-separated event types to filter on.', + }, + tenantRefs: { + ...nonBlankStringSchema, + description: 'Comma-separated tenant refs to filter on.', + }, + limit: { + ...positiveSafeIntegerSchema, + default: JOB_OVERFLOW_LIST_LIMIT_DEFAULT, + }, + }, + additionalProperties: false, + }, +} as const + +const countQueueOverflowSchema = { + description: 'Count created pgBoss jobs in the live queue table.', +} as const + +const backupQueueOverflowSchema = { + description: 'Move created pgBoss jobs into the overflow backup table.', + body: { + type: 'object', + properties: { + name: nonBlankStringSchema, + eventTypes: stringListSchema, + tenantRefs: stringListSchema, + limit: positiveSafeIntegerSchema, + confirmAll: { + type: 'boolean', + description: 'Required when no queue, event type, or tenant filter is supplied.', + }, + }, + anyOf: [ + { required: ['name'] }, + { required: ['eventTypes'] }, + { required: ['tenantRefs'] }, + { + properties: { + confirmAll: { type: 'boolean', enum: [true] }, + }, + required: ['confirmAll'], + }, + ], + additionalProperties: false, + }, +} as const + +const restoreQueueOverflowSchema = { + description: + 'Restore created pgBoss jobs from the overflow backup table in batches. Conflicting rows are dropped from the backup table because the live job table wins.', + body: { + type: 'object', + properties: { + name: nonBlankStringSchema, + eventTypes: stringListSchema, + tenantRefs: stringListSchema, + limit: { + ...positiveSafeIntegerSchema, + default: JOB_OVERFLOW_RESTORE_LIMIT_DEFAULT, + }, + }, + additionalProperties: false, + }, +} as const + interface MoveJobsRequestInterface extends RequestGenericInterface { Body: FromSchema } +interface ListQueueOverflowRequestInterface extends RequestGenericInterface { + Querystring: FromSchema +} + +interface BackupQueueOverflowRequestInterface extends RequestGenericInterface { + Body: FromSchema +} + +interface RestoreQueueOverflowRequestInterface extends RequestGenericInterface { + Body: FromSchema +} + export default async function routes(fastify: FastifyInstance) { registerApiKeyAuth(fastify) @@ -56,4 +182,81 @@ export default async function routes(fastify: FastifyInstance) { return reply.send({ message: 'Move jobs scheduled' }) } ) + + fastify.get( + '/overflow', + { schema: { ...listQueueOverflowSchema, tags: ['queue'] } }, + async (req, reply) => { + if (!pgQueueEnable) { + return reply.status(400).send({ message: 'Queue is not enabled' }) + } + + const store = getQueueOverflowStore() + const data = await store.list({ + source: req.query.source, + groupBy: req.query.groupBy, + name: req.query.name, + eventTypes: parseCommaSeparatedList(req.query.eventTypes), + tenantRefs: parseCommaSeparatedList(req.query.tenantRefs), + limit: req.query.limit, + signal: req.signals.disconnect.signal, + }) + + return reply.send(data) + } + ) + + fastify.get( + '/overflow/count', + { schema: { ...countQueueOverflowSchema, tags: ['queue'] } }, + async (req, reply) => { + if (!pgQueueEnable) { + return reply.status(400).send({ message: 'Queue is not enabled' }) + } + + const store = getQueueOverflowStore() + const data = await store.countCreated({ signal: req.signals.disconnect.signal }) + return reply.send(data) + } + ) + + fastify.post( + '/overflow/backup', + { schema: { ...backupQueueOverflowSchema, tags: ['queue'] } }, + async (req, reply) => { + if (!pgQueueEnable) { + return reply.status(400).send({ message: 'Queue is not enabled' }) + } + + const store = getQueueOverflowStore() + const data = await store.backup({ + ...req.body, + signal: req.signals.disconnect.signal, + }) + return reply.send(data) + } + ) + + fastify.post( + '/overflow/restore', + { schema: { ...restoreQueueOverflowSchema, tags: ['queue'] } }, + async (req, reply) => { + if (!pgQueueEnable) { + return reply.status(400).send({ message: 'Queue is not enabled' }) + } + + if (databaseEngine === 'oriole') { + return reply + .status(400) + .send({ message: 'Queue overflow restore is not supported on OrioleDB' }) + } + + const store = getQueueOverflowStore() + const data = await store.restore({ + ...req.body, + signal: req.signals.disconnect.signal, + }) + return reply.send(data) + } + ) } diff --git a/src/internal/database/pg-connection.test.ts b/src/internal/database/pg-connection.test.ts index 90d0c829c..329666327 100644 --- a/src/internal/database/pg-connection.test.ts +++ b/src/internal/database/pg-connection.test.ts @@ -325,6 +325,28 @@ describe('getPgCancelConnectionTarget', () => { }) describe('PgPoolExecutor', () => { + it('keeps no-signal transaction checkout on the direct pool path', async () => { + const addEventListener = vi.spyOn(AbortSignal.prototype, 'addEventListener') + const client = Object.assign(new EventEmitter(), { + query: vi.fn().mockResolvedValue({ rows: [] }), + release: vi.fn(), + }) as unknown as PoolClient & EventEmitter + const pool = { + connect: vi.fn().mockResolvedValue(client), + } as unknown as Pool + const executor = new PgPoolExecutor(pool) + + try { + const transaction = await executor.beginTransaction() + await transaction.commit() + + expect(pool.connect).toHaveBeenCalledOnce() + expect(addEventListener).not.toHaveBeenCalled() + } finally { + addEventListener.mockRestore() + } + }) + it('tracks checked-out client errors during direct queries', async () => { const socketError = new Error('socket reset') const client = Object.assign(new EventEmitter(), { @@ -558,6 +580,50 @@ describe('PgPoolExecutor', () => { expect(pool.connect).not.toHaveBeenCalled() }) + it('rejects an aborted transaction checkout and releases a late client', async () => { + const controller = new AbortController() + let resolveConnect: (client: PoolClient) => void = () => undefined + const connectPromise = new Promise((resolve) => { + resolveConnect = resolve + }) + const client = Object.assign(new EventEmitter(), { + query: vi.fn().mockResolvedValue({ rows: [] }), + release: vi.fn(), + }) as unknown as PoolClient & EventEmitter + const pool = { + connect: vi.fn(() => connectPromise), + } as unknown as Pool + const executor = new PgPoolExecutor(pool) + + const transactionPromise = executor.beginTransaction({ + signal: controller.signal, + }) + controller.abort() + + const result = await Promise.race([ + transactionPromise.then( + () => ({ status: 'resolved' as const }), + (error) => ({ status: 'rejected' as const, error }) + ), + waitForDrainCheck().then(() => ({ status: 'pending' as const })), + ]) + + resolveConnect(client) + await waitForDrainCheck() + + expect(result).toMatchObject({ + status: 'rejected', + error: { + name: 'AbortError', + code: 'ABORT_ERR', + message: 'Query was aborted', + }, + }) + expect(client.query).not.toHaveBeenCalled() + expect(client.release).toHaveBeenCalledOnce() + expect(client.release).toHaveBeenCalledWith() + }) + it('rejects aborted queries without waiting for the pg query to settle', async () => { const controller = new AbortController() const release = vi.fn() diff --git a/src/internal/database/pg-connection.ts b/src/internal/database/pg-connection.ts index 6d5deeb66..21c5d3eae 100644 --- a/src/internal/database/pg-connection.ts +++ b/src/internal/database/pg-connection.ts @@ -48,7 +48,7 @@ interface PgTransactionOptions { statementTimeoutMs?: number } -type PgBeginTransactionOptions = TransactionOptions & PgTransactionOptions +export type PgBeginTransactionOptions = TransactionOptions & PgTransactionOptions & PgQueryOptions const defaultStatementTimeoutMs = normalizeStatementTimeoutMs(databaseStatementTimeout) @@ -438,9 +438,12 @@ export class PgPoolExecutor implements PgTransactionalExecutor { } async beginTransaction(options?: PgBeginTransactionOptions): Promise { + const signal = options?.signal let client: PoolClient try { - client = await this.pool.connect() + client = signal + ? await acquirePoolClientWithSignal(this.pool, signal) + : await this.pool.connect() } catch (e) { if (isConnectionTimeoutError(e)) { throw ERRORS.DatabaseTimeout(e) @@ -455,7 +458,7 @@ export class PgPoolExecutor implements PgTransactionalExecutor { }) try { - await transaction.runSetupQuery(buildBeginStatement(options)) + await transaction.runSetupQuery(buildBeginStatement(options), signal ? { signal } : undefined) return transaction } catch (e) { if (!transaction.isCompleted()) { @@ -835,6 +838,62 @@ export function createAbortError(): Error & { code: string } { return error } +function acquirePoolClientWithSignal(pool: Pool, signal: AbortSignal): Promise { + assertValidSignal(signal) + + // pg-pool cannot remove a pending checkout. Reject the caller immediately, + // then release a client that arrives after the signal aborts. + return new Promise((resolve, reject) => { + let settled = false + const cleanup = () => signal.removeEventListener('abort', onAbort) + const onAbort = () => { + if (settled) { + return + } + + settled = true + cleanup() + reject(createAbortError()) + } + + signal.addEventListener('abort', onAbort, { once: true }) + + let acquisition: Promise + try { + acquisition = pool.connect() + } catch (error) { + settled = true + cleanup() + reject(error) + return + } + + void acquisition + .then( + (client) => { + if (settled) { + client.release() + return + } + + settled = true + cleanup() + resolve(client) + }, + (error) => { + if (settled) { + return + } + + settled = true + cleanup() + reject(error) + } + ) + .catch(() => undefined) + }) +} + async function runPgQuery( client: PoolClient, statement: string | PgStatement, diff --git a/src/internal/queue/constants.ts b/src/internal/queue/constants.ts index 8c1f68b10..d5dd90a2e 100644 --- a/src/internal/queue/constants.ts +++ b/src/internal/queue/constants.ts @@ -4,3 +4,4 @@ */ export const SYSTEM_TENANT_REF = 'SYSTEM_TENANT' as const export const SYSTEM_TENANT = { ref: '', host: '' } as const +export const PG_BOSS_SCHEMA = 'pgboss_v10' diff --git a/src/internal/queue/database.ts b/src/internal/queue/database.ts index b71789bff..a7ca12e74 100644 --- a/src/internal/queue/database.ts +++ b/src/internal/queue/database.ts @@ -2,7 +2,11 @@ import EventEmitter from 'node:events' import { ERRORS } from '@internal/errors' import pg from 'pg' import { Db } from 'pg-boss' -import { PgExecutor } from '../database/pg-connection' +import { + type PgBeginTransactionOptions, + type PgExecutor, + PgPoolExecutor, +} from '../database/pg-connection' export { quoteIdentifier } from '../database/sql' @@ -32,6 +36,14 @@ export class QueueDB extends EventEmitter implements Db { await this.pool?.end() } + beginTransaction(options?: PgBeginTransactionOptions) { + if (!this.opened || !this.pool) { + throw ERRORS.InternalError(undefined, `QueueDB not opened ${this.opened}`) + } + + return new PgPoolExecutor(this.pool).beginTransaction(options) + } + protected async useTransaction(fn: (client: pg.PoolClient) => Promise): Promise { if (!this.opened || !this.pool) { throw ERRORS.InternalError(undefined, `QueueDB not opened ${this.opened}`) diff --git a/src/internal/queue/index.ts b/src/internal/queue/index.ts index 097a4b411..83ff1aae3 100644 --- a/src/internal/queue/index.ts +++ b/src/internal/queue/index.ts @@ -1,3 +1,4 @@ export * from './constants' export * from './event' +export * from './overflow' export * from './queue' diff --git a/src/internal/queue/overflow.test.ts b/src/internal/queue/overflow.test.ts new file mode 100644 index 000000000..79116a09c --- /dev/null +++ b/src/internal/queue/overflow.test.ts @@ -0,0 +1,358 @@ +import type { PgStatement, PgTransaction, PgTransactionalExecutor } from '@internal/database' +import { + buildQueueOverflowWhereClause, + normalizeQueueOverflowFilters, + QUEUE_OVERFLOW_UNSCOPED_BACKUP_MESSAGE, + QueueOverflowStorePg, +} from './overflow' + +function statementText(statement: string | PgStatement): string { + return typeof statement === 'string' ? statement : statement.text +} + +function statementValues(statement: string | PgStatement): unknown[] | undefined { + return typeof statement === 'string' ? undefined : statement.values +} + +function createMockDatabase( + queryImplementation: (statement: string | PgStatement) => Promise<{ rows: unknown[] }> +) { + const query = vi.fn(queryImplementation) + const commit = vi.fn().mockResolvedValue(undefined) + const rollback = vi.fn().mockResolvedValue(undefined) + const transaction = { query, commit, rollback } as unknown as PgTransaction + const beginTransaction = vi.fn().mockResolvedValue(transaction) + const db = { beginTransaction } as Pick + + return { beginTransaction, commit, db, query, rollback } +} + +describe('normalizeQueueOverflowFilters', () => { + it('trims values and removes empty strings', () => { + expect( + normalizeQueueOverflowFilters({ + name: ' webhooks ', + eventTypes: [' ObjectRemoved:Delete ', ''], + tenantRefs: [' tenant-a ', 'tenant-a', ' '], + }) + ).toEqual({ + name: 'webhooks', + eventTypes: ['ObjectRemoved:Delete'], + tenantRefs: ['tenant-a'], + }) + }) +}) + +describe('buildQueueOverflowWhereClause', () => { + it('always scopes queries to created jobs', () => { + expect(buildQueueOverflowWhereClause({})).toEqual({ + sql: 'state = $1', + values: ['created'], + }) + }) + + it('uses native pg parameters and array bindings in a stable order', () => { + expect( + buildQueueOverflowWhereClause({ + name: ' webhooks ', + eventTypes: ['ObjectRemoved:Delete', ' ObjectCreated:Put '], + tenantRefs: ['tenant-b', 'tenant-a'], + }) + ).toEqual({ + sql: "state = $1 AND name = $2 AND data->'event'->>'type' = ANY($3::text[]) AND data->'tenant'->>'ref' = ANY($4::text[])", + values: [ + 'created', + 'webhooks', + ['ObjectRemoved:Delete', 'ObjectCreated:Put'], + ['tenant-b', 'tenant-a'], + ], + }) + }) +}) + +describe('QueueOverflowStorePg', () => { + it('counts created jobs without grouping or window functions', async () => { + const database = createMockDatabase(async (statement) => { + const text = statementText(statement) + + if (text.includes("set_config('statement_timeout'")) { + return { rows: [] } + } + + return { rows: [{ total_count: '17' }] } + }) + const store = new QueueOverflowStorePg(database.db) + + await expect(store.countCreated()).resolves.toEqual({ totalCount: 17 }) + + const countCall = database.query.mock.calls.find(([statement]) => + statementText(statement).includes('COUNT(*)::bigint AS total_count') + ) + expect(countCall).toBeDefined() + expect(statementText(countCall?.[0] as string | PgStatement)).not.toContain('GROUP BY') + expect(statementText(countCall?.[0] as string | PgStatement)).not.toContain('OVER (') + expect(statementValues(countCall?.[0] as string | PgStatement)).toEqual(['created']) + }) + + it('passes the abort signal while acquiring the maintenance transaction', async () => { + const controller = new AbortController() + const database = createMockDatabase(async (statement) => { + const text = statementText(statement) + + if (text.includes('COUNT(*)::bigint AS total_count')) { + return { rows: [{ total_count: '0' }] } + } + + return { rows: [] } + }) + const store = new QueueOverflowStorePg(database.db) + + await store.countCreated({ signal: controller.signal }) + + expect(database.beginTransaction).toHaveBeenCalledWith({ signal: controller.signal }) + }) + + it('reports exact totals and whether grouped list results were truncated', async () => { + const database = createMockDatabase(async (statement) => { + const text = statementText(statement) + + if (text.includes("set_config('statement_timeout'")) { + return { rows: [] } + } + + return { + rows: [ + { + count: '7', + event_type: 'ObjectRemoved:Delete', + group_count: '3', + name: 'webhooks', + total_count: '11', + }, + ], + } + }) + const store = new QueueOverflowStorePg(database.db) + + await expect(store.list({ limit: 1 })).resolves.toEqual({ + sourceTableExists: true, + data: [{ count: 7, eventType: 'ObjectRemoved:Delete', name: 'webhooks' }], + filters: { eventTypes: undefined, name: undefined, tenantRefs: undefined }, + groupBy: 'summary', + groupCount: 3, + hasMore: true, + source: 'job', + totalCount: 11, + }) + + expect(database.commit).toHaveBeenCalledOnce() + expect(database.rollback).not.toHaveBeenCalled() + expect(statementValues(database.query.mock.calls[0][0])).toEqual(['0', '30s']) + }) + + it('rejects an unfiltered backup unless all jobs were explicitly confirmed', async () => { + const database = createMockDatabase(async () => ({ rows: [] })) + const store = new QueueOverflowStorePg(database.db) + + await expect(store.backup({})).rejects.toThrow(QUEUE_OVERFLOW_UNSCOPED_BACKUP_MESSAGE) + await expect(store.backup({ name: ' ' })).rejects.toThrow( + QUEUE_OVERFLOW_UNSCOPED_BACKUP_MESSAGE + ) + expect(database.beginTransaction).not.toHaveBeenCalled() + }) + + it('moves matching tuple keys atomically', async () => { + const database = createMockDatabase(async (statement) => { + const text = statementText(statement) + + if (text.includes('SELECT to_regclass')) { + return { rows: [{ table_name: null }] } + } + + if (text.includes('WITH selected AS')) { + return { rows: [{ moved_count: '2' }] } + } + + return { rows: [] } + }) + const store = new QueueOverflowStorePg(database.db) + + await expect( + store.backup({ + name: 'webhooks', + eventTypes: ['ObjectRemoved:Delete'], + limit: 2, + }) + ).resolves.toEqual({ + backupTableCreated: true, + filters: { + eventTypes: ['ObjectRemoved:Delete'], + name: 'webhooks', + tenantRefs: undefined, + }, + limit: 2, + movedCount: 2, + }) + + const moveCall = database.query.mock.calls.find(([statement]) => + statementText(statement).includes('WITH selected AS') + ) + if (!moveCall) { + throw new Error('Expected queue overflow move query') + } + + const moveStatement = moveCall[0] + const moveSql = statementText(moveStatement) + expect(moveSql).toContain('SELECT name, id') + expect(moveSql).toContain('source_job.name = selected.name') + expect(moveSql).toContain('source_job.id = selected.id') + expect(moveSql).toContain('INSERT INTO "pgboss_v10"."job_overflow_backup"') + expect(moveSql).toContain('SELECT * FROM moved') + expect(statementValues(moveStatement)).toEqual([ + 'created', + 'webhooks', + ['ObjectRemoved:Delete'], + 2, + ]) + }) + + it('commits a completed move if the signal aborts before finalization', async () => { + const controller = new AbortController() + let markMoveStarted: () => void = () => undefined + let resolveMove: (result: { rows: unknown[] }) => void = () => undefined + const moveStarted = new Promise((resolve) => { + markMoveStarted = resolve + }) + const moveResult = new Promise<{ rows: unknown[] }>((resolve) => { + resolveMove = resolve + }) + const database = createMockDatabase((statement) => { + const text = statementText(statement) + + if (text.includes('SELECT to_regclass')) { + return Promise.resolve({ rows: [{ table_name: null }] }) + } + + if (text.includes('WITH selected AS')) { + markMoveStarted() + return moveResult + } + + return Promise.resolve({ rows: [] }) + }) + const store = new QueueOverflowStorePg(database.db) + const operation = store.backup({ name: 'webhooks', signal: controller.signal }) + + await moveStarted + resolveMove({ rows: [{ moved_count: '1' }] }) + controller.abort() + + await expect(operation).resolves.toMatchObject({ movedCount: 1 }) + expect(database.commit).toHaveBeenCalledOnce() + expect(database.rollback).not.toHaveBeenCalled() + }) + + it('restores non-conflicting rows and reports conflicts dropped from the backup', async () => { + const database = createMockDatabase(async (statement) => { + const text = statementText(statement) + + if (text.includes('SELECT to_regclass')) { + return { rows: [{ table_name: 'pgboss_v10.job_overflow_backup' }] } + } + + if (text.includes('ON CONFLICT DO NOTHING')) { + return { + rows: [ + { + selected_count: '3', + moved_count: '2', + }, + ], + } + } + + return { rows: [] } + }) + const store = new QueueOverflowStorePg(database.db) + + await expect(store.restore({ name: 'webhooks', limit: 3 })).resolves.toEqual({ + backupTableExists: true, + conflictCount: 1, + filters: { eventTypes: undefined, name: 'webhooks', tenantRefs: undefined }, + hasMore: true, + limit: 3, + movedCount: 2, + }) + + const restoreCall = database.query.mock.calls.find(([statement]) => + statementText(statement).includes('ON CONFLICT DO NOTHING') + ) + expect(restoreCall).toBeDefined() + expect(statementText(restoreCall?.[0] as string | PgStatement)).toContain( + 'DELETE FROM "pgboss_v10"."job_overflow_backup"' + ) + expect(statementText(restoreCall?.[0] as string | PgStatement)).toContain('USING selected') + expect(statementText(restoreCall?.[0] as string | PgStatement)).not.toContain('USING inserted') + expect(statementText(restoreCall?.[0] as string | PgStatement)).toContain( + 'source_job.name = selected.name' + ) + expect(statementText(restoreCall?.[0] as string | PgStatement)).toContain( + 'source_job.id = selected.id' + ) + expect(statementText(restoreCall?.[0] as string | PgStatement)).toContain( + '(SELECT COUNT(*) FROM inserted) AS moved_count' + ) + expect(statementText(restoreCall?.[0] as string | PgStatement)).toContain( + 'SELECT * FROM selected' + ) + expect(statementValues(restoreCall?.[0] as string | PgStatement)).toEqual([ + 'created', + 'webhooks', + 3, + ]) + }) + + it('returns the new default limit and no more work when the backup table is missing', async () => { + const database = createMockDatabase(async (statement) => { + const text = statementText(statement) + + if (text.includes('SELECT to_regclass')) { + return { rows: [{ table_name: null }] } + } + + return { rows: [] } + }) + const store = new QueueOverflowStorePg(database.db) + + await expect(store.restore({})).resolves.toEqual({ + backupTableExists: false, + conflictCount: 0, + filters: { eventTypes: undefined, name: undefined, tenantRefs: undefined }, + hasMore: false, + limit: 10000, + movedCount: 0, + }) + }) + + it('rolls back and preserves the primary move failure', async () => { + const moveError = new Error('move failed') + const database = createMockDatabase(async (statement) => { + const text = statementText(statement) + + if (text.includes('SELECT to_regclass')) { + return { rows: [{ table_name: 'pgboss_v10.job_overflow_backup' }] } + } + + if (text.includes('WITH selected AS')) { + throw moveError + } + + return { rows: [] } + }) + const store = new QueueOverflowStorePg(database.db) + + await expect(store.restore({ limit: 1 })).rejects.toBe(moveError) + expect(database.rollback).toHaveBeenCalledOnce() + expect(database.commit).not.toHaveBeenCalled() + }) +}) diff --git a/src/internal/queue/overflow.ts b/src/internal/queue/overflow.ts new file mode 100644 index 000000000..be1858beb --- /dev/null +++ b/src/internal/queue/overflow.ts @@ -0,0 +1,554 @@ +import { normalizeStringList } from '@internal/strings' +import type { QueryResultRow } from 'pg' +import type { PgExecutor, PgTransaction, PgTransactionalExecutor } from '../database/pg-connection' +import { quoteQualifiedIdentifier } from '../database/sql' +import { PG_BOSS_SCHEMA } from './constants' + +const CREATED_STATE = 'created' +const QUEUE_OVERFLOW_ADVISORY_LOCK_KEY = '-5525285245963000612' +export const JOB_OVERFLOW_LIST_LIMIT_DEFAULT = 50 +export const JOB_OVERFLOW_RESTORE_LIMIT_DEFAULT = 10000 +export const QUEUE_OVERFLOW_UNSCOPED_BACKUP_MESSAGE = + 'Backup requires at least one queue, event type, or tenant filter unless confirmAll is true' + +export type QueueOverflowSource = 'job' | 'backup' +export type QueueOverflowGroupBy = 'summary' | 'tenant' + +export interface QueueOverflowFilters { + eventTypes?: readonly string[] + name?: string + tenantRefs?: readonly string[] +} + +interface QueueOverflowOperationOptions { + signal?: AbortSignal +} + +export interface ListQueueOverflowOptions + extends QueueOverflowFilters, + QueueOverflowOperationOptions { + groupBy?: QueueOverflowGroupBy + limit?: number + source?: QueueOverflowSource +} + +export interface MoveQueueOverflowOptions + extends QueueOverflowFilters, + QueueOverflowOperationOptions { + limit?: number +} + +export interface BackupQueueOverflowOptions extends MoveQueueOverflowOptions { + confirmAll?: boolean +} + +export interface QueueOverflowSummaryRow { + count: number + eventType: string | null + name: string +} + +export interface QueueOverflowTenantRow { + count: number + tenantRef: string | null +} + +interface QueueOverflowWhereClause { + sql: string + values: unknown[] +} + +interface QueueOverflowAggregateRow extends QueryResultRow { + count: number | string + group_count: number | string + total_count: number | string +} + +interface QueueOverflowTotalCountRow extends QueryResultRow { + total_count: number | string +} + +interface QueueOverflowSummaryDbRow extends QueueOverflowAggregateRow { + event_type: string | null + name: string +} + +interface QueueOverflowTenantDbRow extends QueueOverflowAggregateRow { + tenant_ref: string | null +} + +interface QueueOverflowCountRow extends QueryResultRow { + moved_count: number | string +} + +interface QueueOverflowRestoreCountRow extends QueueOverflowCountRow { + selected_count: number | string +} + +interface QueueOverflowTableRow extends QueryResultRow { + table_name: string | null +} + +function normalizeOverflowString(value: string | undefined) { + const normalized = value?.trim() + return normalized ? normalized : undefined +} + +export function normalizeQueueOverflowFilters(filters: QueueOverflowFilters): QueueOverflowFilters { + return { + name: normalizeOverflowString(filters.name), + eventTypes: normalizeStringList(filters.eventTypes), + tenantRefs: normalizeStringList(filters.tenantRefs), + } +} + +export function hasQueueOverflowFilters(filters: QueueOverflowFilters): boolean { + const normalized = normalizeQueueOverflowFilters(filters) + return Boolean(normalized.name || normalized.eventTypes?.length || normalized.tenantRefs?.length) +} + +export function isQueueOverflowBackupScoped(options: BackupQueueOverflowOptions): boolean { + return options.confirmAll === true || hasQueueOverflowFilters(options) +} + +export function buildQueueOverflowWhereClause( + filters: QueueOverflowFilters +): QueueOverflowWhereClause { + const normalizedFilters = normalizeQueueOverflowFilters(filters) + const clauses = ['state = $1'] + const values: unknown[] = [CREATED_STATE] + + if (normalizedFilters.name) { + values.push(normalizedFilters.name) + clauses.push(`name = $${values.length}`) + } + + if (normalizedFilters.eventTypes?.length) { + values.push(normalizedFilters.eventTypes) + clauses.push(`data->'event'->>'type' = ANY($${values.length}::text[])`) + } + + if (normalizedFilters.tenantRefs?.length) { + values.push(normalizedFilters.tenantRefs) + clauses.push(`data->'tenant'->>'ref' = ANY($${values.length}::text[])`) + } + + return { + sql: clauses.join(' AND '), + values, + } +} + +export class QueueOverflowStorePg { + private readonly backupTableName: string + private readonly jobTable: string + private readonly backupTable: string + + constructor( + private readonly db: Pick, + schema = PG_BOSS_SCHEMA + ) { + const jobTableName = `${schema}.job` + this.backupTableName = `${schema}.job_overflow_backup` + this.jobTable = quoteQualifiedIdentifier(jobTableName) + this.backupTable = quoteQualifiedIdentifier(this.backupTableName) + } + + async countCreated(options: QueueOverflowOperationOptions = {}) { + return this.withMaintenanceTransaction(options.signal, async (transaction) => { + const result = await transaction.query( + { + text: ` + SELECT COUNT(*)::bigint AS total_count + FROM ${this.jobTable} + WHERE state = $1 + `, + values: [CREATED_STATE], + }, + { signal: options.signal } + ) + + return { + totalCount: Number(result.rows[0]?.total_count ?? 0), + } + }) + } + + async list(options: ListQueueOverflowOptions) { + const source = options.source ?? 'job' + const groupBy = options.groupBy ?? 'summary' + const limit = options.limit ?? JOB_OVERFLOW_LIST_LIMIT_DEFAULT + const filters = normalizeQueueOverflowFilters(options) + + assertPositiveSafeInteger(limit, 'limit') + + return this.withMaintenanceTransaction(options.signal, async (transaction) => { + const sourceTableExists = + source === 'backup' ? await this.backupTableExists(transaction, options.signal) : true + + if (!sourceTableExists) { + return { + sourceTableExists, + data: [] as QueueOverflowSummaryRow[] | QueueOverflowTenantRow[], + filters, + groupBy, + groupCount: 0, + hasMore: false, + source, + totalCount: 0, + } + } + + const tableName = source === 'backup' ? this.backupTable : this.jobTable + const whereClause = buildQueueOverflowWhereClause(filters) + const values = [...whereClause.values, limit] + const limitPlaceholder = `$${values.length}` + + if (groupBy === 'tenant') { + const result = await transaction.query( + { + text: ` + SELECT + data->'tenant'->>'ref' AS tenant_ref, + COUNT(*)::bigint AS count, + COUNT(*) OVER ()::bigint AS group_count, + SUM(COUNT(*)) OVER ()::bigint AS total_count + FROM ${tableName} + WHERE ${whereClause.sql} + GROUP BY data->'tenant'->>'ref' + ORDER BY count DESC, tenant_ref ASC + LIMIT ${limitPlaceholder} + `, + values, + }, + { signal: options.signal } + ) + + const groupCount = Number(result.rows[0]?.group_count ?? 0) + + return { + sourceTableExists, + data: result.rows.map((row) => ({ + count: Number(row.count), + tenantRef: row.tenant_ref, + })), + filters, + groupBy, + groupCount, + hasMore: groupCount > result.rows.length, + source, + totalCount: Number(result.rows[0]?.total_count ?? 0), + } + } + + const result = await transaction.query( + { + text: ` + SELECT + name, + data->'event'->>'type' AS event_type, + COUNT(*)::bigint AS count, + COUNT(*) OVER ()::bigint AS group_count, + SUM(COUNT(*)) OVER ()::bigint AS total_count + FROM ${tableName} + WHERE ${whereClause.sql} + GROUP BY name, data->'event'->>'type' + ORDER BY count DESC, name ASC, event_type ASC + LIMIT ${limitPlaceholder} + `, + values, + }, + { signal: options.signal } + ) + + const groupCount = Number(result.rows[0]?.group_count ?? 0) + + return { + sourceTableExists, + data: result.rows.map((row) => ({ + count: Number(row.count), + eventType: row.event_type, + name: row.name, + })), + filters, + groupBy, + groupCount, + hasMore: groupCount > result.rows.length, + source, + totalCount: Number(result.rows[0]?.total_count ?? 0), + } + }) + } + + async backup(options: BackupQueueOverflowOptions) { + if (!isQueueOverflowBackupScoped(options)) { + throw new Error(QUEUE_OVERFLOW_UNSCOPED_BACKUP_MESSAGE) + } + + assertOptionalPositiveSafeInteger(options.limit, 'limit') + const filters = normalizeQueueOverflowFilters(options) + + return this.withMaintenanceTransaction(options.signal, async (transaction) => { + await this.acquireMaintenanceLock(transaction, options.signal) + const backupTableCreated = await this.ensureBackupTable(transaction, options.signal) + + await transaction.query(`LOCK TABLE ${this.jobTable} IN SHARE ROW EXCLUSIVE MODE`, { + signal: options.signal, + }) + + const movedCount = await this.moveJobs( + transaction, + filters, + this.jobTable, + this.backupTable, + options.limit, + options.signal + ) + + return { + backupTableCreated, + filters, + limit: options.limit ?? null, + movedCount, + } + }) + } + + async restore(options: MoveQueueOverflowOptions) { + const limit = options.limit ?? JOB_OVERFLOW_RESTORE_LIMIT_DEFAULT + assertPositiveSafeInteger(limit, 'limit') + const filters = normalizeQueueOverflowFilters(options) + + return this.withMaintenanceTransaction(options.signal, async (transaction) => { + await this.acquireMaintenanceLock(transaction, options.signal) + const backupTableExists = await this.backupTableExists(transaction, options.signal) + + if (!backupTableExists) { + return { + backupTableExists, + conflictCount: 0, + filters, + hasMore: false, + limit, + movedCount: 0, + } + } + + const { conflictCount, hasMore, movedCount } = await this.restoreJobs( + transaction, + filters, + limit, + options.signal + ) + + return { + backupTableExists, + conflictCount, + filters, + hasMore, + limit, + movedCount, + } + }) + } + + private async backupTableExists(db: PgExecutor, signal?: AbortSignal) { + const result = await db.query( + { + text: 'SELECT to_regclass($1) AS table_name', + values: [this.backupTableName], + }, + { signal } + ) + + return Boolean(result.rows[0]?.table_name) + } + + private async ensureBackupTable(db: PgExecutor, signal?: AbortSignal) { + const existed = await this.backupTableExists(db, signal) + + await db.query( + `CREATE TABLE IF NOT EXISTS ${this.backupTable} (LIKE ${this.jobTable} INCLUDING ALL)`, + { + signal, + } + ) + + return !existed + } + + private async moveJobs( + db: PgExecutor, + filters: QueueOverflowFilters, + sourceTable: string, + targetTable: string, + limit?: number, + signal?: AbortSignal + ) { + const whereClause = buildQueueOverflowWhereClause(filters) + const values = [...whereClause.values] + let selectionBoundary = '' + + if (limit !== undefined) { + values.push(limit) + selectionBoundary = `ORDER BY name, id LIMIT $${values.length}` + } + + const result = await db.query( + { + text: ` + WITH selected AS ( + SELECT name, id + FROM ${sourceTable} + WHERE ${whereClause.sql} + ${selectionBoundary} + ), + moved AS ( + DELETE FROM ${sourceTable} AS source_job + USING selected + WHERE source_job.name = selected.name + AND source_job.id = selected.id + RETURNING source_job.* + ), + inserted AS ( + INSERT INTO ${targetTable} + SELECT * FROM moved + RETURNING 1 + ) + SELECT COUNT(*)::bigint AS moved_count FROM inserted + `, + values, + }, + { signal } + ) + + return Number(result.rows[0]?.moved_count ?? 0) + } + + private async restoreJobs( + db: PgExecutor, + filters: QueueOverflowFilters, + limit: number, + signal?: AbortSignal + ) { + const whereClause = buildQueueOverflowWhereClause(filters) + const values = [...whereClause.values] + values.push(limit) + const limitPlaceholder = `$${values.length}` + const result = await db.query( + { + text: ` + WITH selected AS ( + SELECT * + FROM ${this.backupTable} + WHERE ${whereClause.sql} + ORDER BY name, id + LIMIT ${limitPlaceholder} + ), + inserted AS ( + INSERT INTO ${this.jobTable} + SELECT * FROM selected + ON CONFLICT DO NOTHING + RETURNING 1 + ), + deleted AS ( + DELETE FROM ${this.backupTable} AS source_job + USING selected + WHERE source_job.name = selected.name + AND source_job.id = selected.id + RETURNING 1 + ) + SELECT + (SELECT COUNT(*) FROM selected) AS selected_count, + (SELECT COUNT(*) FROM inserted) AS moved_count + `, + values, + }, + { signal } + ) + + const selectedCount = Number(result.rows[0]?.selected_count ?? 0) + const movedCount = Number(result.rows[0]?.moved_count ?? 0) + + return { + conflictCount: selectedCount - movedCount, + hasMore: selectedCount === limit, + movedCount, + } + } + + private async acquireMaintenanceLock(db: PgExecutor, signal?: AbortSignal): Promise { + await db.query( + { + text: 'SELECT pg_advisory_xact_lock($1::bigint)', + values: [QUEUE_OVERFLOW_ADVISORY_LOCK_KEY], + }, + { signal } + ) + } + + private async withMaintenanceTransaction( + signal: AbortSignal | undefined, + fn: (transaction: PgTransaction) => Promise + ): Promise { + const transaction = await this.db.beginTransaction({ signal }) + + try { + await transaction.query( + { + text: ` + SELECT + set_config('statement_timeout', $1, true), + set_config('lock_timeout', $2, true) + `, + values: ['0', '30s'], + }, + { signal } + ) + + const result = await fn(transaction) + await transaction.commit() + return result + } catch (error) { + try { + await transaction.rollback() + } catch (rollbackError) { + await logQueueOverflowRollbackFailure(error, rollbackError) + } + + throw error + } + } +} + +async function logQueueOverflowRollbackFailure( + originalError: unknown, + rollbackError: unknown +): Promise { + try { + const { logger, logSchema } = await import('@internal/monitoring') + logSchema.warning(logger, '[QueueOverflow] Failed to rollback maintenance transaction', { + type: 'pgboss', + error: rollbackError, + metadata: JSON.stringify({ originalError: String(originalError) }), + }) + } catch (loggingError) { + console.error('[QueueOverflow] Failed to log maintenance transaction rollback failure', { + originalError, + rollbackError, + loggingError, + }) + } +} + +function assertOptionalPositiveSafeInteger(value: number | undefined, field: string): void { + if (value !== undefined) { + assertPositiveSafeInteger(value, field) + } +} + +function assertPositiveSafeInteger(value: number, field: string): void { + if (!Number.isSafeInteger(value) || value <= 0) { + throw new Error(`${field} must be a positive safe integer`) + } +} diff --git a/src/internal/queue/queue.ts b/src/internal/queue/queue.ts index 940cfbea1..09d94d817 100644 --- a/src/internal/queue/queue.ts +++ b/src/internal/queue/queue.ts @@ -10,6 +10,7 @@ import { queueJobError, queueJobRetryFailed, } from '../monitoring/metrics' +import { PG_BOSS_SCHEMA } from './constants' import { Event } from './event' type RegisteredEvent = { @@ -23,13 +24,14 @@ type RegisteredEvent = { name: string } -export const PG_BOSS_SCHEMA = 'pgboss_v10' +export { PG_BOSS_SCHEMA } from './constants' + 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 { diff --git a/src/internal/strings/boolean.test.ts b/src/internal/strings/boolean.test.ts new file mode 100644 index 000000000..dbb325908 --- /dev/null +++ b/src/internal/strings/boolean.test.ts @@ -0,0 +1,19 @@ +import { parseOptionalBoolean } from './boolean' + +const errorMessage = 'value must be either true or false' + +describe('parseOptionalBoolean', () => { + it('returns undefined for absent or empty input', () => { + expect(parseOptionalBoolean(undefined, errorMessage)).toBeUndefined() + expect(parseOptionalBoolean(' ', errorMessage)).toBeUndefined() + }) + + it('parses true and false case-insensitively', () => { + expect(parseOptionalBoolean(' true ', errorMessage)).toBe(true) + expect(parseOptionalBoolean('FALSE', errorMessage)).toBe(false) + }) + + it('rejects other boolean aliases', () => { + expect(() => parseOptionalBoolean('yes', errorMessage)).toThrow(errorMessage) + }) +}) diff --git a/src/internal/strings/boolean.ts b/src/internal/strings/boolean.ts new file mode 100644 index 000000000..11f668a11 --- /dev/null +++ b/src/internal/strings/boolean.ts @@ -0,0 +1,20 @@ +export function parseOptionalBoolean( + value: string | undefined, + errorMessage: string +): boolean | undefined { + const normalized = value?.trim().toLowerCase() + + if (!normalized) { + return undefined + } + + if (normalized === 'true') { + return true + } + + if (normalized === 'false') { + return false + } + + throw new Error(errorMessage) +} diff --git a/src/internal/strings/index.ts b/src/internal/strings/index.ts new file mode 100644 index 000000000..0c5866a09 --- /dev/null +++ b/src/internal/strings/index.ts @@ -0,0 +1,3 @@ +export * from './boolean' +export * from './integer' +export * from './string-list' diff --git a/src/internal/strings/integer.test.ts b/src/internal/strings/integer.test.ts new file mode 100644 index 000000000..2011c23ab --- /dev/null +++ b/src/internal/strings/integer.test.ts @@ -0,0 +1,25 @@ +import { parseNonNegativeInteger, parsePositiveInteger } from './integer' + +const nonNegativeError = 'value must be a non-negative integer' +const positiveError = 'value must be a positive integer' + +describe('parseNonNegativeInteger', () => { + it('parses non-negative safe integers strictly', () => { + expect(parseNonNegativeInteger('0', nonNegativeError)).toBe(0) + expect(parseNonNegativeInteger(' 42 ', nonNegativeError)).toBe(42) + }) + + it.each(['-1', '1.5', '7x', '9007199254740992'])('rejects invalid input %s', (value) => { + expect(() => parseNonNegativeInteger(value, nonNegativeError)).toThrow(nonNegativeError) + }) +}) + +describe('parsePositiveInteger', () => { + it('parses positive safe integers strictly', () => { + expect(parsePositiveInteger(' 42 ', positiveError)).toBe(42) + }) + + it.each(['0', '-1', '1.5', '7x', '9007199254740992'])('rejects invalid input %s', (value) => { + expect(() => parsePositiveInteger(value, positiveError)).toThrow(positiveError) + }) +}) diff --git a/src/internal/strings/integer.ts b/src/internal/strings/integer.ts new file mode 100644 index 000000000..f8d0cfada --- /dev/null +++ b/src/internal/strings/integer.ts @@ -0,0 +1,23 @@ +export function parseNonNegativeInteger(value: string, errorMessage: string): number { + const normalized = value.trim() + + if (!/^\d+$/.test(normalized)) { + throw new Error(errorMessage) + } + + const parsed = Number.parseInt(normalized, 10) + if (!Number.isSafeInteger(parsed)) { + throw new Error(errorMessage) + } + + return parsed +} + +export function parsePositiveInteger(value: string, errorMessage: string): number { + const parsed = parseNonNegativeInteger(value, errorMessage) + if (parsed === 0) { + throw new Error(errorMessage) + } + + return parsed +} diff --git a/src/internal/strings/string-list.test.ts b/src/internal/strings/string-list.test.ts new file mode 100644 index 000000000..cf2b22220 --- /dev/null +++ b/src/internal/strings/string-list.test.ts @@ -0,0 +1,27 @@ +import { normalizeStringList, parseCommaSeparatedList } from './string-list' + +describe('normalizeStringList', () => { + it('returns undefined for absent or empty values', () => { + expect(normalizeStringList(undefined)).toBeUndefined() + expect(normalizeStringList([])).toBeUndefined() + expect(normalizeStringList(['', ' '])).toBeUndefined() + }) + + it('trims, de-duplicates, and preserves order', () => { + expect(normalizeStringList([' tenant-a ', 'tenant-b', 'tenant-a', ''])).toEqual([ + 'tenant-a', + 'tenant-b', + ]) + }) +}) + +describe('parseCommaSeparatedList', () => { + it('returns undefined for absent or empty input', () => { + expect(parseCommaSeparatedList(undefined)).toBeUndefined() + expect(parseCommaSeparatedList(' , , ')).toBeUndefined() + }) + + it('normalizes comma-separated input', () => { + expect(parseCommaSeparatedList('tenant-a, tenant-b,tenant-a')).toEqual(['tenant-a', 'tenant-b']) + }) +}) diff --git a/src/internal/strings/string-list.ts b/src/internal/strings/string-list.ts new file mode 100644 index 000000000..8129d990f --- /dev/null +++ b/src/internal/strings/string-list.ts @@ -0,0 +1,15 @@ +export function normalizeStringList(values: readonly string[] | undefined): string[] | undefined { + if (!values) { + return undefined + } + + const normalized = Array.from( + new Set(values.map((value) => value.trim()).filter((value) => value.length > 0)) + ) + + return normalized.length > 0 ? normalized : undefined +} + +export function parseCommaSeparatedList(value: string | undefined): string[] | undefined { + return value ? normalizeStringList(value.split(',')) : undefined +} diff --git a/src/scripts/README.md b/src/scripts/README.md new file mode 100644 index 000000000..afa4b4230 --- /dev/null +++ b/src/scripts/README.md @@ -0,0 +1,85 @@ +# Jobs client + +`jobs-client.ts` is an operator CLI for the admin queue-overflow API. When the pg-boss +`pgboss_v10.job` table accumulates so many `created` jobs that fetching slows down (high CPU, +starved workers), it lets you inspect the backlog, archive a slice of it into +`pgboss_v10.job_overflow_backup`, and restore it later in self-pacing batches. + +It talks HTTP to the admin API. The server resolves +the same database connection pg-boss uses (including `PG_QUEUE_CONNECTION_URL` when configured), +so the routes require `PG_QUEUE_ENABLE=true`. To keep the maintenance API available while stopping +workers in a process, set `PG_QUEUE_WORKERS_ENABLE=false` rather than disabling the queue. + +All operations only ever touch jobs in state `created`: jobs no worker has picked up. + +## Actions + +| Command | Endpoint | What it does | +| ---------------------- | ------------------------------ | --------------------------------------------------------------- | +| `npm run jobs:list` | `GET /queue/overflow` | Counts `created` jobs grouped by queue/event type or by tenant. | +| `npm run jobs:backup` | `POST /queue/overflow/backup` | Moves matching `created` jobs into the overflow backup table. | +| `npm run jobs:restore` | `POST /queue/overflow/restore` | Moves jobs back in batches until the backup table is drained. | + +## Environment variables + +| Variable | Default | Applies to | Description | +| ------------------------- | --------- | ---------- | ---------------------------------------------------------------------------------------------- | +| `ADMIN_URL` | required | all | Base URL of the admin API. | +| `ADMIN_API_KEY` | required | all | Admin API key (sent as `ApiKey` header). | +| `JOBS_SOURCE` | `job` | list | `job` (live table) or `backup` (overflow backup table). | +| `JOBS_GROUP_BY` | `summary` | list | `summary` (queue + event type) or `tenant`. | +| `JOBS_QUEUE_NAME` | – | all | Exact pg-boss queue name filter. | +| `JOBS_EVENT_TYPES` | – | all | Comma-separated event type filter (`data->event->>type`). | +| `JOBS_TENANT_REFS` | – | all | Comma-separated tenant ref filter (`data->tenant->>ref`). | +| `JOBS_LIMIT` | see below | all | List: max groups (50). Backup: max jobs in one call (unbounded). Restore: batch size (10,000). | +| `JOBS_BACKUP_CONFIRM_ALL` | – | backup | Must be `true` to back up with no filter at all. | +| `JOBS_MAX_PENDING` | `50000` | restore | Pause restoring while the global live `created` backlog exceeds this. | +| `JOBS_SLEEP_MS` | `1000` | restore | Sleep between batches; doubles up to 60s while over `JOBS_MAX_PENDING`. | + +## Examples + +```sh +# Queue/event summary of the live backlog (totalCount, groupCount, hasMore). +ADMIN_URL=https://storage-admin.example.com ADMIN_API_KEY=... npm run jobs:list + +# Top tenants for one queue and event type. +ADMIN_URL=... ADMIN_API_KEY=... \ +JOBS_GROUP_BY=tenant JOBS_QUEUE_NAME=webhooks JOBS_EVENT_TYPES=ObjectRemoved:Delete \ +npm run jobs:list + +# Archive every created webhooks job for two tenants. +ADMIN_URL=... ADMIN_API_KEY=... \ +JOBS_QUEUE_NAME=webhooks JOBS_TENANT_REFS=tenant-a,tenant-b \ +npm run jobs:backup + +# Archive everything (explicit escape hatch required). +ADMIN_URL=... ADMIN_API_KEY=... JOBS_BACKUP_CONFIRM_ALL=true npm run jobs:backup + +# Drain the backup table back into the live queue. +ADMIN_URL=... ADMIN_API_KEY=... JOBS_QUEUE_NAME=webhooks npm run jobs:restore + +# Inspect what is still archived. +ADMIN_URL=... ADMIN_API_KEY=... JOBS_SOURCE=backup npm run jobs:list +``` + +## Semantics + +- Restore is rejected when `DATABASE_ENGINE=oriole`: OrioleDB can misclassify valid rows as + conflicts for pg-boss partial indexes, which would otherwise delete backup rows without + restoring them. +- List responses report `sourceTableExists` for the selected source. The live job table is present + when a live query succeeds; backup-source queries return `false` until the backup table exists. +- Backup needs a filter or `JOBS_BACKUP_CONFIRM_ALL=true`. +- Backup creates the table with `LIKE job INCLUDING ALL`, then moves rows in one transaction. Its + `SHARE ROW EXCLUSIVE` lock pauses queue writes. `JOBS_LIMIT` caps the move. +- There is no schema precheck. Insert errors roll back the move. +- Restore uses batches of 10,000. It pauses while the live backlog exceeds `JOBS_MAX_PENDING`, with + backoff capped at 60 seconds. The throttle uses the count-only `GET /queue/overflow/count` + endpoint rather than the grouped analysis query. +- On conflict, the live job wins. The backup row is removed and counted in `conflictCount`. +- Mutations share an advisory lock. A disconnected request stops waiting for transaction + acquisition and aborts in-flight SQL. Since pg-pool cannot cancel its internal checkout waiter, + that waiter can remain until a client becomes available; a late client is released without + running `BEGIN`. +- Once mutation SQL completes, `COMMIT` is deliberately not canceled by a disconnect. +- Progress goes to stderr; the final JSON summary goes to stdout. diff --git a/src/scripts/jobs-client.test.ts b/src/scripts/jobs-client.test.ts new file mode 100644 index 000000000..b31428021 --- /dev/null +++ b/src/scripts/jobs-client.test.ts @@ -0,0 +1,474 @@ +import { afterEach, describe, expect, it, vi } from 'vitest' +import { + buildJobsCountRequest, + buildJobsRequest, + main, + parseJobsOptions, + resolveJobsAdminUrl, +} from './jobs-client' + +describe('resolveJobsAdminUrl', () => { + it('preserves base URL path prefixes and query params', () => { + const url = resolveJobsAdminUrl('http://localhost:54321/admin/', '/queue/overflow', { + source: 'backup', + groupBy: 'tenant', + }) + + expect(url.toString()).toBe( + 'http://localhost:54321/admin/queue/overflow?source=backup&groupBy=tenant' + ) + }) +}) + +describe('parseJobsOptions', () => { + it('uses defaults when optional env is omitted', () => { + expect(parseJobsOptions({})).toEqual({ + confirmAll: undefined, + queueName: undefined, + eventTypes: undefined, + tenantRefs: undefined, + source: 'job', + groupBy: 'summary', + limit: undefined, + maxPending: 50_000, + sleepMs: 1_000, + }) + }) + + it('rejects invalid JOBS_SOURCE and JOBS_LIMIT values', () => { + expect( + parseJobsOptions({ + JOBS_SOURCE: 'archive', + }) + ).toBe('JOBS_SOURCE must be either job or backup') + + expect( + parseJobsOptions({ + JOBS_LIMIT: '0', + }) + ).toBe('JOBS_LIMIT must be a positive integer') + + expect( + parseJobsOptions({ + JOBS_BACKUP_CONFIRM_ALL: 'yes', + }) + ).toBe('JOBS_BACKUP_CONFIRM_ALL must be either true or false') + }) + + it('parses backlog and sleep settings', () => { + expect( + parseJobsOptions({ + JOBS_MAX_PENDING: '12345', + JOBS_SLEEP_MS: '25', + }) + ).toMatchObject({ maxPending: 12_345, sleepMs: 25 }) + }) + + it('rejects invalid backlog and sleep settings', () => { + expect( + parseJobsOptions({ + JOBS_MAX_PENDING: '0', + }) + ).toBe('JOBS_MAX_PENDING must be a positive integer') + + expect( + parseJobsOptions({ + JOBS_SLEEP_MS: 'later', + }) + ).toBe('JOBS_SLEEP_MS must be a positive integer') + }) +}) + +describe('buildJobsRequest', () => { + const baseConfig = { + adminApiKey: 'test-key', + adminUrl: 'http://localhost:54321/admin', + confirmAll: undefined, + queueName: 'webhooks', + eventTypes: ['ObjectRemoved:Delete'], + tenantRefs: ['tenant-a'], + source: 'backup' as const, + groupBy: 'tenant' as const, + limit: 123, + maxPending: 50_000, + sleepMs: 1_000, + } + + it('builds list requests with query params', () => { + const request = buildJobsRequest('list', baseConfig) + + expect(request.method).toBe('GET') + expect(request.body).toBeUndefined() + expect(request.url.toString()).toBe( + 'http://localhost:54321/admin/queue/overflow?source=backup&groupBy=tenant&name=webhooks&eventTypes=ObjectRemoved%3ADelete&tenantRefs=tenant-a&limit=123' + ) + expect((request.headers as Headers).get('ApiKey')).toBe('test-key') + }) + + it('builds the unfiltered live backlog count request', () => { + const request = buildJobsCountRequest(baseConfig) + + expect(request.method).toBe('GET') + expect(request.url.toString()).toBe('http://localhost:54321/admin/queue/overflow/count') + expect(request.headers.get('ApiKey')).toBe('test-key') + }) + + it('builds backup requests with a JSON body', () => { + const request = buildJobsRequest('backup', baseConfig) + + expect(request.method).toBe('POST') + expect(request.url.toString()).toBe('http://localhost:54321/admin/queue/overflow/backup') + expect(JSON.parse(request.body as string)).toEqual({ + name: 'webhooks', + eventTypes: ['ObjectRemoved:Delete'], + tenantRefs: ['tenant-a'], + limit: 123, + }) + expect((request.headers as Headers).get('ApiKey')).toBe('test-key') + expect((request.headers as Headers).get('Content-Type')).toBe('application/json') + }) + + it('rejects an unfiltered backup unless all jobs are explicitly confirmed', () => { + const unfilteredConfig = { + ...baseConfig, + queueName: undefined, + eventTypes: undefined, + tenantRefs: undefined, + } + + expect(() => buildJobsRequest('backup', unfilteredConfig)).toThrow( + 'Backup requires JOBS_QUEUE_NAME, JOBS_EVENT_TYPES, or JOBS_TENANT_REFS unless JOBS_BACKUP_CONFIRM_ALL=true' + ) + + const request = buildJobsRequest('backup', { + ...unfilteredConfig, + confirmAll: true, + }) + expect(JSON.parse(request.body as string)).toEqual({ + confirmAll: true, + limit: 123, + }) + }) + + it('builds a restore request with configured filters', () => { + const request = buildJobsRequest('restore', baseConfig) + + expect(JSON.parse(request.body as string)).toEqual({ + name: 'webhooks', + eventTypes: ['ObjectRemoved:Delete'], + tenantRefs: ['tenant-a'], + limit: 123, + }) + }) +}) + +describe('main', () => { + afterEach(() => { + vi.restoreAllMocks() + process.exitCode = undefined + }) + + it('sets a non-zero exit code for invalid actions', async () => { + vi.spyOn(console, 'error').mockImplementation(() => {}) + + await expect(main({}, ['node', 'jobs-client.ts', 'invalid'])).resolves.toBe(false) + + expect(process.exitCode).toBe(1) + expect(console.error).toHaveBeenCalledWith('Please provide an action: list, backup, or restore') + }) + + it.each([ + [{}, 'Please provide ADMIN_URL'], + [{ ADMIN_URL: 'http://localhost:54321/admin' }, 'Please provide ADMIN_API_KEY'], + ])('reports missing admin configuration', async (env, message) => { + const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}) + + await expect(main(env, ['node', 'jobs-client.ts', 'list'])).resolves.toBe(false) + + expect(process.exitCode).toBe(1) + expect(errorSpy).toHaveBeenCalledWith(message) + }) + + it('reports option validation errors without logging credentials', async () => { + const adminApiKey = 'super-secret-admin-key' + const fetchMock = vi.fn() + const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}) + + await expect( + main( + { + ADMIN_URL: 'http://localhost:54321/admin', + ADMIN_API_KEY: adminApiKey, + JOBS_LIMIT: '0', + }, + ['node', 'jobs-client.ts', 'list'], + { fetch: fetchMock } + ) + ).resolves.toBe(false) + + expect(fetchMock).not.toHaveBeenCalled() + expect(errorSpy).toHaveBeenCalledWith('JOBS_LIMIT must be a positive integer') + expect(errorSpy.mock.calls.flat().join(' ')).not.toContain(adminApiKey) + }) + + it('refuses a bare backup before making a request', async () => { + const fetchSpy = vi.spyOn(globalThis, 'fetch') + const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}) + + await expect( + main( + { + ADMIN_URL: 'http://localhost:54321/admin', + ADMIN_API_KEY: 'test-key', + }, + ['node', 'jobs-client.ts', 'backup'] + ) + ).resolves.toBe(false) + + expect(fetchSpy).not.toHaveBeenCalled() + expect(process.exitCode).toBe(1) + expect(errorSpy).toHaveBeenCalledWith( + 'Backup requires JOBS_QUEUE_NAME, JOBS_EVENT_TYPES, or JOBS_TENANT_REFS unless JOBS_BACKUP_CONFIRM_ALL=true' + ) + }) + + it('prints the JSON response for successful single-shot requests', async () => { + vi.spyOn(globalThis, 'fetch').mockResolvedValue( + new Response(JSON.stringify({ totalCount: 3 }), { + status: 200, + headers: { + 'Content-Type': 'application/json', + }, + }) + ) + vi.spyOn(console, 'log').mockImplementation(() => {}) + + await expect( + main( + { + ADMIN_URL: 'http://localhost:54321/admin', + ADMIN_API_KEY: 'test-key', + JOBS_QUEUE_NAME: 'webhooks', + }, + ['node', 'jobs-client.ts', 'list'] + ) + ).resolves.toBe(true) + + expect(console.log).toHaveBeenCalledWith('{\n "totalCount": 3\n}') + }) + + it('does not log authenticated response bodies', async () => { + const adminApiKey = 'super-secret-admin-key' + const fetchMock = vi.fn().mockResolvedValue( + new Response(JSON.stringify({ message: `request failed for ${adminApiKey}` }), { + status: 500, + statusText: `Internal Server Error for ${adminApiKey}`, + }) + ) + const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}) + const logSpy = vi.spyOn(console, 'log').mockImplementation(() => {}) + + await expect( + main( + { + ADMIN_URL: 'http://localhost:54321/admin', + ADMIN_API_KEY: adminApiKey, + JOBS_QUEUE_NAME: 'webhooks', + }, + ['node', 'jobs-client.ts', 'list'], + { fetch: fetchMock } + ) + ).resolves.toBe(false) + + expect(process.exitCode).toBe(1) + expect(errorSpy).toHaveBeenCalledOnce() + expect(errorSpy).toHaveBeenCalledWith('Jobs client request failed') + expect(errorSpy.mock.calls.flat().join(' ')).not.toContain(adminApiKey) + expect(logSpy).not.toHaveBeenCalled() + }) + + it.each([ + 'error', + 'string', + ] as const)('does not log %s fetch error details', async (rejectionType) => { + const adminApiKey = 'super-secret-admin-key' + const rejection = + rejectionType === 'error' + ? new Error(`request with ApiKey ${adminApiKey} failed`) + : `request with ApiKey ${adminApiKey} failed` + const fetchMock = vi.fn().mockRejectedValue(rejection) + const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {}) + const logSpy = vi.spyOn(console, 'log').mockImplementation(() => {}) + + await expect( + main( + { + ADMIN_URL: 'http://localhost:54321/admin', + ADMIN_API_KEY: adminApiKey, + JOBS_QUEUE_NAME: 'webhooks', + }, + ['node', 'jobs-client.ts', 'list'], + { fetch: fetchMock } + ) + ).resolves.toBe(false) + + expect(process.exitCode).toBe(1) + expect(errorSpy).toHaveBeenCalledOnce() + expect(errorSpy).toHaveBeenCalledWith('Jobs client request failed') + expect(errorSpy.mock.calls.flat().join(' ')).not.toContain(adminApiKey) + expect(logSpy).not.toHaveBeenCalled() + }) + + it('waits while the global created backlog is above the threshold', async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce(new Response(JSON.stringify({ totalCount: 61_234 }), { status: 200 })) + .mockResolvedValueOnce(new Response(JSON.stringify({ totalCount: 55_000 }), { status: 200 })) + .mockResolvedValueOnce(new Response(JSON.stringify({ totalCount: 50_000 }), { status: 200 })) + .mockResolvedValueOnce( + new Response(JSON.stringify({ movedCount: 3, conflictCount: 1, hasMore: false }), { + status: 200, + }) + ) + const sleep = vi.fn(async (_ms: number) => {}) + vi.spyOn(console, 'error').mockImplementation(() => {}) + vi.spyOn(console, 'log').mockImplementation(() => {}) + + await expect( + main( + { + ADMIN_URL: 'http://localhost:54321/admin', + ADMIN_API_KEY: 'test-key', + JOBS_QUEUE_NAME: 'webhooks', + JOBS_MAX_PENDING: '50000', + JOBS_SLEEP_MS: '2', + }, + ['node', 'jobs-client.ts', 'restore'], + { fetch: fetchMock, sleep } + ) + ).resolves.toBe(true) + + expect(sleep).toHaveBeenNthCalledWith(1, 2) + expect(sleep).toHaveBeenNthCalledWith(2, 4) + expect(fetchMock).toHaveBeenCalledTimes(4) + for (const call of fetchMock.mock.calls.slice(0, 3)) { + expect(call[0].toString()).toBe('http://localhost:54321/admin/queue/overflow/count') + } + expect(console.error).toHaveBeenNthCalledWith(1, 'Queue backlog 61234 above 50000, waiting 2ms') + expect(console.error).toHaveBeenNthCalledWith(2, 'Queue backlog 55000 above 50000, waiting 4ms') + }) + + it('caps backlog backoff and resets it after a successful batch', async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce(new Response(JSON.stringify({ totalCount: 60_000 }), { status: 200 })) + .mockResolvedValueOnce(new Response(JSON.stringify({ totalCount: 60_000 }), { status: 200 })) + .mockResolvedValueOnce(new Response(JSON.stringify({ totalCount: 50_000 }), { status: 200 })) + .mockResolvedValueOnce( + new Response(JSON.stringify({ movedCount: 1, conflictCount: 0, hasMore: true }), { + status: 200, + }) + ) + .mockResolvedValueOnce(new Response(JSON.stringify({ totalCount: 60_000 }), { status: 200 })) + .mockResolvedValueOnce(new Response(JSON.stringify({ totalCount: 50_000 }), { status: 200 })) + .mockResolvedValueOnce( + new Response(JSON.stringify({ movedCount: 1, conflictCount: 0, hasMore: false }), { + status: 200, + }) + ) + const sleep = vi.fn(async (_ms: number) => {}) + vi.spyOn(console, 'error').mockImplementation(() => {}) + vi.spyOn(console, 'log').mockImplementation(() => {}) + + await expect( + main( + { + ADMIN_URL: 'http://localhost:54321/admin', + ADMIN_API_KEY: 'test-key', + JOBS_SLEEP_MS: '40000', + }, + ['node', 'jobs-client.ts', 'restore'], + { fetch: fetchMock, sleep } + ) + ).resolves.toBe(true) + + expect(sleep.mock.calls.map(([ms]) => ms)).toEqual([40_000, 60_000, 40_000, 40_000]) + expect(console.error).toHaveBeenCalledWith('Queue backlog 60000 above 50000, waiting 60s') + expect(console.error).toHaveBeenLastCalledWith('Restore batch 2: moved 1, conflicts 0') + }) + + it('drains restore batches and prints accumulated totals', async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce(new Response(JSON.stringify({ totalCount: 12 }), { status: 200 })) + .mockResolvedValueOnce( + new Response(JSON.stringify({ movedCount: 2, conflictCount: 1, hasMore: true }), { + status: 200, + }) + ) + .mockResolvedValueOnce(new Response(JSON.stringify({ totalCount: 15 }), { status: 200 })) + .mockResolvedValueOnce( + new Response(JSON.stringify({ movedCount: 3, conflictCount: 4, hasMore: false }), { + status: 200, + }) + ) + const sleep = vi.fn(async (_ms: number) => {}) + vi.spyOn(console, 'error').mockImplementation(() => {}) + vi.spyOn(console, 'log').mockImplementation(() => {}) + + await expect( + main( + { + ADMIN_URL: 'http://localhost:54321/admin', + ADMIN_API_KEY: 'test-key', + JOBS_QUEUE_NAME: 'webhooks', + JOBS_SLEEP_MS: '1', + }, + ['node', 'jobs-client.ts', 'restore'], + { fetch: fetchMock, sleep } + ) + ).resolves.toBe(true) + + expect(fetchMock).toHaveBeenCalledTimes(4) + expect(sleep).toHaveBeenCalledOnce() + expect(sleep).toHaveBeenCalledWith(1) + expect(console.error).toHaveBeenCalledWith('Restore batch 1: moved 2, conflicts 1') + expect(console.error).toHaveBeenCalledWith('Restore batch 2: moved 3, conflicts 4') + expect(console.log).toHaveBeenCalledWith( + '{\n "batches": 2,\n "conflictCount": 5,\n "movedCount": 5\n}' + ) + }) + + it('aborts when restore reports more work without making progress', async () => { + const fetchMock = vi + .fn() + .mockResolvedValueOnce(new Response(JSON.stringify({ totalCount: 0 }), { status: 200 })) + .mockResolvedValueOnce( + new Response(JSON.stringify({ movedCount: 0, conflictCount: 0, hasMore: true }), { + status: 200, + }) + ) + const sleep = vi.fn(async (_ms: number) => {}) + vi.spyOn(console, 'error').mockImplementation(() => {}) + vi.spyOn(console, 'log').mockImplementation(() => {}) + + await expect( + main( + { + ADMIN_URL: 'http://localhost:54321/admin', + ADMIN_API_KEY: 'test-key', + }, + ['node', 'jobs-client.ts', 'restore'], + { fetch: fetchMock, sleep } + ) + ).resolves.toBe(false) + + expect(fetchMock).toHaveBeenCalledTimes(2) + expect(sleep).not.toHaveBeenCalled() + expect(process.exitCode).toBe(1) + expect(console.error).toHaveBeenLastCalledWith( + 'Restore reported hasMore=true without moving or dropping any rows' + ) + expect(console.log).not.toHaveBeenCalled() + }) +}) diff --git a/src/scripts/jobs-client.ts b/src/scripts/jobs-client.ts new file mode 100644 index 000000000..f684b7faf --- /dev/null +++ b/src/scripts/jobs-client.ts @@ -0,0 +1,298 @@ +import { + parseCommaSeparatedList, + parseOptionalBoolean, + parsePositiveInteger, +} from '@internal/strings' + +type JobsAction = 'backup' | 'list' | 'restore' +type JobsGroupBy = 'summary' | 'tenant' +type JobsSource = 'backup' | 'job' + +interface JobsClientOptions { + confirmAll?: boolean + eventTypes?: string[] + groupBy: JobsGroupBy + limit?: number + maxPending: number + queueName?: string + sleepMs: number + source: JobsSource + tenantRefs?: string[] +} + +interface JobsClientConfig extends JobsClientOptions { + adminApiKey: string + adminUrl: string +} + +interface JobsClientDependencies { + fetch?: typeof fetch + sleep?: (ms: number) => Promise +} + +const JOBS_MAX_PENDING_DEFAULT = 50_000 +const JOBS_SLEEP_MS_DEFAULT = 1_000 +const JOBS_BACKOFF_MAX_MS = 60_000 +const JOBS_CLIENT_REQUEST_FAILURE_MESSAGE = 'Jobs client request failed' +const JOBS_BACKUP_SCOPE_MESSAGE = + 'Backup requires JOBS_QUEUE_NAME, JOBS_EVENT_TYPES, or JOBS_TENANT_REFS unless JOBS_BACKUP_CONFIRM_ALL=true' + +class UnscopedBackupError extends Error {} + +export function resolveJobsAdminUrl( + baseUrl: string, + requestPath: string, + query?: Record +) { + const url = new URL(`${baseUrl.replace(/\/+$/, '')}/${requestPath.replace(/^\/+/, '')}`) + + for (const [key, value] of Object.entries(query ?? {})) { + if (value !== undefined) { + url.searchParams.set(key, value) + } + } + + return url +} + +export function parseJobsOptions(env: NodeJS.ProcessEnv): JobsClientOptions | string { + const source = (env.JOBS_SOURCE?.trim() || 'job') as JobsSource + const groupBy = (env.JOBS_GROUP_BY?.trim() || 'summary') as JobsGroupBy + + if (source !== 'job' && source !== 'backup') { + return 'JOBS_SOURCE must be either job or backup' + } + + if (groupBy !== 'summary' && groupBy !== 'tenant') { + return 'JOBS_GROUP_BY must be either summary or tenant' + } + + try { + return { + confirmAll: parseOptionalBoolean( + env.JOBS_BACKUP_CONFIRM_ALL, + 'JOBS_BACKUP_CONFIRM_ALL must be either true or false' + ), + queueName: env.JOBS_QUEUE_NAME?.trim() || undefined, + eventTypes: parseCommaSeparatedList(env.JOBS_EVENT_TYPES), + tenantRefs: parseCommaSeparatedList(env.JOBS_TENANT_REFS), + source, + groupBy, + limit: env.JOBS_LIMIT + ? parsePositiveInteger(env.JOBS_LIMIT, 'JOBS_LIMIT must be a positive integer') + : undefined, + maxPending: env.JOBS_MAX_PENDING + ? parsePositiveInteger(env.JOBS_MAX_PENDING, 'JOBS_MAX_PENDING must be a positive integer') + : JOBS_MAX_PENDING_DEFAULT, + sleepMs: env.JOBS_SLEEP_MS + ? parsePositiveInteger(env.JOBS_SLEEP_MS, 'JOBS_SLEEP_MS must be a positive integer') + : JOBS_SLEEP_MS_DEFAULT, + } + } catch (error) { + return error instanceof Error ? error.message : String(error) + } +} + +export function buildJobsRequest(action: JobsAction, config: JobsClientConfig) { + const headers = new Headers({ + ApiKey: config.adminApiKey, + }) + + if (action === 'list') { + return { + method: 'GET', + url: resolveJobsAdminUrl(config.adminUrl, '/queue/overflow', { + source: config.source, + groupBy: config.groupBy, + name: config.queueName, + eventTypes: config.eventTypes?.join(','), + tenantRefs: config.tenantRefs?.join(','), + limit: config.limit?.toString(), + }), + headers, + body: undefined, + } + } + + if ( + action === 'backup' && + !config.confirmAll && + !config.queueName && + !config.eventTypes?.length && + !config.tenantRefs?.length + ) { + throw new UnscopedBackupError(JOBS_BACKUP_SCOPE_MESSAGE) + } + + headers.set('Content-Type', 'application/json') + + return { + method: 'POST', + url: resolveJobsAdminUrl( + config.adminUrl, + action === 'backup' ? '/queue/overflow/backup' : '/queue/overflow/restore' + ), + headers, + body: JSON.stringify({ + name: config.queueName, + eventTypes: config.eventTypes, + tenantRefs: config.tenantRefs, + limit: config.limit, + ...(action === 'backup' && config.confirmAll ? { confirmAll: true } : {}), + }), + } +} + +export function buildJobsCountRequest(config: JobsClientConfig) { + return { + method: 'GET', + url: resolveJobsAdminUrl(config.adminUrl, '/queue/overflow/count'), + headers: new Headers({ + ApiKey: config.adminApiKey, + }), + } +} + +async function assertJsonResponse(response: Response) { + if (response.ok) { + return response.json() + } + + await response.text() + throw new Error(JOBS_CLIENT_REQUEST_FAILURE_MESSAGE) +} + +function fail(message: string) { + process.exitCode = 1 + console.error(message) + return false +} + +function sleep(ms: number) { + return new Promise((resolve) => { + setTimeout(resolve, ms) + }) +} + +function formatWait(ms: number) { + return ms % 1_000 === 0 ? `${ms / 1_000}s` : `${ms}ms` +} + +export async function main( + env: NodeJS.ProcessEnv = process.env, + argv: string[] = process.argv, + dependencies: JobsClientDependencies = {} +): Promise { + const action = argv[2] + + if (action !== 'list' && action !== 'backup' && action !== 'restore') { + return fail('Please provide an action: list, backup, or restore') + } + + const adminUrl = env.ADMIN_URL + if (!adminUrl) { + return fail('Please provide ADMIN_URL') + } + + const adminApiKey = env.ADMIN_API_KEY + if (!adminApiKey) { + return fail('Please provide ADMIN_API_KEY') + } + + const options = parseJobsOptions(env) + if (typeof options === 'string') { + return fail(options) + } + + const config: JobsClientConfig = { ...options, adminApiKey, adminUrl } + + try { + const fetchRequest = dependencies.fetch ?? globalThis.fetch + + if (action === 'restore') { + const wait = dependencies.sleep ?? sleep + const restoreRequest = buildJobsRequest(action, config) + const backlogRequest = buildJobsCountRequest(config) + let backoffMs = Math.min(config.sleepMs, JOBS_BACKOFF_MAX_MS) + let batches = 0 + let conflictCount = 0 + let movedCount = 0 + + while (true) { + while (true) { + const backlogResponse = await fetchRequest(backlogRequest.url, { + method: backlogRequest.method, + headers: backlogRequest.headers, + }) + const backlog = (await assertJsonResponse(backlogResponse)) as { totalCount: number } + + if (backlog.totalCount <= config.maxPending) { + break + } + + console.error( + `Queue backlog ${backlog.totalCount} above ${config.maxPending}, waiting ${formatWait(backoffMs)}` + ) + await wait(backoffMs) + backoffMs = Math.min(backoffMs * 2, JOBS_BACKOFF_MAX_MS) + } + + const response = await fetchRequest(restoreRequest.url, { + method: restoreRequest.method, + headers: restoreRequest.headers, + body: restoreRequest.body, + }) + const data = (await assertJsonResponse(response)) as { + conflictCount: number + hasMore: boolean + movedCount: number + } + + batches += 1 + conflictCount += data.conflictCount + movedCount += data.movedCount + backoffMs = Math.min(config.sleepMs, JOBS_BACKOFF_MAX_MS) + console.error( + `Restore batch ${batches}: moved ${data.movedCount}, conflicts ${data.conflictCount}` + ) + + if (data.hasMore && data.movedCount + data.conflictCount === 0) { + return fail('Restore reported hasMore=true without moving or dropping any rows') + } + + if (!data.hasMore) { + break + } + + await wait(config.sleepMs) + } + + console.log(JSON.stringify({ batches, conflictCount, movedCount }, null, 2)) + return true + } + + const request = buildJobsRequest(action, config) + const response = await fetchRequest(request.url, { + method: request.method, + headers: request.headers, + body: request.body, + }) + + const data = await assertJsonResponse(response) + console.log(JSON.stringify(data, null, 2)) + return true + } catch (error) { + if (error instanceof UnscopedBackupError) { + return fail(JOBS_BACKUP_SCOPE_MESSAGE) + } + + return fail(JOBS_CLIENT_REQUEST_FAILURE_MESSAGE) + } +} + +if (require.main === module) { + main(process.env).catch(() => { + process.exitCode = 1 + console.error(JOBS_CLIENT_REQUEST_FAILURE_MESSAGE) + }) +} diff --git a/src/scripts/pprof-client.ts b/src/scripts/pprof-client.ts index e1d7ad7de..639b2e222 100644 --- a/src/scripts/pprof-client.ts +++ b/src/scripts/pprof-client.ts @@ -2,6 +2,7 @@ import { fetchPprofStream } from '@internal/monitoring/pprof/client-http' import { writePprofCaptureToFile } from '@internal/monitoring/pprof/download' import { generateFlameArtifacts, resolveFlameMdFormat } from '@internal/monitoring/pprof/flame' import type { PprofRequestTargetType } from '@internal/monitoring/pprof/types' +import { parseNonNegativeInteger, parsePositiveInteger } from '@internal/strings' import path from 'path' const ADMIN_URL = process.env.ADMIN_URL @@ -41,21 +42,6 @@ export function parseBooleanEnvWithDefault(value: string | undefined, defaultVal return parseBooleanEnv(value) === true } -function parseUnsignedInteger(value: string, errorMessage: string) { - const normalized = value.trim() - - if (!/^\d+$/.test(normalized)) { - throw new Error(errorMessage) - } - - const parsed = Number.parseInt(normalized, 10) - if (!Number.isSafeInteger(parsed)) { - throw new Error(errorMessage) - } - - return parsed -} - export function parsePositiveIntegerEnv( value: string | undefined, envName: string, @@ -65,12 +51,7 @@ export function parsePositiveIntegerEnv( return defaultValue } - const parsed = parseUnsignedInteger(value, `${envName} must be a positive integer`) - if (parsed <= 0) { - throw new Error(`${envName} must be a positive integer`) - } - - return parsed + return parsePositiveInteger(value, `${envName} must be a positive integer`) } export function parseNonNegativeIntegerEnv(value: string | undefined, envName: string) { @@ -78,7 +59,7 @@ export function parseNonNegativeIntegerEnv(value: string | undefined, envName: s return undefined } - return parseUnsignedInteger(value, `${envName} must be a non-negative integer`) + return parseNonNegativeInteger(value, `${envName} must be a non-negative integer`) } export function parsePprofTarget( @@ -116,10 +97,7 @@ export function parsePprofTarget( } } - const seconds = parseUnsignedInteger(secondsValue, 'seconds must be a positive integer') - if (seconds <= 0) { - throw new Error('seconds must be a positive integer') - } + const seconds = parsePositiveInteger(secondsValue, 'seconds must be a positive integer') return { seconds, diff --git a/src/test/admin-queue-overflow.test.ts b/src/test/admin-queue-overflow.test.ts new file mode 100644 index 000000000..f5a2a5e4c --- /dev/null +++ b/src/test/admin-queue-overflow.test.ts @@ -0,0 +1,712 @@ +import { randomUUID } from 'node:crypto' +import * as migrations from '@internal/database/migrations' +import type { FastifyInstance } from 'fastify' +import pg from 'pg' +import PgBoss from 'pg-boss' +import { getConfig, mergeConfig } from '../config' +import { closeMultitenantPg } from '../internal/database' +import { PgExecutor, PgPoolExecutor } from '../internal/database/pg-connection' +import { PG_BOSS_SCHEMA, Queue, QueueOverflowStorePg } from '../internal/queue' +import { QueueDB } from '../internal/queue/database' +import { createAdminApp } from './common' + +const pgBossJobTable = `${PG_BOSS_SCHEMA}.job` +const pgBossBackupTable = `${PG_BOSS_SCHEMA}.job_overflow_backup` +const testRunId = randomUUID() +const headers = { + apikey: process.env.ADMIN_API_KEYS, +} + +type SeedJobOptions = { + eventType?: string + id?: string + name: string + state?: string + tenantRef?: string +} + +type JobRow = { + id: string + name: string + state: string +} + +let adminApp: FastifyInstance +let queueDbSpy: ReturnType +let routeBoss: PgBoss | undefined +let routeQueueDb: QueueDB | undefined +let queryPool: pg.Pool | undefined +let queryDb: PgPoolExecutor | undefined + +function resolveQueueTestConnectionString(config: ReturnType) { + if (config.pgQueueConnectionURL) { + return config.pgQueueConnectionURL + } + + if (config.isMultitenant) { + return config.multitenantDatabasePoolUrl || config.multitenantDatabaseUrl + } + + return config.databaseURL +} + +async function seedJob(options: SeedJobOptions) { + if (!queryDb) { + throw new Error('Queue test database is not initialized') + } + + const id = options.id ?? randomUUID() + const data: Record = { + queueOverflowTest: testRunId, + } + + if (options.eventType) { + data.event = { + type: options.eventType, + } + } + + if (options.tenantRef) { + data.tenant = { + ref: options.tenantRef, + } + } + + await queryDb.query({ + text: ` + INSERT INTO ${pgBossJobTable} (id, name, state, data) + VALUES ($1, $2, $3, $4::jsonb) + `, + values: [id, options.name, options.state ?? 'created', JSON.stringify(data)], + }) + + return id +} + +async function findJob( + table: string, + name: string, + id: string, + db: PgExecutor | undefined = queryDb +): Promise { + if (!db) { + throw new Error('Queue test database is not initialized') + } + + const result = await db.query({ + text: ` + SELECT id, name, state + FROM ${table} + WHERE name = $1 AND id = $2 + LIMIT 1 + `, + values: [name, id], + }) + + return result.rows[0] +} + +async function findJobData( + table: string, + name: string, + id: string, + db: PgExecutor | undefined = queryDb +): Promise | undefined> { + if (!db) { + throw new Error('Queue test database is not initialized') + } + + const result = await db.query<{ data: Record }>({ + text: ` + SELECT data + FROM ${table} + WHERE name = $1 AND id = $2 + LIMIT 1 + `, + values: [name, id], + }) + + return result.rows[0]?.data +} + +async function cleanupTestJobs(db: PgExecutor | undefined = queryDb) { + if (!db) { + return + } + + const jobTable = await db.query<{ table_name: string | null }>({ + text: 'SELECT to_regclass($1) AS table_name', + values: [pgBossJobTable], + }) + + if (!jobTable.rows[0]?.table_name) { + return + } + + await db.query(`DROP TABLE IF EXISTS ${pgBossBackupTable}`) + await db.query({ + text: `DELETE FROM ${pgBossJobTable} WHERE data->>'queueOverflowTest' = $1`, + values: [testRunId], + }) +} + +// OrioleDB ignores pg-boss partial-index predicates for targetless ON CONFLICT, +// causing restore to discard valid backup rows as conflicts. +describe.skipIf(process.env.DATABASE_ENGINE === 'oriole')('Admin queue overflow routes', () => { + beforeAll(async () => { + getConfig() + mergeConfig({ pgQueueEnable: true }) + await migrations.runMultitenantMigrations() + const config = getConfig() + const connectionString = resolveQueueTestConnectionString(config) + + if (!connectionString) { + throw new Error('Expected a multitenant database URL for queue overflow integration tests') + } + + const inspectionPool = new pg.Pool({ + application_name: config.databaseApplicationName, + connectionString, + connectionTimeoutMillis: config.databaseConnectionTimeout, + max: 1, + statement_timeout: config.pgQueueReadWriteTimeout, + }) + const inspectionDb = new PgPoolExecutor(inspectionPool) + const installation = await inspectionDb.query<{ + job_table: string | null + version_table: string | null + }>({ + text: ` + SELECT + to_regclass($1) AS job_table, + to_regclass($2) AS version_table + `, + values: [`${PG_BOSS_SCHEMA}.job`, `${PG_BOSS_SCHEMA}.version`], + }) + + if (installation.rows[0]?.job_table && !installation.rows[0]?.version_table) { + await inspectionDb.query(`DROP SCHEMA ${PG_BOSS_SCHEMA} CASCADE`) + } + await inspectionPool.end() + + queryPool = new pg.Pool({ + application_name: config.databaseApplicationName, + connectionString, + connectionTimeoutMillis: config.databaseConnectionTimeout, + max: 1, + statement_timeout: config.pgQueueReadWriteTimeout, + }) + queryDb = new PgPoolExecutor(queryPool) + + routeQueueDb = new QueueDB({ + application_name: config.databaseApplicationName, + connectionString, + connectionTimeoutMillis: config.databaseConnectionTimeout, + max: 2, + statement_timeout: config.pgQueueReadWriteTimeout, + }) + routeBoss = new PgBoss({ + connectionString, + db: routeQueueDb, + schema: PG_BOSS_SCHEMA, + schedule: false, + supervise: false, + }) + await routeBoss.start() + await routeBoss.createQueue('webhooks', { name: 'webhooks', policy: 'standard' }) + await routeBoss.createQueue('backup-object', { name: 'backup-object', policy: 'standard' }) + await cleanupTestJobs() + queueDbSpy = vi.spyOn(Queue, 'getDb').mockReturnValue(routeQueueDb) + adminApp = await createAdminApp() + }) + + afterEach(async () => { + await cleanupTestJobs() + }) + + afterAll(async () => { + await adminApp?.close() + queueDbSpy?.mockRestore() + await cleanupTestJobs() + await routeBoss?.stop({ graceful: false, wait: true }).catch(() => routeQueueDb?.close()) + await queryPool?.end() + await closeMultitenantPg() + }) + + it('requires the admin API key', async () => { + const response = await adminApp.inject({ + method: 'GET', + url: '/queue/overflow', + }) + + expect(response.statusCode).toBe(401) + }) + + it('returns exact summary totals and truncation metadata for created jobs', async () => { + const tenantA = `${testRunId}-tenant-a` + const tenantB = `${testRunId}-tenant-b` + const tenantC = `${testRunId}-tenant-c` + await seedJob({ name: 'webhooks', eventType: 'ObjectRemoved:Delete', tenantRef: tenantA }) + await seedJob({ name: 'webhooks', eventType: 'ObjectRemoved:Delete', tenantRef: tenantB }) + await seedJob({ name: 'backup-object', tenantRef: tenantA }) + await seedJob({ + name: 'webhooks', + state: 'active', + eventType: 'ObjectRemoved:Delete', + tenantRef: tenantC, + }) + + const response = await adminApp.inject({ + method: 'GET', + url: `/queue/overflow?limit=1&tenantRefs=${tenantA},${tenantB},${tenantC}`, + headers, + }) + + expect(response.statusCode).toBe(200) + expect(response.json()).toEqual({ + sourceTableExists: true, + source: 'job', + groupBy: 'summary', + groupCount: 2, + hasMore: true, + totalCount: 3, + filters: { tenantRefs: [tenantA, tenantB, tenantC] }, + data: [ + { + count: 2, + eventType: 'ObjectRemoved:Delete', + name: 'webhooks', + }, + ], + }) + }) + + it('returns the global created backlog without grouped list metadata', async () => { + const baselineResponse = await adminApp.inject({ + method: 'GET', + url: '/queue/overflow/count', + headers, + }) + expect(baselineResponse.statusCode).toBe(200) + const baselineCount = baselineResponse.json().totalCount + expect(Number.isInteger(baselineCount)).toBe(true) + + await seedJob({ name: 'webhooks', eventType: 'ObjectRemoved:Delete' }) + await seedJob({ name: 'backup-object' }) + await seedJob({ name: 'webhooks', state: 'active' }) + + const response = await adminApp.inject({ + method: 'GET', + url: '/queue/overflow/count', + headers, + }) + + expect(response.statusCode).toBe(200) + expect(response.json()).toEqual({ totalCount: baselineCount + 2 }) + }) + + it('returns tenant hot spots for a filtered queue and event type', async () => { + const tenantA = `${testRunId}-tenant-a` + const tenantB = `${testRunId}-tenant-b` + const tenantZ = `${testRunId}-tenant-z` + await seedJob({ name: 'webhooks', eventType: 'ObjectRemoved:Delete', tenantRef: tenantB }) + await seedJob({ name: 'webhooks', eventType: 'ObjectRemoved:Delete', tenantRef: tenantB }) + await seedJob({ name: 'webhooks', eventType: 'ObjectRemoved:Delete', tenantRef: tenantA }) + await seedJob({ + name: 'backup-object', + eventType: 'ObjectRemoved:Delete', + tenantRef: tenantZ, + }) + + const response = await adminApp.inject({ + method: 'GET', + url: `/queue/overflow?groupBy=tenant&name=webhooks&eventTypes=ObjectRemoved:Delete&tenantRefs=${tenantA},${tenantB},${tenantZ}`, + headers, + }) + + expect(response.statusCode).toBe(200) + expect(response.json()).toEqual({ + sourceTableExists: true, + source: 'job', + groupBy: 'tenant', + groupCount: 2, + hasMore: false, + totalCount: 3, + filters: { + name: 'webhooks', + eventTypes: ['ObjectRemoved:Delete'], + tenantRefs: [tenantA, tenantB, tenantZ], + }, + data: [ + { count: 2, tenantRef: tenantB }, + { count: 1, tenantRef: tenantA }, + ], + }) + }) + + it('returns an empty result when the overflow backup table does not exist', async () => { + const response = await adminApp.inject({ + method: 'GET', + url: '/queue/overflow?source=backup', + headers, + }) + + expect(response.statusCode).toBe(200) + expect(response.json()).toEqual({ + sourceTableExists: false, + source: 'backup', + groupBy: 'summary', + groupCount: 0, + hasMore: false, + totalCount: 0, + filters: {}, + data: [], + }) + }) + + it('rejects accidental unfiltered and whitespace-only backups', async () => { + const unfiltered = await adminApp.inject({ + method: 'POST', + url: '/queue/overflow/backup', + headers, + payload: {}, + }) + const whitespace = await adminApp.inject({ + method: 'POST', + url: '/queue/overflow/backup', + headers, + payload: { name: ' ' }, + }) + + expect(unfiltered.statusCode).toBe(400) + expect(whitespace.statusCode).toBe(400) + }) + + it('backs up only matching created tuple keys', async () => { + const tenantRef = `${testRunId}-tenant-a` + const sharedId = randomUUID() + await seedJob({ + id: sharedId, + name: 'webhooks', + eventType: 'ObjectRemoved:Delete', + tenantRef, + }) + await seedJob({ id: sharedId, name: 'backup-object', tenantRef }) + const activeId = await seedJob({ + name: 'webhooks', + state: 'active', + eventType: 'ObjectRemoved:Delete', + tenantRef, + }) + + const response = await adminApp.inject({ + method: 'POST', + url: '/queue/overflow/backup', + headers, + payload: { + name: 'webhooks', + eventTypes: ['ObjectRemoved:Delete'], + tenantRefs: [tenantRef], + }, + }) + + expect(response.statusCode).toBe(200) + expect(response.json()).toEqual({ + backupTableCreated: true, + filters: { + name: 'webhooks', + eventTypes: ['ObjectRemoved:Delete'], + tenantRefs: [tenantRef], + }, + limit: null, + movedCount: 1, + }) + await expect(findJob(pgBossJobTable, 'webhooks', sharedId)).resolves.toBeUndefined() + await expect(findJob(pgBossBackupTable, 'webhooks', sharedId)).resolves.toEqual({ + id: sharedId, + name: 'webhooks', + state: 'created', + }) + await expect(findJob(pgBossJobTable, 'backup-object', sharedId)).resolves.toBeDefined() + await expect(findJob(pgBossJobTable, 'webhooks', activeId)).resolves.toBeDefined() + }) + + it('rolls back the backup delete when restoring into the live table fails', async () => { + const tenantRef = `${testRunId}-restore-failure` + const id = await seedJob({ + name: 'webhooks', + eventType: 'ObjectRemoved:Delete', + tenantRef, + }) + + if (!queryDb) { + throw new Error('Queue test database is not initialized') + } + + const backupResponse = await adminApp.inject({ + method: 'POST', + url: '/queue/overflow/backup', + headers, + payload: { name: 'webhooks', tenantRefs: [tenantRef] }, + }) + + expect(backupResponse.statusCode).toBe(200) + await expect(findJob(pgBossJobTable, 'webhooks', id)).resolves.toBeUndefined() + await expect(findJob(pgBossBackupTable, 'webhooks', id)).resolves.toBeDefined() + + await queryDb.query(`ALTER TABLE ${pgBossBackupTable} ALTER COLUMN priority DROP NOT NULL`) + await queryDb.query({ + text: `UPDATE ${pgBossBackupTable} SET priority = NULL WHERE name = $1 AND id = $2`, + values: ['webhooks', id], + }) + + const restoreResponse = await adminApp.inject({ + method: 'POST', + url: '/queue/overflow/restore', + headers, + payload: { name: 'webhooks', tenantRefs: [tenantRef] }, + }) + + expect(restoreResponse.statusCode).toBe(500) + await expect(findJob(pgBossBackupTable, 'webhooks', id)).resolves.toBeDefined() + await expect(findJob(pgBossJobTable, 'webhooks', id)).resolves.toBeUndefined() + }) + + it('requires explicit confirmation before backing up all created jobs', async () => { + await seedJob({ name: 'webhooks' }) + await seedJob({ name: 'backup-object' }) + await seedJob({ name: 'webhooks', state: 'active' }) + + const response = await adminApp.inject({ + method: 'POST', + url: '/queue/overflow/backup', + headers, + payload: { + confirmAll: true, + limit: 1, + }, + }) + + expect(response.statusCode).toBe(200) + expect(response.json()).toMatchObject({ movedCount: 1, limit: 1 }) + }) + + it('restores backed up jobs in deterministic batches', async () => { + const tenantRef = `${testRunId}-restore-batches` + const firstId = await seedJob({ + id: '00000000-0000-0000-0000-000000000041', + name: 'webhooks', + eventType: 'ObjectRemoved:Delete', + tenantRef, + }) + const secondId = await seedJob({ + id: '00000000-0000-0000-0000-000000000042', + name: 'webhooks', + eventType: 'ObjectRemoved:Delete', + tenantRef, + }) + + await adminApp.inject({ + method: 'POST', + url: '/queue/overflow/backup', + headers, + payload: { name: 'webhooks', tenantRefs: [tenantRef] }, + }) + + const batches: unknown[] = [] + let hasMore = true + while (hasMore && batches.length < 10) { + const response = await adminApp.inject({ + method: 'POST', + url: '/queue/overflow/restore', + headers, + payload: { name: 'webhooks', limit: 1 }, + }) + + expect(response.statusCode).toBe(200) + const body = response.json() + batches.push(body) + hasMore = body.hasMore + + if (batches.length === 1) { + await expect(findJob(pgBossJobTable, 'webhooks', firstId)).resolves.toBeDefined() + await expect(findJob(pgBossJobTable, 'webhooks', secondId)).resolves.toBeUndefined() + } + } + + expect(hasMore).toBe(false) + expect(batches).toEqual([ + { + backupTableExists: true, + conflictCount: 0, + filters: { name: 'webhooks' }, + hasMore: true, + limit: 1, + movedCount: 1, + }, + { + backupTableExists: true, + conflictCount: 0, + filters: { name: 'webhooks' }, + hasMore: true, + limit: 1, + movedCount: 1, + }, + { + backupTableExists: true, + conflictCount: 0, + filters: { name: 'webhooks' }, + hasMore: false, + limit: 1, + movedCount: 0, + }, + ]) + await expect(findJob(pgBossJobTable, 'webhooks', firstId)).resolves.toBeDefined() + await expect(findJob(pgBossJobTable, 'webhooks', secondId)).resolves.toBeDefined() + if (!queryDb) { + throw new Error('Queue test database is not initialized') + } + await expect( + queryDb.query<{ count: number }>({ + text: `SELECT COUNT(*)::int AS count FROM ${pgBossBackupTable}`, + }) + ).resolves.toMatchObject({ rows: [{ count: 0 }] }) + }) + + it('drops archived conflicts while preserving the live job', async () => { + const tenantRef = `${testRunId}-conflicts` + const id = await seedJob({ + id: '00000000-0000-0000-0000-000000000001', + name: 'webhooks', + eventType: 'ObjectRemoved:Delete', + tenantRef, + }) + const restorableId = await seedJob({ + id: '00000000-0000-0000-0000-000000000002', + name: 'webhooks', + eventType: 'ObjectRemoved:Delete', + tenantRef, + }) + await adminApp.inject({ + method: 'POST', + url: '/queue/overflow/backup', + headers, + payload: { name: 'webhooks', tenantRefs: [tenantRef] }, + }) + await seedJob({ id, name: 'webhooks', eventType: 'ObjectCreated:Put' }) + + const response = await adminApp.inject({ + method: 'POST', + url: '/queue/overflow/restore', + headers, + payload: { name: 'webhooks', limit: 10 }, + }) + + expect(response.statusCode).toBe(200) + expect(response.json()).toEqual({ + backupTableExists: true, + conflictCount: 1, + filters: { name: 'webhooks' }, + hasMore: false, + limit: 10, + movedCount: 1, + }) + await expect(findJob(pgBossBackupTable, 'webhooks', id)).resolves.toBeUndefined() + await expect(findJob(pgBossJobTable, 'webhooks', id)).resolves.toBeDefined() + await expect(findJobData(pgBossJobTable, 'webhooks', id)).resolves.toEqual({ + event: { type: 'ObjectCreated:Put' }, + queueOverflowTest: testRunId, + }) + await expect(findJob(pgBossBackupTable, 'webhooks', restorableId)).resolves.toBeUndefined() + await expect(findJob(pgBossJobTable, 'webhooks', restorableId)).resolves.toBeDefined() + + const followUp = await adminApp.inject({ + method: 'POST', + url: '/queue/overflow/restore', + headers, + payload: { name: 'webhooks', limit: 10 }, + }) + + expect(followUp.statusCode).toBe(200) + expect(followUp.json()).toEqual({ + backupTableExists: true, + conflictCount: 0, + filters: { name: 'webhooks' }, + hasMore: false, + limit: 10, + movedCount: 0, + }) + }) + + it('moves only the matching tuple through a partitioned pg-boss table', async () => { + const schema = `qo_${randomUUID().replaceAll('-', '')}` + const jobTable = `${schema}.job` + const backupTable = `${schema}.job_overflow_backup` + const id = randomUUID() + const config = getConfig() + const connectionString = resolveQueueTestConnectionString(config) + + if (!connectionString) { + throw new Error('Expected a multitenant database URL for queue overflow integration test') + } + + const queueDb = new QueueDB({ + connectionString, + max: 1, + statement_timeout: 5000, + }) + const boss = new PgBoss({ + connectionString, + db: queueDb, + schema, + schedule: false, + supervise: false, + }) + + try { + await boss.start() + await boss.createQueue('webhooks', { name: 'webhooks', policy: 'standard' }) + await boss.createQueue('backup-object', { name: 'backup-object', policy: 'standard' }) + await boss.createQueue('exactly-once', { name: 'exactly-once', policy: 'exactly_once' }) + await boss.send('webhooks', { event: { type: 'ObjectRemoved:Delete' } }, { id }) + await boss.send('backup-object', {}, { id }) + + if (!queryDb) { + throw new Error('Queue test database is not initialized') + } + + const store = new QueueOverflowStorePg(queueDb, schema) + + await expect(store.backup({ name: 'webhooks' })).resolves.toMatchObject({ movedCount: 1 }) + await expect(findJob(jobTable, 'webhooks', id, queryDb)).resolves.toBeUndefined() + await expect(findJob(jobTable, 'backup-object', id, queryDb)).resolves.toBeDefined() + await expect(findJob(backupTable, 'webhooks', id, queryDb)).resolves.toBeDefined() + + await expect(store.restore({ name: 'webhooks', limit: 1 })).resolves.toMatchObject({ + movedCount: 1, + }) + await expect(findJob(jobTable, 'webhooks', id, queryDb)).resolves.toBeDefined() + + const archivedId = randomUUID() + const replacementId = randomUUID() + await boss.send('exactly-once', {}, { id: archivedId, singletonKey: 'tenant-a' }) + await expect(store.backup({ name: 'exactly-once' })).resolves.toMatchObject({ movedCount: 1 }) + await boss.send('exactly-once', {}, { id: replacementId, singletonKey: 'tenant-a' }) + + await expect(store.restore({ name: 'exactly-once', limit: 1 })).resolves.toMatchObject({ + conflictCount: 1, + hasMore: true, + movedCount: 0, + }) + await expect( + findJob(backupTable, 'exactly-once', archivedId, queryDb) + ).resolves.toBeUndefined() + await expect(findJob(jobTable, 'exactly-once', replacementId, queryDb)).resolves.toBeDefined() + } finally { + await boss.stop({ graceful: false, wait: true }).catch(() => queueDb.close()) + if (queryDb) { + await queryDb.query(`DROP SCHEMA IF EXISTS ${schema} CASCADE`) + } + } + }) +})