feat: add noncurrent lifecycle executor

.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

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' }] }),

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)
 

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
 // ============================================================================

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