diff --git a/.env.sample b/.env.sample index 236681d26..78c4f8c22 100644 --- a/.env.sample +++ b/.env.sample @@ -143,6 +143,10 @@ S3_PROTOCOL_ACCESS_KEY_SECRET=67d161a7a8a46a24a17a75b26e7724f11d56b8d49a119227c6 ####################################### STORAGE_BACKEND=s3 +# Pause lifecycle processing, +# policy APIs (PUT/DELETE/GET) remain available. +STORAGE_LIFECYCLE_PAUSED=false + # Evaluation pages are capped at 480 rows and deletion batches at 48 versions. STORAGE_LIFECYCLE_PAGE_SIZE=480 STORAGE_LIFECYCLE_DELETE_BATCH_SIZE=48 @@ -157,6 +161,7 @@ STORAGE_LIFECYCLE_CLAIM_LEASE_MS=120000 # Enable object versioning and bucket lifecycle configuration APIs. In multitenant # mode the tenant's objectVersioning feature can also enable them. +# The lifecycle engine requires PG_QUEUE_ENABLE=true and workers enabled (PG_QUEUE_WORKERS_ENABLE). STORAGE_VERSIONING_ENABLED=false ####################################### diff --git a/src/config.test.ts b/src/config.test.ts index 95db191cd..d636b10b4 100644 --- a/src/config.test.ts +++ b/src/config.test.ts @@ -24,6 +24,8 @@ const CONFIG_ENV_KEYS = [ 'OTEL_EXPORTER_OTLP_ENDPOINT', 'OTEL_EXPORTER_OTLP_METRICS_ENDPOINT', 'REQUEST_HARD_LIMITS_ENABLED', + 'STORAGE_VERSIONING_ENABLED', + 'STORAGE_LIFECYCLE_PAUSED', 'STORAGE_LIFECYCLE_PAGE_SIZE', 'STORAGE_LIFECYCLE_DELETE_BATCH_SIZE', 'STORAGE_LIFECYCLE_JOB_BUDGET_MS', @@ -78,6 +80,19 @@ function setConfigEnv(env: Partial>) { } describe('configuration parsing', () => { + test.each([ + [undefined, false], + ['', false], + ['false', false], + ['true', true], + ['TRUE', false], + ])('parses STORAGE_LIFECYCLE_PAUSED=%s as %s', async (value, paused) => { + setConfigEnv({}) + vi.stubEnv('STORAGE_LIFECYCLE_PAUSED', value) + const { getConfig } = await import('./config') + expect(getConfig({ reload: true }).storageLifecyclePaused).toBe(paused) + }) + test.each([ ['TENANT_CONFIG', 'tenantConfig', 16384, 3600000], ['TENANT_JWKS', 'tenantJwks', 16384, 3600000], diff --git a/src/config.ts b/src/config.ts index 3e72af1d3..1a3e56367 100644 --- a/src/config.ts +++ b/src/config.ts @@ -107,6 +107,7 @@ type StorageConfigType = { storageS3ForcePathStyle?: boolean storageS3Region: string storageS3ClientTimeout: number + storageLifecyclePaused: boolean storageLifecyclePageSize: number storageLifecycleDeleteBatchSize: number storageLifecycleJobBudgetMs: number @@ -539,6 +540,7 @@ export function getConfig(options?: { reload?: boolean }): StorageConfigType { // Storage storageBackendType: getOptionalConfigFromEnv('STORAGE_BACKEND') as StorageBackendType, emptyBucketMax: parseInt(getOptionalConfigFromEnv('STORAGE_EMPTY_BUCKET_MAX') || '200000', 10), + storageLifecyclePaused: getOptionalConfigFromEnv('STORAGE_LIFECYCLE_PAUSED') === 'true', storageLifecyclePageSize: envStrictPositiveInteger( 'STORAGE_LIFECYCLE_PAGE_SIZE', LIFECYCLE_MAX_PAGE_SIZE, diff --git a/src/internal/database/lifecycle-tenant-store.ts b/src/internal/database/lifecycle-tenant-store.ts index b4713d061..8ecdbcc2f 100644 --- a/src/internal/database/lifecycle-tenant-store.ts +++ b/src/internal/database/lifecycle-tenant-store.ts @@ -1,11 +1,10 @@ import type { DatabaseExecutor } from './connection' -const TENANT_ENROLLMENT_BATCH_SIZE = 1000 - export interface LifecycleTenantClaim { tenantId: string claimId: string claimUntil: Date + reclaimed: boolean } export class LifecycleTenantStorePg { @@ -34,44 +33,6 @@ export class LifecycleTenantStorePg { ) } - async enrollAllTenants(): Promise { - const upperBound = await this.database.query<{ id: string | null }>( - 'SELECT max(id) AS id FROM tenants' - ) - const lastId = upperBound.rows[0].id - if (lastId === null) return - - let cursor: string | null = null - while (true) { - const batch: { rows: { id: string }[] } = await this.database.query<{ id: string }>({ - text: ` - WITH batch AS MATERIALIZED ( - SELECT id FROM tenants - WHERE id >= COALESCE($1::text, '') AND id <= $2 - AND ($1::text IS NULL OR id > $1) - ORDER BY id - LIMIT $3 - ), enrolled AS ( - INSERT INTO lifecycle_tenants (tenant_id, next_dispatch_at) - SELECT tenants.id, clock_timestamp() - FROM tenants JOIN batch ON batch.id = tenants.id - WHERE NOT EXISTS ( - SELECT 1 FROM lifecycle_tenants WHERE tenant_id = tenants.id - ) - ORDER BY tenants.id - FOR KEY SHARE OF tenants - ON CONFLICT (tenant_id) DO NOTHING - ) - SELECT id FROM batch ORDER BY id DESC LIMIT 1 - `, - values: [cursor, lastId, TENANT_ENROLLMENT_BATCH_SIZE], - }) - const next: string | undefined = batch.rows[0]?.id - if (next === undefined || next === lastId) return - cursor = next - } - } - async hasObjectVersioningFeature(tenantId: string): Promise { const result = await this.database.query<{ enabled: boolean }>({ text: ` @@ -84,45 +45,26 @@ export class LifecycleTenantStorePg { return result.rows[0].enabled } - async retireTenant(tenantId: string, claimId: string, deadline?: string): Promise { - const result = await this.database.query({ - text: ` - WITH owned AS ( - SELECT tenant_id FROM lifecycle_tenants - WHERE tenant_id = $1 - AND claim_id = $2::uuid - AND next_dispatch_at = 'infinity' - AND NOT EXISTS ( - SELECT 1 FROM tenants WHERE tenants.id = $1 AND tenants.feature_object_versioning - ) - FOR UPDATE - ), current AS ( - SELECT tenant_id FROM owned - WHERE $3::timestamptz IS NULL OR clock_timestamp() < $3::timestamptz - ) - DELETE FROM lifecycle_tenants AS target - USING current - WHERE target.tenant_id = current.tenant_id - `, - values: [tenantId, claimId, deadline ?? null], - }) - return result.rowCount === 1 - } - - async claimDueTenants(limit: number, leaseMs: number): Promise { - const result = await this.database.query({ - text: ` + async claimDueTenants( + limit: number, + leaseMs: number, + unclaimedLimit = limit, + signal?: AbortSignal + ): Promise { + const result = await this.database.query( + { + text: ` WITH unclaimed AS MATERIALIZED ( - SELECT tenant_id, next_dispatch_at AS due_at + SELECT tenant_id, next_dispatch_at AS due_at, false AS reclaimed FROM lifecycle_tenants WHERE claim_until IS NULL AND next_dispatch_at <= statement_timestamp() ORDER BY next_dispatch_at, tenant_id FOR UPDATE SKIP LOCKED - LIMIT $1 + LIMIT $3 ), expired AS MATERIALIZED ( - SELECT tenant_id, claim_until AS due_at + SELECT tenant_id, claim_until AS due_at, true AS reclaimed FROM lifecycle_tenants WHERE claim_until < statement_timestamp() ORDER BY claim_until, tenant_id @@ -130,17 +72,17 @@ export class LifecycleTenantStorePg { LIMIT $1 ), due AS MATERIALIZED ( - SELECT tenant_id + SELECT tenant_id, reclaimed FROM ( - SELECT tenant_id, due_at FROM unclaimed + SELECT tenant_id, due_at, reclaimed FROM unclaimed UNION ALL - SELECT tenant_id, due_at FROM expired + SELECT tenant_id, due_at, reclaimed FROM expired ) candidates ORDER BY due_at, tenant_id LIMIT $1 ) - -- An expired lease can still have a queued job. Keep its claim id while - -- backing off repeated redispatches; a successful job clears the failure. + -- An expired lease can still have a queued job. Keep its claim id and back off + -- redispatches; the scheduler records a failure only if the re-send creates a new job. UPDATE lifecycle_tenants AS target SET claim_id = COALESCE(target.claim_id, gen_random_uuid()), claim_until = clock_timestamp() + CASE @@ -152,33 +94,51 @@ export class LifecycleTenantStorePg { END, next_dispatch_at = CASE WHEN target.claim_until IS NULL THEN 'infinity'::timestamptz ELSE target.next_dispatch_at END, - failure_count = CASE WHEN target.claim_until IS NULL - THEN target.failure_count ELSE target.failure_count + 1 END, - last_error = CASE WHEN target.claim_until IS NULL THEN target.last_error - ELSE jsonb_build_object( - 'code', 'tenant_dispatch_lease_expired', - 'message', 'Tenant dispatch lease expired before completion' - ) END, last_dispatched_at = clock_timestamp() FROM due WHERE target.tenant_id = due.tenant_id RETURNING target.tenant_id AS "tenantId", target.claim_id AS "claimId", - target.claim_until AS "claimUntil" + target.claim_until AS "claimUntil", + due.reclaimed AS "reclaimed" `, - values: [limit, leaseMs], - }) + values: [limit, leaseMs, unclaimedLimit], + }, + { signal } + ) return result.rows } + // Must run in a transaction: the lock serializes claimers until it ends. + async claimDueTenantsWithinLeaseLimit( + maxLeased: number, + leaseMs: number, + signal?: AbortSignal + ): Promise { + await this.database.query("SELECT pg_advisory_xact_lock(hashtext('lifecycle_tenants_claim'))", { + signal, + }) + const result = await this.database.query<{ leased: number }>( + 'SELECT count(*)::int AS leased FROM lifecycle_tenants WHERE claim_until IS NOT NULL', + { signal } + ) + // An expired lease may still have a queued job: it keeps its slot, and reclaiming it adds no job. + return this.claimDueTenants( + maxLeased, + leaseMs, + Math.max(0, maxLeased - result.rows[0].leased), + signal + ) + } + async completeTenantDispatch( tenantId: string, claimId: string, computedNextDispatchAt: string, - health?: { maxFailureCount: number; oldestInFlightStartedAt: string | null }, - deadline?: string + deadline: string, + health?: { maxFailureCount: number; oldestInFlightStartedAt: string | null } ): Promise { const result = await this.database.query({ text: ` @@ -188,7 +148,7 @@ export class LifecycleTenantStorePg { FOR UPDATE ), current AS ( SELECT tenant_id FROM owned - WHERE $7::timestamptz IS NULL OR clock_timestamp() < $7::timestamptz + WHERE clock_timestamp() < $7::timestamptz ) UPDATE lifecycle_tenants AS target SET next_dispatch_at = LEAST(next_dispatch_at, $3::timestamptz), @@ -211,17 +171,61 @@ export class LifecycleTenantStorePg { health !== undefined, health?.maxFailureCount ?? null, health?.oldestInFlightStartedAt ?? null, - deadline ?? null, + deadline, ], }) return result.rowCount === 1 } + async releaseTenantDispatch( + tenantId: string, + claimId: string, + nextDispatchAt: string, + deadline: string + ): Promise { + const result = await this.database.query({ + text: ` + WITH owned AS ( + SELECT tenant_id FROM lifecycle_tenants + WHERE tenant_id = $1 AND claim_id = $2::uuid + FOR UPDATE + ), current AS ( + SELECT tenant_id FROM owned + WHERE clock_timestamp() < $4::timestamptz + ) + UPDATE lifecycle_tenants AS target + SET next_dispatch_at = LEAST(next_dispatch_at, $3::timestamptz), + claim_id = NULL, + claim_until = NULL + FROM current + WHERE target.tenant_id = current.tenant_id + `, + values: [tenantId, claimId, nextDispatchAt, deadline], + }) + return result.rowCount === 1 + } + + async recordExpiredTenantDispatch(tenantId: string, claimId: string): Promise { + const result = await this.database.query({ + text: ` + UPDATE lifecycle_tenants + SET failure_count = failure_count + 1, + last_error = jsonb_build_object( + 'code', 'tenant_dispatch_lease_expired', + 'message', 'Tenant dispatch lease expired before completion' + ) + WHERE tenant_id = $1 AND claim_id = $2::uuid + `, + values: [tenantId, claimId], + }) + return result.rowCount === 1 + } + async failTenantDispatch( tenantId: string, claimId: string, error: Record, - deadline?: string + deadline: string ): Promise { const result = await this.database.query({ text: ` @@ -231,7 +235,7 @@ export class LifecycleTenantStorePg { FOR UPDATE ), current AS ( SELECT tenant_id FROM owned - WHERE $4::timestamptz IS NULL OR clock_timestamp() < $4::timestamptz + WHERE clock_timestamp() < $4::timestamptz ) UPDATE lifecycle_tenants AS target SET next_dispatch_at = LEAST(next_dispatch_at, @@ -246,7 +250,7 @@ export class LifecycleTenantStorePg { FROM current WHERE target.tenant_id = current.tenant_id `, - values: [tenantId, claimId, JSON.stringify(error), deadline ?? null], + values: [tenantId, claimId, JSON.stringify(error), deadline], }) return result.rowCount === 1 } diff --git a/src/internal/monitoring/metrics.ts b/src/internal/monitoring/metrics.ts index 85d07951a..0bb38dc0d 100644 --- a/src/internal/monitoring/metrics.ts +++ b/src/internal/monitoring/metrics.ts @@ -500,6 +500,16 @@ export const queueJobError = registerMetric('queue_job_error', 'updowncounter', // Object lifecycle metrics. Attribute sets are deliberately closed and never // contain tenant, bucket, object, rule, run, claim, or attempt identifiers. // ============================================================================ +export const lifecycleRunDuration = registerMetric( + 'lifecycle_run_duration_seconds', + 'histogram', + () => + meter.createHistogram('lifecycle_run_duration_seconds', { + description: 'Lifecycle bucket delivery duration', + unit: 's', + }) +) + export const lifecycleVersionsExamined = registerMetric( 'lifecycle_versions_examined_total', 'counter', diff --git a/src/internal/monitoring/otel-metrics.test.ts b/src/internal/monitoring/otel-metrics.test.ts index 4f0e3ffd7..0545cd56b 100644 --- a/src/internal/monitoring/otel-metrics.test.ts +++ b/src/internal/monitoring/otel-metrics.test.ts @@ -267,6 +267,14 @@ describe('otel metrics', () => { instrumentName: '*', aggregation: { type: 'DROP' }, }, + { + meterName: 'storage-api', + instrumentName: 'lifecycle_run_duration_seconds', + aggregation: { + type: 'EXPLICIT_BUCKET_HISTOGRAM', + options: { boundaries: expect.arrayContaining([120, 180, 86400]) }, + }, + }, { meterName: '@opentelemetry/instrumentation-runtime-node', instrumentName: 'v8js.gc.duration', diff --git a/src/internal/monitoring/otel-metrics.ts b/src/internal/monitoring/otel-metrics.ts index b4e9288f8..c527ecd2c 100644 --- a/src/internal/monitoring/otel-metrics.ts +++ b/src/internal/monitoring/otel-metrics.ts @@ -143,6 +143,19 @@ const views = [ instrumentName: 's3_upload_part_seconds', aggregation: histogramAggregation, }, + { + meterName: 'storage-api', + instrumentName: 'lifecycle_run_duration_seconds', + aggregation: { + type: AggregationType.EXPLICIT_BUCKET_HISTOGRAM as const, + options: { + boundaries: [ + 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 20, 30, 45, 60, 90, 120, 180, 300, 600, 1800, 3600, 21600, + 86400, + ], + }, + }, + }, // Override the RuntimeNodeInstrumentation default GC buckets ([10ms, 100ms, // 1s, 10s]) with sub-ms boundaries so incremental/minor pauses are visible. { diff --git a/src/start/server.test.ts b/src/start/server.test.ts index 7dae0ef7c..89bb7df07 100644 --- a/src/start/server.test.ts +++ b/src/start/server.test.ts @@ -15,13 +15,17 @@ const startupState = vi.hoisted((): { shutdownController: AsyncAbortController | const migrationDrainState = vi.hoisted((): { deferred: Deferred } => ({ deferred: Promise.withResolvers(), })) +const lifecycleDrainState = vi.hoisted((): { deferred: Deferred } => ({ + deferred: Promise.withResolvers(), +})) const mockAdminListen = vi.hoisted(() => vi.fn()) const serverConfigState = vi.hoisted(() => ({ isMultitenant: false })) vi.mock('./tracing', () => ({})) vi.mock('@internal/monitoring/otel-metrics', () => ({})) vi.mock('@internal/monitoring', () => ({ - logger: { info: vi.fn() }, + logger: { info: vi.fn(), child: vi.fn(() => ({})) }, + setLogger: vi.fn(), logSchema: { info: vi.fn(), error: vi.fn() }, })) vi.mock('@internal/cluster/cluster', () => ({ @@ -82,7 +86,16 @@ vi.mock('@internal/sharding', () => ({ }, })) vi.mock('@platformatic/globals', () => ({ getGlobal: () => undefined })) -vi.mock('@storage/events', () => ({ registerWorkers: vi.fn() })) +vi.mock('@storage/events', () => ({ + registerWorkers: vi.fn(), + startLifecycleScheduler: vi.fn((signal: AbortSignal) => { + bootOrder.push('lifecycle.start') + signal.addEventListener('abort', () => { + shutdownOrder.push('lifecycle') + return lifecycleDrainState.deferred.promise + }) + }), +})) vi.mock('@storage/events/upgrades/sync-catalog-ids', () => ({ SyncCatalogIds: { invoke: vi.fn(async () => {}) }, })) @@ -109,6 +122,7 @@ vi.mock('../http/plugins/storage', () => ({ }, })) vi.mock('../config', () => ({ + markWorker: vi.fn(), getConfig: () => ({ databaseURL: 'postgres://example', isMultitenant: serverConfigState.isMultitenant, @@ -139,10 +153,12 @@ describe('server boot order', () => { beforeEach(() => { serverConfigState.isMultitenant = false migrationDrainState.deferred = Promise.withResolvers() + lifecycleDrainState.deferred = Promise.withResolvers() }) afterEach(() => { migrationDrainState.deferred.resolve() + lifecycleDrainState.deferred.resolve() bootOrder.length = 0 shutdownOrder.length = 0 startupState.shutdownController = undefined @@ -156,7 +172,7 @@ describe('server boot order', () => { }) await import('./server') - await vi.waitFor(() => expect(bootOrder).toContain('queue.start')) + await vi.waitFor(() => expect(bootOrder).toContain('lifecycle.start')) expect(bootOrder).toEqual([ 'poolManager.setNumWorkers', @@ -164,42 +180,57 @@ describe('server boot order', () => { 'cluster.init', 'cluster.on', 'queue.start', + 'lifecycle.start', ]) expect(exitSpy).not.toHaveBeenCalled() }) - test('drains migration producers before stopping the queue and its dependencies', async () => { - const exitSpy = vi.spyOn(process, 'exit').mockImplementation(() => { - throw new Error('server unexpectedly called process.exit()') - }) - serverConfigState.isMultitenant = true - - await import('./server') - await vi.waitFor(() => expect(startupState.shutdownController).toBeDefined()) - await vi.waitFor(() => expect(bootOrder).toContain('queue.start')) - await vi.waitFor(() => expect(mockAdminListen).toHaveBeenCalledOnce()) - expect(vi.mocked(build).mock.calls.at(-1)?.[0]?.sharedStorageBackend).toBe( - vi.mocked(buildAdmin).mock.calls.at(-1)?.[0]?.sharedStorageBackend - ) - expect(vi.mocked(build).mock.calls.at(-1)?.[0]?.sharedStorageBackend).toBeDefined() - - const shutdownController = startupState.shutdownController - expect(shutdownController).toBeDefined() - if (!shutdownController) { - throw new Error('Server did not register its shutdown controller') + test.each(['server', 'worker'] as const)( + '%s drains lifecycle producers before stopping the queue and its dependencies', + async (entry) => { + const exitSpy = vi.spyOn(process, 'exit').mockImplementation(() => { + throw new Error('server unexpectedly called process.exit()') + }) + serverConfigState.isMultitenant = true + + await (entry === 'server' ? import('./server') : import('./worker')) + await vi.waitFor(() => expect(startupState.shutdownController).toBeDefined()) + await vi.waitFor(() => expect(bootOrder).toContain('queue.start')) + await vi.waitFor(() => expect(mockAdminListen).toHaveBeenCalledOnce()) + if (entry === 'server') { + expect(vi.mocked(build).mock.calls.at(-1)?.[0]?.sharedStorageBackend).toBe( + vi.mocked(buildAdmin).mock.calls.at(-1)?.[0]?.sharedStorageBackend + ) + expect(vi.mocked(build).mock.calls.at(-1)?.[0]?.sharedStorageBackend).toBeDefined() + } + + const shutdownController = startupState.shutdownController + expect(shutdownController).toBeDefined() + if (!shutdownController) { + throw new Error('Server did not register its shutdown controller') + } + + const abortPromise = shutdownController.abortAsync() + await vi.waitFor(() => { + expect(shutdownOrder).toEqual( + entry === 'server' ? ['api', 'admin', 'lifecycle', 'migrations'] : ['lifecycle'] + ) + }) + + migrationDrainState.deferred.resolve() + await Promise.resolve() + expect(shutdownOrder).not.toContain('queue') + lifecycleDrainState.deferred.resolve() + await abortPromise + + expect(shutdownOrder).toEqual( + entry === 'server' + ? ['api', 'admin', 'lifecycle', 'migrations', 'queue', 'cluster', 'pubsub'] + : ['lifecycle', 'queue', 'admin', 'pubsub'] + ) + expect(exitSpy).not.toHaveBeenCalled() } - - const abortPromise = shutdownController.abortAsync() - await vi.waitFor(() => { - expect(shutdownOrder).toEqual(['api', 'admin', 'migrations']) - }) - - migrationDrainState.deferred.resolve() - await abortPromise - - expect(shutdownOrder).toEqual(['api', 'admin', 'migrations', 'queue', 'cluster', 'pubsub']) - expect(exitSpy).not.toHaveBeenCalled() - }) + ) test('exits before starting queue workers when initial discovery times out', async () => { const error = new DOMException('Cluster discovery timed out', 'TimeoutError') diff --git a/src/start/server.ts b/src/start/server.ts index 44089d395..ed9b23b28 100644 --- a/src/start/server.ts +++ b/src/start/server.ts @@ -19,7 +19,7 @@ import { logger, logSchema } from '@internal/monitoring' import { Queue, SYSTEM_TENANT } from '@internal/queue' import { PgShardStoreFactory, ShardCatalog } from '@internal/sharding' import { getGlobal } from '@platformatic/globals' -import { registerWorkers } from '@storage/events' +import { registerWorkers, startLifecycleScheduler } from '@storage/events' import { SyncCatalogIds } from '@storage/events/upgrades/sync-catalog-ids' import { FastifyInstance, LogController } from 'fastify' import buildAdmin from '../admin-app' @@ -72,7 +72,7 @@ async function main() { icebergShards, numWorkers, } = getConfig() - // Abort order: HTTP -> migrations -> queue -> dependencies. + // Abort order: HTTP -> migration/lifecycle producers -> queue -> dependencies. const migrationShutdownGroup = shutdownSignal.nextGroup const queueShutdownGroup = migrationShutdownGroup.nextGroup const dependencyShutdownGroup = queueShutdownGroup.nextGroup @@ -167,6 +167,8 @@ async function main() { }) } + startLifecycleScheduler(migrationShutdownGroup.signal) + // Pubsub await PubSub.start({ signal: dependencyShutdownGroup.signal, diff --git a/src/start/worker.ts b/src/start/worker.ts index 5fe892d29..647c37f54 100644 --- a/src/start/worker.ts +++ b/src/start/worker.ts @@ -3,7 +3,7 @@ import { AsyncAbortController } from '@internal/concurrency' import { listenForTenantUpdate, PubSub } from '@internal/database' import { logger, logSchema, setLogger } from '@internal/monitoring' import { Queue } from '@internal/queue' -import { registerWorkers } from '@storage/events' +import { registerWorkers, startLifecycleScheduler } from '@storage/events' import { LogController } from 'fastify' import adminApp from '../admin-app' import { getConfig } from '../config' @@ -43,6 +43,10 @@ main() */ export async function main() { const { requestTraceHeader, adminPort, host } = getConfig() + // Drain lifecycle producers before the queue, admin server and pubsub. + const queueShutdownGroup = shutdownSignal.nextGroup + const adminShutdownGroup = queueShutdownGroup.nextGroup + const dependencyShutdownGroup = adminShutdownGroup.nextGroup logger.info('[Queue] Starting Queue Worker') @@ -50,7 +54,7 @@ export async function main() { await Promise.all([ Queue.start({ - signal: shutdownSignal.signal, + signal: queueShutdownGroup.signal, registerWorkers, onMessage: (job) => logger.info( @@ -62,10 +66,12 @@ export async function main() { ), }), PubSub.start({ - signal: shutdownSignal.nextGroup.nextGroup.signal, + signal: dependencyShutdownGroup.signal, }), ]) + startLifecycleScheduler(shutdownSignal.signal) + const server = adminApp({ sharedStorageBackend, loggerInstance: logger, @@ -82,12 +88,12 @@ export async function main() { }) }) - shutdownSignal.nextGroup.signal.addEventListener('abort', async () => { + adminShutdownGroup.signal.addEventListener('abort', async () => { logSchema.info(logger, '[Admin Server] Stopping', { type: 'server', }) await shutdownPromise }) - await server.listen({ port: adminPort, host, signal: shutdownSignal.nextGroup.signal }) + await server.listen({ port: adminPort, host, signal: adminShutdownGroup.signal }) } diff --git a/src/storage/events/index.ts b/src/storage/events/index.ts index acf5b3774..a398a2bac 100644 --- a/src/storage/events/index.ts +++ b/src/storage/events/index.ts @@ -8,6 +8,7 @@ export * from './lifecycle/object-created' export * from './lifecycle/object-promoted' export * from './lifecycle/object-removed' export * from './lifecycle/webhook' +export * from './lifecycle-rules' export * from './migrations/reset-migrations' export * from './migrations/run-migrations' export * from './objects/backup-object' diff --git a/src/storage/events/lifecycle-rules/bucket-run.ts b/src/storage/events/lifecycle-rules/bucket-run.ts new file mode 100644 index 000000000..5d3c2e717 --- /dev/null +++ b/src/storage/events/lifecycle-rules/bucket-run.ts @@ -0,0 +1,76 @@ +import { logger, logSchema } from '@internal/monitoring' +import { lifecycleRunDuration } from '@internal/monitoring/metrics' +import { BasePayload } from '@internal/queue' +import { JobWithMetadata, Queue, SendOptions, WorkOptions } from 'pg-boss' +import { getConfig } from '../../../config' +import { NoncurrentLifecycleExecutor } from '../../lifecycle/executor' +import { isLifecycleEnabledForTenant } from '../../lifecycle/guards' +import { BaseEvent } from '../base-event' + +const config = getConfig() + +const LIFECYCLE_BUCKET_CONCURRENCY = 8 + +export interface LifecycleBucketRunPayload extends BasePayload { + bucketId: string +} + +export class LifecycleBucketRun extends BaseEvent { + static queueName = 'lifecycle:bucket:run-v1' + static allowSync = false + + static getQueueOptions(): Queue { + return { name: this.queueName, policy: 'exactly_once' } as const + } + + static getSendOptions(payload: LifecycleBucketRunPayload): SendOptions { + return { + singletonKey: `${payload.tenant.ref}/${payload.bucketId}`, + expireInMinutes: Math.ceil( + (config.storageLifecycleJobBudgetMs + config.storageLifecycleClaimLeaseMs) / 60_000 + ), + retryLimit: 0, + } + } + + static getWorkerOptions(): WorkOptions & { concurrentTaskCount: number } { + return { + includeMetadata: true, + concurrentTaskCount: LIFECYCLE_BUCKET_CONCURRENCY, + batchSize: LIFECYCLE_BUCKET_CONCURRENCY, + } + } + + static async shouldSend(): Promise { + return true + } + + static async handle( + job: JobWithMetadata, + opts?: { signal?: AbortSignal } + ): Promise { + const startedAt = performance.now() + let status = 'ERROR' + let storage: Awaited> | undefined + try { + storage = await this.createStorage(job.data) + const result = await new NoncurrentLifecycleExecutor(storage).run(job.data.bucketId, { + tenantLifecycleEnabled: await isLifecycleEnabledForTenant(job.data.tenant.ref), + signal: opts?.signal, + }) + status = result.status + logSchema.info(logger, '[Lifecycle] Bucket delivery completed', { + type: 'event', + project: job.data.tenant.ref, + metadata: JSON.stringify({ + bucketId: job.data.bucketId, + runId: result.runId, + status: result.status, + }), + }) + } finally { + lifecycleRunDuration.record((performance.now() - startedAt) / 1000, { status }) + storage?.db.destroyConnection() + } + } +} diff --git a/src/storage/events/lifecycle-rules/index.ts b/src/storage/events/lifecycle-rules/index.ts new file mode 100644 index 000000000..1683226c9 --- /dev/null +++ b/src/storage/events/lifecycle-rules/index.ts @@ -0,0 +1,3 @@ +export * from './bucket-run' +export { startLifecycleScheduler } from './scheduler' +export * from './tenant-run' diff --git a/src/storage/events/lifecycle-rules/lifecycle-rules.test.ts b/src/storage/events/lifecycle-rules/lifecycle-rules.test.ts new file mode 100644 index 000000000..a6bc11fb1 --- /dev/null +++ b/src/storage/events/lifecycle-rules/lifecycle-rules.test.ts @@ -0,0 +1,1057 @@ +import { AsyncAbortController } from '@internal/concurrency' +import { getTenantConfig } from '@internal/database' +import { logger, logSchema } from '@internal/monitoring' +import { lifecycleRunDuration } from '@internal/monitoring/metrics' +import { type BasePayload, Queue } from '@internal/queue' +import { vi } from 'vitest' + +const mocks = vi.hoisted(() => ({ + config: { + storageVersioningEnabled: false, + storageLifecyclePaused: false, + storageLifecycleClaimLeaseMs: 120_000, + storageLifecycleJobBudgetMs: 20_000, + isMultitenant: true, + tenantId: 'single-tenant', + storageS3Bucket: 'data', + storageS3MaxSockets: 10, + storageBackendType: 's3', + region: 'test', + pgQueueEnable: true, + pgQueueEnableWorkers: true, + }, + central: { + claimDueTenantsWithinLeaseLimit: vi.fn(), + releaseTenantDispatch: vi.fn(), + recordExpiredTenantDispatch: vi.fn(), + completeTenantDispatch: vi.fn(), + failTenantDispatch: vi.fn(), + hasObjectVersioningFeature: vi.fn(), + }, + getQueue: vi.fn(), + transaction: { query: vi.fn(), commit: vi.fn(), rollback: vi.fn() }, + beginTransaction: vi.fn(), + executorRun: vi.fn(), + retryAbandoned: vi.fn(), + tenantHasFeature: vi.fn(), +})) + +vi.mock('../../../config', () => ({ + getConfig: () => mocks.config, +})) + +vi.mock('@internal/database', () => ({ + LifecycleTenantStorePg: class { + claimDueTenantsWithinLeaseLimit = mocks.central.claimDueTenantsWithinLeaseLimit + releaseTenantDispatch = mocks.central.releaseTenantDispatch + recordExpiredTenantDispatch = mocks.central.recordExpiredTenantDispatch + completeTenantDispatch = mocks.central.completeTenantDispatch + failTenantDispatch = mocks.central.failTenantDispatch + hasObjectVersioningFeature = mocks.central.hasObjectVersioningFeature + }, + multitenantPgExecutor: { beginTransaction: mocks.beginTransaction }, + getPostgresConnection: vi.fn(), + getServiceKeyUser: vi.fn(), + getTenantConfig: vi.fn(), + tenantHasFeature: mocks.tenantHasFeature, +})) + +vi.mock('@internal/monitoring', () => ({ + logger: { + error: vi.fn(), + info: vi.fn(), + warn: vi.fn(), + }, + logSchema: { + error: vi.fn(), + info: vi.fn(), + warning: vi.fn(), + }, +})) + +vi.mock('../../lifecycle/abandoned-artifacts', () => ({ + retryAbandonedLifecycleArtifacts: mocks.retryAbandoned, +})) + +vi.mock('../../lifecycle/executor', () => ({ + LIFECYCLE_EVALUATION_INTERVAL_MS: 24 * 60 * 60 * 1000, + evaluationJitterMs: () => 1_234, + lifecycleErrorMessage: (error: unknown) => + (error instanceof Error ? error.message : String(error)).slice(0, 1000), + NoncurrentLifecycleExecutor: class { + run = mocks.executorRun + }, +})) + +vi.mock('../base-event', async () => { + const { Event } = await vi.importActual('@internal/queue') + + return { + BaseEvent: class> extends Event { + protected static createStorage = vi.fn() + }, + } +}) + +import { LifecycleBucketRun } from './bucket-run' +import { dispatchDueLifecycleTenants, startLifecycleScheduler } from './scheduler' +import { LifecycleTenantRun } from './tenant-run' + +const tenant = { ref: 'tenant-a', host: '' } +const claimId = '5bd8d514-9db1-4b1d-98a4-4a56fc569d95' +const claimUntil = new Date('2026-01-01T00:00:00.000Z') +const bucketId = 'bucket-a' +const dispatchState = (nextDispatchAt: Date | null = null) => ({ + nextDispatchAt, + maxFailureCount: 0, + oldestInFlightStartedAt: null, + hasAbandonedArtifacts: false, +}) +const evaluationIntervalMs = 24 * 60 * 60 * 1000 +const revisitMs = evaluationIntervalMs + 1_234 +const schedulers: AsyncAbortController[] = [] + +function startScheduler(controller = new AsyncAbortController()) { + vi.useFakeTimers() + schedulers.push(controller) + startLifecycleScheduler(controller.signal) + return controller +} + +function job(data: T, retryCount = 0, startedOn = new Date()) { + return { id: 'job-a', data, retryCount, retryLimit: 10, startedOn } as never +} + +function mockCreateStorage(target: unknown, storage: unknown) { + return vi + .spyOn(target as { createStorage(payload: BasePayload): Promise }, 'createStorage') + .mockResolvedValue(storage) +} + +function mockTenantDb(buckets: string[] = [], state: object = dispatchState()) { + const database = { + hasMigration: vi.fn().mockResolvedValue(true), + listDueLifecycleBuckets: vi.fn().mockResolvedValue(buckets), + findLifecycleDispatchState: vi.fn().mockResolvedValue(state), + } + const destroyConnection = vi.fn() + const storage = { db: { asSuperUser: () => database, destroyConnection } } + mockCreateStorage(LifecycleTenantRun, storage) + return { database, destroyConnection, storage } +} + +function mockClaim(reclaimed?: boolean) { + mocks.central.claimDueTenantsWithinLeaseLimit.mockResolvedValue([ + { tenantId: tenant.ref, claimId, claimUntil, reclaimed }, + ]) +} + +function completedNextDispatchAt(call = 0) { + return Date.parse(mocks.central.completeTenantDispatch.mock.calls[call][2]) +} + +describe('lifecycle scheduling and events', () => { + beforeEach(() => { + mocks.config.isMultitenant = true + mocks.config.pgQueueEnable = true + mocks.config.pgQueueEnableWorkers = true + mocks.config.storageVersioningEnabled = false + mocks.config.storageLifecyclePaused = false + mocks.central.claimDueTenantsWithinLeaseLimit.mockResolvedValue([]) + mocks.central.completeTenantDispatch.mockResolvedValue(true) + mocks.central.releaseTenantDispatch.mockResolvedValue(true) + mocks.central.recordExpiredTenantDispatch.mockResolvedValue(true) + mocks.central.failTenantDispatch.mockResolvedValue(true) + mocks.central.hasObjectVersioningFeature.mockResolvedValue(false) + mocks.executorRun.mockResolvedValue({ status: 'COMPLETED' }) + mocks.tenantHasFeature.mockResolvedValue(true) + mocks.getQueue.mockResolvedValue({ name: 'lifecycle:tenant:run-v1' }) + mocks.transaction = { query: vi.fn(), commit: vi.fn(), rollback: vi.fn() } + mocks.beginTransaction.mockResolvedValue(mocks.transaction) + vi.spyOn(Queue, 'getInstance').mockReturnValue({ getQueue: mocks.getQueue } as never) + vi.spyOn(Math, 'random').mockReturnValue(0.5) + }) + + afterEach(async () => { + await Promise.all(schedulers.splice(0).map((controller) => controller.abortAsync())) + vi.useRealTimers() + }) + + test('claims central tenants and sends only the central dispatch fence', async () => { + mockClaim() + const send = vi.spyOn(LifecycleTenantRun, 'send').mockResolvedValue('tenant-job') + + await dispatchDueLifecycleTenants(new AbortController().signal) + + expect(mocks.central.claimDueTenantsWithinLeaseLimit).toHaveBeenCalledWith( + 100, + 180_000, + expect.any(AbortSignal) + ) + expect(send).toHaveBeenCalledWith({ tenant, centralClaim: { claimId } }) + expect(mocks.central.recordExpiredTenantDispatch).not.toHaveBeenCalled() + expect(mocks.central.releaseTenantDispatch).not.toHaveBeenCalled() + expect(logSchema.warning).not.toHaveBeenCalled() + }) + + test('single-tenant deployments dispatch the configured tenant without a central claim', async () => { + mocks.config.isMultitenant = false + const send = vi.spyOn(LifecycleTenantRun, 'send').mockResolvedValue('tenant-job') + + await dispatchDueLifecycleTenants(new AbortController().signal) + + expect(send).toHaveBeenCalledWith({ tenant: { ref: 'single-tenant', host: '' } }) + expect(mocks.central.claimDueTenantsWithinLeaseLimit).not.toHaveBeenCalled() + }) + + test('releases a central claim with backoff when the tenant enqueue fails', async () => { + mockClaim(false) + vi.spyOn(LifecycleTenantRun, 'send').mockRejectedValue(new Error('queue unavailable')) + + await dispatchDueLifecycleTenants(new AbortController().signal) + + expect(mocks.central.failTenantDispatch).toHaveBeenCalledWith( + 'tenant-a', + claimId, + { code: 'tenant_dispatch_enqueue_failed', message: 'queue unavailable' }, + claimUntil.toISOString() + ) + }) + + test('keeps a reclaimed claim when re-sending fails', async () => { + const error = new Error('queue unavailable') + mockClaim(true) + vi.spyOn(LifecycleTenantRun, 'send').mockRejectedValue(error) + + await dispatchDueLifecycleTenants(new AbortController().signal) + + expect(logSchema.error).toHaveBeenCalledWith( + logger, + '[Lifecycle] Failed to enqueue tenant dispatcher', + { type: 'event', project: 'tenant-a', error } + ) + expect(mocks.central.failTenantDispatch).not.toHaveBeenCalled() + expect(mocks.central.releaseTenantDispatch).not.toHaveBeenCalled() + }) + + test('releases a fresh claim for the next poll when exactly-once drops its job', async () => { + mockClaim(false) + vi.spyOn(LifecycleTenantRun, 'send').mockResolvedValue(null) + vi.useFakeTimers({ toFake: ['Date'] }) + + await dispatchDueLifecycleTenants(new AbortController().signal) + + expect(mocks.central.releaseTenantDispatch).toHaveBeenCalledExactlyOnceWith( + 'tenant-a', + claimId, + new Date(Date.now() + 60_000).toISOString(), + claimUntil.toISOString() + ) + expect(mocks.central.recordExpiredTenantDispatch).not.toHaveBeenCalled() + expect(mocks.central.failTenantDispatch).not.toHaveBeenCalled() + expect(logSchema.error).not.toHaveBeenCalled() + expect(logSchema.warning).not.toHaveBeenCalled() + }) + + test('leaves a reclaimed claim to its queued job when exactly-once drops the re-send', async () => { + mockClaim(true) + vi.spyOn(LifecycleTenantRun, 'send').mockResolvedValue(null) + vi.useFakeTimers({ toFake: ['Date'] }) + + await dispatchDueLifecycleTenants(new AbortController().signal) + + expect(mocks.central.releaseTenantDispatch).not.toHaveBeenCalled() + expect(mocks.central.recordExpiredTenantDispatch).not.toHaveBeenCalled() + expect(mocks.central.failTenantDispatch).not.toHaveBeenCalled() + expect(logSchema.error).not.toHaveBeenCalled() + expect(logSchema.warning).not.toHaveBeenCalled() + }) + + test("records a lease expiry only when a reclaimed claim's re-send creates a new job", async () => { + mockClaim(true) + vi.spyOn(LifecycleTenantRun, 'send').mockResolvedValue('tenant-job') + + await dispatchDueLifecycleTenants(new AbortController().signal) + + expect(mocks.central.recordExpiredTenantDispatch).toHaveBeenCalledExactlyOnceWith( + 'tenant-a', + claimId + ) + expect(mocks.central.releaseTenantDispatch).not.toHaveBeenCalled() + expect(mocks.central.failTenantDispatch).not.toHaveBeenCalled() + expect(logSchema.error).not.toHaveBeenCalled() + expect(logSchema.warning).toHaveBeenCalledExactlyOnceWith( + logger, + '[Lifecycle] Tenant dispatch re-enqueued after claim expiry', + { type: 'event', project: 'tenant-a' } + ) + }) + + test.each([ + { + name: 'fresh-claim release', + reclaimed: false, + jobId: null, + write: 'releaseTenantDispatch', + }, + { + name: 'reclaimed-claim expiry record', + reclaimed: true, + jobId: 'tenant-job', + write: 'recordExpiredTenantDispatch', + }, + ] as const)( + 'logs a failed $name without rejecting the poll', + async ({ reclaimed, jobId, write }) => { + const writeError = new Error('central database unavailable') + mockClaim(reclaimed) + mocks.central[write].mockRejectedValue(writeError) + vi.spyOn(LifecycleTenantRun, 'send').mockResolvedValue(jobId) + + await expect( + dispatchDueLifecycleTenants(new AbortController().signal) + ).resolves.toBeUndefined() + + expect(logSchema.error).toHaveBeenCalledWith( + logger, + '[Lifecycle] Failed to update tenant dispatch claim', + { type: 'event', project: 'tenant-a', error: writeError } + ) + } + ) + + test.each([ + { lifecycle: true, queue: false, workers: true, warns: true }, + { lifecycle: true, queue: false, workers: false, warns: true }, + { lifecycle: true, queue: true, workers: false, warns: false }, + { lifecycle: false, queue: false, workers: true, warns: false }, + ])( + 'does not start a timer and warns=$warns when %o', + async ({ lifecycle, queue, workers, warns }) => { + mocks.config.storageVersioningEnabled = lifecycle + mocks.config.pgQueueEnable = queue + mocks.config.pgQueueEnableWorkers = workers + const send = vi.spyOn(LifecycleTenantRun, 'send') + startScheduler() + await vi.advanceTimersByTimeAsync(180_000) + + if (warns) { + expect(logSchema.warning).toHaveBeenCalledExactlyOnceWith( + logger, + expect.stringContaining('PG_QUEUE_ENABLE'), + { type: 'event' } + ) + } else { + expect(logSchema.warning).not.toHaveBeenCalled() + } + expect(mocks.central.claimDueTenantsWithinLeaseLimit).not.toHaveBeenCalled() + expect(send).not.toHaveBeenCalled() + expect(vi.getTimerCount()).toBe(0) + } + ) + + test('does not claim tenants or start a timer when lifecycle is paused', async () => { + mocks.config.storageLifecyclePaused = true + const send = vi.spyOn(LifecycleTenantRun, 'send') + startScheduler() + await vi.advanceTimersByTimeAsync(180_000) + + expect(logSchema.warning).toHaveBeenCalledExactlyOnceWith( + logger, + expect.stringContaining('STORAGE_LIFECYCLE_PAUSED'), + { type: 'event' } + ) + expect(mocks.central.claimDueTenantsWithinLeaseLimit).not.toHaveBeenCalled() + expect(send).not.toHaveBeenCalled() + expect(vi.getTimerCount()).toBe(0) + }) + + test('does not dispatch or start a timer with an already aborted signal', async () => { + const controller = new AsyncAbortController() + controller.abort() + startScheduler(controller) + await dispatchDueLifecycleTenants(controller.signal) + + expect(mocks.central.claimDueTenantsWithinLeaseLimit).not.toHaveBeenCalled() + expect(vi.getTimerCount()).toBe(0) + }) + + test('polls immediately, then about every minute with jitter, and stops on abort', async () => { + vi.mocked(Math.random).mockReturnValueOnce(0).mockReturnValueOnce(0.9) + const controller = startScheduler() + await vi.advanceTimersByTimeAsync(0) + expect(mocks.central.claimDueTenantsWithinLeaseLimit).toHaveBeenCalledTimes(1) + + await vi.advanceTimersByTimeAsync(54_999) + expect(mocks.central.claimDueTenantsWithinLeaseLimit).toHaveBeenCalledTimes(1) + await vi.advanceTimersByTimeAsync(1) + expect(mocks.central.claimDueTenantsWithinLeaseLimit).toHaveBeenCalledTimes(2) + await vi.advanceTimersByTimeAsync(64_000) + expect(mocks.central.claimDueTenantsWithinLeaseLimit).toHaveBeenCalledTimes(3) + + await controller.abortAsync() + expect(vi.getTimerCount()).toBe(0) + await vi.advanceTimersByTimeAsync(180_000) + expect(mocks.central.claimDueTenantsWithinLeaseLimit).toHaveBeenCalledTimes(3) + }) + + test('retries a failed poll on the next timer', async () => { + const error = new Error('central database unavailable') + mocks.central.claimDueTenantsWithinLeaseLimit.mockRejectedValueOnce(error) + startScheduler() + await vi.advanceTimersByTimeAsync(0) + + expect(logSchema.error).toHaveBeenCalledWith(logger, '[Lifecycle] Scheduler poll failed', { + type: 'event', + error, + }) + await vi.advanceTimersByTimeAsync(60_000) + expect(mocks.central.claimDueTenantsWithinLeaseLimit).toHaveBeenCalledTimes(2) + expect(vi.getTimerCount()).toBe(1) + }) + + test('never overlaps a slow poll', async () => { + const pendingClaim = Promise.withResolvers<[]>() + mocks.central.claimDueTenantsWithinLeaseLimit.mockReturnValueOnce(pendingClaim.promise) + startScheduler() + try { + await vi.advanceTimersByTimeAsync(180_000) + expect(mocks.central.claimDueTenantsWithinLeaseLimit).toHaveBeenCalledTimes(1) + expect(vi.getTimerCount()).toBe(0) + } finally { + pendingClaim.resolve([]) + } + await vi.advanceTimersByTimeAsync(60_000) + expect(mocks.central.claimDueTenantsWithinLeaseLimit).toHaveBeenCalledTimes(2) + }) + + test('drains all claimed sends before queue shutdown, even if releasing a failed claim fails', async () => { + const recoveryError = new Error('central database unavailable') + const pendingSend = Promise.withResolvers() + mocks.central.claimDueTenantsWithinLeaseLimit.mockResolvedValue([ + { tenantId: 'tenant-a', claimId, claimUntil }, + { tenantId: 'tenant-b', claimId, claimUntil }, + ]) + mocks.central.failTenantDispatch.mockRejectedValueOnce(recoveryError) + const send = vi + .spyOn(LifecycleTenantRun, 'send') + .mockRejectedValueOnce(new Error('enqueue failed')) + .mockReturnValueOnce(pendingSend.promise) + const controller = startScheduler() + const stopQueue = vi.fn() + controller.nextGroup.signal.addEventListener('abort', stopQueue) + + try { + await vi.advanceTimersByTimeAsync(0) + expect(send).toHaveBeenCalledTimes(2) + const shutdown = controller.abortAsync() + await vi.advanceTimersByTimeAsync(0) + expect(stopQueue).not.toHaveBeenCalled() + pendingSend.resolve('tenant-job') + await shutdown + + expect(stopQueue).toHaveBeenCalledOnce() + expect(logSchema.error).toHaveBeenCalledWith( + logger, + '[Lifecycle] Failed to update tenant dispatch claim', + { type: 'event', project: 'tenant-a', error: recoveryError } + ) + expect(vi.getTimerCount()).toBe(0) + } finally { + pendingSend.resolve('tenant-job') + } + }) + + test('claims within the lease limit in a committed central transaction', async () => { + mockClaim() + const send = vi.spyOn(LifecycleTenantRun, 'send').mockResolvedValue('tenant-job') + + await dispatchDueLifecycleTenants(new AbortController().signal) + + expect(mocks.beginTransaction).toHaveBeenCalledExactlyOnceWith({ timeout: 10_000 }) + expect(mocks.transaction.query).toHaveBeenCalledExactlyOnceWith( + "SET LOCAL idle_in_transaction_session_timeout = '30000ms'", + { signal: expect.any(AbortSignal) } + ) + expect(mocks.central.claimDueTenantsWithinLeaseLimit).toHaveBeenCalledExactlyOnceWith( + 100, + 180_000, + expect.any(AbortSignal) + ) + expect(mocks.transaction.commit).toHaveBeenCalled() + expect(mocks.transaction.query.mock.invocationCallOrder[0]).toBeLessThan( + mocks.central.claimDueTenantsWithinLeaseLimit.mock.invocationCallOrder[0] + ) + expect(mocks.transaction.commit.mock.invocationCallOrder[0]).toBeLessThan( + send.mock.invocationCallOrder[0] + ) + }) + + test('rolls back when the idle transaction timeout cannot be set', async () => { + const error = new Error('timeout setup failed') + mocks.transaction.query.mockRejectedValueOnce(error) + + await expect(dispatchDueLifecycleTenants(new AbortController().signal)).rejects.toBe(error) + + expect(mocks.transaction.rollback).toHaveBeenCalledOnce() + expect(mocks.transaction.commit).not.toHaveBeenCalled() + expect(mocks.central.claimDueTenantsWithinLeaseLimit).not.toHaveBeenCalled() + }) + + test('rolls back the claim transaction when claiming fails', async () => { + const error = new Error('claim failed') + mocks.central.claimDueTenantsWithinLeaseLimit.mockRejectedValue(error) + const send = vi.spyOn(LifecycleTenantRun, 'send') + + await expect(dispatchDueLifecycleTenants(new AbortController().signal)).rejects.toBe(error) + + expect(mocks.transaction.rollback).toHaveBeenCalled() + expect(mocks.transaction.commit).not.toHaveBeenCalled() + expect(send).not.toHaveBeenCalled() + }) + + test('rolls back claims if shutdown arrives before commit', async () => { + const controller = new AbortController() + mocks.central.claimDueTenantsWithinLeaseLimit.mockImplementationOnce(async () => { + controller.abort() + return [{ tenantId: tenant.ref, claimId, claimUntil }] + }) + const send = vi.spyOn(LifecycleTenantRun, 'send') + + await dispatchDueLifecycleTenants(controller.signal) + + expect(mocks.transaction.rollback).toHaveBeenCalledOnce() + expect(mocks.transaction.commit).not.toHaveBeenCalled() + expect(send).not.toHaveBeenCalled() + }) + + test('cancels a blocked claim before stopping the queue', async () => { + mocks.central.claimDueTenantsWithinLeaseLimit.mockImplementation( + (_limit: number, _leaseMs: number, signal: AbortSignal) => + new Promise((_resolve, reject) => { + signal.addEventListener('abort', () => reject(new Error('claim aborted')), { once: true }) + }) + ) + const controller = startScheduler() + const stopQueue = vi.fn() + controller.nextGroup.signal.addEventListener('abort', stopQueue) + + await vi.advanceTimersByTimeAsync(0) + expect(mocks.central.claimDueTenantsWithinLeaseLimit).toHaveBeenCalledOnce() + await controller.abortAsync() + + expect(mocks.transaction.rollback).toHaveBeenCalledOnce() + expect(mocks.transaction.commit).not.toHaveBeenCalled() + expect(stopQueue).toHaveBeenCalledOnce() + expect(logSchema.error).not.toHaveBeenCalled() + }) + + test.each([true, false])( + 'fails the poll without claiming when the tenant queue is missing (multitenant=%s)', + async (multitenant) => { + mocks.config.isMultitenant = multitenant + mocks.getQueue.mockResolvedValue(null) + const send = vi.spyOn(LifecycleTenantRun, 'send') + + await expect(dispatchDueLifecycleTenants(new AbortController().signal)).rejects.toThrow( + /does not exist/ + ) + + expect(mocks.getQueue).toHaveBeenCalledWith('lifecycle:tenant:run-v1') + expect(mocks.beginTransaction).not.toHaveBeenCalled() + expect(mocks.central.claimDueTenantsWithinLeaseLimit).not.toHaveBeenCalled() + expect(send).not.toHaveBeenCalled() + } + ) + + test('tenant dispatcher sends bounded bucket ids', async () => { + const observation = { + ...dispatchState(new Date('2026-08-16T00:00:00.000Z')), + maxFailureCount: 7, + oldestInFlightStartedAt: '2026-08-15T00:00:00.000Z', + } + const { database, destroyConnection } = mockTenantDb([bucketId], observation) + const send = vi.spyOn(LifecycleBucketRun, 'send').mockResolvedValue('bucket-job') + + await LifecycleTenantRun.handle(job({ tenant, centralClaim: { claimId } })) + + expect(database.listDueLifecycleBuckets).toHaveBeenCalledWith(8) + expect(database.findLifecycleDispatchState).toHaveBeenCalledWith({ runBudgetMs: 20_000 }) + expect(send).toHaveBeenCalledWith({ tenant, bucketId }) + expect(mocks.central.completeTenantDispatch).toHaveBeenCalledWith( + 'tenant-a', + claimId, + expect.any(String), + expect.any(String), + observation + ) + expect(destroyConnection).toHaveBeenCalledOnce() + }) + + test.each([true, false])( + 'reports a missing bucket queue when a send is dropped (multitenant=%s)', + async (multitenant) => { + mocks.config.isMultitenant = multitenant + mocks.config.storageVersioningEnabled = !multitenant + mocks.getQueue.mockResolvedValue(null) + const { destroyConnection } = mockTenantDb([bucketId, 'bucket-b']) + vi.spyOn(LifecycleBucketRun, 'send') + .mockResolvedValueOnce('bucket-job') + .mockResolvedValueOnce(null) + const payload = multitenant ? { tenant, centralClaim: { claimId } } : { tenant } + + if (multitenant) { + await LifecycleTenantRun.handle(job(payload)) + expect(mocks.central.failTenantDispatch).toHaveBeenCalledWith( + 'tenant-a', + claimId, + { + code: 'tenant_dispatch_failed', + message: 'Lifecycle queue lifecycle:bucket:run-v1 does not exist', + }, + expect.any(String) + ) + expect(mocks.central.completeTenantDispatch).not.toHaveBeenCalled() + } else { + await expect(LifecycleTenantRun.handle(job(payload))).rejects.toThrow( + 'Lifecycle queue lifecycle:bucket:run-v1 does not exist' + ) + } + expect(mocks.getQueue).toHaveBeenCalledWith('lifecycle:bucket:run-v1') + expect(destroyConnection).toHaveBeenCalledOnce() + } + ) + + test('treats a dropped bucket send as a duplicate when its queue exists', async () => { + mockTenantDb([bucketId]) + vi.spyOn(LifecycleBucketRun, 'send').mockResolvedValue(null) + + await LifecycleTenantRun.handle(job({ tenant, centralClaim: { claimId } })) + + expect(mocks.getQueue).toHaveBeenCalledWith('lifecycle:bucket:run-v1') + expect(mocks.central.completeTenantDispatch).toHaveBeenCalledOnce() + expect(mocks.central.failTenantDispatch).not.toHaveBeenCalled() + }) + + test('sends every due bucket and caps a distant next dispatch at the evaluation interval', async () => { + vi.useFakeTimers({ toFake: ['Date'] }) + const second = 'bucket-b' + mockTenantDb([bucketId, second], dispatchState(new Date(Date.now() + 3 * evaluationIntervalMs))) + const send = vi.spyOn(LifecycleBucketRun, 'send').mockResolvedValue('bucket-job') + + await LifecycleTenantRun.handle(job({ tenant, centralClaim: { claimId } })) + + expect(send).toHaveBeenCalledTimes(2) + expect(send).toHaveBeenCalledWith({ tenant, bucketId }) + expect(send).toHaveBeenCalledWith({ tenant, bucketId: second }) + expect(completedNextDispatchAt()).toBe(Date.now() + revisitMs) + }) + + test.each([ + { due: 3, floored: true }, + { due: 8, floored: false }, + ])('floors the redispatch after $due sent buckets: $floored', async ({ due, floored }) => { + vi.useFakeTimers({ toFake: ['Date'] }) + const past = new Date(Date.now() - 1000) + mockTenantDb( + Array.from({ length: due }, (_, index) => `bucket-${index}`), + dispatchState(past) + ) + vi.spyOn(LifecycleBucketRun, 'send').mockResolvedValue('bucket-job') + + await LifecycleTenantRun.handle(job({ tenant, centralClaim: { claimId } })) + + expect(completedNextDispatchAt()).toBe( + floored ? Date.now() + mocks.config.storageLifecycleJobBudgetMs : past.getTime() + ) + }) + + test('does not floor a near-term revisit when no buckets were due', async () => { + vi.useFakeTimers({ toFake: ['Date'] }) + const next = new Date(Date.now() + 1_000) + mockTenantDb([], dispatchState(next)) + const send = vi.spyOn(LifecycleBucketRun, 'send') + + await LifecycleTenantRun.handle(job({ tenant, centralClaim: { claimId } })) + + expect(send).not.toHaveBeenCalled() + expect(completedNextDispatchAt()).toBe(next.getTime()) + }) + + test('fences central completion with the job deadline', async () => { + mockTenantDb() + mocks.tenantHasFeature.mockResolvedValue(false) + const startedOn = new Date(Date.now() - 180_000) + const deadline = new Date(startedOn.getTime() + 180_000).toISOString() + + await LifecycleTenantRun.handle(job({ tenant, centralClaim: { claimId } }, 0, startedOn)) + + expect(mocks.central.completeTenantDispatch.mock.calls[0][3]).toBe(deadline) + }) + + test('fails the central dispatch with tenant_dispatch_failed fenced by the job deadline', async () => { + const failure = new Error('tenant offline') + mockCreateStorage(LifecycleTenantRun, {}).mockRejectedValue(failure) + const startedOn = new Date(Date.now() - 180_000) + + await LifecycleTenantRun.handle(job({ tenant, centralClaim: { claimId } }, 0, startedOn)) + + expect(logSchema.error).toHaveBeenCalledWith(logger, '[Lifecycle] Tenant dispatcher failed', { + type: 'event', + project: 'tenant-a', + error: failure, + }) + expect(mocks.central.failTenantDispatch).toHaveBeenCalledWith( + 'tenant-a', + claimId, + { code: 'tenant_dispatch_failed', message: 'tenant offline' }, + new Date(startedOn.getTime() + 180_000).toISOString() + ) + expect(mocks.central.completeTenantDispatch).not.toHaveBeenCalled() + }) + + test('logs the original error before failTenantDispatch can throw', async () => { + const failure = new Error('tenant offline') + mockCreateStorage(LifecycleTenantRun, {}).mockRejectedValue(failure) + mocks.central.failTenantDispatch.mockRejectedValueOnce(new Error('central down')) + + await expect( + LifecycleTenantRun.handle(job({ tenant, centralClaim: { claimId } }, 0)) + ).rejects.toThrow('central down') + + expect(logSchema.error).toHaveBeenCalledWith(logger, '[Lifecycle] Tenant dispatcher failed', { + type: 'event', + project: 'tenant-a', + error: failure, + }) + }) + + test.each([true, false])( + 'warns when completing the central dispatch finds no claim (enabled=%s)', + async (enabled) => { + mocks.tenantHasFeature.mockResolvedValueOnce(enabled) + mockTenantDb() + mocks.central.completeTenantDispatch.mockResolvedValue(false) + + await LifecycleTenantRun.handle(job({ tenant, centralClaim: { claimId } })) + + expect(logSchema.warning).toHaveBeenCalledWith( + logger, + '[Lifecycle] Tenant dispatch lost its claim', + { type: 'event', project: 'tenant-a' } + ) + } + ) + + test.each([true, false])( + 'bounds the revisit for a tenant (versioning=%s)', + async (versioning) => { + mocks.tenantHasFeature.mockResolvedValueOnce(versioning) + const { database, destroyConnection } = mockTenantDb([bucketId]) + const send = vi.spyOn(LifecycleBucketRun, 'send').mockResolvedValue('bucket-job') + vi.useFakeTimers({ toFake: ['Date'] }) + + await LifecycleTenantRun.handle(job({ tenant, centralClaim: { claimId } })) + + expect(mocks.tenantHasFeature).toHaveBeenCalledWith('tenant-a', 'objectVersioning') + expect(database.findLifecycleDispatchState).toHaveBeenCalledWith({ runBudgetMs: 20_000 }) + if (versioning) { + expect(database.listDueLifecycleBuckets).toHaveBeenCalledWith(8) + expect(send).toHaveBeenCalledWith({ tenant, bucketId }) + } else { + expect(database.listDueLifecycleBuckets).not.toHaveBeenCalled() + expect(send).not.toHaveBeenCalled() + } + expect(mocks.central.failTenantDispatch).not.toHaveBeenCalled() + expect(completedNextDispatchAt()).toBe(Date.now() + (versioning ? revisitMs : 60 * 60_000)) + expect(destroyConnection).toHaveBeenCalledOnce() + } + ) + + test('retries abandoned artifacts at most hourly', async () => { + const { database, storage } = mockTenantDb() + const signal = new AbortController().signal + vi.useFakeTimers({ toFake: ['Date'] }) + + await LifecycleTenantRun.handle(job({ tenant, centralClaim: { claimId } }), { signal }) + + expect(mocks.retryAbandoned).toHaveBeenCalledExactlyOnceWith({ + storage, + database, + attemptedBefore: new Date(Date.now() - 60 * 60_000).toISOString(), + budgetMs: 60_000, + signal, + }) + }) + + test.each([ + { label: 'no bucket is due', next: null, expected: 60 * 60_000 }, + { label: 'a bucket is due sooner', next: 10 * 60_000, expected: 10 * 60_000 }, + ])( + 'retries abandoned artifacts of an enabled tenant within the hour when $label', + async ({ next, expected }) => { + vi.useFakeTimers({ toFake: ['Date'] }) + mockTenantDb([], { + ...dispatchState(next === null ? null : new Date(Date.now() + next)), + hasAbandonedArtifacts: true, + }) + + await LifecycleTenantRun.handle(job({ tenant, centralClaim: { claimId } })) + + expect(completedNextDispatchAt()).toBe(Date.now() + expected) + } + ) + + test.each([true, false])( + 'retries soon when the feature cache is stale (artifacts=%s)', + async (hasAbandonedArtifacts) => { + mocks.tenantHasFeature.mockResolvedValueOnce(false) + mocks.central.hasObjectVersioningFeature.mockResolvedValueOnce(true) + mockTenantDb([], { ...dispatchState(), hasAbandonedArtifacts }) + vi.useFakeTimers({ toFake: ['Date'] }) + + await LifecycleTenantRun.handle(job({ tenant, centralClaim: { claimId } })) + + expect(mocks.central.hasObjectVersioningFeature).toHaveBeenCalledExactlyOnceWith('tenant-a') + expect(completedNextDispatchAt()).toBe(Date.now() + 5 * 60_000) + } + ) + + test('revisits a disabled tenant when outbox work appears after its empty observation', async () => { + vi.useFakeTimers({ toFake: ['Date'] }) + mocks.tenantHasFeature.mockResolvedValue(false) + const { database } = mockTenantDb() + + await LifecycleTenantRun.handle(job({ tenant, centralClaim: { claimId } })) + + const nextDispatchAt = completedNextDispatchAt() + expect(nextDispatchAt).toBe(Date.now() + 60 * 60_000) + + // A previously started bucket worker commits outbox rows after the first observation. + database.findLifecycleDispatchState.mockResolvedValue({ + ...dispatchState(), + hasAbandonedArtifacts: true, + }) + vi.setSystemTime(nextDispatchAt) + await LifecycleTenantRun.handle(job({ tenant, centralClaim: { claimId } })) + + expect(mocks.retryAbandoned).toHaveBeenCalledTimes(2) + expect(mocks.central.completeTenantDispatch).toHaveBeenCalledTimes(2) + expect(completedNextDispatchAt(1)).toBe(Date.now() + 60 * 60_000) + }) + + test.each([ + { multitenant: true, enabled: true, interval: 60_000 }, + { multitenant: true, enabled: false, interval: 60_000 }, + { multitenant: false, enabled: true, interval: 60_000 }, + { multitenant: false, enabled: false, interval: 600_000 }, + ])('polls every $interval ms when %o', async (c) => { + mocks.config.isMultitenant = c.multitenant + mocks.config.storageVersioningEnabled = c.enabled + const send = vi.spyOn(LifecycleTenantRun, 'send').mockResolvedValue('tenant-job') + const polls = () => + c.multitenant + ? mocks.central.claimDueTenantsWithinLeaseLimit.mock.calls.length + : send.mock.calls.length + startScheduler() + await vi.advanceTimersByTimeAsync(0) + + expect(polls()).toBe(1) + await vi.advanceTimersByTimeAsync(c.interval - 5_001) + expect(polls()).toBe(1) + await vi.advanceTimersByTimeAsync(10_001) + expect(polls()).toBe(2) + }) + + test('defers enrolled tenants until promotion fencing is installed', async () => { + const database = { + hasMigration: vi.fn().mockResolvedValue(false), + listDueLifecycleBuckets: vi.fn(), + } + const destroyConnection = vi.fn() + mockCreateStorage(LifecycleTenantRun, { + db: { asSuperUser: () => database, destroyConnection }, + }) + const send = vi.spyOn(LifecycleBucketRun, 'send').mockResolvedValue('bucket-job') + vi.useFakeTimers({ toFake: ['Date'] }) + + await LifecycleTenantRun.handle(job({ tenant, centralClaim: { claimId } })) + + expect(database.hasMigration).toHaveBeenCalledWith('lifecycle-writer-coordination') + expect(database.listDueLifecycleBuckets).not.toHaveBeenCalled() + expect(send).not.toHaveBeenCalled() + expect(mocks.central.failTenantDispatch).not.toHaveBeenCalled() + expect(completedNextDispatchAt()).toBe(Date.now() + revisitMs) + expect(destroyConnection).toHaveBeenCalledOnce() + }) + + test('queue options pin identity, expiry, retries, batch size and unconditional sending', async () => { + expect(LifecycleBucketRun.getSendOptions({ tenant, bucketId })).toEqual({ + singletonKey: 'tenant-a/bucket-a', + expireInMinutes: Math.ceil( + (mocks.config.storageLifecycleJobBudgetMs + mocks.config.storageLifecycleClaimLeaseMs) / + 60_000 + ), + retryLimit: 0, + }) + expect(LifecycleBucketRun.getSendOptions({ tenant, bucketId }).expireInMinutes).toBe(3) + expect(LifecycleTenantRun.getSendOptions({ tenant })).toEqual({ + singletonKey: 'tenant-a', + expireInMinutes: 3, + retryLimit: 0, + }) + + for (const event of [LifecycleTenantRun, LifecycleBucketRun]) { + expect(event.getQueueOptions()).toMatchObject({ policy: 'exactly_once' }) + expect(event.allowSync).toBe(false) + expect(event.getWorkerOptions().pollingIntervalSeconds).toBeUndefined() + } + + const { concurrentTaskCount, batchSize } = LifecycleBucketRun.getWorkerOptions() + expect(batchSize).toBe(concurrentTaskCount) + const tenantOptions = LifecycleTenantRun.getWorkerOptions() + expect(tenantOptions.batchSize).toBe(tenantOptions.concurrentTaskCount) + + vi.mocked(getTenantConfig).mockResolvedValue({ + disableEvents: [LifecycleTenantRun.eventName(), LifecycleBucketRun.eventName()], + } as Awaited>) + await expect(LifecycleTenantRun.shouldSend()).resolves.toBe(true) + await expect(LifecycleBucketRun.shouldSend()).resolves.toBe(true) + }) + + test.each([true, false])( + 'bucket delivery passes tenantLifecycleEnabled=%s from the objectVersioning feature', + async (enabled) => { + mocks.tenantHasFeature.mockResolvedValue(enabled) + const destroyConnection = vi.fn() + mockCreateStorage(LifecycleBucketRun, { db: { destroyConnection } }) + const duration = vi.spyOn(lifecycleRunDuration, 'record') + + await LifecycleBucketRun.handle(job({ tenant, bucketId })) + + expect(mocks.executorRun).toHaveBeenCalledWith(bucketId, { + tenantLifecycleEnabled: enabled, + }) + expect(mocks.tenantHasFeature).toHaveBeenCalledWith('tenant-a', 'objectVersioning') + expect(destroyConnection).toHaveBeenCalledOnce() + expect(duration).toHaveBeenCalledExactlyOnceWith(expect.any(Number), { status: 'COMPLETED' }) + } + ) + + test.each([true, false])( + 'global versioning enables bucket delivery (multitenant=%s)', + async (multitenant) => { + mocks.config.isMultitenant = multitenant + mocks.config.storageVersioningEnabled = true + mocks.tenantHasFeature.mockResolvedValue(false) + mockCreateStorage(LifecycleBucketRun, { db: { destroyConnection: vi.fn() } }) + + await LifecycleBucketRun.handle(job({ tenant, bucketId })) + + expect(mocks.executorRun).toHaveBeenCalledWith(bucketId, { tenantLifecycleEnabled: true }) + expect(mocks.tenantHasFeature).not.toHaveBeenCalled() + } + ) + + test('bucket delivery records the ineligible status', async () => { + mocks.executorRun.mockResolvedValue({ status: 'INELIGIBLE' }) + mockCreateStorage(LifecycleBucketRun, { db: { destroyConnection: vi.fn() } }) + const duration = vi.spyOn(lifecycleRunDuration, 'record') + + await LifecycleBucketRun.handle(job({ tenant, bucketId })) + + expect(duration).toHaveBeenCalledExactlyOnceWith(expect.any(Number), { status: 'INELIGIBLE' }) + }) + + test('bucket delivery records the run duration in seconds', async () => { + mockCreateStorage(LifecycleBucketRun, { db: { destroyConnection: vi.fn() } }) + vi.spyOn(performance, 'now').mockReturnValueOnce(1_000).mockReturnValueOnce(2_500) + const duration = vi.spyOn(lifecycleRunDuration, 'record') + + await LifecycleBucketRun.handle(job({ tenant, bucketId })) + + expect(duration).toHaveBeenCalledExactlyOnceWith(1.5, { status: 'COMPLETED' }) + }) + + test('bucket delivery forwards the abort signal to the executor', async () => { + mockCreateStorage(LifecycleBucketRun, { db: { destroyConnection: vi.fn() } }) + const { signal } = new AbortController() + + await LifecycleBucketRun.handle(job({ tenant, bucketId }), { signal }) + + expect(mocks.executorRun).toHaveBeenCalledWith(bucketId, { + tenantLifecycleEnabled: true, + signal, + }) + }) + + test('bucket delivery failure propagates, closes the connection and records an ERROR run', async () => { + const failure = new Error('executor failed') + mocks.executorRun.mockRejectedValue(failure) + const destroyConnection = vi.fn() + mockCreateStorage(LifecycleBucketRun, { db: { destroyConnection } }) + const duration = vi.spyOn(lifecycleRunDuration, 'record') + + await expect(LifecycleBucketRun.handle(job({ tenant, bucketId }))).rejects.toBe(failure) + + expect(destroyConnection).toHaveBeenCalledOnce() + expect(duration).toHaveBeenCalledExactlyOnceWith(expect.any(Number), { status: 'ERROR' }) + }) + + test('records an ERROR duration if bucket storage creation fails', async () => { + const failure = new Error('storage unavailable') + mockCreateStorage(LifecycleBucketRun, {}).mockRejectedValue(failure) + const duration = vi.spyOn(lifecycleRunDuration, 'record') + + await expect(LifecycleBucketRun.handle(job({ tenant, bucketId }))).rejects.toBe(failure) + expect(duration).toHaveBeenCalledExactlyOnceWith(expect.any(Number), { status: 'ERROR' }) + }) + + test.each([true, false])( + 'single-tenant dispatch (versioning=%s) sweeps abandoned artifacts without computing the unused dispatch state', + async (versioning) => { + mocks.config.isMultitenant = false + mocks.config.storageVersioningEnabled = versioning + const { database } = mockTenantDb([bucketId]) + const send = vi.spyOn(LifecycleBucketRun, 'send').mockResolvedValue('bucket-job') + + await LifecycleTenantRun.handle(job({ tenant })) + + expect(mocks.retryAbandoned).toHaveBeenCalledOnce() + if (versioning) { + expect(database.listDueLifecycleBuckets).toHaveBeenCalled() + expect(send).toHaveBeenCalledWith({ tenant, bucketId }) + } else { + expect(database.listDueLifecycleBuckets).not.toHaveBeenCalled() + expect(send).not.toHaveBeenCalled() + } + expect(database.findLifecycleDispatchState).not.toHaveBeenCalled() + expect(mocks.central.completeTenantDispatch).not.toHaveBeenCalled() + expect(mocks.tenantHasFeature).not.toHaveBeenCalled() + } + ) + + test('single-tenant tenant dispatcher failure is rethrown without a central failure', async () => { + mocks.config.isMultitenant = false + const failure = new Error('tenant offline') + mockCreateStorage(LifecycleTenantRun, {}).mockRejectedValue(failure) + + await expect(LifecycleTenantRun.handle(job({ tenant }))).rejects.toBe(failure) + expect(mocks.central.failTenantDispatch).not.toHaveBeenCalled() + }) + + test('a failed bucket enqueue fails the central dispatch instead of completing it', async () => { + mockTenantDb([bucketId]) + vi.spyOn(LifecycleBucketRun, 'send').mockRejectedValue(new Error('enqueue offline')) + + await LifecycleTenantRun.handle(job({ tenant, centralClaim: { claimId } })) + + expect(mocks.central.failTenantDispatch).toHaveBeenCalledWith( + 'tenant-a', + claimId, + { code: 'tenant_dispatch_failed', message: 'enqueue offline' }, + expect.any(String) + ) + expect(mocks.central.completeTenantDispatch).not.toHaveBeenCalled() + }) +}) diff --git a/src/storage/events/lifecycle-rules/scheduler.ts b/src/storage/events/lifecycle-rules/scheduler.ts new file mode 100644 index 000000000..c2bdab4be --- /dev/null +++ b/src/storage/events/lifecycle-rules/scheduler.ts @@ -0,0 +1,169 @@ +import { + type LifecycleTenantClaim, + LifecycleTenantStorePg, + multitenantPgExecutor, +} from '@internal/database' +import { logger, logSchema } from '@internal/monitoring' +import { Queue } from '@internal/queue' +import { getConfig } from '../../../config' +import { lifecycleErrorMessage } from '../../lifecycle/executor' +import { LIFECYCLE_TENANT_LEASE_MS, LifecycleTenantRun } from './tenant-run' + +const config = getConfig() +const pollIntervalMs = 60_000 +const idlePollIntervalMs = 600_000 +const pollJitterMs = 5_000 +const dispatchLimit = 100 +const claimTimeoutMs = 10_000 +const claimIdleTimeoutMs = 30_000 + +// The signal shutdown group must drain before the queue's group. +export function startLifecycleScheduler(signal: AbortSignal): void { + if (!config.pgQueueEnable || !config.pgQueueEnableWorkers) { + if (config.storageVersioningEnabled && !config.pgQueueEnable) { + logSchema.warning( + logger, + '[Lifecycle] Lifecycle is enabled but PG_QUEUE_ENABLE is false; no rules will be applied', + { type: 'event' } + ) + } + return + } + if (config.storageLifecyclePaused) { + logSchema.warning( + logger, + '[Lifecycle] Lifecycle is paused by STORAGE_LIFECYCLE_PAUSED; no rules will be applied', + { type: 'event' } + ) + return + } + + let timer: NodeJS.Timeout | undefined + let inFlight = Promise.resolve() + const idle = !config.isMultitenant && !config.storageVersioningEnabled + const intervalMs = idle ? idlePollIntervalMs : pollIntervalMs + + const poll = () => { + inFlight = dispatchDueLifecycleTenants(signal) + .catch((error) => { + if (!signal.aborted) { + logSchema.error(logger, '[Lifecycle] Scheduler poll failed', { type: 'event', error }) + } + }) + .finally(() => { + if (signal.aborted) return + const delay = intervalMs + (Math.random() * 2 - 1) * pollJitterMs + timer = setTimeout(poll, delay) + timer.unref() + }) + } + + signal.addEventListener( + 'abort', + () => { + clearTimeout(timer) + return inFlight + }, + { once: true } + ) + poll() +} + +export async function dispatchDueLifecycleTenants(signal: AbortSignal): Promise { + if (signal.aborted) return + if (!(await Queue.getInstance().getQueue(LifecycleTenantRun.getQueueName()))) { + throw new Error(`Lifecycle queue ${LifecycleTenantRun.getQueueName()} does not exist`) + } + if (!config.isMultitenant) { + await LifecycleTenantRun.send({ tenant: { ref: config.tenantId, host: '' } }) + return + } + + const claimSignal = AbortSignal.any([signal, AbortSignal.timeout(claimTimeoutMs)]) + const transaction = await multitenantPgExecutor.beginTransaction({ timeout: claimTimeoutMs }) + let claims: LifecycleTenantClaim[] + try { + await transaction.query( + `SET LOCAL idle_in_transaction_session_timeout = '${claimIdleTimeoutMs}ms'`, + { + signal: claimSignal, + } + ) + // Leases start at claim time and cap outstanding claims across all schedulers. + claims = await new LifecycleTenantStorePg(transaction).claimDueTenantsWithinLeaseLimit( + dispatchLimit, + LIFECYCLE_TENANT_LEASE_MS, + claimSignal + ) + if (signal.aborted) return + await transaction.commit() + } finally { + await transaction.rollback() + } + const store = new LifecycleTenantStorePg(multitenantPgExecutor) + + const settle = async (claim: LifecycleTenantClaim, write: Promise) => { + try { + await write + } catch (error) { + logSchema.error(logger, '[Lifecycle] Failed to update tenant dispatch claim', { + type: 'event', + project: claim.tenantId, + error, + }) + } + } + + // Finish every claimed dispatch even if persisting one enqueue failure fails. + await Promise.all( + claims.map(async (claim) => { + try { + const jobId = await LifecycleTenantRun.send({ + tenant: { ref: claim.tenantId, host: '' }, + centralClaim: { claimId: claim.claimId }, + }) + if (jobId === null) { + // A null send may be a duplicate, + // leave a reclaimed claim to its job and release a fresh one. + if (!claim.reclaimed) { + await settle( + claim, + store.releaseTenantDispatch( + claim.tenantId, + claim.claimId, + new Date(Date.now() + pollIntervalMs).toISOString(), + claim.claimUntil.toISOString() + ) + ) + } + } else if (claim.reclaimed) { + logSchema.warning(logger, '[Lifecycle] Tenant dispatch re-enqueued after claim expiry', { + type: 'event', + project: claim.tenantId, + }) + await settle(claim, store.recordExpiredTenantDispatch(claim.tenantId, claim.claimId)) + } + } catch (error) { + logSchema.error(logger, '[Lifecycle] Failed to enqueue tenant dispatcher', { + type: 'event', + project: claim.tenantId, + error, + }) + if (!claim.reclaimed) { + await settle( + claim, + store.failTenantDispatch( + claim.tenantId, + claim.claimId, + { + code: 'tenant_dispatch_enqueue_failed', + message: lifecycleErrorMessage(error), + }, + claim.claimUntil.toISOString() + ) + ) + } + } + }) + ) +} diff --git a/src/storage/events/lifecycle-rules/tenant-run.ts b/src/storage/events/lifecycle-rules/tenant-run.ts new file mode 100644 index 000000000..d07997ccf --- /dev/null +++ b/src/storage/events/lifecycle-rules/tenant-run.ts @@ -0,0 +1,167 @@ +import { LifecycleTenantStorePg, multitenantPgExecutor } from '@internal/database' +import { logger, logSchema } from '@internal/monitoring' +import { BasePayload, Queue as EventQueue } from '@internal/queue' +import { JobWithMetadata, Queue, SendOptions, WorkOptions } from 'pg-boss' +import { getConfig } from '../../../config' +import { retryAbandonedLifecycleArtifacts } from '../../lifecycle/abandoned-artifacts' +import { + evaluationJitterMs, + LIFECYCLE_EVALUATION_INTERVAL_MS, + lifecycleErrorMessage, +} from '../../lifecycle/executor' +import { isLifecycleEnabledForTenant } from '../../lifecycle/guards' +import type { LifecycleDispatchState } from '../../schemas' +import { BaseEvent } from '../base-event' +import { LifecycleBucketRun } from './bucket-run' + +const LIFECYCLE_TENANT_FANOUT = 8 +const LIFECYCLE_TENANT_CONCURRENCY = 4 +const LIFECYCLE_TENANT_EXPIRE_MINUTES = 3 +const LIFECYCLE_FEATURE_CACHE_RETRY_MS = 5 * 60_000 +const LIFECYCLE_ABANDONED_RETRY_INTERVAL_MS = 60 * 60_000 +const LIFECYCLE_ABANDONED_RETRY_BUDGET_MS = 60_000 +export const LIFECYCLE_TENANT_LEASE_MS = LIFECYCLE_TENANT_EXPIRE_MINUTES * 60_000 + +export interface LifecycleTenantRunPayload extends BasePayload { + centralClaim?: { + claimId: string + } +} + +export class LifecycleTenantRun extends BaseEvent { + static queueName = 'lifecycle:tenant:run-v1' + static allowSync = false + + static getQueueOptions(): Queue { + return { name: this.queueName, policy: 'exactly_once' } as const + } + + static getSendOptions(payload: LifecycleTenantRunPayload): SendOptions { + return { + singletonKey: payload.tenant.ref, + expireInMinutes: LIFECYCLE_TENANT_EXPIRE_MINUTES, + retryLimit: 0, + } + } + + static getWorkerOptions(): WorkOptions & { concurrentTaskCount: number } { + return { + includeMetadata: true, + concurrentTaskCount: LIFECYCLE_TENANT_CONCURRENCY, + batchSize: LIFECYCLE_TENANT_CONCURRENCY, + } + } + + static async shouldSend(): Promise { + return true + } + + static async handle( + job: JobWithMetadata, + opts?: { signal?: AbortSignal } + ): Promise { + const { tenant, centralClaim } = job.data + const centralStore = centralClaim + ? new LifecycleTenantStorePg(multitenantPgExecutor) + : undefined + + // After expiry the scheduler may have re-queued this claim + // but the store rejects writes past the deadline. + const deadline = new Date(job.startedOn.getTime() + LIFECYCLE_TENANT_LEASE_MS).toISOString() + + let storage: Awaited> | undefined + try { + let dispatchState: LifecycleDispatchState | undefined + let dispatchFloor = 0 + const gateClosed = !(await isLifecycleEnabledForTenant(tenant.ref)) + storage = await this.createStorage(job.data) + const database = storage.db.asSuperUser() + if (await database.hasMigration('lifecycle-writer-coordination')) { + if (!gateClosed) { + const bucketIds = await database.listDueLifecycleBuckets(LIFECYCLE_TENANT_FANOUT) + const sent = await Promise.all( + bucketIds.map((bucketId) => LifecycleBucketRun.send({ tenant, bucketId })) + ) + if ( + sent.some((jobId) => jobId === null) && + !(await EventQueue.getInstance().getQueue(LifecycleBucketRun.getQueueName())) + ) { + throw new Error(`Lifecycle queue ${LifecycleBucketRun.getQueueName()} does not exist`) + } + if (bucketIds.length > 0 && bucketIds.length < LIFECYCLE_TENANT_FANOUT) { + dispatchFloor = Date.now() + getConfig().storageLifecycleJobBudgetMs + } + } + await retryAbandonedLifecycleArtifacts({ + storage, + database, + attemptedBefore: new Date( + Date.now() - LIFECYCLE_ABANDONED_RETRY_INTERVAL_MS + ).toISOString(), + budgetMs: LIFECYCLE_ABANDONED_RETRY_BUDGET_MS, + signal: opts?.signal, + }) + if (centralClaim) { + dispatchState = await database.findLifecycleDispatchState({ + runBudgetMs: getConfig().storageLifecycleJobBudgetMs, + }) + } + } + + if (!centralStore || !centralClaim) return + + const staleFeatureCache = + gateClosed && (await centralStore.hasObjectVersioningFeature(tenant.ref)) + const now = Date.now() + let nextDispatchAt = dispatchState?.nextDispatchAt?.getTime() ?? Number.POSITIVE_INFINITY + if (gateClosed) { + // A bucket worker that saw the gate open can still create outbox rows after this check. + // Keep the tenant enrolled so those rows are swept even if the worker crashes. + nextDispatchAt = + now + + (staleFeatureCache + ? LIFECYCLE_FEATURE_CACHE_RETRY_MS + : LIFECYCLE_ABANDONED_RETRY_INTERVAL_MS) + } else { + nextDispatchAt = Math.max(nextDispatchAt, dispatchFloor) + if (dispatchState?.hasAbandonedArtifacts) { + nextDispatchAt = Math.min(nextDispatchAt, now + LIFECYCLE_ABANDONED_RETRY_INTERVAL_MS) + } + } + const completed = await centralStore.completeTenantDispatch( + tenant.ref, + centralClaim.claimId, + new Date( + Math.min( + nextDispatchAt, + now + LIFECYCLE_EVALUATION_INTERVAL_MS + evaluationJitterMs(tenant.ref) + ) + ).toISOString(), + deadline, + dispatchState + ) + if (!completed) { + logSchema.warning(logger, '[Lifecycle] Tenant dispatch lost its claim', { + type: 'event', + project: tenant.ref, + }) + } + } catch (error) { + if (!centralStore || !centralClaim) throw error + + logSchema.error(logger, '[Lifecycle] Tenant dispatcher failed', { + type: 'event', + project: tenant.ref, + error, + }) + await centralStore.failTenantDispatch( + tenant.ref, + centralClaim.claimId, + { code: 'tenant_dispatch_failed', message: lifecycleErrorMessage(error) }, + deadline + ) + } finally { + storage?.db.destroyConnection() + } + } +} diff --git a/src/storage/events/workers.test.ts b/src/storage/events/workers.test.ts new file mode 100644 index 000000000..bf6ab97b9 --- /dev/null +++ b/src/storage/events/workers.test.ts @@ -0,0 +1,20 @@ +import { Queue } from '@internal/queue' +import * as config from '../../config' +import { LifecycleBucketRun, LifecycleTenantRun } from './lifecycle-rules' +import { registerWorkers } from './workers' + +vi.mock('./base-event', () => ({ BaseEvent: class {} })) + +test.each([false, true])('registers lifecycle consumers: paused=%s', (paused) => { + const register = vi.spyOn(Queue, 'register').mockImplementation(() => {}) + vi.spyOn(config, 'getConfig').mockReturnValue({ + ...config.getConfig(), + storageLifecyclePaused: paused, + }) + + registerWorkers() + + expect(register).toHaveBeenCalled() + expect(register.mock.calls.some(([event]) => event === LifecycleTenantRun)).toBe(!paused) + expect(register.mock.calls.some(([event]) => event === LifecycleBucketRun)).toBe(!paused) +}) diff --git a/src/storage/events/workers.ts b/src/storage/events/workers.ts index dac31cc46..a5b500119 100644 --- a/src/storage/events/workers.ts +++ b/src/storage/events/workers.ts @@ -1,4 +1,5 @@ import { Queue } from '@internal/queue' +import { getConfig } from '../../config' import { PurgeCdnCache } from './cdn/purge-cdn-cache' import { DeleteIcebergResources } from './iceberg/delete-iceberg-resources' import { ReclaimIcebergShardSlots } from './iceberg/reclaim-shard-slots' @@ -6,6 +7,7 @@ import { ReconcileIcebergCatalog } from './iceberg/reconcile-catalog' import { JwksCreateSigningSecret } from './jwks/jwks-create-signing-secret' import { JwksRollUrlSigningKey } from './jwks/jwks-roll-url-signing-key' import { Webhook } from './lifecycle/webhook' +import { LifecycleBucketRun, LifecycleTenantRun } from './lifecycle-rules' import { ResetMigrationsOnTenant } from './migrations/reset-migrations' import { RunMigrationsOnTenants } from './migrations/run-migrations' import { BackupObjectEvent } from './objects/backup-object' @@ -31,4 +33,9 @@ export function registerWorkers() { Queue.register(ReclaimIcebergShardSlots) Queue.register(DeleteIcebergResources) Queue.register(SyncCatalogIds) + + if (!getConfig().storageLifecyclePaused) { + Queue.register(LifecycleTenantRun) + Queue.register(LifecycleBucketRun) + } } diff --git a/src/test/lifecycle-control-plane.test.ts b/src/test/lifecycle-control-plane.test.ts index 9170e007f..b3c198964 100644 --- a/src/test/lifecycle-control-plane.test.ts +++ b/src/test/lifecycle-control-plane.test.ts @@ -1,4 +1,5 @@ import { randomUUID } from 'node:crypto' +import * as tenants from '@internal/database' import { closeMultitenantPg, LifecycleTenantStorePg, @@ -7,6 +8,7 @@ import { import { runMultitenantMigrations } from '@internal/database/migrations' import { logSchema } from '@internal/monitoring' import { StoragePgDB } from '@storage/database' +import { LifecycleBucketRun, LifecycleTenantRun } from '@storage/events/lifecycle-rules' import { Storage } from '@storage/storage' import { Client } from 'pg' import * as config from '../config' @@ -96,6 +98,86 @@ describe('lifecycle policy enrollment across tenant and control databases', () = ) }) + it.each(['active', 'expired', 'stale feature cache'] as const)( + 'dispatches against the real central store with %s', + async (mode) => { + const { helper, bucketId } = fixtures + await fixtures.setStatus(bucketId, 'ENABLED') + await helper.database.putLifecycleConfiguration(bucketId, policy) + await helper.database.connection.query({ + text: 'UPDATE storage.bucket_lifecycle_states SET failure_count = 3 WHERE bucket_id = $1', + values: [bucketId], + }) + await multitenantPgExecutor.query({ + text: 'UPDATE public.tenants SET feature_object_versioning = true WHERE id = $1', + values: [tenantId], + }) + const store = new LifecycleTenantStorePg(multitenantPgExecutor) + const claims = await store.claimDueTenants(100, 180_000) + const claim = claims.find((entry) => entry.tenantId === tenantId)! + expect(claim).toBeDefined() + const database = new StoragePgDB(helper.database.connection.asSuperUser(), { + tenantId, + host: 'localhost', + }) + const storage = new Storage(helper.adapter, database, helper.storage.location) + vi.spyOn(config, 'getConfig').mockReturnValue({ + ...config.getConfig(), + isMultitenant: true, + storageVersioningEnabled: false, + }) + vi.spyOn(tenants, 'tenantHasFeature').mockResolvedValue(mode !== 'stale feature cache') + vi.spyOn( + LifecycleTenantRun as unknown as { createStorage: () => Promise }, + 'createStorage' + ).mockResolvedValue(storage) + const send = vi.spyOn(LifecycleBucketRun, 'send').mockResolvedValue('job') + const job = { + data: { + tenant: { ref: tenantId, host: 'localhost' }, + centralClaim: { claimId: claim.claimId }, + }, + startedOn: new Date(Date.now() - (mode === 'expired' ? 185_000 : 0)), + } as never + + const budget = config.getConfig().storageLifecycleJobBudgetMs + const startedAt = Date.now() + await LifecycleTenantRun.handle(job) + const finishedAt = Date.now() + const result = await multitenantPgExecutor.query({ + text: `SELECT claim_id, next_dispatch_at, last_success_at, failure_count, + max_bucket_failure_count, bucket_status_observed_at + FROM public.lifecycle_tenants WHERE tenant_id = $1`, + values: [tenantId], + }) + if (mode === 'expired') { + expect(result.rows[0].claim_id).toBe(claim.claimId) + expect(result.rows[0].last_success_at).toBeNull() + } else { + expect(result.rows[0].claim_id).toBeNull() + expect(result.rows[0].last_success_at).not.toBeNull() + expect(result.rows[0].next_dispatch_at).toBeInstanceOf(Date) + expect(result.rows[0].max_bucket_failure_count).toBe(3) + if (mode === 'active') { + const nextDispatchAt = result.rows[0].next_dispatch_at.getTime() + expect(nextDispatchAt).toBeGreaterThanOrEqual(startedAt + budget) + expect(nextDispatchAt).toBeLessThanOrEqual(finishedAt + budget) + } + expect(result.rows[0].bucket_status_observed_at).toBeInstanceOf(Date) + } + if (mode === 'stale feature cache') { + expect(send).not.toHaveBeenCalled() + expect(result.rows[0].next_dispatch_at.getTime() - Date.now()).toBeGreaterThan(299_000) + expect(result.rows[0].next_dispatch_at.getTime() - Date.now()).toBeLessThanOrEqual(300_000) + } else { + expect(send).toHaveBeenCalledWith({ + tenant: { ref: tenantId, host: 'localhost' }, + bucketId, + }) + } + } + ) + it('keeps the policy when the central tenant is missing and enrollment fails', async () => { const { helper, bucketId } = fixtures await fixtures.setStatus(bucketId, 'ENABLED') diff --git a/src/test/lifecycle-tenant-store.test.ts b/src/test/lifecycle-tenant-store.test.ts index 91eb0f24f..d24441e0d 100644 --- a/src/test/lifecycle-tenant-store.test.ts +++ b/src/test/lifecycle-tenant-store.test.ts @@ -14,6 +14,7 @@ describe('lifecycle tenant retry scheduling', () => { let transaction: DatabaseTransaction | undefined let store: LifecycleTenantStorePg const claimId = '49f3e225-7455-4b2e-bf49-2ca7abc80b49' + const deadline = () => new Date(Date.now() + 60_000).toISOString() const seedClaimed = () => transaction!.query({ text: `INSERT INTO lifecycle_tenants (tenant_id, next_dispatch_at, claim_id, claim_until) @@ -55,9 +56,9 @@ describe('lifecycle tenant retry scheduling', () => { values: [claimId, failureCount], }) - await expect(store.failTenantDispatch('tenant-a', claimId, { code: 'offline' })).resolves.toBe( - true - ) + await expect( + store.failTenantDispatch('tenant-a', claimId, { code: 'offline' }, deadline()) + ).resolves.toBe(true) const result = await transaction!.query(` SELECT extract(epoch FROM next_dispatch_at - last_completed_at)::float8 AS delay, @@ -79,12 +80,12 @@ describe('lifecycle tenant retry scheduling', () => { VALUES ('tenant-a', '2026-01-01T00:00:00Z', $1, 4)`, values: [claimId], }) - await expect(store.failTenantDispatch('tenant-a', claimId, { code: 'offline' })).resolves.toBe( - true - ) - await expect(store.failTenantDispatch('tenant-a', claimId, { code: 'stale' })).resolves.toBe( - false - ) + await expect( + store.failTenantDispatch('tenant-a', claimId, { code: 'offline' }, deadline()) + ).resolves.toBe(true) + await expect( + store.failTenantDispatch('tenant-a', claimId, { code: 'stale' }, deadline()) + ).resolves.toBe(false) const result = await transaction!.query(`SELECT * FROM lifecycle_tenants`) expect(result.rows[0]).toMatchObject({ next_dispatch_at: new Date('2026-01-01T00:00:00Z'), @@ -101,7 +102,12 @@ describe('lifecycle tenant retry scheduling', () => { await store.wakeTenant('tenant-a') await expect(store.claimDueTenants(1, 60_000)).resolves.toEqual([]) await expect( - store.completeTenantDispatch('tenant-a', claim.claimId, '2099-01-01T00:00:00Z') + store.completeTenantDispatch( + 'tenant-a', + claim.claimId, + '2099-01-01T00:00:00Z', + claim.claimUntil.toISOString() + ) ).resolves.toBe(true) const result = await transaction!.query(` @@ -115,6 +121,7 @@ describe('lifecycle tenant retry scheduling', () => { await store.wakeTenant('tenant-a') const [first] = await store.claimDueTenants(1, 60_000) expect(first.tenantId).toBe('tenant-a') + expect(first.reclaimed).toBe(false) await transaction!.query(` UPDATE lifecycle_tenants SET claim_until = clock_timestamp() - interval '1 second' WHERE tenant_id = 'tenant-a' @@ -123,17 +130,23 @@ describe('lifecycle tenant retry scheduling', () => { const [recovered] = await store.claimDueTenants(1, 60_000) expect(recovered.tenantId).toBe('tenant-a') expect(recovered.claimId).toBe(first.claimId) + expect(recovered.reclaimed).toBe(true) await expect( - store.completeTenantDispatch('tenant-a', first.claimId, new Date().toISOString()) + store.completeTenantDispatch( + 'tenant-a', + first.claimId, + new Date().toISOString(), + recovered.claimUntil.toISOString() + ) ).resolves.toBe(true) }) - it('backs off repeated expired dispatch leases without invalidating a queued job', async () => { + it('backs off expired leases by recorded failures without counting the reclaim', async () => { await store.wakeTenant('tenant-a') const [first] = await store.claimDueTenants(1, 180_000) for (const [failureCount, delaySeconds] of [ - [1, 300], - [2, 600], + [0, 300], + [1, 600], ] as const) { await transaction!.query(` UPDATE lifecycle_tenants SET claim_until = clock_timestamp() - interval '1 second' @@ -146,17 +159,20 @@ describe('lifecycle tenant retry scheduling', () => { extract(epoch FROM claim_until - clock_timestamp())::float8 AS delay FROM lifecycle_tenants WHERE tenant_id = 'tenant-a' `) - expect(rows[0]).toMatchObject({ - failure_count: failureCount, - last_error: { - code: 'tenant_dispatch_lease_expired', - message: 'Tenant dispatch lease expired before completion', - }, - }) + expect(rows[0].failure_count).toBe(failureCount) + if (failureCount === 0) expect(rows[0].last_error).toBeNull() expect(rows[0].delay).toBeGreaterThan(delaySeconds - 2) expect(rows[0].delay).toBeLessThanOrEqual(delaySeconds) await expect(store.claimDueTenants(1, 180_000)).resolves.toEqual([]) + await expect(store.recordExpiredTenantDispatch('tenant-a', first.claimId)).resolves.toBe(true) } + expect( + ( + await transaction!.query( + "SELECT failure_count FROM lifecycle_tenants WHERE tenant_id = 'tenant-a'" + ) + ).rows + ).toEqual([{ failure_count: 2 }]) await transaction!.query(` UPDATE lifecycle_tenants SET claim_until = clock_timestamp() - interval '1 second', failure_count = 100 WHERE tenant_id = 'tenant-a' @@ -168,7 +184,12 @@ describe('lifecycle tenant retry scheduling', () => { await store.wakeTenant('tenant-a') await expect(store.claimDueTenants(1, 180_000)).resolves.toEqual([]) await expect( - store.completeTenantDispatch('tenant-a', first.claimId, '2099-01-01T00:00:00Z') + store.completeTenantDispatch( + 'tenant-a', + first.claimId, + '2099-01-01T00:00:00Z', + capped.claimUntil.toISOString() + ) ).resolves.toBe(true) const { rows } = await transaction!.query(` SELECT failure_count, last_error FROM lifecycle_tenants WHERE tenant_id = 'tenant-a' @@ -206,7 +227,33 @@ describe('lifecycle tenant retry scheduling', () => { `) expect(result.rows[0].claim_id).toBe(first.claimId) expect(result.rows[0].claim_until).toEqual(recovered.claimUntil) - expect(result.rows[0].failure_count).toBe(1) + expect(result.rows[0].failure_count).toBe(0) + }) + + it('records an expired dispatch only for its claim', async () => { + await seedClaimed() + const read = async () => + ( + await transaction!.query( + "SELECT failure_count, last_error FROM lifecycle_tenants WHERE tenant_id = 'tenant-a'" + ) + ).rows + + await expect( + store.recordExpiredTenantDispatch('tenant-a', '00000000-0000-4000-8000-000000000000') + ).resolves.toBe(false) + expect(await read()).toEqual([{ failure_count: 0, last_error: null }]) + + await expect(store.recordExpiredTenantDispatch('tenant-a', claimId)).resolves.toBe(true) + expect(await read()).toEqual([ + { + failure_count: 1, + last_error: { + code: 'tenant_dispatch_lease_expired', + message: 'Tenant dispatch lease expired before completion', + }, + }, + ]) }) it('reports whether the central feature flag is on', async () => { @@ -231,7 +278,12 @@ describe('lifecycle tenant retry scheduling', () => { const [recovered] = await store.claimDueTenants(1, 60_000) expect(recovered.claimId).toBe(first.claimId) await expect( - store.completeTenantDispatch('tenant-a', first.claimId, '2099-01-01T00:00:00Z') + store.completeTenantDispatch( + 'tenant-a', + first.claimId, + '2099-01-01T00:00:00Z', + recovered.claimUntil.toISOString() + ) ).resolves.toBe(true) const result = await transaction!.query(` @@ -246,7 +298,12 @@ describe('lifecycle tenant retry scheduling', () => { const [first] = await store.claimDueTenants(1, 60_000) const revisitAt = new Date(Date.now() + 24 * 60 * 60 * 1000) await expect( - store.completeTenantDispatch('tenant-a', first.claimId, revisitAt.toISOString()) + store.completeTenantDispatch( + 'tenant-a', + first.claimId, + revisitAt.toISOString(), + first.claimUntil.toISOString() + ) ).resolves.toBe(true) const result = await transaction!.query(` SELECT next_dispatch_at, claim_id FROM lifecycle_tenants WHERE tenant_id = 'tenant-a' @@ -271,7 +328,7 @@ describe('lifecycle tenant retry scheduling', () => { }) await expect( - store.completeTenantDispatch('tenant-a', claimId, '2099-01-01T00:00:00Z') + store.completeTenantDispatch('tenant-a', claimId, '2099-01-01T00:00:00Z', deadline()) ).resolves.toBe(true) const result = await transaction!.query(` @@ -303,98 +360,296 @@ describe('lifecycle tenant retry scheduling', () => { 'SELECT max_bucket_failure_count, oldest_in_flight_started_at, bucket_status_observed_at FROM lifecycle_tenants' ) ).rows[0] - expect(await store.completeTenantDispatch('tenant-a', claim.claimId, next, health)).toBe(true) + expect( + await store.completeTenantDispatch( + 'tenant-a', + claim.claimId, + next, + claim.claimUntil.toISOString(), + health + ) + ).toBe(true) const observed = await read() expect(observed).toEqual({ max_bucket_failure_count: 7, oldest_in_flight_started_at: new Date(health.oldestInFlightStartedAt), bucket_status_observed_at: expect.any(Date), }) - expect(await store.completeTenantDispatch('tenant-a', claim.claimId, next, healthy)).toBe(false) + expect( + await store.completeTenantDispatch( + 'tenant-a', + claim.claimId, + next, + claim.claimUntil.toISOString(), + healthy + ) + ).toBe(false) await store.wakeTenant('tenant-a') const [unobserved] = await store.claimDueTenants(1, 60_000) - await store.completeTenantDispatch('tenant-a', unobserved.claimId, next) + await store.completeTenantDispatch( + 'tenant-a', + unobserved.claimId, + next, + unobserved.claimUntil.toISOString() + ) expect(await read()).toEqual(observed) await store.wakeTenant('tenant-a') const [recovered] = await store.claimDueTenants(1, 60_000) - await store.completeTenantDispatch('tenant-a', recovered.claimId, next, healthy) + await store.completeTenantDispatch( + 'tenant-a', + recovered.claimId, + next, + recovered.claimUntil.toISOString(), + healthy + ) expect(await read()).toMatchObject({ max_bucket_failure_count: 0, oldest_in_flight_started_at: null, }) }) - it('does not postpone an already due row when it is woken again', async () => { + it('counts expired leases toward the limit but still reclaims them', async () => { await transaction!.query(` - INSERT INTO lifecycle_tenants (tenant_id, next_dispatch_at) - VALUES ('tenant-a', '2000-01-01T00:00:00Z') + INSERT INTO lifecycle_tenants (tenant_id, next_dispatch_at, claim_id, claim_until) VALUES + ('claimed', 'infinity', gen_random_uuid(), now() + interval '1 hour'), + ('expired-old', 'infinity', gen_random_uuid(), now() - interval '1 minute') + `) + await transaction!.query(` + INSERT INTO lifecycle_tenants (tenant_id, next_dispatch_at) VALUES + ('due-old', now() - interval '1 day'), + ('due-new', now() - interval '2 hours') `) - await store.wakeTenant('tenant-a') + const claims = await store.claimDueTenantsWithinLeaseLimit(3, 60_000) - const result = await transaction!.query('SELECT next_dispatch_at FROM lifecycle_tenants') - expect(result.rows).toEqual([{ next_dispatch_at: new Date('2000-01-01T00:00:00Z') }]) + expect( + claims + .map(({ tenantId, reclaimed }) => ({ tenantId, reclaimed })) + .sort((a, b) => a.tenantId.localeCompare(b.tenantId)) + ).toEqual([ + { tenantId: 'due-old', reclaimed: false }, + { tenantId: 'expired-old', reclaimed: true }, + ]) + await expect(store.claimDueTenantsWithinLeaseLimit(3, 60_000)).resolves.toEqual([]) }) - it('retires a tenant only with the matching claim id', async () => { - await seedClaimed() + it('admits no new tenants while expired leases fill the limit', async () => { + await transaction!.query(` + INSERT INTO lifecycle_tenants (tenant_id, next_dispatch_at, claim_id, claim_until) VALUES + ('expired-old', 'infinity', gen_random_uuid(), now() - interval '2 minutes'), + ('expired-new', 'infinity', gen_random_uuid(), now() - interval '1 minute') + `) + await transaction!.query( + "INSERT INTO lifecycle_tenants (tenant_id, next_dispatch_at) VALUES ('due-old', now() - interval '1 day')" + ) - await expect( - store.retireTenant('tenant-a', '00000000-0000-4000-8000-000000000000') - ).resolves.toBe(false) - const rows = () => - transaction!.query("SELECT 1 FROM lifecycle_tenants WHERE tenant_id = 'tenant-a'") - expect((await rows()).rowCount).toBe(1) + const claims = await store.claimDueTenantsWithinLeaseLimit(2, 60_000) - await expect(store.retireTenant('tenant-a', claimId)).resolves.toBe(true) - expect((await rows()).rowCount).toBe(0) - await expect(store.retireTenant('tenant-a', claimId)).resolves.toBe(false) + expect( + claims + .map(({ tenantId, reclaimed }) => ({ tenantId, reclaimed })) + .sort((a, b) => a.tenantId.localeCompare(b.tenantId)) + ).toEqual([ + { tenantId: 'expired-new', reclaimed: true }, + { tenantId: 'expired-old', reclaimed: true }, + ]) + const row = await transaction!.query( + "SELECT claim_id FROM lifecycle_tenants WHERE tenant_id = 'due-old'" + ) + expect(row.rows).toEqual([{ claim_id: null }]) }) - it('does not retire a tenant whose central feature flag is on', async () => { + it('reclaims expired leases without admitting new tenants when leases exceed the limit', async () => { + await transaction!.query(` + INSERT INTO lifecycle_tenants (tenant_id, next_dispatch_at, claim_id, claim_until) VALUES + ('claimed', 'infinity', gen_random_uuid(), now() + interval '1 hour'), + ('expired-old', 'infinity', gen_random_uuid(), now() - interval '1 minute') + `) await transaction!.query( - "UPDATE tenants SET feature_object_versioning = true WHERE id = 'tenant-a'" + "INSERT INTO lifecycle_tenants (tenant_id, next_dispatch_at) VALUES ('due-old', now() - interval '1 day')" ) - await transaction!.query('DELETE FROM lifecycle_tenants') - await seedClaimed() - await expect(store.retireTenant('tenant-a', claimId)).resolves.toBe(false) - const result = await transaction!.query( - "SELECT 1 FROM lifecycle_tenants WHERE tenant_id = 'tenant-a'" + const claims = await store.claimDueTenantsWithinLeaseLimit(1, 60_000) + + expect(claims.map(({ tenantId, reclaimed }) => ({ tenantId, reclaimed }))).toEqual([ + { tenantId: 'expired-old', reclaimed: true }, + ]) + const row = await transaction!.query( + "SELECT claim_id FROM lifecycle_tenants WHERE tenant_id = 'due-old'" ) - expect(result.rowCount).toBe(1) + expect(row.rows).toEqual([{ claim_id: null }]) + }) + + it('cancels a claim blocked on the scheduler advisory lock', async () => { + const pool = new Pool({ + connectionString: getConfig().multitenantDatabaseUrl, + options: '-c statement_timeout=5000', + max: 3, + }) + const executor = new PgPoolExecutor(pool) + let holder: DatabaseTransaction | undefined + let waiter: DatabaseTransaction | undefined + try { + holder = await executor.beginTransaction() + await holder.query("SELECT pg_advisory_xact_lock(hashtext('lifecycle_tenants_claim'))") + waiter = await executor.beginTransaction() + const controller = new AbortController() + const pending = new LifecycleTenantStorePg(waiter).claimDueTenantsWithinLeaseLimit( + 100, + 180_000, + controller.signal + ) + await expect + .poll(async () => { + const result = await pool.query( + "SELECT count(*)::int AS count FROM pg_stat_activity WHERE wait_event_type = 'Lock' AND query LIKE '%lifecycle_tenants_claim%'" + ) + return result.rows[0].count + }) + .toBeGreaterThan(0) + + controller.abort() + await expect(pending).rejects.toMatchObject({ name: 'AbortError' }) + expect(holder.isCompleted()).toBe(false) + } finally { + if (waiter && !waiter.isCompleted()) await waiter.rollback() + if (holder && !holder.isCompleted()) await holder.rollback() + await pool.end() + } + }) + + it('serializes concurrent claimers so outstanding claims stay within the limit', async () => { + const ids = [1, 2, 3].map(() => `lease-limit-${randomUUID()}`) + const pool = new Pool({ + connectionString: getConfig().multitenantDatabaseUrl, + options: '-c statement_timeout=5000', + max: 3, + }) + const executor = new PgPoolExecutor(pool) + let first: DatabaseTransaction | undefined + let second: DatabaseTransaction | undefined + try { + await pool.query( + "INSERT INTO tenants (id, anon_key, database_url, jwt_secret, service_key) SELECT id, 'anon', 'url', 'jwt', 'service' FROM unnest($1::text[]) AS id", + [ids] + ) + await pool.query( + "INSERT INTO lifecycle_tenants (tenant_id, next_dispatch_at) SELECT id, '2000-01-01' FROM unnest($1::text[]) AS id", + [ids] + ) + const baseline = ( + await pool.query( + 'SELECT count(*)::int AS n FROM lifecycle_tenants WHERE claim_until IS NOT NULL' + ) + ).rows[0].n + const limit = baseline + 2 + + first = await executor.beginTransaction() + const claims1 = await new LifecycleTenantStorePg(first).claimDueTenantsWithinLeaseLimit( + limit, + 60_000 + ) + expect(claims1.filter((claim) => ids.includes(claim.tenantId))).toHaveLength(2) + + second = await executor.beginTransaction() + let settled = false + const pending = new LifecycleTenantStorePg(second) + .claimDueTenantsWithinLeaseLimit(limit, 60_000) + .finally(() => { + settled = true + }) + await new Promise((resolve) => setTimeout(resolve, 300)) + expect(settled).toBe(false) + + await first.commit() + expect((await pending).filter((claim) => ids.includes(claim.tenantId))).toEqual([]) + await second.commit() + + const claimed = await pool.query( + 'SELECT count(*)::int AS n FROM lifecycle_tenants WHERE tenant_id = ANY($1::text[]) AND claim_id IS NOT NULL', + [ids] + ) + expect(claimed.rows[0].n).toBe(2) + } finally { + if (first && !first.isCompleted()) await first.rollback() + if (second && !second.isCompleted()) await second.rollback() + await pool.query('DELETE FROM lifecycle_tenants WHERE tenant_id = ANY($1::text[])', [ids]) + await pool.query('DELETE FROM tenants WHERE id = ANY($1::text[])', [ids]) + await pool.end() + } }) - it('keeps a tenant woken during its claim instead of retiring it', async () => { + it('does not postpone an already due row when it is woken again', async () => { + await transaction!.query(` + INSERT INTO lifecycle_tenants (tenant_id, next_dispatch_at) + VALUES ('tenant-a', '2000-01-01T00:00:00Z') + `) + await store.wakeTenant('tenant-a') - const [claim] = await store.claimDueTenants(1, 60_000) + + const result = await transaction!.query('SELECT next_dispatch_at FROM lifecycle_tenants') + expect(result.rows).toEqual([{ next_dispatch_at: new Date('2000-01-01T00:00:00Z') }]) + }) + + it('releases a claim without counting a failure and keeps a wake from its lease', async () => { await store.wakeTenant('tenant-a') + const [claim] = await store.claimDueTenants(1, 60_000) + expect(claim.reclaimed).toBe(false) + await transaction!.query( + "UPDATE lifecycle_tenants SET failure_count = 2 WHERE tenant_id = 'tenant-a'" + ) - await expect(store.retireTenant('tenant-a', claim.claimId)).resolves.toBe(false) await expect( - store.completeTenantDispatch('tenant-a', claim.claimId, '2099-01-01T00:00:00Z') + store.releaseTenantDispatch( + 'tenant-a', + claim.claimId, + '2099-01-01T00:00:00Z', + claim.claimUntil.toISOString() + ) ).resolves.toBe(true) + const row = await transaction!.query( + "SELECT claim_id, claim_until, next_dispatch_at, failure_count, last_error FROM lifecycle_tenants WHERE tenant_id = 'tenant-a'" + ) + expect(row.rows).toEqual([ + { + claim_id: null, + claim_until: null, + next_dispatch_at: new Date('2099-01-01T00:00:00Z'), + failure_count: 2, + last_error: null, + }, + ]) - const result = await transaction!.query(` - SELECT next_dispatch_at <= clock_timestamp() AS due FROM lifecycle_tenants - WHERE tenant_id = 'tenant-a' - `) - expect(result.rows).toEqual([{ due: true }]) + await transaction!.query( + "UPDATE lifecycle_tenants SET next_dispatch_at = clock_timestamp() WHERE tenant_id = 'tenant-a'" + ) + const [again] = await store.claimDueTenants(1, 60_000) + await store.wakeTenant('tenant-a') + await expect( + store.releaseTenantDispatch( + 'tenant-a', + again.claimId, + '2099-01-01T00:00:00Z', + again.claimUntil.toISOString() + ) + ).resolves.toBe(true) + const due = await transaction!.query( + "SELECT next_dispatch_at <= clock_timestamp() AS due FROM lifecycle_tenants WHERE tenant_id = 'tenant-a'" + ) + expect(due.rows).toEqual([{ due: true }]) }) it.each([ [ 'complete', (deadline: string) => - store.completeTenantDispatch( - 'tenant-a', - claimId, - new Date().toISOString(), - undefined, - deadline - ), + store.completeTenantDispatch('tenant-a', claimId, new Date().toISOString(), deadline), + ], + [ + 'release', + (deadline: string) => + store.releaseTenantDispatch('tenant-a', claimId, new Date().toISOString(), deadline), ], - ['retire', (deadline: string) => store.retireTenant('tenant-a', claimId, deadline)], [ 'fail', (deadline: string) => @@ -407,6 +662,8 @@ describe('lifecycle tenant retry scheduling', () => { .rows const before = await snapshot() + await expect(write(undefined as never)).resolves.toBe(false) + expect(await snapshot()).toEqual(before) await expect(write(new Date(Date.now() - 1000).toISOString())).resolves.toBe(false) expect(await snapshot()).toEqual(before) @@ -442,7 +699,6 @@ describe('lifecycle tenant retry scheduling', () => { tenantId, claimId, new Date().toISOString(), - undefined, new Date(Date.now() + 1000).toISOString() ) await new Promise((resolve) => setTimeout(resolve, 1500)) @@ -466,18 +722,12 @@ describe('lifecycle tenant retry scheduling', () => { [ 'complete', (store: LifecycleTenantStorePg, tenantId: string, deadline: string) => - store.completeTenantDispatch( - tenantId, - claimId, - new Date().toISOString(), - undefined, - deadline - ), + store.completeTenantDispatch(tenantId, claimId, new Date().toISOString(), deadline), ], [ - 'retire', + 'release', (store: LifecycleTenantStorePg, tenantId: string, deadline: string) => - store.retireTenant(tenantId, claimId, deadline), + store.releaseTenantDispatch(tenantId, claimId, new Date().toISOString(), deadline), ], [ 'fail', @@ -533,185 +783,6 @@ describe('lifecycle tenant retry scheduling', () => { } }) - it('enrolls every tenant missing a row without touching existing ones', async () => { - await transaction!.query(`INSERT INTO lifecycle_tenants (tenant_id, next_dispatch_at) - VALUES ('future', '2099-01-01T00:00:00Z')`) - await transaction!.query({ - text: `INSERT INTO lifecycle_tenants (tenant_id, next_dispatch_at, claim_id, claim_until) - VALUES ('claimed', 'infinity', $1, now() + interval '1 hour')`, - values: [claimId], - }) - - await store.enrollAllTenants() - - const result = await transaction!.query(` - SELECT tenant_id, next_dispatch_at <= clock_timestamp() AS due, claim_id - FROM lifecycle_tenants ORDER BY tenant_id - `) - expect(result.rows.map((row) => row.tenant_id)).toEqual( - expect.arrayContaining(['tenant-a', 'due-old', 'claimed', 'future']) - ) - const due = Object.fromEntries(result.rows.map((row) => [row.tenant_id, row.due])) - expect(due.future).toBe(false) - expect(due['tenant-a']).toBe(true) - expect(due.claimed).toBe(false) - expect(result.rows.find((row) => row.tenant_id === 'claimed')?.claim_id).toBe(claimId) - }) - - it('batches enrollment, retries partial progress, and tolerates concurrent scans', async () => { - const prefix = `enroll-batch-${randomUUID()}-` - const schema = `enroll_batch_${randomUUID().replaceAll('-', '')}` - const pool = new Pool({ - connectionString: getConfig().multitenantDatabaseUrl, - options: `-c search_path=${schema} -c statement_timeout=5000`, - max: 3, - }) - const executor = new PgPoolExecutor(pool) - const firstId = `${prefix}0000` - const failure = new Error('batch failed') - let pages = 0 - let failSecondPage = true - let addedAfterBound = false - const query = executor.query.bind(executor) - const spy = vi.spyOn(executor, 'query').mockImplementation((statement, options) => { - if (typeof statement === 'string' && !failSecondPage && !addedAfterBound) { - return query(statement, options).then(async (result) => { - await pool.query( - `INSERT INTO tenants (id, anon_key, database_url, jwt_secret, service_key) - VALUES ($1, 'anon', 'url', 'jwt', 'service')`, - [`${prefix}9999`] - ) - addedAfterBound = true - return result - }) - } - if (typeof statement !== 'string' && statement.text.includes('WITH batch AS MATERIALIZED')) { - pages++ - if (failSecondPage && pages === 2) return Promise.reject(failure) - } - return query(statement, options) - }) - - try { - await pool.query(`CREATE SCHEMA ${schema}`) - await pool.query('CREATE TABLE tenants (LIKE public.tenants INCLUDING ALL)') - await pool.query( - 'CREATE TABLE lifecycle_tenants (LIKE public.lifecycle_tenants INCLUDING ALL)' - ) - await pool.query( - 'ALTER TABLE lifecycle_tenants ADD FOREIGN KEY (tenant_id) REFERENCES tenants(id) ON DELETE CASCADE' - ) - await pool.query( - `INSERT INTO tenants (id, anon_key, database_url, jwt_secret, service_key) - SELECT $1 || lpad(i::text, 4, '0'), 'anon', 'url', 'jwt', 'service' - FROM generate_series(0, 2000) AS i`, - [prefix] - ) - await pool.query({ - text: `INSERT INTO lifecycle_tenants (tenant_id, next_dispatch_at, claim_id, claim_until) - VALUES ($1, 'infinity', $2, now() + interval '1 hour')`, - values: [firstId, claimId], - }) - const store = new LifecycleTenantStorePg(executor) - - await expect(store.enrollAllTenants()).rejects.toBe(failure) - const partial = await pool.query( - 'SELECT count(*)::int AS count FROM lifecycle_tenants WHERE tenant_id LIKE $1', - [`${prefix}%`] - ) - expect(partial.rows[0].count).toBe(1000) - - pages = 0 - failSecondPage = false - await store.enrollAllTenants() - expect(pages).toBe(3) - - const enrolled = await pool.query( - 'SELECT count(*)::int AS count FROM lifecycle_tenants WHERE tenant_id LIKE $1', - [`${prefix}%`] - ) - expect(enrolled.rows[0].count).toBe(2001) - const existing = await pool.query( - 'SELECT next_dispatch_at, claim_id FROM lifecycle_tenants WHERE tenant_id = $1', - [firstId] - ) - expect(existing.rows).toEqual([ - { next_dispatch_at: Number.POSITIVE_INFINITY, claim_id: claimId }, - ]) - - expect(addedAfterBound).toBe(true) - const late = await pool.query('SELECT 1 FROM lifecycle_tenants WHERE tenant_id = $1', [ - `${prefix}9999`, - ]) - expect(late.rowCount).toBe(0) - await store.enrollAllTenants() - expect( - ( - await pool.query('SELECT 1 FROM lifecycle_tenants WHERE tenant_id = $1', [ - `${prefix}9999`, - ]) - ).rowCount - ).toBe(1) - await pool.query('DELETE FROM lifecycle_tenants') - await Promise.all([ - store.enrollAllTenants(), - new LifecycleTenantStorePg(new PgPoolExecutor(pool)).enrollAllTenants(), - ]) - expect( - (await pool.query('SELECT count(*)::int AS count FROM lifecycle_tenants')).rows - ).toEqual([{ count: 2002 }]) - } finally { - spy.mockRestore() - await pool.query(`DROP SCHEMA IF EXISTS ${schema} CASCADE`) - await pool.end() - } - }) - - it('enrolls the surviving tenants when one is deleted concurrently', async () => { - const suffix = randomUUID() - const [kept, deleted] = [`enroll-kept-${suffix}`, `enroll-deleted-${suffix}`] - const pool = new Pool({ - connectionString: getConfig().multitenantDatabaseUrl, - options: '-c statement_timeout=5000', - max: 3, - }) - const executor = new PgPoolExecutor(pool) - let deleter: DatabaseTransaction | undefined - try { - await pool.query( - `INSERT INTO tenants (id, anon_key, database_url, jwt_secret, service_key) - SELECT id, 'anon', 'url', 'jwt', 'service' FROM unnest($1::text[]) AS id`, - [[kept, deleted]] - ) - deleter = await executor.beginTransaction() - await deleter.query({ text: 'DELETE FROM tenants WHERE id = $1', values: [deleted] }) - - const enrolling = new LifecycleTenantStorePg(executor).enrollAllTenants() - await expect - .poll( - async () => - ( - await pool.query( - "SELECT count(*)::int AS count FROM pg_stat_activity WHERE datname = current_database() AND wait_event_type = 'Lock' AND query LIKE '%INSERT INTO lifecycle_tenants%'" - ) - ).rows[0].count - ) - .toBe(1) - await deleter.commit() - - await expect(enrolling).resolves.toBeUndefined() - const enrolled = await pool.query('SELECT 1 FROM lifecycle_tenants WHERE tenant_id = $1', [ - kept, - ]) - expect(enrolled.rowCount).toBe(1) - } finally { - if (deleter && !deleter.isCompleted()) await deleter.rollback() - await pool.query('DELETE FROM lifecycle_tenants WHERE tenant_id = ANY($1)', [[kept, deleted]]) - await pool.query('DELETE FROM tenants WHERE id = ANY($1)', [[kept, deleted]]) - await pool.end() - } - }) - it('claims the oldest due or expired rows first and leases them', async () => { await transaction!.query(` INSERT INTO lifecycle_tenants (tenant_id, next_dispatch_at, claim_id, claim_until) VALUES diff --git a/src/test/noncurrent-lifecycle-executor.test.ts b/src/test/noncurrent-lifecycle-executor.test.ts index b25a92657..af7ddda75 100644 --- a/src/test/noncurrent-lifecycle-executor.test.ts +++ b/src/test/noncurrent-lifecycle-executor.test.ts @@ -4,12 +4,14 @@ import { text } from 'node:stream/consumers' import { logSchema } from '@internal/monitoring' import { isMissingBackendObject, withOptionalVersion } from '@storage/backend' import { StoragePgDB } from '@storage/database' +import type { LifecycleBucketRunPayload } from '@storage/events/lifecycle-rules/bucket-run' import { retryAbandonedLifecycleArtifacts } from '@storage/lifecycle/abandoned-artifacts' import { type LifecycleExecutorOptions, NoncurrentLifecycleExecutor, } from '@storage/lifecycle/executor' import { Client } from 'pg' +import PgBoss from 'pg-boss' import { getConfig } from '../config' import { batchOf, newContinuation } from './utils/lifecycle-continuation' import { useLifecycleStorage } from './utils/lifecycle-versioning' @@ -549,6 +551,92 @@ describe('noncurrent lifecycle executor against version-aware writes', () => { } ) + it('deduplicates real queue deliveries and re-sends after an expired job fails without queue retries', async () => { + const { LifecycleBucketRun } = await import('@storage/events/lifecycle-rules/bucket-run') + await activate() + const name = 'history/queue-recovery.txt' + const archived = randomUUID() + const current = randomUUID() + const archivedAt = threeDaysAgo() + await seedHistory(name, archived, current, { archivedAt }) + const schema = `lifecycle_queue_${randomUUID().replaceAll('-', '')}` + const config = getConfig() + // pg-boss uses dynamic SQL unsupported by Multigres, use the control DB for this test. + const queueUrl = + config.pgQueueConnectionURL || + (config.databaseEngine === 'multigres' + ? config.multitenantDatabaseUrl || config.multitenantDatabasePoolUrl + : config.databaseURL) + if (!queueUrl) throw new Error('A PostgreSQL queue database URL is required for this test') + const queueClient = new Client({ connectionString: queueUrl }) + await queueClient.connect() + const options = { + connectionString: queueUrl, + schema, + supervise: false, + schedule: false, + } + let boss = new PgBoss(options) + const payload: LifecycleBucketRunPayload = { + bucketId, + tenant: { ref: tenantId, host: '' }, + } + const queue = LifecycleBucketRun.getQueueOptions() + try { + await boss.start() + await boss.createQueue(queue.name, queue) + const send = () => boss.send(queue.name, payload, LifecycleBucketRun.getSendOptions(payload)) + const id = await send() + expect(id).toBeTypeOf('string') + await expect(send()).resolves.toBeNull() + const [first] = await boss.fetch(queue.name) + expect(first.id).toBe(id) + const input = await prepareAttempt(name, archived, archivedAt) + expect(await helper.database.armLifecycleAttempt(input)).toMatchObject({ status: 'armed' }) + await expect(runExecutor()).resolves.toMatchObject({ status: 'NOT_CLAIMED' }) + await expect(send()).resolves.toBeNull() + await boss.stop() + + // Model elapsed lease time without sleeping for the expiry budgets. + await queueClient.query( + `UPDATE ${schema}.job SET started_on = '2000-01-01', retry_delay = 0 WHERE id = $1`, + [id] + ) + await helper.database.connection.query( + "UPDATE storage.bucket_lifecycle_states SET claim_until = '2000-01-01', next_run_at = '2000-01-01' WHERE bucket_id = $1", + [bucketId] + ) + boss = new PgBoss(options) + await boss.start() + await boss.maintain() + await expect( + boss.fetch(queue.name, { includeMetadata: true }) + ).resolves.toEqual([]) + const resentId = await send() + expect(resentId).toBeTypeOf('string') + expect(resentId).not.toBe(id) + const [recovered] = await boss.fetch(queue.name) + expect(recovered.id).toBe(resentId) + const deletes = vi.spyOn(helper.adapter, 'deleteObjectsDetailed') + await expect( + new NoncurrentLifecycleExecutor(helper.storage).run(recovered.data.bucketId, { + tenantLifecycleEnabled: true, + }) + ).resolves.toMatchObject({ status: 'COMPLETED' }) + await boss.complete(queue.name, recovered.id) + expect(deletes).not.toHaveBeenCalled() + expect(await abandonedKeys()).toHaveLength(2) + expect(await versions(name)).toEqual([{ version: current, current: true }]) + } finally { + try { + await boss.stop() + await queueClient.query(`DROP SCHEMA IF EXISTS ${schema} CASCADE`) + } finally { + await queueClient.end() + } + } + }) + it('expires history produced by complete uploads and preserves the replacement body', async () => { await activate() const name = 'history/writer.txt'