From 62eb4797583620a10ba094f60f80c2cb15b0031b Mon Sep 17 00:00:00 2001 From: Ferhat Elmas Date: Mon, 9 Nov 2026 18:47:56 +0100 Subject: [PATCH] feat: add noncurrent lifecycle executor Signed-off-by: Ferhat Elmas --- .env.sample | 12 + src/config.test.ts | 119 ++ src/config.ts | 51 + src/internal/monitoring/metrics.ts | 59 + src/storage/database/pg.ts | 2 + .../lifecycle/abandoned-artifacts.test.ts | 221 +++ src/storage/lifecycle/abandoned-artifacts.ts | 55 + .../lifecycle/artifact-deletion.test.ts | 147 ++ src/storage/lifecycle/artifact-deletion.ts | 66 + src/storage/lifecycle/executor.test.ts | 864 +++++++++ src/storage/lifecycle/executor.ts | 331 ++++ .../noncurrent-lifecycle-database.test.ts | 11 +- .../noncurrent-lifecycle-executor.test.ts | 1593 +++++++++++++++++ 13 files changed, 3527 insertions(+), 4 deletions(-) create mode 100644 src/storage/lifecycle/abandoned-artifacts.test.ts create mode 100644 src/storage/lifecycle/abandoned-artifacts.ts create mode 100644 src/storage/lifecycle/artifact-deletion.test.ts create mode 100644 src/storage/lifecycle/artifact-deletion.ts create mode 100644 src/storage/lifecycle/executor.test.ts create mode 100644 src/storage/lifecycle/executor.ts create mode 100644 src/test/noncurrent-lifecycle-executor.test.ts diff --git a/.env.sample b/.env.sample index 805914a1b..236681d26 100644 --- a/.env.sample +++ b/.env.sample @@ -143,6 +143,18 @@ S3_PROTOCOL_ACCESS_KEY_SECRET=67d161a7a8a46a24a17a75b26e7724f11d56b8d49a119227c6 ####################################### STORAGE_BACKEND=s3 +# Evaluation pages are capped at 480 rows and deletion batches at 48 versions. +STORAGE_LIFECYCLE_PAGE_SIZE=480 +STORAGE_LIFECYCLE_DELETE_BATCH_SIZE=48 + +# Stop starting new evaluation work after this budget. +# In-flight work may finish later. +# The claim lease prevents another worker from claiming the bucket and is renewed +# when a deletion batch is armed. The budget must be shorter than the lease and +# their sum must not exceed 23h 59m. Invalid values prevent startup. +STORAGE_LIFECYCLE_JOB_BUDGET_MS=50000 +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. STORAGE_VERSIONING_ENABLED=false diff --git a/src/config.test.ts b/src/config.test.ts index 75a98b4dd..95db191cd 100644 --- a/src/config.test.ts +++ b/src/config.test.ts @@ -24,6 +24,10 @@ const CONFIG_ENV_KEYS = [ 'OTEL_EXPORTER_OTLP_ENDPOINT', 'OTEL_EXPORTER_OTLP_METRICS_ENDPOINT', 'REQUEST_HARD_LIMITS_ENABLED', + 'STORAGE_LIFECYCLE_PAGE_SIZE', + 'STORAGE_LIFECYCLE_DELETE_BATCH_SIZE', + 'STORAGE_LIFECYCLE_JOB_BUDGET_MS', + 'STORAGE_LIFECYCLE_CLAIM_LEASE_MS', 'STORAGE_S3_REQUEST_CHECKSUM_CALCULATION', 'STORAGE_S3_RESPONSE_CHECKSUM_VALIDATION', 'STORAGE_S3_CLIENT_TIMEOUT', @@ -276,6 +280,121 @@ describe('configuration parsing', () => { expect(config.otlpMetricsEndpoint).toBe('http://metrics-collector:4317') }) + test.each([ + [ + 'lifecycle job budget and claim 1 ms above the queue expiry limit', + { + STORAGE_LIFECYCLE_JOB_BUDGET_MS: '100000', + STORAGE_LIFECYCLE_CLAIM_LEASE_MS: '86240001', + }, + 'must not exceed 23 hours 59 minutes', + ], + [ + 'lifecycle job budget at the largest accepted knob value', + { STORAGE_LIFECYCLE_JOB_BUDGET_MS: '2147483647' }, + 'STORAGE_LIFECYCLE_JOB_BUDGET_MS must be shorter than STORAGE_LIFECYCLE_CLAIM_LEASE_MS', + ], + [ + 'lifecycle job budget outliving its claim', + { STORAGE_LIFECYCLE_JOB_BUDGET_MS: '120000', STORAGE_LIFECYCLE_CLAIM_LEASE_MS: '120000' }, + 'STORAGE_LIFECYCLE_JOB_BUDGET_MS must be shorter than STORAGE_LIFECYCLE_CLAIM_LEASE_MS', + ], + ])('rejects %s at startup', async (_label, env, message) => { + setConfigEnv(env) + + const { getConfig } = await import('./config') + + expect(() => getConfig({ reload: true })).toThrow(message) + }) + + test('accepts lifecycle timers at the maximum budget and claim total', async () => { + setConfigEnv({ + STORAGE_LIFECYCLE_JOB_BUDGET_MS: '100000', + STORAGE_LIFECYCLE_CLAIM_LEASE_MS: '86240000', + }) + const { getConfig } = await import('./config') + + expect(() => getConfig({ reload: true })).not.toThrow() + }) + + test.each([ + ['STORAGE_LIFECYCLE_PAGE_SIZE', 480], + ['STORAGE_LIFECYCLE_DELETE_BATCH_SIZE', 48], + ['STORAGE_LIFECYCLE_JOB_BUDGET_MS', 2147483647], + ['STORAGE_LIFECYCLE_CLAIM_LEASE_MS', 2147483647], + ] as const)('rejects invalid values of %s naming the variable', async (name, maximum) => { + const { getConfig } = await import('./config') + + for (const value of [ + '120s', + '0', + '-1', + '04', + '1.5', + 'NaN', + '5e4', + String(maximum + 1), + '9007199254740992', + '050000', + '50000.0', + '+50000', + '0x10', + ]) { + setConfigEnv({ [name]: value }) + expect(() => getConfig({ reload: true })).toThrow( + `${name} must be a positive integer of at most ${maximum}` + ) + } + }) + + test('defaults unset lifecycle settings and parses valid ones', async () => { + const { getConfig, setEnvPaths } = await import('./config') + + setEnvPaths([]) + setConfigEnv({}) + expect(getConfig({ reload: true })).toMatchObject({ + storageLifecyclePageSize: 480, + storageLifecycleDeleteBatchSize: 48, + storageLifecycleJobBudgetMs: 50_000, + storageLifecycleClaimLeaseMs: 120_000, + }) + + setConfigEnv({ + STORAGE_LIFECYCLE_PAGE_SIZE: '', + STORAGE_LIFECYCLE_DELETE_BATCH_SIZE: '', + STORAGE_LIFECYCLE_JOB_BUDGET_MS: '', + STORAGE_LIFECYCLE_CLAIM_LEASE_MS: '', + }) + expect(getConfig({ reload: true })).toMatchObject({ + storageLifecyclePageSize: 480, + storageLifecycleDeleteBatchSize: 48, + storageLifecycleJobBudgetMs: 50_000, + storageLifecycleClaimLeaseMs: 120_000, + }) + + setConfigEnv({ + STORAGE_LIFECYCLE_PAGE_SIZE: '200', + STORAGE_LIFECYCLE_DELETE_BATCH_SIZE: '12', + STORAGE_LIFECYCLE_JOB_BUDGET_MS: '10000', + STORAGE_LIFECYCLE_CLAIM_LEASE_MS: '90000', + }) + expect(getConfig({ reload: true })).toMatchObject({ + storageLifecyclePageSize: 200, + storageLifecycleDeleteBatchSize: 12, + storageLifecycleJobBudgetMs: 10_000, + storageLifecycleClaimLeaseMs: 90_000, + }) + + setConfigEnv({ + STORAGE_LIFECYCLE_PAGE_SIZE: ' 200\n', + STORAGE_LIFECYCLE_DELETE_BATCH_SIZE: ' \t', + }) + expect(getConfig({ reload: true })).toMatchObject({ + storageLifecyclePageSize: 200, + storageLifecycleDeleteBatchSize: 48, + }) + }) + test('freezes JWT JWKS configuration and its keys', async () => { setConfigEnv({ JWT_JWKS: JSON.stringify({ keys: [{ kty: 'oct', k: 'secret' }] }), diff --git a/src/config.ts b/src/config.ts index 6eaae75e6..3e72af1d3 100644 --- a/src/config.ts +++ b/src/config.ts @@ -1,6 +1,10 @@ import type { DBMigration } from '@internal/database/migrations' import { loadOptionalEnvFile } from '@internal/env-file' import { SignJWT } from 'jose' +import { + LIFECYCLE_MAX_DELETE_BATCH_SIZE, + LIFECYCLE_MAX_PAGE_SIZE, +} from './storage/schemas/lifecycle' export type StorageBackendType = 'file' | 's3' export type VectorBucketProvider = 's3' | 'pgvector' @@ -103,6 +107,10 @@ type StorageConfigType = { storageS3ForcePathStyle?: boolean storageS3Region: string storageS3ClientTimeout: number + storageLifecyclePageSize: number + storageLifecycleDeleteBatchSize: number + storageLifecycleJobBudgetMs: number + storageLifecycleClaimLeaseMs: number isMultitenant: boolean jwtSecret: string jwtAlgorithm: JwtAlgorithm @@ -531,6 +539,26 @@ export function getConfig(options?: { reload?: boolean }): StorageConfigType { // Storage storageBackendType: getOptionalConfigFromEnv('STORAGE_BACKEND') as StorageBackendType, emptyBucketMax: parseInt(getOptionalConfigFromEnv('STORAGE_EMPTY_BUCKET_MAX') || '200000', 10), + storageLifecyclePageSize: envStrictPositiveInteger( + 'STORAGE_LIFECYCLE_PAGE_SIZE', + LIFECYCLE_MAX_PAGE_SIZE, + LIFECYCLE_MAX_PAGE_SIZE + ), + storageLifecycleDeleteBatchSize: envStrictPositiveInteger( + 'STORAGE_LIFECYCLE_DELETE_BATCH_SIZE', + LIFECYCLE_MAX_DELETE_BATCH_SIZE, + LIFECYCLE_MAX_DELETE_BATCH_SIZE + ), + storageLifecycleJobBudgetMs: envStrictPositiveInteger( + 'STORAGE_LIFECYCLE_JOB_BUDGET_MS', + 50_000, + MAX_TIMER_DELAY_MS + ), + storageLifecycleClaimLeaseMs: envStrictPositiveInteger( + 'STORAGE_LIFECYCLE_CLAIM_LEASE_MS', + 120_000, + MAX_TIMER_DELAY_MS + ), // Storage - File storageFilePath: getOptionalConfigFromEnv( @@ -929,6 +957,21 @@ export function getConfig(options?: { reload?: boolean }): StorageConfigType { ) } + if (config.storageLifecycleJobBudgetMs >= config.storageLifecycleClaimLeaseMs) { + throw new Error( + 'STORAGE_LIFECYCLE_JOB_BUDGET_MS must be shorter than STORAGE_LIFECYCLE_CLAIM_LEASE_MS' + ) + } + + if ( + config.storageLifecycleJobBudgetMs + config.storageLifecycleClaimLeaseMs > + 24 * 60 * 60_000 - 60_000 + ) { + throw new Error( + 'STORAGE_LIFECYCLE_JOB_BUDGET_MS plus STORAGE_LIFECYCLE_CLAIM_LEASE_MS must not exceed 23 hours 59 minutes' + ) + } + const serviceKey = getOptionalConfigFromEnv('SERVICE_KEY') || '' if (!config.isMultitenant && !serviceKey) { config.serviceKeyAsync = new SignJWT({ role: config.dbServiceRole }) @@ -1019,6 +1062,14 @@ function envBoundedPositiveInteger( return Number.isSafeInteger(parsed) && parsed > 0 && parsed <= maximum ? parsed : defaultValue } +function envStrictPositiveInteger(name: string, defaultValue: number, maximum: number): number { + const value = getOptionalConfigFromEnv(name)?.trim() + if (!value) return defaultValue + const parsed = envBoundedPositiveInteger(value, 0, maximum) + if (!parsed) throw new Error(`${name} must be a positive integer of at most ${maximum}`) + return parsed +} + function envNonNegativeInteger(value: string | undefined, defaultValue: number): number { const parsed = envNumber(value, defaultValue) diff --git a/src/internal/monitoring/metrics.ts b/src/internal/monitoring/metrics.ts index 3da60e608..85d07951a 100644 --- a/src/internal/monitoring/metrics.ts +++ b/src/internal/monitoring/metrics.ts @@ -496,6 +496,65 @@ 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 lifecycleVersionsExamined = registerMetric( + 'lifecycle_versions_examined_total', + 'counter', + () => + meter.createCounter('lifecycle_versions_examined_total', { + description: 'Historical version evaluations, including repeated evaluations of a version', + }) +) + +export const lifecycleVersionsEligible = registerMetric( + 'lifecycle_versions_eligible_total', + 'counter', + () => + meter.createCounter('lifecycle_versions_eligible_total', { + description: + 'Eligible historical version evaluations, including repeated evaluations of a version', + }) +) + +export const lifecycleObjectVersionsDeleted = registerMetric( + 'lifecycle_object_versions_deleted_total', + 'counter', + () => + meter.createCounter('lifecycle_object_versions_deleted_total', { + description: 'Acknowledged object-version metadata deletions by lifecycle', + }) +) + +export const lifecycleDeleteMarkersDeleted = registerMetric( + 'lifecycle_delete_markers_deleted_total', + 'counter', + () => + meter.createCounter('lifecycle_delete_markers_deleted_total', { + description: 'Acknowledged delete-marker metadata deletions by lifecycle', + }) +) + +export const lifecycleBytesDeleted = registerMetric( + 'lifecycle_bytes_deleted_total', + 'counter', + () => + meter.createCounter('lifecycle_bytes_deleted_total', { + description: 'Object sizes from acknowledged lifecycle metadata deletions', + }) +) + +export const lifecycleArtifactOutcomes = registerMetric( + 'lifecycle_artifact_outcomes_total', + 'counter', + () => + meter.createCounter('lifecycle_artifact_outcomes_total', { + description: 'Lifecycle artifact outcomes observed across backend attempts', + }) +) + // ============================================================================ // S3 Metrics // ============================================================================ diff --git a/src/storage/database/pg.ts b/src/storage/database/pg.ts index 96ef69a44..3fd5aa390 100644 --- a/src/storage/database/pg.ts +++ b/src/storage/database/pg.ts @@ -3565,6 +3565,7 @@ export class StoragePgDB implements Database { THEN clock_timestamp() + $4::bigint * interval '1 millisecond' ELSE NULL END, next_run_at = clock_timestamp(), + continuation = $14::jsonb, updated_at = clock_timestamp() FROM storage.buckets AS bucket, eligible_targets WHERE ${LIFECYCLE_CLAIM_IDENTITY} @@ -3588,6 +3589,7 @@ export class StoragePgDB implements Database { maximumNewerProbe, lockedIds, batch.versions.map((version) => version.id), + JSON.stringify(continuation), ], }, signal diff --git a/src/storage/lifecycle/abandoned-artifacts.test.ts b/src/storage/lifecycle/abandoned-artifacts.test.ts new file mode 100644 index 000000000..8c4eb1c49 --- /dev/null +++ b/src/storage/lifecycle/abandoned-artifacts.test.ts @@ -0,0 +1,221 @@ +import { logger, logSchema } from '@internal/monitoring' +import * as config from '../../config' +import type { Database } from '../database' +import type { Storage } from '../storage' +import { retryAbandonedLifecycleArtifacts } from './abandoned-artifacts' + +const attemptedBefore = '2026-08-15T00:00:00.000Z' + +function createHarness(outcomesByCall: Array> = []) { + let call = 0 + const database = { + tenantId: 'tenant', + listLifecycleAbandonedArtifacts: vi.fn().mockResolvedValue([]), + resolveLifecycleAbandonedArtifacts: vi.fn().mockResolvedValue(undefined), + } + const backend = { + deleteObjectsDetailed: vi + .fn() + .mockImplementation((_bucket: string, keys: string[], _options?: { signal: AbortSignal }) => { + const outcomes = outcomesByCall[call++] ?? [] + return keys.map((key, index) => ({ key, outcome: outcomes[index] ?? 'DELETED' })) + }), + } + const storage = { + backend, + location: { + getRootLocation: () => 'root', + getKeyLocation: ({ tenantId, bucketId, objectName }: Record) => + `${tenantId}/${bucketId}/${objectName}`, + }, + } as unknown as Storage + const retry = (signal?: AbortSignal, budgetMs = 60_000) => + retryAbandonedLifecycleArtifacts({ + storage, + database: database as unknown as Database, + attemptedBefore, + budgetMs, + signal, + }) + return { backend, database, retry } +} + +describe('retryAbandonedLifecycleArtifacts', () => { + it('deletes per-bucket keys, resolves outcomes and stops when a page deletes nothing', async () => { + const warning = vi.spyOn(logSchema, 'warning').mockImplementation(() => {}) + const harness = createHarness([['DELETED', 'FAILED'], ['FAILED']]) + harness.database.listLifecycleAbandonedArtifacts + .mockResolvedValueOnce([ + { bucketId: 'one', key: 'a/v1', live: false }, + { bucketId: 'two', key: 'b/v2.info', live: false }, + ]) + .mockResolvedValueOnce([{ bucketId: 'two', key: 'b/v2.info', live: false }]) + + await harness.retry() + + expect(harness.database.listLifecycleAbandonedArtifacts).toHaveBeenCalledTimes(2) + expect(harness.database.listLifecycleAbandonedArtifacts).toHaveBeenCalledWith( + attemptedBefore, + 1000 + ) + expect(harness.backend.deleteObjectsDetailed.mock.calls.map(([, keys]) => keys)).toEqual([ + ['tenant/one/a/v1', 'tenant/two/b/v2.info'], + ['tenant/two/b/v2.info'], + ]) + expect(harness.database.resolveLifecycleAbandonedArtifacts.mock.calls).toEqual([ + [ + [{ bucketId: 'one', key: 'a/v1', live: false }], + [{ bucketId: 'two', key: 'b/v2.info', live: false }], + ], + [[], [{ bucketId: 'two', key: 'b/v2.info', live: false }]], + ]) + expect(warning).toHaveBeenCalledTimes(2) + expect(warning).toHaveBeenCalledWith( + logger, + '[Lifecycle] Abandoned artifacts remain after retry', + { + type: 'event', + tenantId: 'tenant', + project: 'tenant', + metadata: JSON.stringify({ failed: 1 }), + } + ) + }) + + it('stops after the budget even when a page was deleted', async () => { + const clock = vi.spyOn(Date, 'now').mockReturnValue(1000) + const harness = createHarness() + harness.database.listLifecycleAbandonedArtifacts.mockResolvedValueOnce([ + { bucketId: 'one', key: 'a/v1', live: false }, + ]) + harness.backend.deleteObjectsDetailed.mockImplementationOnce( + async (_bucket, keys: string[]) => { + clock.mockReturnValue(2000) + return keys.map((key) => ({ key, outcome: 'DELETED' })) + } + ) + + await harness.retry(undefined, 1000) + + expect(harness.database.listLifecycleAbandonedArtifacts).toHaveBeenCalledOnce() + expect(harness.database.resolveLifecycleAbandonedArtifacts).toHaveBeenCalledOnce() + }) + + it('awaits the resolution of each page before reporting', async () => { + const harness = createHarness() + harness.database.listLifecycleAbandonedArtifacts.mockResolvedValueOnce([ + { bucketId: 'one', key: 'a/v1', live: false }, + ]) + const failure = new Error('resolve failed') + harness.database.resolveLifecycleAbandonedArtifacts.mockRejectedValueOnce(failure) + + await expect(harness.retry()).rejects.toBe(failure) + }) + + it('bounds each deletion by the remaining budget with a minimum send time', async () => { + vi.useFakeTimers({ toFake: ['Date'] }) + try { + const timeout = vi.spyOn(AbortSignal, 'timeout') + const harness = createHarness() + harness.database.listLifecycleAbandonedArtifacts + .mockResolvedValueOnce([{ bucketId: 'one', key: 'a/v1', live: false }]) + .mockResolvedValueOnce([{ bucketId: 'one', key: 'a/v2', live: false }]) + .mockResolvedValueOnce([{ bucketId: 'one', key: 'a/v3', live: false }]) + for (const advance of [20_000, 35_000]) { + harness.backend.deleteObjectsDetailed.mockImplementationOnce( + async (_bucket, keys: string[]) => { + vi.advanceTimersByTime(advance) + return keys.map((key) => ({ key, outcome: 'DELETED' })) + } + ) + } + + await harness.retry() + + expect(timeout.mock.calls.map(([ms]) => ms)).toEqual([60_000, 40_000, 10_000]) + } finally { + vi.useRealTimers() + } + }) + + it('never sends live entries to the backend and warns with the dropped count', async () => { + const warning = vi.spyOn(logSchema, 'warning').mockImplementation(() => {}) + const harness = createHarness() + harness.database.listLifecycleAbandonedArtifacts.mockResolvedValueOnce([ + { bucketId: 'one', key: 'a/v1', live: true }, + { bucketId: 'one', key: 'a/v1.info', live: true }, + { bucketId: 'two', key: 'b/v2', live: false }, + ]) + + await harness.retry() + + expect(harness.backend.deleteObjectsDetailed.mock.calls.map(([, keys]) => keys)).toEqual([ + ['tenant/two/b/v2'], + ]) + expect(harness.database.resolveLifecycleAbandonedArtifacts).toHaveBeenCalledExactlyOnceWith( + [{ bucketId: 'two', key: 'b/v2', live: false }], + [] + ) + expect(warning).toHaveBeenCalledExactlyOnceWith( + logger, + '[Lifecycle] Dropped abandoned artifacts owned by live objects', + { + type: 'event', + tenantId: 'tenant', + project: 'tenant', + metadata: JSON.stringify({ dropped: 2 }), + } + ) + }) + + it('continues to the next page when a page holds only live entries', async () => { + const warning = vi.spyOn(logSchema, 'warning').mockImplementation(() => {}) + const harness = createHarness() + harness.database.listLifecycleAbandonedArtifacts + .mockResolvedValueOnce([{ bucketId: 'one', key: 'a/v1', live: true }]) + .mockResolvedValueOnce([{ bucketId: 'one', key: 'b/v2', live: false }]) + + await harness.retry() + + expect(harness.database.listLifecycleAbandonedArtifacts).toHaveBeenCalledTimes(3) + expect(harness.backend.deleteObjectsDetailed.mock.calls.map(([, keys]) => keys)).toEqual([ + ['tenant/one/b/v2'], + ]) + expect(warning).toHaveBeenCalledOnce() + }) + + it('keeps paging after a page that only held keys beyond the S3 limit', async () => { + vi.spyOn(config, 'getConfig').mockReturnValue({ + ...config.getConfig(), + storageBackendType: 's3', + }) + const harness = createHarness() + harness.database.listLifecycleAbandonedArtifacts + .mockResolvedValueOnce([{ bucketId: 'one', key: 'a'.repeat(1100), live: false }]) + .mockResolvedValueOnce([{ bucketId: 'one', key: 'b/v2', live: false }]) + + await harness.retry() + + expect(harness.database.listLifecycleAbandonedArtifacts).toHaveBeenCalledTimes(3) + expect(harness.backend.deleteObjectsDetailed.mock.calls.map(([, keys]) => keys)).toEqual([ + ['tenant/one/b/v2'], + ]) + }) + + it('returns immediately when nothing is abandoned', async () => { + const harness = createHarness() + + await harness.retry() + + expect(harness.backend.deleteObjectsDetailed).not.toHaveBeenCalled() + expect(harness.database.resolveLifecycleAbandonedArtifacts).not.toHaveBeenCalled() + }) + + it('does nothing once aborted', async () => { + const harness = createHarness() + + await harness.retry(AbortSignal.abort()) + + expect(harness.database.listLifecycleAbandonedArtifacts).not.toHaveBeenCalled() + }) +}) diff --git a/src/storage/lifecycle/abandoned-artifacts.ts b/src/storage/lifecycle/abandoned-artifacts.ts new file mode 100644 index 000000000..95165466a --- /dev/null +++ b/src/storage/lifecycle/abandoned-artifacts.ts @@ -0,0 +1,55 @@ +import { logger, logSchema } from '@internal/monitoring' +import type { Database } from '../database' +import type { Storage } from '../storage' +import { deleteLifecycleArtifacts } from './artifact-deletion' + +const ABANDONED_ARTIFACT_PAGE_SIZE = 1000 +const ABANDONED_ARTIFACT_MIN_SEND_MS = 10_000 + +export async function retryAbandonedLifecycleArtifacts({ + storage, + database, + attemptedBefore, + budgetMs, + signal, +}: { + storage: Storage + database: Database + attemptedBefore: string + budgetMs: number + signal?: AbortSignal +}): Promise { + const deadline = Date.now() + budgetMs + while (Date.now() < deadline && !signal?.aborted) { + const page = await database.listLifecycleAbandonedArtifacts( + attemptedBefore, + ABANDONED_ARTIFACT_PAGE_SIZE + ) + if (page.length === 0) return + const artifacts = page.filter(({ live }) => !live) + if (artifacts.length < page.length) { + logSchema.warning(logger, '[Lifecycle] Dropped abandoned artifacts owned by live objects', { + type: 'event', + tenantId: database.tenantId, + project: database.tenantId, + metadata: JSON.stringify({ dropped: page.length - artifacts.length }), + }) + } + if (artifacts.length === 0) continue + const { absent, deleted, failed, error } = await deleteLifecycleArtifacts({ + storage, + database, + artifacts, + signal: AbortSignal.timeout(Math.max(ABANDONED_ARTIFACT_MIN_SEND_MS, deadline - Date.now())), + }) + if (failed > 0) { + logSchema.warning(logger, '[Lifecycle] Abandoned artifacts remain after retry', { + type: 'event', + tenantId: database.tenantId, + project: database.tenantId, + metadata: JSON.stringify({ failed, error }), + }) + } + if (absent + deleted === 0) return + } +} diff --git a/src/storage/lifecycle/artifact-deletion.test.ts b/src/storage/lifecycle/artifact-deletion.test.ts new file mode 100644 index 000000000..7d0c596ce --- /dev/null +++ b/src/storage/lifecycle/artifact-deletion.test.ts @@ -0,0 +1,147 @@ +import { lifecycleArtifactOutcomes } from '@internal/monitoring/metrics' +import * as config from '../../config' +import type { Database } from '../database' +import type { Storage } from '../storage' +import { deleteLifecycleArtifacts } from './artifact-deletion' + +vi.mock('@internal/monitoring/metrics', async (importOriginal) => ({ + ...(await importOriginal()), + lifecycleArtifactOutcomes: { add: vi.fn() }, +})) + +type Result = { key: string; outcome: string; error?: { message?: string; code?: string } } + +function setup(storageBackendType: 's3' | 'file', results?: (keys: string[]) => Result[]) { + vi.spyOn(config, 'getConfig').mockReturnValue({ ...config.getConfig(), storageBackendType }) + const backend = { + deleteObjectsDetailed: vi + .fn() + .mockImplementation((_bucket: string, keys: string[]) => + results ? results(keys) : keys.map((key) => ({ key, outcome: 'DELETED' })) + ), + } + const database = { tenantId: 'tenant', resolveLifecycleAbandonedArtifacts: vi.fn() } + const storage = { + backend, + location: { + getRootLocation: () => 'root', + getKeyLocation: ({ tenantId, bucketId, objectName }: Record) => + `${tenantId}/${bucketId}/${objectName}`, + }, + } as unknown as Storage + const run = ( + artifacts: Array<{ bucketId: string; key: string }>, + signal = new AbortController().signal + ) => + deleteLifecycleArtifacts({ + storage, + database: database as unknown as Database, + artifacts, + signal, + }) + return { backend, database, run } +} + +const artifact = (key: string) => ({ bucketId: 'one', key }) +const keyOfBytes = (bytes: number) => 'a'.repeat(bytes - 'tenant/one/'.length) + +describe('deleteLifecycleArtifacts', () => { + it('resolves S3 keys beyond the key limit as absent without sending them', async () => { + const { backend, database, run } = setup('s3') + const exact = artifact(keyOfBytes(1024)) + const over = artifact(keyOfBytes(1025)) + const short = artifact('a/v1') + + await expect(run([over, exact, short])).resolves.toMatchObject({ absent: 1, deleted: 2 }) + + expect(backend.deleteObjectsDetailed).toHaveBeenCalledExactlyOnceWith( + 'root', + [`tenant/one/${exact.key}`, 'tenant/one/a/v1'], + { signal: expect.any(AbortSignal) } + ) + expect(database.resolveLifecycleAbandonedArtifacts).toHaveBeenCalledExactlyOnceWith( + [over, exact, short], + [] + ) + expect(lifecycleArtifactOutcomes.add).toHaveBeenCalledWith(1, { outcome: 'ABSENT' }) + expect(lifecycleArtifactOutcomes.add).toHaveBeenCalledWith(2, { outcome: 'DELETED' }) + }) + + it('measures the S3 key limit in bytes, not characters', async () => { + const { backend, run } = setup('s3') + const multibyte = artifact('€'.repeat(342)) + + expect(multibyte.key.length + 'tenant/one/'.length).toBeLessThan(1024) + await expect(run([multibyte])).resolves.toMatchObject({ absent: 1, deleted: 0 }) + expect(backend.deleteObjectsDetailed).not.toHaveBeenCalled() + }) + + it('does not call the backend when every key is over the limit', async () => { + const { backend, run } = setup('s3') + + await expect(run([artifact(keyOfBytes(1025))])).resolves.toEqual({ + absent: 1, + deleted: 0, + failed: 0, + error: undefined, + }) + expect(backend.deleteObjectsDetailed).not.toHaveBeenCalled() + }) + + it('sends a file backend key beyond the S3 limit', async () => { + const { backend, run } = setup('file') + const long = artifact(keyOfBytes(1025)) + const { signal } = new AbortController() + + await expect(run([long], signal)).resolves.toMatchObject({ absent: 0, deleted: 1 }) + expect(backend.deleteObjectsDetailed).toHaveBeenCalledExactlyOnceWith( + 'root', + [`tenant/one/${long.key}`], + { signal } + ) + }) + + it.each([ + { label: 'swapped', results: (keys: string[]) => [...keys].reverse(), deleted: 0 }, + { label: 'truncated', results: (keys: string[]) => keys.slice(0, 1), deleted: 1 }, + { label: 'mismatched', results: (keys: string[]) => [keys[0], 'other'], deleted: 1 }, + ])('counts $label results as failed by key position', async ({ results, deleted }) => { + const { database, run } = setup('file', (keys) => + results(keys).map((key) => ({ key, outcome: 'DELETED' })) + ) + const [first, second] = [artifact('a'), artifact('b')] + + await expect(run([first, second])).resolves.toMatchObject({ deleted, failed: 2 - deleted }) + expect(database.resolveLifecycleAbandonedArtifacts).toHaveBeenCalledExactlyOnceWith( + deleted ? [first] : [], + deleted ? [second] : [first, second] + ) + }) + + it.each(['FAILED', 'UNKNOWN'])('counts %s as failed with a FAILED outcome', async (outcome) => { + const { run } = setup('file', (keys) => keys.map((key) => ({ key, outcome }))) + + await expect(run([artifact('a')])).resolves.toMatchObject({ failed: 1, deleted: 0 }) + expect(lifecycleArtifactOutcomes.add).toHaveBeenCalledExactlyOnceWith(1, { outcome: 'FAILED' }) + }) + + it.each([ + { error: { message: 'denied', code: 'AccessDenied' }, expected: 'denied' }, + { error: { message: '', code: 'AccessDenied' }, expected: 'AccessDenied' }, + ])('reports the first failed result error $expected', async ({ error, expected }) => { + const { run } = setup('file', (keys) => + keys.map((key, index) => ({ + key, + outcome: 'FAILED', + error: index === 0 ? error : { message: 'second' }, + })) + ) + + await expect(run([artifact('a'), artifact('b')])).resolves.toEqual({ + absent: 0, + deleted: 0, + failed: 2, + error: expected, + }) + }) +}) diff --git a/src/storage/lifecycle/artifact-deletion.ts b/src/storage/lifecycle/artifact-deletion.ts new file mode 100644 index 000000000..1608df0f1 --- /dev/null +++ b/src/storage/lifecycle/artifact-deletion.ts @@ -0,0 +1,66 @@ +import { lifecycleArtifactOutcomes } from '@internal/monitoring/metrics' +import { getConfig } from '../../config' +import type { Database } from '../database' +import type { Storage } from '../storage' + +const MAX_S3_KEY_BYTES = 1024 + +export async function deleteLifecycleArtifacts({ + storage, + database, + artifacts, + signal, +}: { + storage: Storage + database: Database + artifacts: Array<{ bucketId: string; key: string }> + signal: AbortSignal +}): Promise<{ absent: number; deleted: number; failed: number; error?: string }> { + const isS3Backend = getConfig().storageBackendType === 's3' + const absent: typeof artifacts = [] + const sendable: typeof artifacts = [] + const physicalKeys: string[] = [] + for (const artifact of artifacts) { + const physicalKey = storage.location.getKeyLocation({ + tenantId: database.tenantId, + bucketId: artifact.bucketId, + objectName: artifact.key, + }) + // A key beyond the S3 limit cannot exist and the backend rejects it on every attempt. + if (isS3Backend && Buffer.byteLength(physicalKey) > MAX_S3_KEY_BYTES) { + absent.push(artifact) + } else { + sendable.push(artifact) + physicalKeys.push(physicalKey) + } + } + const results = + physicalKeys.length === 0 + ? [] + : await storage.backend.deleteObjectsDetailed( + storage.location.getRootLocation(), + physicalKeys, + { signal } + ) + const deleted: typeof artifacts = [] + const failed: typeof artifacts = [] + let error: string | undefined + sendable.forEach((artifact, index) => { + const result = results[index] + if (result?.key === physicalKeys[index] && result.outcome === 'DELETED') { + deleted.push(artifact) + return + } + failed.push(artifact) + error ??= result?.error?.message || result?.error?.code + }) + await database.resolveLifecycleAbandonedArtifacts([...absent, ...deleted], failed) + for (const [outcome, count] of [ + ['DELETED', deleted.length], + ['ABSENT', absent.length], + ['FAILED', failed.length], + ] as const) { + if (count) lifecycleArtifactOutcomes.add(count, { outcome }) + } + return { absent: absent.length, deleted: deleted.length, failed: failed.length, error } +} diff --git a/src/storage/lifecycle/executor.test.ts b/src/storage/lifecycle/executor.test.ts new file mode 100644 index 000000000..afd280ed6 --- /dev/null +++ b/src/storage/lifecycle/executor.test.ts @@ -0,0 +1,864 @@ +import { randomUUID } from 'node:crypto' +import { ERRORS } from '@internal/errors' +import { logSchema } from '@internal/monitoring' +import { + lifecycleBytesDeleted, + lifecycleDeleteMarkersDeleted, + lifecycleObjectVersionsDeleted, + lifecycleVersionsEligible, + lifecycleVersionsExamined, +} from '@internal/monitoring/metrics' +import { newContinuation } from '../../test/utils/lifecycle-continuation' +import type { Database } from '../database' +import type { + LifecycleBatch, + LifecycleBucket, + LifecycleContinuation, + LifecycleEvaluatedCandidate, + LifecycleState, +} from '../schemas' +import type { Storage } from '../storage' +import { advanceLifecycleBatch } from './continuation' +import { evaluationJitterMs, lifecycleErrorMessage, NoncurrentLifecycleExecutor } from './executor' + +vi.mock('@internal/monitoring/metrics', async (importOriginal) => ({ + ...(await importOriginal()), + lifecycleBytesDeleted: { add: vi.fn() }, + lifecycleDeleteMarkersDeleted: { add: vi.fn() }, + lifecycleObjectVersionsDeleted: { add: vi.fn() }, + lifecycleVersionsEligible: { add: vi.fn() }, + lifecycleVersionsExamined: { add: vi.fn() }, +})) + +const DAY_MS = 24 * 60 * 60 * 1000 +const now = new Date('2026-08-15T00:00:00.000Z') +const pageEnd = { name: 'folder/object.txt', archivedAt: '2026-07-02T00:00:00.000Z' } +const unfinishedPage = { + rawRowsExamined: 1, + candidates: [], + pageEnd: { name: 'folder/skipped.txt', archivedAt: '2026-07-02T00:00:00.000Z' }, + exhausted: false, +} +const emptyPage = { rawRowsExamined: 0, candidates: [], exhausted: true } + +function configuration(generation: string) { + return { + id: 'bucket', + name: 'bucket', + type: 'STANDARD' as const, + versioning_status: 'ENABLED' as const, + lifecycle_configuration_generation: generation, + lifecycle_configuration: { + rules: [ + { + status: 'Enabled' as const, + filter: {}, + noncurrentVersionExpiration: { noncurrentDays: 1 }, + }, + ], + }, + } +} + +function candidate(): LifecycleEvaluatedCandidate { + return { + id: randomUUID(), + name: 'folder/object.txt', + version: randomUUID(), + isDeleteMarker: false, + archivedAt: '2026-07-02T00:00:00.000Z', + rawRowNumber: 1, + } +} + +function fullPage(count: number) { + const candidates = Array.from({ length: count }, (_, index) => ({ + ...candidate(), + name: `object-${String(index).padStart(2, '0')}`, + rawRowNumber: index + 1, + })) + return { + rawRowsExamined: candidates.length, + candidates, + pageEnd: { name: candidates.at(-1)!.name, archivedAt: candidates.at(-1)!.archivedAt }, + exhausted: true, + } +} + +function state(continuation: LifecycleContinuation | null = null): LifecycleState { + return { + bucketId: 'bucket', + claimId: randomUUID(), + continuation, + failureCount: 0, + } +} + +function outboxOf(batch: LifecycleBatch) { + return batch.versions.flatMap(({ name, version, artifacts }) => + artifacts.map((key) => ({ bucketId: 'bucket', key, name, version })) + ) +} + +function createHarness(input: { + continuation?: LifecycleContinuation + deleteOutcome?: 'DELETED' | 'FAILED' | 'UNKNOWN' + failureCount?: number + jobBudgetMs?: number + now?: () => Date +}) { + const generation = input.continuation?.generation ?? randomUUID() + let persisted = input.continuation ?? null + let bucket: LifecycleBucket = configuration(generation) + const claimed = state(persisted) + claimed.failureCount = input.failureCount ?? 0 + const pageCandidate = candidate() + const calls: string[] = [] + const database = { + tenantId: 'tenant', + hasMigration: vi.fn().mockResolvedValue(true), + asSuperUser: vi.fn(() => database), + claimLifecycleBucket: vi + .fn() + .mockImplementation(() => ({ ...claimed, continuation: persisted })), + findLifecycleBucket: vi.fn().mockImplementation(() => bucket), + evaluateNoncurrentLifecyclePage: vi.fn().mockResolvedValue({ + rawRowsExamined: 1, + candidates: [pageCandidate], + pageEnd: { + name: pageCandidate.name, + archivedAt: pageCandidate.archivedAt, + }, + exhausted: true, + }), + armLifecycleAttempt: vi + .fn() + .mockImplementation( + (attempt: { continuation: LifecycleContinuation; batch: LifecycleBatch }) => { + const objectVersions = attempt.batch.versions.filter( + (version) => version.artifacts.length > 0 + ).length + persisted = advanceLifecycleBatch(attempt.continuation, attempt.batch, { + objectVersions, + deleteMarkers: attempt.batch.versions.length - objectVersions, + bytes: objectVersions * 12, + }) + calls.push('arm') + return { + status: 'armed' as const, + state: { ...claimed, continuation: persisted }, + artifacts: outboxOf(attempt.batch), + } + } + ), + revalidateLifecycleClaim: vi.fn().mockImplementation(() => ({ + ...claimed, + continuation: persisted, + })), + resolveLifecycleAbandonedArtifacts: vi.fn().mockImplementation(() => { + calls.push('resolve') + }), + releaseLifecycleClaim: vi.fn().mockImplementation((input) => { + if (input.continuation !== undefined) persisted = input.continuation + if (input.error) claimed.failureCount++ + return true + }), + completeLifecycleRun: vi.fn().mockImplementation(() => { + calls.push('complete') + return true + }), + } + const backend = { + deleteObjectsDetailed: vi.fn().mockImplementation((_bucket: string, keys: string[]) => { + calls.push('delete') + return keys.map((key) => { + const outcome = input.deleteOutcome ?? 'DELETED' + return { + key, + outcome, + ...(outcome === 'DELETED' ? {} : { error: { message: outcome.toLowerCase() } }), + } + }) + }), + } + const storage = { + backend, + db: database as unknown as Database, + location: { + getRootLocation: () => 'root', + getKeyLocation: ({ tenantId, bucketId, objectName }: Record) => + `${tenantId}/${bucketId}/${objectName}`, + }, + } as unknown as Storage + const executor = new NoncurrentLifecycleExecutor(storage, { + now: input.now ?? (() => now), + pageSize: 480, + jobBudgetMs: input.jobBudgetMs ?? 20_000, + claimLeaseMs: 120_000, + fairnessDelayMs: 1000, + }) + return { + backend, + calls, + database, + executor, + run: () => executor.run(bucketId, { tenantLifecycleEnabled: true }), + generation, + pageCandidate, + setBucket: (next: LifecycleBucket) => { + bucket = next + }, + getPersisted: () => persisted, + } +} + +const bucketId = 'bucket' + +test.each([false, true])( + 'preserves a claim decoding failure when release fails=%s', + async (releaseFails) => { + const harness = createHarness({}) + const failure = new TypeError('Unsupported lifecycle continuation version') + const releaseFailure = new Error('database offline') + harness.database.claimLifecycleBucket.mockRejectedValueOnce(failure) + if (releaseFails) harness.database.releaseLifecycleClaim.mockRejectedValueOnce(releaseFailure) + const warning = vi.spyOn(logSchema, 'warning').mockImplementation(() => {}) + + await expect(harness.run()).rejects.toBe(failure) + expect(harness.database.releaseLifecycleClaim).toHaveBeenCalledWith( + expect.objectContaining({ + bucketId, + claimId: expect.any(String), + nextRunAt: expect.any(String), + error: { name: 'TypeError', message: failure.message }, + }) + ) + expect(harness.database.releaseLifecycleClaim.mock.calls[0][0]).not.toHaveProperty( + 'continuation' + ) + if (releaseFails) { + expect(warning).toHaveBeenCalledWith( + expect.anything(), + expect.stringContaining('release lifecycle claim'), + expect.objectContaining({ + error: releaseFailure, + tenantId: 'tenant', + metadata: JSON.stringify({ + bucketId, + claimId: harness.database.releaseLifecycleClaim.mock.calls[0][0].claimId, + }), + }) + ) + } else { + expect(warning).not.toHaveBeenCalled() + } + } +) + +it('truncates error messages without splitting a surrogate pair', () => { + const message = lifecycleErrorMessage(new Error(`${'a'.repeat(999)}😀tail`)) + expect(message.isWellFormed()).toBe(true) + expect(message.length).toBeLessThanOrEqual(1000) +}) + +describe('NoncurrentLifecycleExecutor', () => { + it('counts a committed deletion once even when the backend deletion fails', async () => { + const versions = vi.spyOn(lifecycleObjectVersionsDeleted, 'add') + const bytes = vi.spyOn(lifecycleBytesDeleted, 'add') + const harness = createHarness({ deleteOutcome: 'FAILED' }) + + await harness.run() + + expect(versions).toHaveBeenCalledExactlyOnceWith(1) + expect(bytes).toHaveBeenCalledExactlyOnceWith(12) + }) + + it('does not export a deletion when its metadata commit is not acknowledged', async () => { + const versions = vi.spyOn(lifecycleObjectVersionsDeleted, 'add') + const bytes = vi.spyOn(lifecycleBytesDeleted, 'add') + const harness = createHarness({}) + harness.database.armLifecycleAttempt.mockRejectedValueOnce(new Error('lost acknowledgement')) + await expect(harness.run()).rejects.toThrow('lost acknowledgement') + expect(versions).not.toHaveBeenCalled() + expect(bytes).not.toHaveBeenCalled() + }) + + it('exports only the arm delta of a delete marker after earlier batches', async () => { + const initial = newContinuation(randomUUID()) + initial.counters.objectVersionsDeleted = 5 + initial.counters.deleteMarkersDeleted = 3 + initial.counters.bytesDeleted = 60 + const harness = createHarness({ continuation: initial }) + harness.database.evaluateNoncurrentLifecyclePage.mockResolvedValue({ + rawRowsExamined: 1, + candidates: [{ ...candidate(), isDeleteMarker: true }], + pageEnd, + exhausted: true, + }) + + await harness.run() + + expect(lifecycleObjectVersionsDeleted.add).toHaveBeenCalledExactlyOnceWith(0) + expect(lifecycleDeleteMarkersDeleted.add).toHaveBeenCalledExactlyOnceWith(1) + expect(lifecycleBytesDeleted.add).toHaveBeenCalledExactlyOnceWith(0) + expect(harness.backend.deleteObjectsDetailed).not.toHaveBeenCalled() + expect(harness.database.resolveLifecycleAbandonedArtifacts).not.toHaveBeenCalled() + }) + + test('does not claim work before writer coordination is installed', async () => { + const harness = createHarness({}) + harness.database.hasMigration.mockResolvedValue(false) + await expect(harness.run()).resolves.toEqual({ status: 'PAUSED' }) + expect(harness.database.hasMigration).toHaveBeenCalledExactlyOnceWith( + 'lifecycle-writer-coordination' + ) + expect(harness.database.claimLifecycleBucket).not.toHaveBeenCalled() + expect(harness.backend.deleteObjectsDetailed).not.toHaveBeenCalled() + }) + + it.each([ + { state: 'versioning disabled', change: { versioning_status: 'DISABLED' as const } }, + { state: 'no configuration', change: { lifecycle_configuration: null } }, + { state: 'no generation', change: { lifecycle_configuration_generation: null } }, + ])('pauses without evaluating when the claimed bucket has $state', async ({ change }) => { + const continuation = newContinuation(randomUUID()) + const harness = createHarness({ continuation }) + harness.setBucket({ ...configuration(harness.generation), ...change }) + + await expect(harness.run()).resolves.toEqual({ status: 'PAUSED', runId: continuation.runId }) + expect(harness.database.releaseLifecycleClaim).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ continuation: null, nextRunAt: null }) + ) + + expect(harness.database.evaluateNoncurrentLifecyclePage).not.toHaveBeenCalled() + expect(harness.backend.deleteObjectsDetailed).not.toHaveBeenCalled() + }) + + it('pauses under the closed tenant gate without changing its continuation', async () => { + const continuation = { ...newContinuation(randomUUID()), cursor: pageEnd } + const harness = createHarness({ continuation }) + + await expect( + harness.executor.run(bucketId, { tenantLifecycleEnabled: false }) + ).resolves.toEqual({ + status: 'PAUSED', + runId: continuation.runId, + }) + + expect(harness.database.releaseLifecycleClaim).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ + continuation, + nextRunAt: new Date(now.getTime() + 1000).toISOString(), + }) + ) + expect(harness.database.findLifecycleBucket).not.toHaveBeenCalled() + expect(harness.database.evaluateNoncurrentLifecyclePage).not.toHaveBeenCalled() + expect(harness.database.armLifecycleAttempt).not.toHaveBeenCalled() + expect(harness.backend.deleteObjectsDetailed).not.toHaveBeenCalled() + }) + + it('schedules the next evaluation a day out plus a deterministic jitter below a tenth of a day', async () => { + const harness = createHarness({}) + + await harness.run() + + const [identity, result] = harness.database.completeLifecycleRun.mock.calls[0] + expect(identity.generation).toBe(result.generation) + expect(identity.generation).toBeTruthy() + const nextRunAt = new Date(harness.database.completeLifecycleRun.mock.calls[0][2]).getTime() + const day = 24 * 60 * 60 * 1000 + expect(nextRunAt).toBeGreaterThan(now.getTime() + day) + expect(nextRunAt).toBeLessThan(now.getTime() + day + day / 10) + }) + + it('completes with the run identity and counters and schedules from the completion clock', async () => { + let clock = now.getTime() + const harness = createHarness({ now: () => new Date(clock) }) + harness.backend.deleteObjectsDetailed.mockImplementationOnce((_bucket, keys: string[]) => { + clock += 3000 + return keys.map((key) => ({ key, outcome: 'DELETED' })) + }) + + await expect(harness.run()).resolves.toMatchObject({ + status: 'COMPLETED', + runId: expect.any(String), + }) + + const armed = harness.getPersisted()! + const [identity, result, nextRunAt] = harness.database.completeLifecycleRun.mock.calls[0] + expect(identity).toEqual({ + bucketId, + claimId: harness.database.claimLifecycleBucket.mock.calls[0][0].claimId, + generation: armed.generation, + }) + expect(harness.database.armLifecycleAttempt.mock.calls[0][0].claimId).toBe(identity.claimId) + expect(result).toEqual({ + runId: armed.runId, + generation: armed.generation, + snapshotAt: now.toISOString(), + counters: armed.counters, + completedAt: new Date(clock).toISOString(), + }) + expect(nextRunAt).toBe( + new Date(clock + DAY_MS + evaluationJitterMs(`tenant/${bucketId}`)).toISOString() + ) + }) + + it('reports NOT_CLAIMED without releasing when nothing can be claimed', async () => { + const harness = createHarness({}) + harness.database.claimLifecycleBucket.mockResolvedValueOnce(undefined) + + await expect(harness.run()).resolves.toEqual({ status: 'NOT_CLAIMED' }) + expect(harness.database.claimLifecycleBucket).toHaveBeenCalledExactlyOnceWith({ + bucketId, + claimId: expect.any(String), + leaseMs: 120_000, + }) + expect(harness.database.releaseLifecycleClaim).not.toHaveBeenCalled() + }) + + it('reports a lost claim without releasing when the bucket was deleted mid-run', async () => { + const harness = createHarness({}) + harness.database.findLifecycleBucket.mockRejectedValueOnce(ERRORS.NoSuchBucket(bucketId)) + + await expect(harness.run()).resolves.toEqual({ status: 'LOST_CLAIM' }) + expect(harness.database.releaseLifecycleClaim).not.toHaveBeenCalled() + }) + + it('caps the persisted backend delete error', async () => { + const harness = createHarness({ deleteOutcome: 'FAILED' }) + harness.backend.deleteObjectsDetailed.mockImplementation((_bucket: string, keys: string[]) => + keys.map((key) => ({ key, outcome: 'FAILED', error: { message: 'x'.repeat(5000) } })) + ) + + const warning = vi.spyOn(logSchema, 'warning').mockImplementation(() => {}) + + await harness.run() + + expect(harness.database.releaseLifecycleClaim.mock.calls[0][0].error.message).toHaveLength(1000) + expect(warning.mock.calls[0][2]).toMatchObject({ tenantId: 'tenant', project: 'tenant' }) + const logged = JSON.parse(warning.mock.calls[0][2].metadata as string) + expect(logged.error).toHaveLength(1000) + expect(logged).toMatchObject({ bucketId, failedArtifacts: 2 }) + }) + + it('reports an ineligible batch without revalidation or backend deletion', async () => { + const harness = createHarness({}) + harness.database.armLifecycleAttempt.mockResolvedValueOnce({ status: 'ineligible' }) + + await expect(harness.run()).resolves.toMatchObject({ + status: 'INELIGIBLE', + runId: expect.any(String), + }) + expect(harness.database.revalidateLifecycleClaim).not.toHaveBeenCalled() + expect(harness.database.releaseLifecycleClaim).not.toHaveBeenCalled() + expect(harness.backend.deleteObjectsDetailed).not.toHaveBeenCalled() + }) + + it.each(['lost', 'unchanged'] as const)( + 'handles an arm rejection with a %s claim', + async (claim) => { + const harness = createHarness({}) + harness.database.armLifecycleAttempt.mockResolvedValueOnce({ status: 'fenced' }) + if (claim === 'lost') + harness.database.revalidateLifecycleClaim.mockResolvedValueOnce(undefined) + + const run = harness.run() + if (claim === 'unchanged') { + await expect(run).rejects.toThrow('Lifecycle arm fence rejected') + } else { + await expect(run).resolves.toMatchObject({ + status: 'LOST_CLAIM', + runId: expect.any(String), + }) + } + expect(harness.database.armLifecycleAttempt).toHaveBeenCalledOnce() + expect(harness.database.revalidateLifecycleClaim).toHaveBeenCalledOnce() + expect(harness.backend.deleteObjectsDetailed).not.toHaveBeenCalled() + if (claim === 'lost') { + expect(harness.database.releaseLifecycleClaim).not.toHaveBeenCalled() + } else { + expect(harness.database.releaseLifecycleClaim).toHaveBeenCalledWith( + expect.objectContaining({ + nextRunAt: expect.any(String), + error: expect.any(Object), + }) + ) + } + } + ) + + it('restarts from a fresh continuation over a run of a changed configuration generation', async () => { + const initial = { ...newContinuation(randomUUID()), cursor: pageEnd } + const harness = createHarness({ continuation: initial }) + const generation = randomUUID() + harness.setBucket(configuration(generation)) + + await expect(harness.run()).resolves.toMatchObject({ status: 'COMPLETED' }) + expect(harness.database.findLifecycleBucket).toHaveBeenCalledOnce() + expect(harness.database.evaluateNoncurrentLifecyclePage.mock.calls[0][0].cursor).toBeUndefined() + const armed = harness.database.armLifecycleAttempt.mock.calls[0][0].continuation + expect(armed.generation).toBe(generation) + expect(armed.runId).not.toBe(initial.runId) + expect(armed.snapshotAt).toBe(now.toISOString()) + expect(harness.database.evaluateNoncurrentLifecyclePage.mock.calls[0][0].snapshotAt).toBe( + now.toISOString() + ) + }) + + it('pauses instead of failing when the configuration is regenerated between evaluation and arm', async () => { + const harness = createHarness({}) + const bucket = configuration(harness.generation) + harness.setBucket(bucket) + harness.database.findLifecycleBucket.mockResolvedValueOnce(bucket).mockResolvedValueOnce({ + ...bucket, + lifecycle_configuration_generation: randomUUID(), + }) + harness.database.armLifecycleAttempt.mockResolvedValueOnce({ status: 'fenced' }) + + await expect(harness.run()).resolves.toEqual({ + status: 'PAUSED', + runId: expect.any(String), + }) + expect(harness.database.findLifecycleBucket).toHaveBeenCalledTimes(2) + expect(harness.database.releaseLifecycleClaim).toHaveBeenCalledOnce() + expect(harness.database.releaseLifecycleClaim.mock.calls[0][0]).not.toHaveProperty('error') + expect(harness.database.releaseLifecycleClaim.mock.calls[0][0].continuation).toBeNull() + expect(harness.database.releaseLifecycleClaim.mock.calls[0][0].nextRunAt).toBe( + new Date(now.getTime() + 1000).toISOString() + ) + expect(harness.backend.deleteObjectsDetailed).not.toHaveBeenCalled() + }) + + it('bounds the backend delete and arms with the claim lease', async () => { + const timeout = vi.spyOn(AbortSignal, 'timeout') + const harness = createHarness({}) + + await harness.run() + + expect(timeout).toHaveBeenCalledWith(120_000) + expect(harness.database.armLifecycleAttempt.mock.calls[0][0].leaseMs).toBe(120_000) + }) + + it('records the examined and eligible versions of each evaluated page', async () => { + const harness = createHarness({}) + harness.database.evaluateNoncurrentLifecyclePage.mockResolvedValueOnce({ + ...fullPage(1), + rawRowsExamined: 5, + }) + + await harness.run() + + expect(lifecycleVersionsExamined.add).toHaveBeenCalledExactlyOnceWith(5) + expect(lifecycleVersionsEligible.add).toHaveBeenCalledExactlyOnceWith(1) + }) + + it('stamps completion with the clock at completion rather than the snapshot', async () => { + let tick = 0 + const harness = createHarness({ + now: () => new Date(now.getTime() + tick++ * 1000), + }) + + await harness.run() + + const completion = harness.database.completeLifecycleRun.mock.calls[0][1] + expect(Date.parse(completion.completedAt)).toBeGreaterThan(Date.parse(completion.snapshotAt)) + }) + + it('releases idle without scanning when no rule compiles', async () => { + const continuation = newContinuation(randomUUID()) + const harness = createHarness({ continuation }) + const bucket = configuration(harness.generation) + harness.setBucket({ + ...bucket, + lifecycle_configuration: { + rules: [{ ...bucket.lifecycle_configuration.rules[0], status: 'Disabled' as const }], + }, + }) + + await expect(harness.run()).resolves.toEqual({ status: 'PAUSED', runId: continuation.runId }) + expect(harness.database.releaseLifecycleClaim).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ continuation: null, nextRunAt: null }) + ) + expect(harness.database.evaluateNoncurrentLifecyclePage).not.toHaveBeenCalled() + }) + + it('advances the cursor past a non-exhausted empty page and continues', async () => { + const harness = createHarness({}) + const pageEnd = { name: 'folder/skipped.txt', archivedAt: '2026-07-02T00:00:00.000Z' } + harness.database.evaluateNoncurrentLifecyclePage.mockResolvedValueOnce({ + rawRowsExamined: 1, + candidates: [], + pageEnd, + exhausted: false, + }) + + await expect(harness.run()).resolves.toMatchObject({ status: 'COMPLETED' }) + expect(harness.database.evaluateNoncurrentLifecyclePage).toHaveBeenCalledTimes(2) + expect(harness.database.evaluateNoncurrentLifecyclePage.mock.calls[1][0].cursor).toEqual( + pageEnd + ) + }) + + it('sends only the artifacts the arm returns when it drops vetoed versions', async () => { + const harness = createHarness({}) + const kept = harness.pageCandidate + harness.database.evaluateNoncurrentLifecyclePage.mockResolvedValueOnce({ + rawRowsExamined: 2, + candidates: [{ ...candidate(), name: 'folder/vetoed.txt' }, kept], + pageEnd: { name: kept.name, archivedAt: kept.archivedAt }, + exhausted: true, + }) + harness.database.armLifecycleAttempt.mockImplementationOnce(({ continuation, batch }) => ({ + status: 'armed', + state: { ...state(), continuation }, + artifacts: outboxOf({ ...batch, versions: batch.versions.slice(1) }), + })) + + await expect(harness.run()).resolves.toMatchObject({ status: 'COMPLETED' }) + expect(harness.backend.deleteObjectsDetailed).toHaveBeenCalledExactlyOnceWith( + 'root', + [ + `tenant/bucket/${kept.name}/${kept.version}`, + `tenant/bucket/${kept.name}/${kept.version}.info`, + ], + { signal: expect.any(AbortSignal) } + ) + }) + + it('continues with the next page without a backend call when the arm deletes nothing', async () => { + const harness = createHarness({}) + harness.database.evaluateNoncurrentLifecyclePage + .mockResolvedValueOnce({ + rawRowsExamined: 1, + candidates: [harness.pageCandidate], + pageEnd, + exhausted: false, + }) + .mockResolvedValueOnce(emptyPage) + harness.database.armLifecycleAttempt.mockImplementationOnce(({ continuation, batch }) => ({ + status: 'armed', + state: { + ...state(), + continuation: advanceLifecycleBatch(continuation, batch, { + objectVersions: 0, + deleteMarkers: 0, + bytes: 0, + }), + }, + artifacts: [], + })) + + await expect(harness.run()).resolves.toMatchObject({ status: 'COMPLETED' }) + expect(harness.backend.deleteObjectsDetailed).not.toHaveBeenCalled() + expect(harness.database.resolveLifecycleAbandonedArtifacts).not.toHaveBeenCalled() + expect(harness.database.evaluateNoncurrentLifecyclePage).toHaveBeenCalledTimes(2) + expect(harness.database.evaluateNoncurrentLifecyclePage.mock.calls[1][0].cursor).toEqual( + pageEnd + ) + }) + + it('reports a lost claim when completing the run is rejected', async () => { + const harness = createHarness({}) + harness.database.completeLifecycleRun.mockResolvedValueOnce(false) + + await expect(harness.run()).resolves.toMatchObject({ status: 'LOST_CLAIM' }) + }) + + it.each(['idle', 'tenant gate', 'budget', 'incomplete deletion'] as const)( + 'reports a lost claim when release fails during %s', + async (path) => { + const harness = createHarness({ + jobBudgetMs: path === 'budget' ? 0 : undefined, + deleteOutcome: path === 'incomplete deletion' ? 'FAILED' : undefined, + }) + if (path === 'idle') { + harness.setBucket({ ...configuration(harness.generation), lifecycle_configuration: null }) + } + if (path === 'budget') { + harness.database.evaluateNoncurrentLifecyclePage.mockResolvedValueOnce(unfinishedPage) + } + harness.database.releaseLifecycleClaim.mockResolvedValueOnce(false) + + await expect( + harness.executor.run(bucketId, { tenantLifecycleEnabled: path !== 'tenant gate' }) + ).resolves.toMatchObject({ status: 'LOST_CLAIM' }) + } + ) + + it('releases a run that exhausts its job budget after the fairness delay', async () => { + const harness = createHarness({ jobBudgetMs: 0 }) + harness.database.evaluateNoncurrentLifecyclePage.mockResolvedValueOnce(unfinishedPage) + + await expect(harness.run()).resolves.toMatchObject({ + status: 'PARTIAL', + runId: expect.any(String), + }) + expect(harness.database.releaseLifecycleClaim).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ + continuation: expect.objectContaining({ cursor: unfinishedPage.pageEnd }), + nextRunAt: new Date(now.getTime() + 1000).toISOString(), + }) + ) + expect(harness.database.evaluateNoncurrentLifecyclePage).toHaveBeenCalledOnce() + }) + + it('processes the first batch when evaluation consumes the remaining budget', async () => { + let clock = now.getTime() + const harness = createHarness({ jobBudgetMs: 1000, now: () => new Date(clock) }) + const evaluate = harness.database.evaluateNoncurrentLifecyclePage.getMockImplementation()! + harness.database.evaluateNoncurrentLifecyclePage.mockImplementationOnce(async (...args) => { + const page = await evaluate(...args) + clock += 1000 + return page + }) + + await expect(harness.run()).resolves.toMatchObject({ status: 'COMPLETED' }) + expect(harness.database.armLifecycleAttempt).toHaveBeenCalledOnce() + expect(harness.backend.deleteObjectsDetailed).toHaveBeenCalledOnce() + }) + + it('stops starting new batches once the signal aborts and releases as PARTIAL', async () => { + const controller = new AbortController() + const harness = createHarness({}) + const page = { + rawRowsExamined: 1, + candidates: [harness.pageCandidate], + pageEnd: { name: harness.pageCandidate.name, archivedAt: harness.pageCandidate.archivedAt }, + exhausted: false, + } + harness.database.evaluateNoncurrentLifecyclePage + .mockResolvedValueOnce(page) + .mockResolvedValue(emptyPage) + harness.database.resolveLifecycleAbandonedArtifacts.mockImplementationOnce(() => { + controller.abort() + }) + + await expect( + harness.executor.run(bucketId, { + tenantLifecycleEnabled: true, + signal: controller.signal, + }) + ).resolves.toMatchObject({ status: 'PARTIAL' }) + expect(harness.database.armLifecycleAttempt).toHaveBeenCalledOnce() + expect(harness.database.evaluateNoncurrentLifecyclePage).toHaveBeenCalledOnce() + expect(harness.database.releaseLifecycleClaim).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ nextRunAt: new Date(now.getTime() + 1000).toISOString() }) + ) + }) + + it.each([ + { deleteOutcome: 'FAILED', failureCount: 0, delayMinutes: 5 }, + { deleteOutcome: 'UNKNOWN', failureCount: 6, delayMinutes: 15 }, + ] as const)( + 'stops after a $deleteOutcome backend key without arming another batch and backs off', + async ({ deleteOutcome, failureCount, delayMinutes }) => { + const harness = createHarness({ deleteOutcome, failureCount }) + harness.database.evaluateNoncurrentLifecyclePage.mockResolvedValueOnce(fullPage(49)) + + await expect(harness.run()).resolves.toMatchObject({ status: 'PARTIAL' }) + expect(harness.database.armLifecycleAttempt).toHaveBeenCalledOnce() + expect(harness.database.completeLifecycleRun).not.toHaveBeenCalled() + const [deleted, failed] = harness.database.resolveLifecycleAbandonedArtifacts.mock.calls[0] + expect(deleted).toEqual([]) + expect(failed).toHaveLength(96) + const released = harness.database.releaseLifecycleClaim.mock.calls[0][0] + expect(released).toMatchObject({ + nextRunAt: new Date(now.getTime() + delayMinutes * 60_000).toISOString(), + error: { + name: 'LifecycleDeletionIncomplete', + message: deleteOutcome.toLowerCase(), + pendingArtifacts: 96, + }, + }) + expect(released).not.toHaveProperty('continuation') + } + ) + + it('compiles rules from the persisted snapshot instead of the current time', async () => { + const continuation = newContinuation(randomUUID(), { snapshotAt: '2026-08-12T00:00:00.000Z' }) + const harness = createHarness({ continuation }) + + await harness.run() + + expect(harness.database.evaluateNoncurrentLifecyclePage.mock.calls[0][0]).toMatchObject({ + bucketId, + snapshotAt: '2026-08-12T00:00:00.000Z', + pageSize: 480, + rules: [{ cutoffAt: '2026-08-11T00:00:00.000Z' }], + }) + }) + + it.each([ + { failureCount: 6, delayMinutes: 320 }, + { failureCount: 10, delayMinutes: 360 }, + ])( + 'releases without rereading the continuation after $failureCount failures when the send throws', + async ({ failureCount, delayMinutes }) => { + const harness = createHarness({ failureCount }) + const failure = new Error('connection reset') + harness.backend.deleteObjectsDetailed.mockRejectedValueOnce(failure) + + await expect(harness.run()).rejects.toBe(failure) + expect(harness.database.revalidateLifecycleClaim).not.toHaveBeenCalled() + const release = harness.database.releaseLifecycleClaim.mock.calls.at(-1)![0] + expect(release).toMatchObject({ + nextRunAt: new Date(now.getTime() + delayMinutes * 60_000).toISOString(), + error: expect.objectContaining({ message: failure.message }), + }) + expect(release).not.toHaveProperty('continuation') + expect(harness.database.resolveLifecycleAbandonedArtifacts).not.toHaveBeenCalled() + } + ) + + it.each([null, '856d94e5-5241-42ae-9c0b-d06332a7ae15'])( + 'deletes exactly the physical version %s for an unversioned data row', + async (version) => { + const harness = createHarness({}) + harness.pageCandidate.version = version + const name = harness.pageCandidate.name + const keys = version === null ? [name] : [`${name}/${version}`, `${name}/${version}.info`] + + await expect(harness.run()).resolves.toMatchObject({ status: 'COMPLETED' }) + expect(harness.backend.deleteObjectsDetailed).toHaveBeenCalledWith( + 'root', + keys.map((key) => `tenant/bucket/${key}`), + { signal: expect.any(AbortSignal) } + ) + expect(harness.calls).toEqual(['arm', 'delete', 'resolve', 'complete']) + } + ) + + it('reports a non-empty error for an empty backend error', async () => { + const harness = createHarness({}) + harness.backend.deleteObjectsDetailed.mockImplementationOnce((_bucket, keys: string[]) => + keys.map((key) => ({ key, outcome: 'UNKNOWN', error: { message: '' } })) + ) + const warning = vi.spyOn(logSchema, 'warning').mockImplementation(() => {}) + + await expect(harness.run()).resolves.toMatchObject({ + status: 'PARTIAL', + runId: expect.any(String), + }) + expect(harness.database.releaseLifecycleClaim.mock.calls[0][0].error.message).toBe( + 'Lifecycle deletion remains incomplete' + ) + expect(JSON.parse(warning.mock.calls[0][2].metadata as string)).not.toHaveProperty('error') + }) + + it('counts only the failed artifacts as pending after a partial backend deletion', async () => { + const harness = createHarness({}) + harness.backend.deleteObjectsDetailed.mockImplementationOnce((_bucket, keys: string[]) => + keys.map((key, index) => ({ + key, + outcome: index === 0 ? 'DELETED' : 'FAILED', + error: { message: 'nope' }, + })) + ) + const warning = vi.spyOn(logSchema, 'warning').mockImplementation(() => {}) + + await expect(harness.run()).resolves.toMatchObject({ status: 'PARTIAL' }) + expect(harness.database.releaseLifecycleClaim.mock.calls[0][0].error.pendingArtifacts).toBe(1) + expect(JSON.parse(warning.mock.calls[0][2].metadata as string).failedArtifacts).toBe(1) + }) +}) diff --git a/src/storage/lifecycle/executor.ts b/src/storage/lifecycle/executor.ts new file mode 100644 index 000000000..3e98e6039 --- /dev/null +++ b/src/storage/lifecycle/executor.ts @@ -0,0 +1,331 @@ +import { randomUUID } from 'node:crypto' +import { ErrorCode, isStorageError } from '@internal/errors' +import { hashStringToInt } from '@internal/hashing' +import { logger, logSchema } from '@internal/monitoring' +import { + lifecycleBytesDeleted, + lifecycleDeleteMarkersDeleted, + lifecycleObjectVersionsDeleted, + lifecycleVersionsEligible, + lifecycleVersionsExamined, +} from '@internal/monitoring/metrics' +import { getConfig } from '../../config' +import type { Database } from '../database' +import { + type LifecycleBatch, + type LifecycleClaimIdentity, + type LifecycleContinuation, + type LifecycleState, +} from '../schemas' +import type { Storage } from '../storage' +import { deleteLifecycleArtifacts } from './artifact-deletion' +import { compileLifecycleEvaluationRules } from './configuration' +import { + advanceEmptyLifecyclePage, + createLifecycleContinuation, + freezeLifecycleBatchVersion, +} from './continuation' +import { lifecycleEvaluationBatches } from './evaluation' + +const configured = getConfig() +export const LIFECYCLE_EVALUATION_INTERVAL_MS = 24 * 60 * 60 * 1000 +const MAX_DELETION_RETRY_MS = 15 * 60 * 1000 + +export interface LifecycleExecutorOptions { + pageSize: number + deleteBatchSize: number + jobBudgetMs: number + claimLeaseMs: number + fairnessDelayMs: number + now: () => Date +} + +export interface LifecycleRunOptions { + // False pauses new deletion work for this tenant. + tenantLifecycleEnabled: boolean + signal?: AbortSignal +} + +export interface LifecycleRunResult { + status: 'NOT_CLAIMED' | 'COMPLETED' | 'PARTIAL' | 'PAUSED' | 'LOST_CLAIM' | 'INELIGIBLE' + runId?: string +} + +const defaultOptions: LifecycleExecutorOptions = { + pageSize: configured.storageLifecyclePageSize, + deleteBatchSize: configured.storageLifecycleDeleteBatchSize, + jobBudgetMs: configured.storageLifecycleJobBudgetMs, + claimLeaseMs: configured.storageLifecycleClaimLeaseMs, + fairnessDelayMs: 5000, + now: () => new Date(), +} + +export class NoncurrentLifecycleExecutor { + private readonly database: Database + private readonly options: LifecycleExecutorOptions + + constructor( + private readonly storage: Storage, + options: Partial = {} + ) { + this.database = storage.db.asSuperUser() + this.options = { ...defaultOptions, ...options } + } + + async run(bucketId: string, runOptions: LifecycleRunOptions): Promise { + if (!(await this.database.hasMigration('lifecycle-writer-coordination'))) { + return { status: 'PAUSED' } + } + + const identity = { bucketId, claimId: randomUUID() } + let claimed: LifecycleState | undefined + try { + // The claim can commit before its persisted continuation fails decoding. + claimed = await this.database.claimLifecycleBucket({ + ...identity, + leaseMs: this.options.claimLeaseMs, + }) + if (!claimed) return { status: 'NOT_CLAIMED' } + return await this.runClaimedBucket(claimed, identity, runOptions) + } catch (error) { + // The bucket delete cascaded the state row, so there is no claim to release. + if (isStorageError(ErrorCode.NoSuchBucket, error)) return { status: 'LOST_CLAIM' } + try { + const now = this.options.now() + await this.database.releaseLifecycleClaim({ + ...identity, + // A failed acknowledgement may follow a committed arm. + // Preserve the database continuation when releasing after an error. + nextRunAt: retryAt( + now, + claimed?.failureCount ?? (error as { failureCount?: number }).failureCount ?? 0 + ), + error: { + name: error instanceof Error ? error.name : 'Error', + message: lifecycleErrorMessage(error), + }, + }) + } catch (releaseError) { + logSchema.warning(logger, '[Lifecycle] Failed to release lifecycle claim', { + type: 'event', + tenantId: this.database.tenantId, + project: this.database.tenantId, + error: releaseError, + metadata: JSON.stringify(identity), + }) + } + throw error + } + } + + private async runClaimedBucket( + claimed: LifecycleState, + identity: LifecycleClaimIdentity, + runOptions: LifecycleRunOptions + ): Promise { + const startedAt = this.options.now().getTime() + let continuation = claimed.continuation + + if (!runOptions.tenantLifecycleEnabled) { + const released = await this.releaseAfterDelay(identity, continuation) + return { status: released ? 'PAUSED' : 'LOST_CLAIM', runId: continuation?.runId } + } + + // The claim may have waited for a policy update while reading an older bucket snapshot. + const bucket = await this.database.findLifecycleBucket(identity.bucketId) + if ( + bucket.versioning_status === 'DISABLED' || + !bucket.lifecycle_configuration || + !bucket.lifecycle_configuration_generation + ) { + const released = await this.releaseIdle(identity) + return { status: released ? 'PAUSED' : 'LOST_CLAIM', runId: continuation?.runId } + } + + if (!continuation || continuation.generation !== bucket.lifecycle_configuration_generation) { + continuation = createLifecycleContinuation({ + runId: randomUUID(), + generation: bucket.lifecycle_configuration_generation, + snapshotAt: this.options.now().toISOString(), + }) + } + + const rules = compileLifecycleEvaluationRules( + bucket.lifecycle_configuration, + new Date(continuation.snapshotAt) + ) + if (rules.length === 0) { + const released = await this.releaseIdle(identity) + return { status: released ? 'PAUSED' : 'LOST_CLAIM', runId: continuation.runId } + } + + let evaluated = false + while ( + (!evaluated || this.options.now().getTime() - startedAt < this.options.jobBudgetMs) && + !runOptions.signal?.aborted + ) { + evaluated = true + const page = await this.database.evaluateNoncurrentLifecyclePage({ + bucketId: claimed.bucketId, + snapshotAt: continuation.snapshotAt, + cursor: continuation.cursor, + rules, + pageSize: this.options.pageSize, + }) + lifecycleVersionsExamined.add(page.rawRowsExamined) + lifecycleVersionsEligible.add(page.candidates.length) + + if (page.candidates.length === 0) { + continuation = advanceEmptyLifecyclePage(continuation, page) + } else { + let first = true + for (const batch of lifecycleEvaluationBatches(page, this.options.deleteBatchSize)) { + // Allow one chunk after evaluation so a slow scan can still make progress. + if ( + runOptions.signal?.aborted || + (!first && this.options.now().getTime() - startedAt >= this.options.jobBudgetMs) + ) { + const released = await this.releaseAfterDelay(identity, continuation) + return { status: released ? 'PARTIAL' : 'LOST_CLAIM', runId: continuation.runId } + } + first = false + const processed = await this.deleteBatch(identity, continuation, { + pageEnd: batch.pageEnd!, + rawRowsExamined: batch.rawRowsExamined, + versions: batch.candidates.map(freezeLifecycleBatchVersion), + }) + if ('status' in processed) return processed + continuation = processed + } + } + + if (page.exhausted) { + const completed = await this.completeRun(identity, continuation) + return { status: completed ? 'COMPLETED' : 'LOST_CLAIM', runId: continuation.runId } + } + } + + const released = await this.releaseAfterDelay(identity, continuation) + return { status: released ? 'PARTIAL' : 'LOST_CLAIM', runId: continuation.runId } + } + + private async deleteBatch( + identity: LifecycleClaimIdentity, + continuation: LifecycleContinuation, + batch: LifecycleBatch + ): Promise { + const result = await this.database.armLifecycleAttempt({ + ...identity, + leaseMs: this.options.claimLeaseMs, + continuation, + batch, + }) + if (result.status === 'fenced') return this.resolveArmRejection(identity, continuation) + if (result.status === 'ineligible') return { status: 'INELIGIBLE', runId: continuation.runId } + const armed = result.state.continuation! + const before = continuation.counters + const after = armed.counters + lifecycleObjectVersionsDeleted.add(after.objectVersionsDeleted - before.objectVersionsDeleted) + lifecycleDeleteMarkersDeleted.add(after.deleteMarkersDeleted - before.deleteMarkersDeleted) + lifecycleBytesDeleted.add(after.bytesDeleted - before.bytesDeleted) + + if (result.artifacts.length === 0) return armed + const { failed, error } = await deleteLifecycleArtifacts({ + storage: this.storage, + database: this.database, + artifacts: result.artifacts, + signal: AbortSignal.timeout(this.options.claimLeaseMs), + }) + if (failed === 0) return armed + + logSchema.warning(logger, '[Lifecycle] Backend rejected lifecycle deletions', { + type: 'event', + tenantId: this.database.tenantId, + project: this.database.tenantId, + metadata: JSON.stringify({ + bucketId: identity.bucketId, + failedArtifacts: failed, + error: error && lifecycleErrorMessage(error), + }), + }) + const released = await this.database.releaseLifecycleClaim({ + ...identity, + nextRunAt: retryAt(this.options.now(), result.state.failureCount, MAX_DELETION_RETRY_MS), + error: { + name: 'LifecycleDeletionIncomplete', + message: lifecycleErrorMessage(error ?? 'Lifecycle deletion remains incomplete'), + pendingArtifacts: failed, + }, + }) + return { status: released ? 'PARTIAL' : 'LOST_CLAIM', runId: armed.runId } + } + + private async resolveArmRejection( + identity: LifecycleClaimIdentity, + continuation: LifecycleContinuation + ): Promise { + const current = await this.database.revalidateLifecycleClaim(identity) + if (!current) return { status: 'LOST_CLAIM', runId: continuation.runId } + const bucket = await this.database.findLifecycleBucket(identity.bucketId) + if (bucket.lifecycle_configuration_generation !== continuation.generation) { + const released = await this.releaseAfterDelay(identity, null) + return { status: released ? 'PAUSED' : 'LOST_CLAIM', runId: continuation.runId } + } + + throw new Error('Lifecycle arm fence rejected a batch with an unchanged claim') + } + + private async releaseIdle(identity: LifecycleClaimIdentity): Promise { + return this.database.releaseLifecycleClaim({ + ...identity, + continuation: null, + nextRunAt: null, + }) + } + + private async releaseAfterDelay( + identity: LifecycleClaimIdentity, + continuation: LifecycleContinuation | null + ): Promise { + return this.database.releaseLifecycleClaim({ + ...identity, + continuation, + nextRunAt: new Date( + this.options.now().getTime() + this.options.fairnessDelayMs + ).toISOString(), + }) + } + + private completeRun(identity: LifecycleClaimIdentity, continuation: LifecycleContinuation) { + return this.database.completeLifecycleRun( + { ...identity, generation: continuation.generation }, + { + runId: continuation.runId, + generation: continuation.generation, + snapshotAt: continuation.snapshotAt, + counters: continuation.counters, + completedAt: this.options.now().toISOString(), + }, + nextEvaluationAt(this.options.now(), this.database.tenantId, identity.bucketId) + ) + } +} + +export function lifecycleErrorMessage(error: unknown): string { + const value = error instanceof Error ? error.message : String(error) + return value.slice(0, 1000).toWellFormed() +} + +function retryAt(now: Date, failureCount: number, maxDelayMs = 6 * 60 * 60 * 1000): string { + const delay = Math.min(maxDelayMs, 5 * 60 * 1000 * 2 ** failureCount) + return new Date(now.getTime() + delay).toISOString() +} + +export function evaluationJitterMs(key: string): number { + return Math.abs(hashStringToInt(key)) % (LIFECYCLE_EVALUATION_INTERVAL_MS / 10) +} + +function nextEvaluationAt(now: Date, tenantId: string, bucketId: string): string { + const jitter = evaluationJitterMs(`${tenantId}/${bucketId}`) + return new Date(now.getTime() + LIFECYCLE_EVALUATION_INTERVAL_MS + jitter).toISOString() +} diff --git a/src/test/noncurrent-lifecycle-database.test.ts b/src/test/noncurrent-lifecycle-database.test.ts index b1ec69707..2813f8acd 100644 --- a/src/test/noncurrent-lifecycle-database.test.ts +++ b/src/test/noncurrent-lifecycle-database.test.ts @@ -514,7 +514,7 @@ describe('noncurrent lifecycle database API', () => { claimId: randomUUID(), leaseMs: 60_000, }) - ).resolves.toMatchObject({ continuation: null }) + ).resolves.toMatchObject({ continuation: { snapshotAt: fixture.evaluation.snapshotAt } }) await expect( tHelper.database.evaluateNoncurrentLifecyclePage(fixture.evaluation) ).resolves.toMatchObject({ candidates: [] }) @@ -929,7 +929,10 @@ describe('noncurrent lifecycle database API', () => { status: claimKept ? 'fenced' : 'ineligible', }) await expect(fixtures.readState(bucketId, 'continuation, claim_id')).resolves.toEqual([ - { continuation: null, claim_id: claimKept ? fixture.identity.claimId : null }, + { + continuation: claimKept ? null : fixture.continuation, + claim_id: claimKept ? fixture.identity.claimId : null, + }, ]) }) @@ -944,7 +947,7 @@ describe('noncurrent lifecycle database API', () => { await expect(fixture.arm()).resolves.toEqual({ status: 'ineligible' }) await expect(fixtures.readState(bucketId, 'continuation, claim_id')).resolves.toEqual([ - { continuation: null, claim_id: null }, + { continuation: fixture.continuation, claim_id: null }, ]) }) @@ -1115,7 +1118,7 @@ describe('noncurrent lifecycle database API', () => { expect(armed.status).toBe(arms ? 'armed' : 'ineligible') if (!arms) { await expect(fixtures.readState(bucketId, 'continuation, claim_id')).resolves.toEqual([ - { continuation: null, claim_id: null }, + { continuation: fixture.continuation, claim_id: null }, ]) } }) diff --git a/src/test/noncurrent-lifecycle-executor.test.ts b/src/test/noncurrent-lifecycle-executor.test.ts new file mode 100644 index 000000000..b25a92657 --- /dev/null +++ b/src/test/noncurrent-lifecycle-executor.test.ts @@ -0,0 +1,1593 @@ +import { randomUUID } from 'node:crypto' +import { Readable } from 'node:stream' +import { text } from 'node:stream/consumers' +import { logSchema } from '@internal/monitoring' +import { isMissingBackendObject, withOptionalVersion } from '@storage/backend' +import { StoragePgDB } from '@storage/database' +import { retryAbandonedLifecycleArtifacts } from '@storage/lifecycle/abandoned-artifacts' +import { + type LifecycleExecutorOptions, + NoncurrentLifecycleExecutor, +} from '@storage/lifecycle/executor' +import { Client } from 'pg' +import { getConfig } from '../config' +import { batchOf, newContinuation } from './utils/lifecycle-continuation' +import { useLifecycleStorage } from './utils/lifecycle-versioning' +import { withDeleteEnabled } from './utils/storage' + +const { storageBackendType, storageS3Bucket, tenantId } = getConfig() +const HOUR_MS = 60 * 60 * 1000 +const DAY_MS = 24 * HOUR_MS +const threeDaysAgo = () => new Date(Date.now() - 3 * DAY_MS) + +describe('noncurrent lifecycle executor against version-aware writes', () => { + const fixtures = useLifecycleStorage('lifecycle-executor') + const helper = fixtures.helper + const uploaded: Array<{ name: string; version: string | null }> = [] + let bucketId: string + const configuration = { + rules: [ + { + id: 'expire-history', + status: 'Enabled' as const, + filter: {}, + noncurrentVersionExpiration: { noncurrentDays: 1 }, + }, + ], + } + + beforeEach(() => { + bucketId = fixtures.bucketId + }) + + afterEach(async () => { + const cleanup = await Promise.allSettled( + uploaded + .splice(0) + .map(({ name, version }) => + helper.adapter.deleteObject(storageS3Bucket, `${tenantId}/${bucketId}/${name}`, version) + ) + ) + expect(cleanup.filter((result) => result.status === 'rejected')).toEqual([]) + }) + + async function uploadFixture(name: string, version: string | null) { + uploaded.push({ name, version }) + return helper.adapter.uploadObject( + storageS3Bucket, + `${tenantId}/${bucketId}/${name}`, + version, + Readable.from(Buffer.from('x')), + 'text/plain', + 'no-cache', + undefined, + 1 + ) + } + + async function seedHistory( + name: string, + archivedVersion: string | null, + currentVersion: string, + options: { isVersioned?: boolean; archivedAt?: Date } = {} + ) { + const archivedMetadata = await uploadFixture(name, archivedVersion) + if (archivedVersion === null) { + // Legacy unsuffixed blobs predate the UUID writer. + await helper.database.connection.query( + `INSERT INTO storage.objects (bucket_id, name, version, is_versioned, metadata) + VALUES ($1, $2, NULL, false, $3::jsonb)`, + [bucketId, name, JSON.stringify(archivedMetadata)] + ) + } else { + if (options.isVersioned !== false) await fixtures.setStatus(bucketId, 'ENABLED') + await helper.uploader.completeUpload({ + bucketId, + objectName: name, + version: archivedVersion, + objectMetadata: archivedMetadata, + uploadType: 'standard', + isUpsert: true, + }) + } + await fixtures.setStatus(bucketId, 'ENABLED') + // File storage cannot retain an unsuffixed blob beside a revision directory. + let currentMetadata = archivedMetadata + if (archivedVersion !== null || storageBackendType === 's3') { + currentMetadata = await uploadFixture(name, currentVersion) + } + await helper.uploader.completeUpload({ + bucketId, + objectName: name, + version: currentVersion, + objectMetadata: currentMetadata, + uploadType: 'standard', + isUpsert: true, + }) + if (options.archivedAt) { + await helper.database.connection.query( + `UPDATE storage.objects SET archived_at = $4 + WHERE bucket_id = $1 AND name = $2 AND version IS NOT DISTINCT FROM $3`, + [bucketId, name, archivedVersion, options.archivedAt] + ) + } + } + + const historyRows = (names: string[]) => + names.map((name) => ({ name, archived: randomUUID(), current: randomUUID() })) + + async function activate(status: 'ENABLED' | 'SUSPENDED' = 'ENABLED') { + await fixtures.setStatus(bucketId, status) + await helper.database.putLifecycleConfiguration(bucketId, configuration) + } + + function runExecutor( + options: Partial & { + tenantLifecycleEnabled?: boolean + signal?: AbortSignal + } = {} + ) { + const { tenantLifecycleEnabled = true, signal, ...executorOptions } = options + return new NoncurrentLifecycleExecutor(helper.storage, { + now: () => new Date(Date.now() + 3 * DAY_MS), + ...executorOptions, + }).run(bucketId, { tenantLifecycleEnabled, signal }) + } + + const makeDue = () => + helper.database.connection.query( + "UPDATE storage.bucket_lifecycle_states SET next_run_at = '2000-01-01' WHERE bucket_id = $1", + [bucketId] + ) + + it('evaluates a retained-history page once while deleting one candidate per transaction', async () => { + await activate() + await helper.database.putLifecycleConfiguration(bucketId, { + rules: [ + { + status: 'Enabled', + filter: {}, + noncurrentVersionExpiration: { noncurrentDays: 1, newerNoncurrentVersions: 1 }, + }, + ], + }) + await helper.database.connection.query( + `INSERT INTO storage.objects (bucket_id, name, version, archived_at, is_versioned, is_delete_marker) + SELECT $1, 'evaluated/' || lpad(n::text, 3, '0'), gen_random_uuid()::text, at, true, true + FROM generate_series(1, 101) AS n, + unnest(ARRAY['2000-01-01'::timestamptz, '2000-01-02'::timestamptz]) AS at`, + [bucketId] + ) + const evaluation = vi.spyOn(StoragePgDB.prototype, 'evaluateNoncurrentLifecyclePage') + await expect(runExecutor({ pageSize: 480, deleteBatchSize: 1 })).resolves.toMatchObject({ + status: 'COMPLETED', + }) + expect(evaluation).toHaveBeenCalledTimes(1) + const [state] = await fixtures.readState(bucketId, 'last_result') + expect(state.last_result.counters).toMatchObject({ + versionsExamined: 202, + versionsEligible: 101, + deleteMarkersDeleted: 101, + batchesCompleted: 101, + }) + const remaining = await helper.database.connection.query( + `SELECT count(*)::integer AS count FROM storage.objects + WHERE bucket_id = $1 AND archived_at = '2000-01-02'`, + [bucketId] + ) + expect(remaining.rows).toEqual([{ count: 101 }]) + }, 30_000) + + it('resumes within one object key without rounding a chunk cursor to milliseconds', async () => { + await activate() + const name = 'buffer/microseconds' + const archived = [randomUUID(), randomUUID(), randomUUID()] + for (const [index, version] of archived.entries()) { + const metadata = await uploadFixture(name, version) + await helper.database.connection.query( + `INSERT INTO storage.objects(bucket_id, name, version, archived_at, is_versioned, metadata) + VALUES ($1, $2, $3, $4, true, $5)`, + [ + bucketId, + name, + version, + `2000-01-01T00:00:00.00000${index + 1}Z`, + JSON.stringify(metadata), + ] + ) + } + const controller = new AbortController() + const original = helper.adapter.deleteObjectsDetailed.bind(helper.adapter) + vi.spyOn(helper.adapter, 'deleteObjectsDetailed').mockImplementationOnce(async (...args) => { + const result = await original(...args) + controller.abort() + return result + }) + await expect( + runExecutor({ pageSize: 100, deleteBatchSize: 1, signal: controller.signal }) + ).resolves.toMatchObject({ status: 'PARTIAL' }) + const [state] = await fixtures.readState(bucketId, 'continuation') + expect(state.continuation.cursor).toEqual({ name, archivedAt: '2000-01-01T00:00:00.000001Z' }) + await makeDue() + await expect(runExecutor({ pageSize: 100, deleteBatchSize: 1 })).resolves.toMatchObject({ + status: 'COMPLETED', + }) + expect(await versions(name)).toEqual([]) + const [completed] = await fixtures.readState(bucketId, 'last_result') + expect(completed.last_result.counters).toMatchObject({ + versionsExamined: 3, + objectVersionsDeleted: 3, + }) + for (const version of archived) await expectBlobAbsent(name, version) + }) + + it.each(['budget', 'abort'] as const)( + 'resumes the buffered suffix after %s stops between chunks', + async (stop) => { + await activate() + const rows = historyRows(['buffer/a', 'buffer/b', 'buffer/c']) + for (const row of rows) + await seedHistory(row.name, row.archived, row.current, { archivedAt: threeDaysAgo() }) + let now = Date.now() + 3 * DAY_MS + const controller = new AbortController() + const original = helper.adapter.deleteObjectsDetailed.bind(helper.adapter) + vi.spyOn(helper.adapter, 'deleteObjectsDetailed').mockImplementationOnce(async (...args) => { + const result = await original(...args) + if (stop === 'abort') controller.abort() + else now += 1000 + return result + }) + const evaluation = vi.spyOn(StoragePgDB.prototype, 'evaluateNoncurrentLifecyclePage') + await expect( + runExecutor({ + pageSize: 100, + deleteBatchSize: 1, + jobBudgetMs: 1000, + now: () => new Date(now), + signal: controller.signal, + }) + ).resolves.toMatchObject({ status: 'PARTIAL' }) + expect(evaluation).toHaveBeenCalledTimes(1) + const [state] = await fixtures.readState(bucketId, 'continuation') + expect(state.continuation).toMatchObject({ + cursor: { name: rows[0].name }, + counters: { versionsExamined: 1, versionsEligible: 1, objectVersionsDeleted: 1 }, + }) + expect(state.continuation.batch).toBeUndefined() + await expectBlobAbsent(rows[0].name, rows[0].archived) + for (const row of rows.slice(1)) expect(await versions(row.name)).toHaveLength(2) + await makeDue() + await expect(runExecutor({ pageSize: 480, deleteBatchSize: 2 })).resolves.toMatchObject({ + status: 'COMPLETED', + }) + expect(evaluation).toHaveBeenCalledTimes(2) + const [completed] = await fixtures.readState(bucketId, 'last_result') + expect(completed.last_result.counters).toMatchObject({ + versionsExamined: 3, + versionsEligible: 3, + objectVersionsDeleted: 3, + batchesCompleted: 2, + }) + for (const row of rows) await expectBlobAbsent(row.name, row.archived) + } + ) + + it.each(['promotion', 'foreign key'] as const)( + 'rechecks buffered candidates after a concurrent %s', + async (change) => { + await activate() + const rows = historyRows(['buffer-change/a', 'buffer-change/b']) + for (const row of rows) + await seedHistory(row.name, row.archived, row.current, { archivedAt: threeDaysAgo() }) + const target = rows[1] + const original = helper.adapter.deleteObjectsDetailed.bind(helper.adapter) + vi.spyOn(helper.adapter, 'deleteObjectsDetailed').mockImplementationOnce(async (...args) => { + if (change === 'promotion') { + const transaction = await helper.database.connection.transaction() + try { + await transaction.query( + 'UPDATE storage.objects SET archived_at = clock_timestamp() WHERE bucket_id = $1 AND name = $2 AND archived_at IS NULL', + [bucketId, target.name] + ) + await transaction.query( + 'UPDATE storage.objects SET archived_at = NULL WHERE bucket_id = $1 AND name = $2 AND version = $3', + [bucketId, target.name, target.archived] + ) + await transaction.commit() + } catch (error) { + await transaction.rollback() + throw error + } + } else { + await helper.database.connection.query( + 'CREATE TABLE public.lifecycle_buffer_refs (id uuid REFERENCES storage.objects(id))' + ) + await helper.database.connection.query( + 'INSERT INTO public.lifecycle_buffer_refs SELECT id FROM storage.objects WHERE bucket_id = $1 AND name = $2 AND version = $3', + [bucketId, target.name, target.archived] + ) + } + return original(...args) + }) + try { + await expect(runExecutor({ pageSize: 100, deleteBatchSize: 1 })).resolves.toMatchObject({ + status: change === 'promotion' ? 'INELIGIBLE' : 'COMPLETED', + }) + await expectBlobAbsent(rows[0].name, rows[0].archived) + expect(await versions(target.name)).toContainEqual({ + version: target.archived, + current: change === 'promotion', + }) + await expectBlobPresent(target.name, target.archived) + } finally { + await helper.database.connection.query('DROP TABLE IF EXISTS public.lifecycle_buffer_refs') + } + } + ) + + it.each([ + { size: 48, nullVeto: false, statementVeto: false, keepB: false }, + { size: 1, nullVeto: false, statementVeto: false, keepB: true }, + { size: 48, nullVeto: true, statementVeto: false, keepB: false }, + { size: 1, nullVeto: true, statementVeto: false, keepB: true }, + { size: 48, nullVeto: false, statementVeto: true, keepB: true }, + ])( + 'honors state-dependent trigger decisions: batch $size, NULL=$nullVeto, statement=$statementVeto', + async ({ size, nullVeto, statementVeto, keepB }) => { + await activate() + const names = [ + 'dependent/B', + 'dependent/A', + ...Array.from({ length: 47 }, (_, index) => `dependent/F${String(index).padStart(2, '0')}`), + 'dependent/Z', + ] + const revisions = new Map() + for (const [index, name] of names.entries()) { + const version = randomUUID() + revisions.set(name, version) + const metadata = await uploadFixture(name, version) + await helper.database.connection.query( + `INSERT INTO storage.objects(id, bucket_id, name, version, archived_at, is_versioned, metadata) + VALUES ($1, $2, $3, $4, '2000-01-01', true, $5)`, + [ + `00000000-0000-0000-0000-${String(index + 1).padStart(12, '0')}`, + bucketId, + name, + version, + JSON.stringify(metadata), + ] + ) + } + try { + await helper.database.connection.query(` + CREATE TABLE public.lifecycle_dependency_permitted (name text); + ALTER FUNCTION storage.delete_lifecycle_objects(uuid[]) SET enable_indexscan = off; + ALTER FUNCTION storage.delete_lifecycle_objects(uuid[]) SET enable_bitmapscan = off; + ALTER FUNCTION storage.delete_lifecycle_objects(uuid[]) SET enable_indexonlyscan = off; + CREATE FUNCTION public.lifecycle_dependency_veto() RETURNS trigger LANGUAGE plpgsql SECURITY DEFINER AS $$ + BEGIN + IF OLD.name = 'dependent/Z' THEN RAISE EXCEPTION 'keep Z' USING ERRCODE = '42501'; END IF; + IF OLD.name = 'dependent/B' AND NOT EXISTS ( + SELECT 1 FROM storage.objects WHERE bucket_id = OLD.bucket_id AND name = 'dependent/A' + ) THEN ${nullVeto ? 'RETURN NULL;' : "RAISE EXCEPTION 'keep B without A' USING ERRCODE = '42501';"} END IF; + IF OLD.name = 'dependent/B' THEN INSERT INTO public.lifecycle_dependency_permitted VALUES (OLD.name); END IF; + RETURN OLD; + END $$; + CREATE TRIGGER lifecycle_dependency_veto BEFORE DELETE ON storage.objects + FOR EACH ROW EXECUTE FUNCTION public.lifecycle_dependency_veto(); + `) + if (statementVeto) + await helper.database.connection.query(` + CREATE FUNCTION public.lifecycle_dependency_statement() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + IF (SELECT count(*) FROM removed) > 1 THEN RAISE EXCEPTION 'one row only' USING ERRCODE = '42501'; END IF; + RETURN NULL; + END $$; + CREATE TRIGGER lifecycle_dependency_statement AFTER DELETE ON storage.objects + REFERENCING OLD TABLE AS removed FOR EACH STATEMENT EXECUTE FUNCTION public.lifecycle_dependency_statement(); + `) + await expect(runExecutor({ pageSize: 100, deleteBatchSize: size })).resolves.toMatchObject({ + status: 'COMPLETED', + }) + const permitted = await helper.database.connection.query( + 'SELECT name FROM public.lifecycle_dependency_permitted' + ) + // Only decisions surviving the actual committed statement authorize bytes. + // Row fallback visits A before B; bulk visitation is engine/plan dependent. + if (keepB) expect(permitted.rows).toEqual([]) + const bWasPermitted = permitted.rows.length > 0 + const table = await helper.database.connection.query( + `SELECT am.amname FROM pg_class AS c JOIN pg_am AS am ON am.oid = c.relam + WHERE c.oid = 'storage.objects'::regclass` + ) + if (table.rows[0].amname === 'heap') expect(bWasPermitted).toBe(!keepB) + for (const name of ['dependent/A', 'dependent/B', 'dependent/Z']) { + const kept = name === 'dependent/Z' || (name === 'dependent/B' && !bWasPermitted) + expect(await versions(name)).toHaveLength(kept ? 1 : 0) + if (kept) await expectBlobPresent(name, revisions.get(name)!) + else await expectBlobAbsent(name, revisions.get(name)!) + } + } finally { + await helper.database.connection.query(` + DROP TRIGGER IF EXISTS lifecycle_dependency_veto ON storage.objects; + DROP TRIGGER IF EXISTS lifecycle_dependency_statement ON storage.objects; + DROP FUNCTION IF EXISTS public.lifecycle_dependency_veto(); + DROP FUNCTION IF EXISTS public.lifecycle_dependency_statement(); + DROP TABLE IF EXISTS public.lifecycle_dependency_permitted; + ALTER FUNCTION storage.delete_lifecycle_objects(uuid[]) RESET enable_indexscan; + ALTER FUNCTION storage.delete_lifecycle_objects(uuid[]) RESET enable_bitmapscan; + ALTER FUNCTION storage.delete_lifecycle_objects(uuid[]) RESET enable_indexonlyscan; + `) + } + } + ) + + async function versions(name: string) { + const result = await helper.database.connection.query<{ + version: string | null + current: boolean + }>( + `SELECT version, archived_at IS NULL AS current + FROM storage.objects WHERE bucket_id = $1 AND name = $2 ORDER BY created_at`, + [bucketId, name] + ) + return result.rows + } + + async function abandonedKeys() { + const result = await helper.database.connection.query<{ key: string }>( + 'SELECT key FROM storage.lifecycle_abandoned_artifacts WHERE bucket_id = $1 ORDER BY key', + [bucketId] + ) + return result.rows.map((row) => row.key) + } + + async function abandonedAttempts() { + const result = await helper.database.connection.query<{ attempted_at: Date }>( + 'SELECT attempted_at FROM storage.lifecycle_abandoned_artifacts WHERE bucket_id = $1', + [bucketId] + ) + return result.rows.map((row) => row.attempted_at.getTime()) + } + + async function sweep() { + await retryAbandonedLifecycleArtifacts({ + storage: helper.storage, + database: helper.database, + attemptedBefore: new Date(Date.now() + HOUR_MS).toISOString(), + budgetMs: 60_000, + }) + } + + async function expectBlobAbsent(name: string, version: string | null) { + const missing = await helper.adapter + .headObject(storageS3Bucket, `${tenantId}/${bucketId}/${name}`, version, { + confirmMissing: true, + }) + .then( + () => false, + (error: unknown) => isMissingBackendObject(error) + ) + expect(missing).toBe(true) + } + + async function expectBlobPresent(name: string, version: string | null) { + await expect( + helper.adapter.headObject(storageS3Bucket, `${tenantId}/${bucketId}/${name}`, version) + ).resolves.toMatchObject({ size: 1 }) + } + + async function prepareAttempt(name: string, version: string | null, archivedAt: Date) { + const bucket = await helper.database.findLifecycleBucket(bucketId) + const generation = bucket.lifecycle_configuration_generation! + const identity = { bucketId, claimId: randomUUID() } + expect( + await helper.database.claimLifecycleBucket({ ...identity, leaseMs: 60_000 }) + ).toBeDefined() + const { id } = ( + await helper.database.connection.query<{ id: string }>( + 'SELECT id FROM storage.objects WHERE bucket_id = $1 AND name = $2 AND version IS NOT DISTINCT FROM $3', + [bucketId, name, version] + ) + ).rows[0] + return { + ...identity, + leaseMs: 60_000, + continuation: newContinuation(generation), + batch: batchOf([{ id, name, version, isDeleteMarker: false }], { + name, + archivedAt: archivedAt.toISOString(), + }), + } + } + + it.each([{ legacy: false }, { legacy: true }])( + 'serializes direct promotion with arming (legacy=$legacy)', + async ({ legacy }) => { + await activate() + const name = 'history/row-lock.txt' + const archived = legacy ? null : randomUUID() + const current = randomUUID() + const archivedAt = threeDaysAgo() + await seedHistory(name, archived, current, { archivedAt }) + const input = await prepareAttempt(name, archived, archivedAt) + const writer = new Client({ connectionString: getConfig().databaseURL }) + await writer.connect() + let arming: Promise | undefined + try { + await writer.query('BEGIN') + const pid = (await writer.query('SELECT pg_backend_pid() AS pid')).rows[0].pid + await writer.query( + 'UPDATE storage.objects SET archived_at = clock_timestamp() WHERE bucket_id = $1 AND name = $2 AND version = $3', + [bucketId, name, current] + ) + await writer.query( + 'UPDATE storage.objects SET archived_at = NULL WHERE bucket_id = $1 AND name = $2 AND version IS NOT DISTINCT FROM $3', + [bucketId, name, archived] + ) + arming = helper.database.armLifecycleAttempt(input) + await expect + .poll( + async () => + ( + await helper.database.connection.query( + 'SELECT EXISTS (SELECT 1 FROM pg_stat_activity WHERE $1 = ANY(pg_blocking_pids(pid))) AS blocked', + [pid] + ) + ).rows[0].blocked + ) + .toBe(true) + await writer.query('COMMIT') + await expect(arming).resolves.toEqual({ status: 'ineligible' }) + expect(await versions(name)).toContainEqual({ version: archived, current: true }) + await expectBlobPresent(name, archived) + } finally { + await writer.query('ROLLBACK') + await Promise.allSettled([arming]) + await writer.end() + } + } + ) + + it('expires history produced by complete uploads and preserves the replacement body', async () => { + await activate() + const name = 'history/writer.txt' + const written: string[] = [] + for (const content of ['previous', 'replacement']) { + const { obj } = await helper.uploader.upload({ + bucketId, + objectName: name, + uploadType: 'standard', + isUpsert: true, + file: { + body: Readable.from(content), + mimeType: 'text/plain', + cacheControl: 'no-cache', + contentLength: Buffer.byteLength(content), + isTruncated: () => false, + }, + }) + const version = obj.version as string + uploaded.push({ name, version }) + written.push(version) + } + const [archived, current] = written + expect(await versions(name)).toEqual([ + { version: archived, current: false }, + { version: current, current: true }, + ]) + + await expect(runExecutor()).resolves.toMatchObject({ status: 'COMPLETED' }) + expect(await versions(name)).toEqual([{ version: current, current: true }]) + await expectBlobAbsent(name, archived) + const object = await helper.adapter.getObject( + storageS3Bucket, + `${tenantId}/${bucketId}/${name}`, + current + ) + if (!(object.body instanceof Readable)) throw new Error('Expected a readable object body') + expect(await text(object.body)).toBe('replacement') + }) + + it('preserves malformed continuation JSON and releases its committed claim after decoding fails', async () => { + await activate() + const continuation = { continuationVersion: 999 } + await helper.database.connection.query( + 'UPDATE storage.bucket_lifecycle_states SET continuation = $2::jsonb WHERE bucket_id = $1', + [bucketId, JSON.stringify(continuation)] + ) + + const now = new Date() + for (const [index, delayMinutes] of [5, 10].entries()) { + await expect(runExecutor({ now: () => now })).rejects.toThrow() + const rows = await fixtures.readState( + bucketId, + 'claim_id, claim_until, failure_count, last_error, continuation, next_run_at' + ) + expect(rows).toEqual([ + expect.objectContaining({ + claim_id: null, + claim_until: null, + failure_count: index + 1, + continuation, + last_error: expect.objectContaining({ name: 'LifecycleContinuationDecodeError' }), + next_run_at: new Date(now.getTime() + delayMinutes * 60_000), + }), + ]) + await makeDue() + } + }) + + it('pauses a due bucket of a versioning-disabled bucket without failing', async () => { + await helper.database.putLifecycleConfiguration(bucketId, configuration) + const name = 'history/disabled.txt' + const current = randomUUID() + await helper.uploader.completeUpload({ + bucketId, + objectName: name, + version: current, + objectMetadata: await uploadFixture(name, current), + uploadType: 'standard', + }) + await makeDue() + const deletes = vi.spyOn(helper.adapter, 'deleteObjectsDetailed') + + await expect(runExecutor()).resolves.toMatchObject({ status: 'PAUSED' }) + expect(deletes).not.toHaveBeenCalled() + + const rows = await fixtures.readState( + bucketId, + 'failure_count, last_error, next_run_at, continuation' + ) + expect(rows).toEqual([ + { failure_count: 0, last_error: null, next_run_at: null, continuation: null }, + ]) + expect(await versions(name)).toEqual([{ version: current, current: true }]) + }) + + it('observes a policy committed while its bucket claim waits for the state row', async () => { + await fixtures.setStatus(bucketId, 'ENABLED') + await helper.database.putLifecycleConfiguration(bucketId, { + rules: [{ ...configuration.rules[0], status: 'Disabled' }], + }) + await makeDue() + + const writer = await helper.database.connection.transaction() + const observer = new Client({ connectionString: getConfig().databaseURL }) + await observer.connect() + let running: ReturnType | undefined + try { + await helper.database.connection.setScope(writer) + await writer.query( + `UPDATE storage.buckets SET lifecycle_configuration = $2::jsonb, + lifecycle_configuration_generation = $3::uuid WHERE id = $1`, + [bucketId, JSON.stringify(configuration), randomUUID()] + ) + running = runExecutor() + await vi.waitFor(async () => { + const { rows } = await observer.query(` + SELECT EXISTS ( + SELECT 1 FROM pg_stat_activity + WHERE wait_event_type = 'Lock' + AND query LIKE '%UPDATE storage.bucket_lifecycle_states AS state%' + ) AS blocked + `) + expect(rows[0].blocked).toBe(true) + }) + await writer.commit() + + await expect(running).resolves.toMatchObject({ status: 'COMPLETED' }) + const [state] = await fixtures.readState(bucketId, 'next_run_at, last_result') + expect(state.next_run_at).not.toBeNull() + expect(state.last_result).not.toBeNull() + } finally { + await writer.rollback().catch(() => undefined) + await observer.end() + await running?.catch(() => undefined) + } + }) + + it('leaves the outbox empty and the backend bytes gone after a successful run', async () => { + await activate() + const name = 'history/outbox-success.txt' + const archived = randomUUID() + const current = randomUUID() + await seedHistory(name, archived, current, { archivedAt: threeDaysAgo() }) + const original = helper.adapter.deleteObjectsDetailed.bind(helper.adapter) + let duringSend: string[] = [] + vi.spyOn(helper.adapter, 'deleteObjectsDetailed').mockImplementationOnce(async (...args) => { + duringSend = await abandonedKeys() + return original(...args) + }) + + await expect(runExecutor()).resolves.toMatchObject({ status: 'COMPLETED' }) + + expect(duringSend).toHaveLength(2) + expect(await abandonedKeys()).toEqual([]) + await expectBlobAbsent(name, archived) + await expectBlobPresent(name, current) + }) + + it('stops after a rejected backend deletion, keeps its outbox rows and leaves the rest to the sweeper', async () => { + await activate() + const rows = ['history/rejected-a.txt', 'history/rejected-b.txt'].map((name, index) => ({ + name, + archived: randomUUID(), + current: randomUUID(), + archivedAt: new Date(Date.now() - (5 - index) * DAY_MS), + })) + for (const row of rows) { + await seedHistory(row.name, row.archived, row.current, { archivedAt: row.archivedAt }) + } + let armedAttempts: number[] = [] + const deletes = vi + .spyOn(helper.adapter, 'deleteObjectsDetailed') + .mockImplementation(async (_bucket, keys) => { + armedAttempts = await abandonedAttempts() + return keys.map((key) => ({ + key, + outcome: 'FAILED' as const, + error: { code: 'AccessDenied', message: 'deletion denied' }, + })) + }) + const warning = vi.spyOn(logSchema, 'warning').mockImplementation(() => {}) + const now = new Date(Date.now() + 3 * DAY_MS) + + try { + await expect( + runExecutor({ pageSize: 100, deleteBatchSize: 1, now: () => now }) + ).resolves.toMatchObject({ status: 'PARTIAL' }) + expect(deletes).toHaveBeenCalledOnce() + expect(warning).toHaveBeenCalledOnce() + expect(warning).toHaveBeenCalledWith( + expect.anything(), + '[Lifecycle] Backend rejected lifecycle deletions', + expect.objectContaining({ + metadata: JSON.stringify({ bucketId, failedArtifacts: 2, error: 'deletion denied' }), + }) + ) + } finally { + deletes.mockRestore() + warning.mockRestore() + } + + const [state] = await fixtures.readState( + bucketId, + 'failure_count, next_run_at, claim_id, continuation' + ) + expect(state).toMatchObject({ + failure_count: 1, + claim_id: null, + continuation: { counters: { objectVersionsDeleted: 1 } }, + }) + expect(state.next_run_at.getTime()).toBeLessThanOrEqual(now.getTime() + 15 * 60_000) + expect(state.continuation).not.toHaveProperty('batch') + expect(await versions(rows[0].name)).toEqual([{ version: rows[0].current, current: true }]) + expect(await versions(rows[1].name)).toHaveLength(2) + expect(await abandonedKeys()).toHaveLength(2) + expect(Math.min(...(await abandonedAttempts()))).toBeGreaterThan(Math.max(...armedAttempts)) + await expectBlobPresent(rows[0].name, rows[0].archived) + + await sweep() + expect(await abandonedKeys()).toEqual([]) + await expectBlobAbsent(rows[0].name, rows[0].archived) + }) + + it('keeps the outbox rows and does not block bucket deletion when the send throws after arm', async () => { + await activate() + const name = 'history/crash-after-arm.txt' + const archived = randomUUID() + const current = randomUUID() + await seedHistory(name, archived, current, { archivedAt: threeDaysAgo() }) + const failure = new Error('connection reset') + vi.spyOn(helper.adapter, 'deleteObjectsDetailed').mockRejectedValueOnce(failure) + const now = new Date() + + await expect(runExecutor({ now: () => now })).rejects.toBe(failure) + + const [state] = await fixtures.readState( + bucketId, + 'failure_count, next_run_at, claim_id, continuation' + ) + expect(state).toMatchObject({ + failure_count: 1, + claim_id: null, + next_run_at: new Date(now.getTime() + 5 * 60_000), + continuation: { counters: { objectVersionsDeleted: 1 } }, + }) + expect(state.continuation).not.toHaveProperty('batch') + expect(await versions(name)).toEqual([{ version: current, current: true }]) + expect(await abandonedKeys()).toHaveLength(2) + await expectBlobPresent(name, archived) + + await expect( + withDeleteEnabled(helper.database.connection, async (transaction) => { + await transaction.query('DELETE FROM storage.objects WHERE bucket_id = $1', [bucketId]) + return transaction.query('DELETE FROM storage.buckets WHERE id = $1', [bucketId]) + }) + ).resolves.toMatchObject({ rowCount: 1 }) + expect(await abandonedKeys()).toHaveLength(2) + + await sweep() + expect(await abandonedKeys()).toEqual([]) + await expectBlobAbsent(name, archived) + }) + + it('prevents new foreign-key references after metadata is deleted and before backend deletion', async () => { + await activate() + const name = 'history/late-reference.txt' + const archived = randomUUID() + const current = randomUUID() + await seedHistory(name, archived, current, { archivedAt: threeDaysAgo() }) + const rows = await helper.database.connection.query<{ id: string }>( + 'SELECT id FROM storage.objects WHERE bucket_id = $1 AND name = $2 AND version = $3', + [bucketId, name, archived] + ) + await helper.database.connection.query( + 'CREATE TABLE public.lifecycle_late_ref (object_id uuid REFERENCES storage.objects(id))' + ) + const original = helper.adapter.deleteObjectsDetailed.bind(helper.adapter) + const deletion = vi + .spyOn(helper.adapter, 'deleteObjectsDetailed') + .mockImplementation(async (root, keys, options) => { + await expect( + helper.database.connection.query('INSERT INTO public.lifecycle_late_ref VALUES ($1)', [ + rows.rows[0].id, + ]) + ).rejects.toMatchObject({ code: '23503' }) + return original(root, keys, options) + }) + try { + await expect(runExecutor()).resolves.toMatchObject({ status: 'COMPLETED' }) + expect(deletion).toHaveBeenCalledOnce() + await expectBlobAbsent(name, archived) + } finally { + await helper.database.connection.query('DROP TABLE public.lifecycle_late_ref') + } + }) + + function longName(length: number) { + return Array.from({ length }, (_, index) => (index % 200 === 199 ? '/' : 'x')).join('') + } + + it.skipIf(storageBackendType !== 's3')( + 'expires a version whose .info sibling exceeds the S3 key limit and releases the promotion fence', + async () => { + await activate() + const archived = randomUUID() + const current = randomUUID() + const prefix = `${tenantId}/${bucketId}/` + // The data key is 1020 bytes, so its .info sibling is 1025 and cannot exist in S3. + const name = longName(1020 - prefix.length - 37) + await seedHistory(name, archived, current, { archivedAt: threeDaysAgo() }) + const deletes = vi.spyOn(helper.adapter, 'deleteObjectsDetailed') + + await expect(runExecutor()).resolves.toMatchObject({ status: 'COMPLETED' }) + for (const [, keys] of deletes.mock.calls) { + expect(keys.every((key) => Buffer.byteLength(key) <= 1024)).toBe(true) + } + + const [state] = await fixtures.readState(bucketId, 'continuation, failure_count, last_error') + expect(state).toEqual({ continuation: null, failure_count: 0, last_error: null }) + expect(await versions(name)).toEqual([{ version: current, current: true }]) + await expectBlobAbsent(name, archived) + await helper.storage.from(bucketId).deleteObject(name, current) + expect(await versions(name)).toEqual([]) + } + ) + + it('expires an archived delete marker without a backend call', async () => { + await activate() + const name = 'history/marker.txt' + const version = randomUUID() + await helper.database.connection.query( + `INSERT INTO storage.objects (bucket_id, name, version, is_versioned, is_delete_marker, metadata, archived_at) + VALUES ($1, $2, $3, true, true, NULL, $4)`, + [bucketId, name, version, threeDaysAgo()] + ) + const deletes = vi.spyOn(helper.adapter, 'deleteObjectsDetailed') + + await expect(runExecutor()).resolves.toMatchObject({ status: 'COMPLETED' }) + expect(deletes).not.toHaveBeenCalled() + + const [state] = await fixtures.readState(bucketId, 'continuation, last_result') + expect(state.continuation).toBeNull() + expect(state.last_result.counters).toMatchObject({ + deleteMarkersDeleted: 1, + objectVersionsDeleted: 0, + }) + expect(await versions(name)).toEqual([]) + }) + + it('loses its idle claim when direct SQL removes the policy between evaluation and arm', async () => { + await activate() + const name = 'history/config-removed.txt' + const archived = randomUUID() + await seedHistory(name, archived, randomUUID(), { + archivedAt: threeDaysAgo(), + }) + const original = StoragePgDB.prototype.armLifecycleAttempt + vi.spyOn(StoragePgDB.prototype, 'armLifecycleAttempt').mockImplementationOnce(async function ( + this: StoragePgDB, + ...args + ) { + await fixtures.withServiceOperation('storage.s3.bucket.put_lifecycle', (transaction) => + transaction.query( + 'UPDATE storage.buckets SET lifecycle_configuration = NULL, lifecycle_configuration_generation = NULL WHERE id = $1', + [bucketId] + ) + ) + return original.apply(this, args) + }) + const deletes = vi.spyOn(helper.adapter, 'deleteObjectsDetailed') + + await expect(runExecutor()).resolves.toMatchObject({ status: 'LOST_CLAIM' }) + expect(deletes).not.toHaveBeenCalled() + await expect(fixtures.readState(bucketId, '1')).resolves.toEqual([]) + }) + + it.each(['findLifecycleBucket', 'armLifecycleAttempt'] as const)( + 'loses its claim without an error when the bucket is deleted before %s', + async (method) => { + await activate() + await helper.database.connection.query( + `INSERT INTO storage.objects (bucket_id, name, version, archived_at, is_versioned, is_delete_marker) + VALUES ($1, 'gone/a', gen_random_uuid()::text, '2000-01-01', true, true)`, + [bucketId] + ) + const original = StoragePgDB.prototype[method] as (...args: unknown[]) => Promise + vi.spyOn(StoragePgDB.prototype, method).mockImplementationOnce(async function ( + this: StoragePgDB, + ...args: unknown[] + ) { + await withDeleteEnabled(helper.database.connection, async (transaction) => { + await transaction.query('DELETE FROM storage.objects WHERE bucket_id = $1', [bucketId]) + await transaction.query('DELETE FROM storage.buckets WHERE id = $1', [bucketId]) + }) + return original.apply(this, args) as never + }) + + await expect(runExecutor()).resolves.toEqual({ status: 'LOST_CLAIM' }) + await expect(fixtures.readState(bucketId, '1')).resolves.toEqual([]) + } + ) + + it('leaves fresh history and its continuation untouched under a closed tenant gate', async () => { + await activate() + const name = 'history/paused.txt' + const archived = randomUUID() + const current = randomUUID() + await seedHistory(name, archived, current) + const before = await versions(name) + + await expect(runExecutor({ tenantLifecycleEnabled: false })).resolves.toMatchObject({ + status: 'PAUSED', + }) + + expect(await versions(name)).toEqual(before) + for (const version of [archived, current]) { + await expectBlobPresent(name, version) + } + const rows = await fixtures.readState(bucketId, 'continuation, claim_id, next_run_at') + expect(rows).toEqual([{ continuation: null, claim_id: null, next_run_at: expect.any(Date) }]) + }) + + it.each([ + { + label: 'enabled history', + status: 'ENABLED', + legacyNull: false, + isVersioned: true, + realClock: false, + }, + { + label: 'suspended history', + status: 'SUSPENDED', + legacyNull: false, + isVersioned: true, + realClock: false, + }, + { + label: 'legacy physical null version', + status: 'ENABLED', + legacyNull: true, + isVersioned: false, + realClock: false, + }, + { + label: 'logical null version under a physical UUID', + status: 'ENABLED', + legacyNull: false, + isVersioned: false, + realClock: false, + }, + { + label: 'two-day-old history at the real clock', + status: 'ENABLED', + legacyNull: false, + isVersioned: true, + realClock: true, + }, + ] as const)( + 'expires $label and preserves current bytes and sidecar-like siblings', + async ({ status, legacyNull, isVersioned, realClock }) => { + const name = 'history/expiry.txt' + const archived = legacyNull ? null : randomUUID() + const current = randomUUID() + await seedHistory(name, archived, current, { + isVersioned, + ...(realClock && { archivedAt: new Date(Date.now() - 2 * DAY_MS) }), + }) + if (legacyNull && storageBackendType === 'file') { + // Model missing legacy bytes followed by a real replacement upload. + await helper.adapter.deleteObject(storageS3Bucket, `${tenantId}/${bucketId}/${name}`, null) + await uploadFixture(name, current) + } + await activate(status) + if (legacyNull && storageBackendType === 's3') await uploadFixture(`${name}.info`, null) + expect(await versions(name)).toContainEqual({ version: archived, current: false }) + + await expect(runExecutor(realClock ? { now: () => new Date() } : {})).resolves.toMatchObject({ + status: 'COMPLETED', + }) + expect(await versions(name)).toEqual([{ version: current, current: true }]) + await expectBlobAbsent(name, archived) + await expectBlobPresent(name, current) + const currentObject = await helper.adapter.getObject( + storageS3Bucket, + `${tenantId}/${bucketId}/${name}`, + current + ) + if (!(currentObject.body instanceof Readable)) + throw new Error('Expected a readable object body') + expect(await text(currentObject.body)).toBe('x') + if (legacyNull && storageBackendType === 's3') { + await expect( + helper.adapter.headObject(storageS3Bucket, `${tenantId}/${bucketId}/${name}.info`, null) + ).resolves.toMatchObject({ size: 1 }) + } + } + ) + + it('expires an intact legacy file archived by a delete marker', async () => { + const name = 'history/legacy.txt' + const metadata = await uploadFixture(name, null) + await helper.database.connection.query( + `INSERT INTO storage.objects (bucket_id, name, version, is_versioned, metadata) + VALUES ($1, $2, NULL, false, $3::jsonb)`, + [bucketId, name, JSON.stringify(metadata)] + ) + await activate() + await helper.storage.from(bucketId).deleteObject(name) + const rows = await versions(name) + const marker = rows.find((row) => row.current)! + expect(marker.version).toEqual(expect.any(String)) + expect(rows).toContainEqual({ version: null, current: false }) + await expectBlobPresent(name, null) + + await expect(runExecutor()).resolves.toMatchObject({ status: 'COMPLETED' }) + expect(await versions(name)).toEqual([marker]) + await expectBlobAbsent(name, null) + }) + + it.each([ + { label: 'null metadata', metadata: null, bytes: 0 }, + { label: 'numeric string size', metadata: { size: '1' }, bytes: 1 }, + ])('continues past $label and preserves byte accounting', async ({ metadata, bytes }) => { + await activate() + const archivedAt = threeDaysAgo() + const first = { name: 'history/a.txt', archived: randomUUID(), current: randomUUID() } + const second = { name: 'history/b.txt', archived: randomUUID(), current: randomUUID() } + for (const row of [first, second]) { + await seedHistory(row.name, row.archived, row.current, { archivedAt }) + } + await helper.database.connection.query( + 'UPDATE storage.objects SET metadata = $4::jsonb WHERE bucket_id = $1 AND name = $2 AND version = $3', + [bucketId, first.name, first.archived, metadata === null ? null : JSON.stringify(metadata)] + ) + + const warning = vi.spyOn(logSchema, 'warning').mockImplementation(() => undefined) + await expect(runExecutor({ pageSize: 1 })).resolves.toMatchObject({ status: 'COMPLETED' }) + if (bytes === 0) { + expect(warning).toHaveBeenCalledExactlyOnceWith( + expect.anything(), + '[Lifecycle] Missing or invalid object sizes counted as zero bytes', + expect.objectContaining({ + tenantId, + metadata: JSON.stringify({ bucketId, invalidSizeCount: 1 }), + }) + ) + } else { + expect(warning).not.toHaveBeenCalled() + } + + const [result] = await fixtures.readState(bucketId, 'continuation, last_result') + expect(result).toMatchObject({ + continuation: null, + last_result: { counters: { objectVersionsDeleted: 2, bytesDeleted: bytes + 1 } }, + }) + for (const row of [first, second]) { + expect(await versions(row.name)).toEqual([{ version: row.current, current: true }]) + await expectBlobAbsent(row.name, row.archived) + await expectBlobPresent(row.name, row.current) + } + }) + + it.each([1, 2])('re-reads unstored work after a failure before arm %i', async (failAt) => { + await activate() + const archivedAt = threeDaysAgo() + const rows = historyRows(['retry/a', 'retry/b']) + for (const row of rows) await seedHistory(row.name, row.archived, row.current, { archivedAt }) + const now = new Date() + const failure = new Error('worker failed before arming') + const arm = StoragePgDB.prototype.armLifecycleAttempt + const attempts = new Set() + const arming = vi + .spyOn(StoragePgDB.prototype, 'armLifecycleAttempt') + .mockImplementation(async function (this: StoragePgDB, input) { + // The public method delegates into a transaction with the same input. + attempts.add(input.batch) + if (attempts.size === failAt) throw failure + return arm.call(this, input) + }) + await expect(runExecutor({ pageSize: 100, deleteBatchSize: 1, now: () => now })).rejects.toBe( + failure + ) + arming.mockRestore() + const [state] = await fixtures.readState(bucketId, 'continuation') + if (failAt === 1) expect(state.continuation).toBeNull() + else { + expect(state.continuation).toMatchObject({ + snapshotAt: now.toISOString(), + cursor: { name: rows[0].name }, + counters: { objectVersionsDeleted: 1 }, + }) + } + await makeDue() + const retryNow = new Date(now.getTime() + 15 * 60_000) + await expect( + runExecutor({ pageSize: 100, deleteBatchSize: 1, now: () => retryNow }) + ).resolves.toMatchObject({ + status: 'COMPLETED', + }) + const [completed] = await fixtures.readState(bucketId, 'last_result') + expect(completed.last_result).toMatchObject({ + snapshotAt: (failAt === 1 ? retryNow : now).toISOString(), + counters: { + versionsExamined: 2, + versionsEligible: 2, + objectVersionsDeleted: 2, + batchesCompleted: 2, + }, + }) + for (const row of rows) { + expect(await versions(row.name)).toEqual([{ version: row.current, current: true }]) + await expectBlobAbsent(row.name, row.archived) + } + }) + + it('keeps the outbox rows when arm commits but its acknowledgement is lost', async () => { + await activate() + const name = 'history/lost-ack.txt' + const archived = randomUUID() + const current = randomUUID() + await seedHistory(name, archived, current, { archivedAt: threeDaysAgo() }) + const original = StoragePgDB.prototype.armLifecycleAttempt + const lost = new Error('acknowledgement lost') + vi.spyOn(StoragePgDB.prototype, 'armLifecycleAttempt').mockImplementationOnce(async function ( + this: StoragePgDB, + input + ) { + await original.call(this, input) + throw lost + }) + const deletes = vi.spyOn(helper.adapter, 'deleteObjectsDetailed') + const now = new Date() + await expect(runExecutor({ now: () => now })).rejects.toBe(lost) + expect(deletes).not.toHaveBeenCalled() + const [state] = await fixtures.readState( + bucketId, + 'claim_id, failure_count, next_run_at, continuation' + ) + expect(state).toMatchObject({ + claim_id: null, + failure_count: 1, + next_run_at: new Date(now.getTime() + 5 * 60_000), + }) + expect(state.continuation).not.toHaveProperty('batch') + expect(await versions(name)).toEqual([{ version: current, current: true }]) + expect(await abandonedKeys()).toHaveLength(2) + + await makeDue() + await expect( + runExecutor({ now: () => new Date(now.getTime() + 5 * 60_000) }) + ).resolves.toMatchObject({ status: 'COMPLETED' }) + await sweep() + expect(await abandonedKeys()).toEqual([]) + await expectBlobAbsent(name, archived) + }) + + it('keeps the committed cursor and counters when an evaluated row is deleted before arming', async () => { + await activate() + const archivedAt = threeDaysAgo() + const rows = historyRows(['resume/a', 'resume/b', 'resume/c']) + for (const row of rows) await seedHistory(row.name, row.archived, row.current, { archivedAt }) + const arm = StoragePgDB.prototype.armLifecycleAttempt + const attempts = new Set() + const arming = vi + .spyOn(StoragePgDB.prototype, 'armLifecycleAttempt') + .mockImplementation(async function (this: StoragePgDB, input) { + if (attempts.size === 1 && !attempts.has(input.batch)) { + await helper.storage.from(bucketId).deleteObject(rows[1].name, rows[1].archived) + } + attempts.add(input.batch) + return arm.call(this, input) + }) + try { + await expect(runExecutor({ pageSize: 1 })).resolves.toMatchObject({ status: 'INELIGIBLE' }) + } finally { + arming.mockRestore() + } + const [state] = await fixtures.readState(bucketId, 'continuation, claim_id, failure_count') + expect(state).toMatchObject({ + claim_id: null, + failure_count: 0, + continuation: { cursor: { name: rows[0].name }, counters: { objectVersionsDeleted: 1 } }, + }) + + await expect(runExecutor({ pageSize: 1 })).resolves.toMatchObject({ status: 'COMPLETED' }) + const [completed] = await fixtures.readState(bucketId, 'last_result') + expect(completed.last_result.counters.objectVersionsDeleted).toBe(2) + for (const row of rows) { + expect(await versions(row.name)).toEqual([{ version: row.current, current: true }]) + } + }) + + it('persists the scanned cursor when arming finds a candidate ineligible', async () => { + await activate() + await helper.database.putLifecycleConfiguration(bucketId, { + rules: [ + { + status: 'Enabled', + filter: {}, + noncurrentVersionExpiration: { noncurrentDays: 1, newerNoncurrentVersions: 1 }, + }, + ], + }) + await helper.database.connection.query( + `INSERT INTO storage.objects (bucket_id, name, version, archived_at, is_versioned, is_delete_marker) + SELECT $1, 'keep/' || lpad(n::text, 3, '0'), gen_random_uuid()::text, '2000-01-01'::timestamptz, true, true + FROM generate_series(1, 10) AS n`, + [bucketId] + ) + const newer = randomUUID() + await helper.database.connection.query( + `INSERT INTO storage.objects (bucket_id, name, version, archived_at, is_versioned, is_delete_marker) + VALUES ($1, 'zcand', gen_random_uuid()::text, '2000-01-02', true, true), + ($1, 'zcand', $2, '2000-01-03', true, true)`, + [bucketId, newer] + ) + const evaluation = vi.spyOn(StoragePgDB.prototype, 'evaluateNoncurrentLifecyclePage') + const arm = StoragePgDB.prototype.armLifecycleAttempt + vi.spyOn(StoragePgDB.prototype, 'armLifecycleAttempt').mockImplementationOnce(async function ( + this: StoragePgDB, + input + ) { + await withDeleteEnabled(helper.database.connection, (tx) => + tx.query('DELETE FROM storage.objects WHERE bucket_id = $1 AND version = $2', [ + bucketId, + newer, + ]) + ) + return arm.call(this, input) + }) + + await expect(runExecutor({ pageSize: 2 })).resolves.toMatchObject({ status: 'INELIGIBLE' }) + const firstRunPages = evaluation.mock.calls.length + const [state] = await fixtures.readState( + bucketId, + 'claim_id, next_run_at <= clock_timestamp() AS due, continuation' + ) + expect(state).toMatchObject({ claim_id: null, due: true }) + expect(state.continuation?.cursor?.name).toBe('keep/010') + + evaluation.mockClear() + await expect(runExecutor({ pageSize: 2 })).resolves.toMatchObject({ status: 'COMPLETED' }) + expect(evaluation.mock.calls.length).toBeLessThan(firstRunPages) + }) + + const foreignKey = (options = '') => ({ + install: async (ids: string[]) => { + await helper.database.connection.query( + `CREATE TABLE public.lifecycle_probe_refs (object_id uuid REFERENCES storage.objects(id) ${options})` + ) + await helper.database.connection.query( + 'INSERT INTO public.lifecycle_probe_refs VALUES ($1)', + [ids[0]] + ) + }, + uninstall: () => + helper.database.connection.query('DROP TABLE IF EXISTS public.lifecycle_probe_refs'), + }) + const keepTrigger = (body: (ids: string[]) => string) => ({ + install: async (ids: string[]) => { + await helper.database.connection.query( + `CREATE FUNCTION public.lifecycle_probe_keep() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN ${body(ids)} RETURN OLD; END $$` + ) + await helper.database.connection.query( + `CREATE TRIGGER lifecycle_probe_keep BEFORE DELETE ON storage.objects + FOR EACH ROW EXECUTE FUNCTION public.lifecycle_probe_keep()` + ) + }, + uninstall: async () => { + await helper.database.connection.query( + 'DROP TRIGGER IF EXISTS lifecycle_probe_keep ON storage.objects' + ) + await helper.database.connection.query( + 'DROP FUNCTION IF EXISTS public.lifecycle_probe_keep()' + ) + }, + }) + + it.each<{ + veto: string + kept: number + pageSize?: number + error?: string + install: (ids: string[]) => Promise + uninstall: () => Promise + }>([ + { veto: 'a foreign key', kept: 1, ...foreignKey() }, + { veto: 'a foreign key on the oldest page', kept: 1, pageSize: 1, ...foreignKey() }, + { + veto: 'a deferred foreign key', + kept: 1, + ...foreignKey('DEFERRABLE INITIALLY DEFERRED'), + }, + { + veto: 'a BEFORE DELETE trigger returning NULL for one row of the batch', + kept: 1, + ...keepTrigger((ids) => `IF OLD.id = '${ids[0]}' THEN RETURN NULL; END IF;`), + }, + { + veto: 'a BEFORE DELETE trigger returning NULL', + kept: 2, + ...keepTrigger(() => `IF OLD.bucket_id = '${bucketId}' THEN RETURN NULL; END IF;`), + }, + { + veto: 'a BEFORE DELETE trigger raising a transient error', + kept: 2, + error: 'ResourceLocked', + ...keepTrigger( + (ids) => + `IF OLD.id = '${ids[0]}' THEN RAISE EXCEPTION 'busy' USING ERRCODE = '55P03'; END IF;` + ), + }, + ])( + 'keeps the bytes of versions held by $veto and deletes the rest', + async ({ kept, pageSize = 100, error, install, uninstall }) => { + await activate() + const rows = ['veto/a.txt', 'veto/b.txt'].map((name) => ({ + name, + archived: randomUUID(), + })) + for (const row of rows) { + await seedHistory(row.name, row.archived, randomUUID(), { + archivedAt: threeDaysAgo(), + }) + } + const ids = ( + await helper.database.connection.query<{ id: string }>( + 'SELECT id FROM storage.objects WHERE bucket_id = $1 AND name = $2 AND version = $3', + [bucketId, rows[0].name, rows[0].archived] + ) + ).rows.map((row) => row.id) + + try { + await install(ids) + const run = runExecutor({ pageSize }) + if (error) await expect(run).rejects.toMatchObject({ code: error }) + else await expect(run).resolves.toMatchObject({ status: 'COMPLETED' }) + for (const row of rows.slice(0, kept)) { + await expectBlobPresent(row.name, row.archived) + expect(await versions(row.name)).toHaveLength(2) + } + for (const row of rows.slice(kept)) { + expect(await versions(row.name)).toEqual([{ version: expect.any(String), current: true }]) + await expectBlobAbsent(row.name, row.archived) + } + const [state] = await fixtures.readState( + bucketId, + 'failure_count, continuation, last_error, last_result' + ) + expect(state).toMatchObject({ failure_count: error ? 1 : 0 }) + expect(state.continuation?.batch).toBeUndefined() + if (!error) { + expect(state.last_result.counters.objectVersionsDeleted).toBe(rows.length - kept) + } + expect(Boolean(state.last_error)).toBe(Boolean(error)) + } finally { + await uninstall() + } + + await makeDue() + await expect( + runExecutor({ now: () => new Date(Date.now() + 4 * DAY_MS) }) + ).resolves.toMatchObject({ status: 'COMPLETED' }) + for (const row of rows) { + expect(await versions(row.name)).toEqual([{ version: expect.any(String), current: true }]) + await expectBlobAbsent(row.name, row.archived) + } + } + ) + + it('finishes a 100-row page in bounded delete transactions without overflowing on vetoes', async (context) => { + const server = await helper.database.connection.query<{ version: string }>( + `SELECT current_setting('server_version_num') AS version` + ) + if (Number(server.rows[0].version) < 160000) context.skip() + await activate() + await helper.database.connection.query( + `INSERT INTO storage.objects (bucket_id, name, version, archived_at, is_versioned, is_delete_marker) + SELECT $1, 'bounded/' || lpad(n::text, 3, '0'), gen_random_uuid()::text, + '2000-01-01'::timestamptz, true, true + FROM generate_series(1, 100) AS n`, + [bucketId] + ) + try { + await helper.database.connection.query(` + CREATE TABLE public.lifecycle_batch_audit (transaction_id text, overflowed boolean); + CREATE FUNCTION public.lifecycle_batch_veto() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + IF OLD.name = 'bounded/001' THEN + RAISE EXCEPTION 'keep this marker' USING ERRCODE = '42501'; + END IF; + RETURN OLD; + END $$; + CREATE TRIGGER lifecycle_batch_veto BEFORE DELETE ON storage.objects + FOR EACH ROW EXECUTE FUNCTION public.lifecycle_batch_veto(); + CREATE FUNCTION public.lifecycle_batch_audit_fn() RETURNS trigger + LANGUAGE plpgsql SECURITY DEFINER AS $$ + BEGIN + IF (SELECT count(*) FROM removed) > 1 THEN + RAISE EXCEPTION 'delete rows individually' USING ERRCODE = '42501'; + END IF; + PERFORM pg_stat_clear_snapshot(); + INSERT INTO public.lifecycle_batch_audit + SELECT pg_current_xact_id()::text, subxact.subxact_overflowed + FROM pg_stat_get_backend_idset() AS backend, + LATERAL pg_stat_get_backend_subxact(backend) AS subxact + WHERE pg_stat_get_backend_pid(backend) = pg_backend_pid(); + RETURN NULL; + END $$; + CREATE TRIGGER lifecycle_batch_audit AFTER DELETE ON storage.objects + REFERENCING OLD TABLE AS removed FOR EACH STATEMENT + EXECUTE FUNCTION public.lifecycle_batch_audit_fn(); + `) + + await expect(runExecutor({ pageSize: 96 })).resolves.toMatchObject({ status: 'COMPLETED' }) + const audit = await helper.database.connection.query( + `SELECT count(*)::integer AS deleted, bool_or(overflowed) AS overflowed + FROM public.lifecycle_batch_audit GROUP BY transaction_id ORDER BY deleted` + ) + expect(audit.rows).toEqual([ + { deleted: 4, overflowed: false }, + { deleted: 47, overflowed: false }, + { deleted: 48, overflowed: false }, + ]) + const remaining = await helper.database.connection.query( + 'SELECT name FROM storage.objects WHERE bucket_id = $1', + [bucketId] + ) + expect(remaining.rows).toEqual([{ name: 'bounded/001' }]) + const [state] = await fixtures.readState(bucketId, 'continuation, last_result') + expect(state.continuation).toBeNull() + expect(state.last_result.counters).toMatchObject({ + versionsExamined: 100, + versionsEligible: 100, + deleteMarkersDeleted: 99, + batchesCompleted: 3, + }) + } finally { + await helper.database.connection.query(` + DROP TRIGGER IF EXISTS lifecycle_batch_veto ON storage.objects; + DROP TRIGGER IF EXISTS lifecycle_batch_audit ON storage.objects; + DROP FUNCTION IF EXISTS public.lifecycle_batch_veto(); + DROP FUNCTION IF EXISTS public.lifecycle_batch_audit_fn(); + DROP TABLE IF EXISTS public.lifecycle_batch_audit; + `) + } + }) + + it('counts a version once when a second worker takes over during the first worker delete', async () => { + await activate() + const name = 'history/lease-race.txt' + const archived = randomUUID() + const current = randomUUID() + await seedHistory(name, archived, current, { archivedAt: threeDaysAgo() }) + const deleteObjects = helper.adapter.deleteObjectsDetailed.bind(helper.adapter) + let second: unknown + vi.spyOn(helper.adapter, 'deleteObjectsDetailed').mockImplementationOnce( + async (bucket, keys, options) => { + await helper.database.connection.query( + "UPDATE storage.bucket_lifecycle_states SET claim_until = '2000-01-01' WHERE bucket_id = $1", + [bucketId] + ) + second = await runExecutor({ now: () => new Date(Date.now() + 3 * DAY_MS + 60_000) }) + return deleteObjects(bucket, keys, options) + } + ) + + await expect(runExecutor()).resolves.toMatchObject({ status: 'LOST_CLAIM' }) + + expect(second).toMatchObject({ status: 'COMPLETED' }) + const [state] = await fixtures.readState(bucketId, 'continuation, last_result') + expect(state.continuation).toBeNull() + expect(state.last_result.counters.objectVersionsDeleted).toBe(1) + expect(await versions(name)).toEqual([{ version: current, current: true }]) + await expectBlobAbsent(name, archived) + }) + + it('loses its claim on the next arm when a policy write lands during a batch, then finishes under the new generation', async () => { + await activate() + const names = ['history/policy-a.txt', 'history/policy-b.txt'] + const archived = names.map(() => randomUUID()) + for (const [index, name] of names.entries()) { + await seedHistory(name, archived[index], randomUUID(), { + archivedAt: new Date(Date.now() - (4 - index) * DAY_MS), + }) + } + const deleteObjects = helper.adapter.deleteObjectsDetailed.bind(helper.adapter) + vi.spyOn(helper.adapter, 'deleteObjectsDetailed').mockImplementationOnce( + async (bucket, keys, options) => { + await helper.database.putLifecycleConfiguration(bucketId, { + rules: [ + { ...configuration.rules[0], noncurrentVersionExpiration: { noncurrentDays: 2 } }, + ], + }) + return deleteObjects(bucket, keys, options) + } + ) + + await expect(runExecutor({ pageSize: 100, deleteBatchSize: 1 })).resolves.toMatchObject({ + status: 'LOST_CLAIM', + }) + + const [state] = await fixtures.readState(bucketId, 'continuation, failure_count') + expect(state).toEqual({ continuation: null, failure_count: 0 }) + expect(await versions(names[0])).toHaveLength(1) + await expectBlobAbsent(names[0], archived[0]) + expect(await versions(names[1])).toHaveLength(2) + + await makeDue() + await expect(runExecutor({ pageSize: 100, deleteBatchSize: 1 })).resolves.toMatchObject({ + status: 'COMPLETED', + }) + expect(await versions(names[1])).toHaveLength(1) + await expectBlobAbsent(names[1], archived[1]) + }) + + it('keeps the blob of an abandoned record whose version belongs to a live object', async () => { + const name = 'history/reused.txt' + const version = randomUUID() + await uploadFixture(name, version) + await helper.database.connection.query( + `INSERT INTO storage.objects (bucket_id, name, version, is_versioned, metadata) + VALUES ($1, $2, $3, true, '{"size":1}'::jsonb)`, + [bucketId, name, version] + ) + await helper.database.connection.query( + `INSERT INTO storage.lifecycle_abandoned_artifacts (bucket_id, key, name, version) + VALUES ($1, $2, $3, $4)`, + [bucketId, withOptionalVersion(name, version), name, version] + ) + const warning = vi.spyOn(logSchema, 'warning').mockImplementation(() => {}) + + await sweep() + + expect(await abandonedKeys()).toEqual([]) + await expectBlobPresent(name, version) + expect(warning).toHaveBeenCalledOnce() + }) +})