diff --git a/src/storage/database/adapter.ts b/src/storage/database/adapter.ts index 67a7e9c26..f5684f8ae 100644 --- a/src/storage/database/adapter.ts +++ b/src/storage/database/adapter.ts @@ -10,10 +10,8 @@ import { LifecycleArmAttemptInput, LifecycleArmAttemptResult, LifecycleBucket, - LifecycleClaimedState, LifecycleClaimIdentity, LifecycleClaimInput, - LifecycleCommitAttemptInput, LifecycleCompleteRunInput, LifecycleDispatchState, LifecycleEvaluationPage, @@ -219,14 +217,11 @@ export interface Database { deleteLifecycleConfiguration(bucketId: string): Promise - listDueLifecycleBuckets(limit: number, options?: { inFlightOnly?: boolean }): Promise + listDueLifecycleBuckets(limit: number): Promise - findLifecycleDispatchState(options: { - inFlightOnly?: boolean - runBudgetMs: number - }): Promise + findLifecycleDispatchState(options: { runBudgetMs: number }): Promise - claimLifecycleBucket(input: LifecycleClaimInput): Promise + claimLifecycleBucket(input: LifecycleClaimInput): Promise revalidateLifecycleClaim(input: LifecycleClaimIdentity): Promise @@ -236,14 +231,6 @@ export interface Database { armLifecycleAttempt(input: LifecycleArmAttemptInput): Promise - revalidateLifecycleRecoveryAttempt( - input: LifecycleClaimIdentity, - attemptId: string, - leaseMs: number - ): Promise - - commitLifecycleAttempt(input: LifecycleCommitAttemptInput): Promise - listLifecycleAbandonedArtifacts( attemptedBefore: string, limit: number diff --git a/src/storage/database/pg.ts b/src/storage/database/pg.ts index 4cdd3cbc6..96ef69a44 100644 --- a/src/storage/database/pg.ts +++ b/src/storage/database/pg.ts @@ -1,5 +1,4 @@ import { randomUUID } from 'node:crypto' -import { isDeepStrictEqual } from 'node:util' import { type DatabaseExecutor, type DatabaseStatement, @@ -16,16 +15,12 @@ import { logger, logSchema } from '@internal/monitoring' import { dbQueryPerformance } from '@internal/monitoring/metrics' import { ObjectMetadata } from '@storage/backend' import { + advanceLifecycleBatch, assertLifecycleSchemaReady, - commitLifecycleBatchResults, compileLifecycleEvaluationRules, decodeLifecycleContinuation, - encodeLifecycleContinuation, + lifecycleArtifactKeys, lifecycleConfigurationsEqual, - pendingArtifacts, - recordLifecycleArtifactOutcomes, - recordLifecycleMetadataDeletions, - skipLifecycleBatch, } from '@storage/lifecycle' import { type LifecycleDeletedObjectRow, @@ -44,12 +39,9 @@ import { LifecycleArmAttemptInput, LifecycleArmAttemptResult, LifecycleBucket, - LifecycleClaimedState, LifecycleClaimIdentity, LifecycleClaimInput, - LifecycleCommitAttemptInput, LifecycleCompleteRunInput, - LifecycleContinuation, LifecycleDispatchState, LifecycleEvaluationPage, LifecycleReleaseClaimInput, @@ -163,7 +155,6 @@ const HEALTHCHECK_QUERY_OPTIONS: UnscopedQueryOptions = Object.freeze({ timeoutMs: databaseStatementTimeout, }) const LIFECYCLE_LOCK_TIMEOUT_MS = 5000 -const IN_FLIGHT_ONLY = "AND state.continuation #> '{batch,inFlight}' IS NOT NULL" const LIFECYCLE_BUCKET_MATCH = `bucket.type = 'STANDARD'` const LIFECYCLE_CLAIM_IDENTITY = `state.bucket_id = $1 AND state.claim_id = $2::uuid` const LIFECYCLE_BUCKET_COLUMNS = [ @@ -184,13 +175,6 @@ interface LifecycleStateRow extends QueryResultRow { failure_count: number } -interface LifecycleClaimRow - extends LifecycleStateRow, - Pick< - LifecycleBucket, - 'versioning_status' | 'lifecycle_configuration' | 'lifecycle_configuration_generation' - > {} - async function executeQuery( db: DatabaseExecutor, statement: string | DatabaseStatement, @@ -3329,7 +3313,7 @@ export class StoragePgDB implements Database { forShare: true, }) assertStandardLifecycleBucket(bucket) - // Row locks also serialize current promotions with journal publication. + // Row locks also serialize current promotions with the metadata delete. // Eligibility is checked in the next statement after any writer has committed. const locked = await this.runQuery('LockLifecycleObjects', (db, signal) => this.query<{ id: string }>( @@ -3365,10 +3349,7 @@ export class StoragePgDB implements Database { } } - async listDueLifecycleBuckets( - limit: number, - options: { inFlightOnly?: boolean } = {} - ): Promise { + async listDueLifecycleBuckets(limit: number): Promise { const result = await this.runQuery('ListDueLifecycleBuckets', async (db, signal) => this.query<{ bucket_id: string }>( db, @@ -3379,7 +3360,6 @@ export class StoragePgDB implements Database { AND ${LIFECYCLE_BUCKET_MATCH} WHERE state.next_run_at <= statement_timestamp() AND (state.claim_until IS NULL OR state.claim_until < statement_timestamp()) - ${options.inFlightOnly ? IN_FLIGHT_ONLY : ''} ORDER BY state.next_run_at, state.bucket_id LIMIT $1`, values: [limit], @@ -3391,14 +3371,13 @@ export class StoragePgDB implements Database { } async findLifecycleDispatchState(options: { - inFlightOnly?: boolean runBudgetMs: number }): Promise { const result = await this.runQuery('FindLifecycleDispatchState', async (db, signal) => { return this.query<{ next_dispatch_at: Date | null max_failure_count: number - oldest_in_flight_started_at: string | null + oldest_in_flight_started_at: Date | null has_abandoned_artifacts: boolean }>( db, @@ -3410,12 +3389,9 @@ export class StoragePgDB implements Database { state.claim_until, state.last_started_at + $1::bigint * interval '1 millisecond' ) END - )) FILTER ( - WHERE state.next_run_at IS NOT NULL - ${options.inFlightOnly ? IN_FLIGHT_ONLY : ''} - ) AS next_dispatch_at, + )) FILTER (WHERE state.next_run_at IS NOT NULL) AS next_dispatch_at, COALESCE(MAX(state.failure_count), 0) AS max_failure_count, - MIN(state.continuation #>> '{batch,inFlight,startedAt}') + (SELECT MIN(attempted_at) FROM storage.lifecycle_abandoned_artifacts) AS oldest_in_flight_started_at, EXISTS (SELECT 1 FROM storage.lifecycle_abandoned_artifacts) AS has_abandoned_artifacts @@ -3431,23 +3407,17 @@ export class StoragePgDB implements Database { }) const row = result.rows[0] - const startedAt = row.oldest_in_flight_started_at return { nextDispatchAt: row.next_dispatch_at, maxFailureCount: row.max_failure_count, - oldestInFlightStartedAt: - startedAt && Number.isFinite(Date.parse(startedAt)) - ? new Date(startedAt).toISOString() - : null, + oldestInFlightStartedAt: row.oldest_in_flight_started_at?.toISOString() ?? null, hasAbandonedArtifacts: row.has_abandoned_artifacts, } } - async claimLifecycleBucket( - input: LifecycleClaimInput - ): Promise { + async claimLifecycleBucket(input: LifecycleClaimInput): Promise { const result = await this.runQuery('ClaimLifecycleBucket', async (db, signal) => { - return this.query( + return this.query( db, { text: ` @@ -3462,10 +3432,7 @@ export class StoragePgDB implements Database { AND (state.claim_until IS NULL OR state.claim_until < clock_timestamp()) AND bucket.id = state.bucket_id AND ${LIFECYCLE_BUCKET_MATCH} - RETURNING state.*, - bucket.versioning_status, - bucket.lifecycle_configuration, - bucket.lifecycle_configuration_generation + RETURNING state.* `, values: [input.bucketId, input.claimId, input.leaseMs], }, @@ -3476,22 +3443,14 @@ export class StoragePgDB implements Database { const row = result.rows[0] if (!row) return undefined try { - return { - ...mapLifecycleState(row), - bucket: { - versioning_status: row.versioning_status, - lifecycle_configuration: row.lifecycle_configuration, - lifecycle_configuration_generation: row.lifecycle_configuration_generation, - }, - } + return mapLifecycleState(row) } catch (error) { throw Object.assign(error as Error, { failureCount: row.failure_count }) } } async revalidateLifecycleClaim( - input: LifecycleClaimIdentity, - options: { attemptId?: string; forUpdate?: boolean } = {} + input: LifecycleClaimIdentity ): Promise { const result = await this.runQuery('RevalidateLifecycleClaim', async (db, signal) => { return this.query( @@ -3505,10 +3464,8 @@ export class StoragePgDB implements Database { AND ${LIFECYCLE_BUCKET_MATCH} WHERE ${LIFECYCLE_CLAIM_IDENTITY} AND state.claim_until > clock_timestamp() - AND ($3::text IS NULL OR state.continuation #>> '{batch,inFlight,attemptId}' = $3) - ${options.forUpdate ? 'FOR UPDATE OF state' : ''} `, - values: [...lifecycleClaimIdentityValues(input), options.attemptId ?? null], + values: lifecycleClaimIdentityValues(input), }, signal ) @@ -3535,9 +3492,9 @@ export class StoragePgDB implements Database { return this.withTransaction((db) => db.armLifecycleAttempt(input)) } const continuation = decodeLifecycleContinuation(input.continuation) - const batch = continuation.batch - if (!batch?.inFlight || batch.versions.length === 0) { - throw ERRORS.InvalidParameter('lifecycle attempt continuation') + const { batch } = input + if (batch.versions.length === 0) { + throw ERRORS.InvalidParameter('lifecycle attempt batch') } if (batch.versions.length > LIFECYCLE_MAX_DELETE_BATCH_SIZE) { throw ERRORS.InvalidParameter( @@ -3560,8 +3517,7 @@ export class StoragePgDB implements Database { const { cutoffs, newerCounts, maximumNewerProbe } = lifecycleRuleParameters(rules) - // Recheck the exact archived rows and policy under the writer locks before - // durably authorizing backend deletion. + // Recheck the exact archived rows and policy under the writer locks before deleting. const result = await this.runQuery('ArmLifecycleAttempt', (db, signal) => this.query( db, @@ -3569,23 +3525,23 @@ export class StoragePgDB implements Database { text: ` WITH targets AS MATERIALIZED ( SELECT * - FROM unnest($14::uuid[], $6::text[], $7::text[], $8::boolean[]) + FROM unnest($13::uuid[], $5::text[], $6::text[], $7::boolean[]) AS target(id, name, version, is_delete_marker) ), rules(cutoff_at, newer_noncurrent_versions) AS MATERIALIZED ( - SELECT * FROM unnest($9::timestamptz[], $10::integer[]) + SELECT * FROM unnest($8::timestamptz[], $9::integer[]) ), eligible_targets AS MATERIALIZED ( SELECT count(*)::integer AS eligible_count FROM targets AS target JOIN storage.objects AS object_row ON object_row.id = target.id - WHERE object_row.id = ANY($13::uuid[]) + WHERE object_row.id = ANY($12::uuid[]) AND object_row.bucket_id = $1 AND (object_row.name, object_row.version) IS NOT DISTINCT FROM (target.name, target.version) AND object_row.archived_at IS NOT NULL AND object_row.is_delete_marker = target.is_delete_marker - AND object_row.archived_at <= $11::timestamptz + AND object_row.archived_at <= $10::timestamptz AND EXISTS ( SELECT 1 FROM rules WHERE object_row.archived_at < rules.cutoff_at @@ -3595,19 +3551,17 @@ export class StoragePgDB implements Database { WHERE newer.bucket_id = $1 AND newer.name COLLATE "C" = object_row.name COLLATE "C" AND newer.archived_at > object_row.archived_at - AND newer.archived_at <= $11::timestamptz - ORDER BY newer.archived_at LIMIT $12::integer + AND newer.archived_at <= $10::timestamptz + ORDER BY newer.archived_at LIMIT $11::integer ) AS bounded_newer ) >= rules.newer_noncurrent_versions) ) ) UPDATE storage.bucket_lifecycle_states AS state - SET continuation = CASE WHEN eligible_targets.eligible_count = cardinality($6::text[]) - THEN $5::jsonb ELSE state.continuation END, - claim_id = CASE WHEN eligible_targets.eligible_count = cardinality($6::text[]) + SET claim_id = CASE WHEN eligible_targets.eligible_count = cardinality($5::text[]) THEN state.claim_id ELSE NULL END, - claim_until = CASE WHEN eligible_targets.eligible_count = cardinality($6::text[]) + claim_until = CASE WHEN eligible_targets.eligible_count = cardinality($5::text[]) THEN clock_timestamp() + $4::bigint * interval '1 millisecond' ELSE NULL END, next_run_at = clock_timestamp(), @@ -3619,14 +3573,12 @@ export class StoragePgDB implements Database { AND ${LIFECYCLE_BUCKET_MATCH} AND bucket.versioning_status <> 'DISABLED' AND bucket.lifecycle_configuration_generation = $3::uuid - AND state.continuation #> '{batch,inFlight}' IS NULL RETURNING state.* `, values: [ ...lifecycleClaimIdentityValues(input), continuation.generation, input.leaseMs, - JSON.stringify(encodeLifecycleContinuation(continuation)), batch.versions.map((version) => version.name), batch.versions.map((version) => version.version), batch.versions.map((version) => version.artifacts.length === 0), @@ -3645,7 +3597,7 @@ export class StoragePgDB implements Database { if (!result.rows[0]) return { status: 'fenced' } if (!result.rows[0].claim_id) return { status: 'ineligible' } - // The metadata delete and armed journal commit together. + // The metadata delete, outbox rows and cursor advance commit together. // No new foreign-key reference can be added after this transaction commits. const deleted = await serviceDatabase.runQuery('DeleteLifecycleMetadata', (db, signal) => serviceDatabase.query( @@ -3665,27 +3617,47 @@ export class StoragePgDB implements Database { metadata: JSON.stringify({ bucketId: input.bucketId, vetoed }), }) } - const next = - versions.length === 0 - ? skipLifecycleBatch(continuation) - : recordLifecycleMetadataDeletions( - { ...continuation, batch: { ...batch, versions } }, - lifecycleDeletionCounters(deleted.rows, { - tenantId: this.tenantId, - bucketId: input.bucketId, - }) - ) + const artifacts = versions.flatMap(({ name, version, artifacts }) => + artifacts.length === 0 + ? [] + : lifecycleArtifactKeys(name, version).map((key) => ({ + bucketId: input.bucketId, + key, + name, + version, + })) + ) + const next = advanceLifecycleBatch( + continuation, + batch, + lifecycleDeletionCounters(deleted.rows, { + tenantId: this.tenantId, + bucketId: input.bucketId, + }) + ) const saved = await serviceDatabase.runQuery('SaveLifecycleMetadataDeletions', (db, signal) => serviceDatabase.query( db, { - text: `UPDATE storage.bucket_lifecycle_states AS state - SET continuation = $3::jsonb, updated_at = clock_timestamp() - WHERE ${LIFECYCLE_CLAIM_IDENTITY} - RETURNING state.*`, + text: `WITH outbox AS ( + INSERT INTO storage.lifecycle_abandoned_artifacts (bucket_id, key, name, version) + SELECT $1, artifact.key, artifact.name, artifact.version + FROM unnest($5::text[], $6::text[], $7::text[]) AS artifact(key, name, version) + ON CONFLICT (bucket_id, key) DO NOTHING + ) + UPDATE storage.bucket_lifecycle_states AS state + SET continuation = $3::jsonb, + claim_until = clock_timestamp() + $4::bigint * interval '1 millisecond', + updated_at = clock_timestamp() + WHERE ${LIFECYCLE_CLAIM_IDENTITY} + RETURNING state.*`, values: [ ...lifecycleClaimIdentityValues(input), - JSON.stringify(encodeLifecycleContinuation(next)), + JSON.stringify(next), + input.leaseMs, + artifacts.map(({ key }) => key), + artifacts.map(({ name }) => name), + artifacts.map(({ version }) => version), ], }, signal @@ -3693,119 +3665,7 @@ export class StoragePgDB implements Database { ) if (!saved.rows[0]) throw new Error('Lifecycle claim disappeared while arming metadata deletions') - return { status: 'armed', state: mapLifecycleState(saved.rows[0]) } - } - - async revalidateLifecycleRecoveryAttempt( - input: LifecycleClaimIdentity, - attemptId: string, - leaseMs: number - ): Promise { - const result = await this.runQuery('RevalidateLifecycleRecoveryAttempt', (db, signal) => - this.query( - db, - { - text: ` - UPDATE storage.bucket_lifecycle_states AS state - SET claim_until = clock_timestamp() + $4::bigint * interval '1 millisecond', - next_run_at = clock_timestamp(), - updated_at = clock_timestamp() - FROM storage.buckets AS bucket - WHERE ${LIFECYCLE_CLAIM_IDENTITY} - AND state.claim_until > clock_timestamp() - AND state.continuation #>> '{batch,inFlight,attemptId}' = $3 - AND bucket.id = state.bucket_id - AND ${LIFECYCLE_BUCKET_MATCH} - RETURNING state.* - `, - values: [...lifecycleClaimIdentityValues(input), attemptId, leaseMs], - }, - signal - ) - ) - return result.rows[0] ? mapLifecycleState(result.rows[0]) : undefined - } - - async commitLifecycleAttempt( - input: LifecycleCommitAttemptInput - ): Promise { - if (!this.options.tnx) { - return this.withTransaction((database) => database.commitLifecycleAttempt(input)) - } - - const { continuation } = input - if (continuation.batch?.inFlight?.attemptId !== input.attemptId) { - throw ERRORS.InvalidParameter('lifecycle attempt identity') - } - - const serviceDatabase = this.asSuperUser() - - const state = await serviceDatabase.revalidateLifecycleClaim(input, { - attemptId: input.attemptId, - forUpdate: true, - }) - if (!state) return undefined - if (!state.continuation || !sameFrozenLifecycleBatch(state.continuation, continuation)) { - throw ERRORS.InvalidRequest('Lifecycle attempt artifact set changed before completion') - } - - const outcomes = new Map( - continuation.batch!.versions.flatMap(({ artifacts }) => - artifacts.map(({ key, ...outcome }) => [key, outcome] as const) - ) - ) - const merged = recordLifecycleArtifactOutcomes(state.continuation, outcomes) - const persisted = commitLifecycleBatchResults(merged, input.abandonPending) - - const result = await serviceDatabase.runQuery('CommitLifecycleAttemptState', (db, signal) => - serviceDatabase.query( - db, - { - text: ` - UPDATE storage.bucket_lifecycle_states AS state - SET continuation = $4::jsonb, - next_run_at = clock_timestamp(), - updated_at = clock_timestamp() - WHERE ${LIFECYCLE_CLAIM_IDENTITY} - AND continuation #>> '{batch,inFlight,attemptId}' = $3 - RETURNING * - `, - values: [ - ...lifecycleClaimIdentityValues(input), - input.attemptId, - JSON.stringify(encodeLifecycleContinuation(persisted)), - ], - }, - signal - ) - ) - if (!result.rows[0]) return undefined - if (input.abandonPending) { - const pending = pendingArtifacts(merged) - if (pending.length > 0) { - await serviceDatabase.runQuery('PersistLifecycleAbandonedArtifacts', (db, signal) => - serviceDatabase.query( - db, - { - text: ` - INSERT INTO storage.lifecycle_abandoned_artifacts (bucket_id, key, name, version) - SELECT $1, artifact.key, artifact.name, artifact.version - FROM unnest($2::text[], $3::text[], $4::text[]) AS artifact(key, name, version) - ON CONFLICT DO NOTHING - `, - values: [ - input.bucketId, - pending.map(({ key }) => key), - pending.map(({ name }) => name), - pending.map(({ version }) => version), - ], - }, - signal - ) - ) - } - } - return mapLifecycleState(result.rows[0]) + return { status: 'armed', state: mapLifecycleState(saved.rows[0]), artifacts } } async listLifecycleAbandonedArtifacts( @@ -3894,11 +3754,8 @@ export class StoragePgDB implements Database { async releaseLifecycleClaim(input: LifecycleReleaseClaimInput): Promise { const { continuation } = input - if (continuation?.batch?.inFlight) { - throw ERRORS.InvalidParameter('lifecycle armed journal') - } if (continuation === undefined && input.nextRunAt === null) { - throw ERRORS.InvalidParameter('lifecycle recovery schedule') + throw ERRORS.InvalidParameter('lifecycle release without a schedule') } const result = await this.runQuery('ReleaseLifecycleClaim', (db, signal) => this.query( @@ -3907,38 +3764,26 @@ export class StoragePgDB implements Database { text: ` UPDATE storage.bucket_lifecycle_states AS state SET continuation = CASE WHEN $6::boolean THEN $3::jsonb ELSE continuation END, - next_run_at = CASE - WHEN NOT $6::boolean AND continuation #> '{batch,inFlight}' IS NOT NULL - THEN LEAST($4::timestamptz, $7::timestamptz) - ELSE $4::timestamptz - END, + next_run_at = $4::timestamptz, claim_id = NULL, claim_until = NULL, last_completed_at = clock_timestamp(), - failure_count = CASE - WHEN $5::jsonb IS NOT NULL THEN failure_count + 1 - WHEN (CASE WHEN $6::boolean THEN $3::jsonb ELSE continuation END) #> '{batch,inFlight}' IS NULL THEN 0 - ELSE failure_count - END, + failure_count = CASE WHEN $5::jsonb IS NOT NULL THEN failure_count + 1 ELSE 0 END, last_error = COALESCE($5::jsonb, last_error), updated_at = clock_timestamp() WHERE ${LIFECYCLE_CLAIM_IDENTITY} AND ( NOT $6::boolean OR state.continuation IS NULL - OR ( - state.continuation @> '{"continuationVersion":1}'::jsonb - AND state.continuation #> '{batch,inFlight}' IS NULL - ) + OR state.continuation @> '{"continuationVersion":1}'::jsonb ) `, values: [ ...lifecycleClaimIdentityValues(input), - continuation == null ? null : JSON.stringify(encodeLifecycleContinuation(continuation)), + continuation == null ? null : JSON.stringify(continuation), input.nextRunAt, input.error === undefined ? null : JSON.stringify(input.error), continuation !== undefined, - input.maxInFlightRetryAt ?? null, ], }, signal @@ -3980,10 +3825,7 @@ export class StoragePgDB implements Database { AND bucket.id = state.bucket_id AND ( state.continuation IS NULL - OR ( - state.continuation @> '{"continuationVersion":1}'::jsonb - AND state.continuation #> '{batch,inFlight}' IS NULL - ) + OR state.continuation @> '{"continuationVersion":1}'::jsonb ) `, values: [ @@ -4286,20 +4128,3 @@ function mapLifecycleState(row: LifecycleStateRow): LifecycleState { function lifecycleClaimIdentityValues(input: LifecycleClaimIdentity): unknown[] { return [input.bucketId, input.claimId] } - -function sameFrozenLifecycleBatch( - persisted: LifecycleContinuation, - proposed: LifecycleContinuation -): boolean { - const frozen = ({ batch }: LifecycleContinuation) => - batch && [ - batch.keySeparator, - batch.versions.map(({ id, name, version, artifacts }) => [ - id, - name, - version, - artifacts.length, - ]), - ] - return isDeepStrictEqual(frozen(persisted), frozen(proposed)) -} diff --git a/src/storage/lifecycle/continuation.test.ts b/src/storage/lifecycle/continuation.test.ts index eb9312ce2..81f726e12 100644 --- a/src/storage/lifecycle/continuation.test.ts +++ b/src/storage/lifecycle/continuation.test.ts @@ -1,127 +1,31 @@ import { randomUUID } from 'node:crypto' import { withOptionalVersion } from '@storage/backend/adapter' -import { - armedContinuation, - newContinuation, - stagedContinuation, -} from '../../test/utils/lifecycle-continuation' +import { newContinuation } from '../../test/utils/lifecycle-continuation' import { advanceEmptyLifecyclePage, - armLifecycleAttempt, - commitLifecycleBatchResults, + advanceLifecycleBatch, decodeLifecycleContinuation, - encodeLifecycleContinuation, freezeLifecycleBatchVersion, LifecycleContinuationDecodeError, - recordLifecycleArtifactOutcomes, - recordLifecycleMetadataDeletions, - skipLifecycleBatch, - stageLifecycleBatch, } from './continuation' function validContinuation() { - const generation = randomUUID() return { - ...newContinuation(generation, { snapshotAt: '2026-08-06T00:00:00.000Z' }), - pageEnd: { - name: 'logs/example.json', - archivedAt: '2026-08-05T12:00:00.000Z', - }, - counters: { - versionsExamined: 1, - versionsEligible: 1, - objectVersionsDeleted: 0, - deleteMarkersDeleted: 0, - bytesDeleted: 0, - batchesCompleted: 0, - }, - batch: { - keySeparator: '/' as const, - versions: [ - { - id: randomUUID(), - name: 'logs/example.json', - version: 'version', - artifacts: [{ outcome: 'UNRESOLVED' }, { outcome: 'UNRESOLVED' }], - }, - ], - inFlight: { - attemptId: randomUUID(), - startedAt: '2026-08-06T00:01:00.000Z', - }, - }, + ...newContinuation(randomUUID(), { snapshotAt: '2026-08-06T00:00:00.000Z' }), + cursor: { name: 'logs/example.json', archivedAt: '2026-08-05T12:00:00.000Z' }, } } describe('decodeLifecycleContinuation', () => { - test('decodes a bounded v1 continuation', () => { + test('decodes a v1 continuation', () => { const input = validContinuation() - const decoded = decodeLifecycleContinuation(input) - expect(decoded.batch?.versions[0].artifacts).toEqual([ - { key: 'logs/example.json/version', outcome: 'UNRESOLVED' }, - { key: 'logs/example.json/version.info', outcome: 'UNRESOLVED' }, - ]) - expect(decoded).toMatchObject({ ...input, batch: { inFlight: input.batch.inFlight } }) - }) - - test('ignores a stored artifact key and derives it from the name and version', () => { - const input = validContinuation() - input.batch.versions[0].artifacts[0] = { key: 'tampered', outcome: 'UNRESOLVED' } as never - expect(decodeLifecycleContinuation(input).batch?.versions[0].artifacts[0].key).toBe( - 'logs/example.json/version' - ) - }) - - test('encodes without artifact keys and decodes back to the same continuation', () => { - const decoded = decodeLifecycleContinuation(validContinuation()) - const encoded = JSON.parse(JSON.stringify(encodeLifecycleContinuation(decoded))) - expect(encoded.batch.versions[0].artifacts).toEqual([ - { outcome: 'UNRESOLVED' }, - { outcome: 'UNRESOLVED' }, - ]) - expect(decodeLifecycleContinuation(encoded)).toEqual(decoded) + expect(decodeLifecycleContinuation(input)).toEqual(input) }) - test('recovers keys using the separator frozen in the journal', () => { - const input = validContinuation() - const encoded = { ...input, batch: { ...input.batch, keySeparator: '-$v-' } } - const recovered = decodeLifecycleContinuation(encoded) - expect(recovered.batch?.versions[0].artifacts.map(({ key }) => key)).toEqual([ - 'logs/example.json-$v-version', - 'logs/example.json-$v-version.info', - ]) - expect(decodeLifecycleContinuation(encodeLifecycleContinuation(recovered))).toEqual(recovered) + test('rejects a continuation that still carries a batch', () => { expect(() => - decodeLifecycleContinuation({ ...input, batch: { ...input.batch, keySeparator: undefined } }) - ).toThrow('batch.keySeparator') - }) - - test('encodes a continuation without a batch unchanged', () => { - const { batch: _batch, ...input } = validContinuation() - const decoded = decodeLifecycleContinuation(input) - expect(encodeLifecycleContinuation(decoded)).toEqual(decoded) - }) - - test('rejects a resolved artifact with a missing companion', () => { - const input = validContinuation() - const artifacts = input.batch.versions[0].artifacts - artifacts[0].outcome = 'DELETED' - artifacts.pop() - - expect(() => decodeLifecycleContinuation(input)).toThrow( - 'batch.versions[0].artifacts must contain 0 or 2 entries' - ) - }) - - test('rejects an omitted physical version selector in a durable batch', () => { - const input = validContinuation() - const { version: _version, ...entry } = input.batch.versions[0] - expect(() => - decodeLifecycleContinuation({ - ...input, - batch: { ...input.batch, versions: [{ ...entry, artifacts: entry.artifacts.slice(0, 1) }] }, - }) - ).toThrow('batch.versions[0].version') + decodeLifecycleContinuation({ ...validContinuation(), batch: { versions: [] } }) + ).toThrow(LifecycleContinuationDecodeError) }) test.each([2, 3])( @@ -137,13 +41,7 @@ describe('decodeLifecycleContinuation', () => { [ 'invalid UUID', (value: ReturnType) => { - value.runId = 'not-a-uuid' as ReturnType['runId'] - }, - ], - [ - 'batch without page end', - (value: ReturnType) => { - delete (value as { pageEnd?: unknown }).pageEnd + value.runId = 'not-a-uuid' }, ], [ @@ -158,83 +56,15 @@ describe('decodeLifecycleContinuation', () => { value.counters.versionsExamined = -1 }, ], - [ - 'unknown artifact outcome', - (value: ReturnType) => { - value.batch.versions[0].artifacts[0].outcome = 'MAYBE' - }, - ], ])('rejects %s', (_name, mutate) => { const value = validContinuation() mutate(value) expect(() => decodeLifecycleContinuation(value)).toThrow(LifecycleContinuationDecodeError) }) - - test.each([ - '00000000-0000-0000-0000-000000000000', - '6ba7b810-9dad-31d1-70b4-00c04fd430c8', - ] as const)('decodes the database uuid %s as a batch version id', (id) => { - const input = validContinuation() - input.batch.versions[0].id = id - expect(decodeLifecycleContinuation(input).batch?.versions[0].id).toBe(id) - }) - - test.each([ - '6ba7b810-9dad-31d1-70b4-00c04fd430c', - '6ba7b810-9dad-31d1-70b4-00c04fd430cg', - ] as const)('rejects the malformed batch version id %s', (id) => { - const input = validContinuation() - input.batch.versions[0].id = id - expect(() => decodeLifecycleContinuation(input)).toThrow('batch.versions[0].id must be a UUID') - }) - - test('round-trips every journal shape through the encoded JSON form', () => { - const pageEnd = { name: 'a/b', archivedAt: '2026-08-05T12:00:00.000Z' } - const versions = [ - { id: randomUUID(), name: 'a/b', version: randomUUID(), isDeleteMarker: false }, - { id: randomUUID(), name: 'a/b', version: null, isDeleteMarker: false }, - { id: randomUUID(), name: 'a/b', version: 'null', isDeleteMarker: false }, - { id: randomUUID(), name: 'd/-$v-x.info/é', version: randomUUID(), isDeleteMarker: false }, - { id: randomUUID(), name: 'a/b', version: randomUUID(), isDeleteMarker: true }, - ] - const armed = armedContinuation(randomUUID(), versions, pageEnd) - expect(armed.batch!.versions[1]).toMatchObject({ - version: null, - artifacts: [{ key: 'a/b', outcome: 'UNRESOLVED' }], - }) - const [data, info] = armed.batch!.versions[0].artifacts - const recorded = recordLifecycleArtifactOutcomes( - armed, - new Map([ - [data.key, { outcome: 'UNRESOLVED' as const, error: 'timeout' }], - [info.key, { outcome: 'FAILED' as const, error: 'denied' }], - [armed.batch!.versions[1].artifacts[0].key, { outcome: 'DELETED' as const }], - [armed.batch!.versions[2].artifacts[0].key, { outcome: 'ABSENT' as const }], - ]) - ) - for (const continuation of [ - newContinuation(randomUUID()), - { ...newContinuation(randomUUID()), cursor: pageEnd }, - stagedContinuation(randomUUID(), versions, pageEnd), - armed, - recorded, - commitLifecycleBatchResults(recorded), - ]) { - const encoded = JSON.parse(JSON.stringify(encodeLifecycleContinuation(continuation))) - expect(JSON.stringify(encoded)).not.toContain('"key"') - expect(decodeLifecycleContinuation(encoded)).toEqual(continuation) - } - }) - - test('bounds the durable logical-version batch', () => { - const value = validContinuation() - value.batch.versions = Array.from({ length: 481 }, () => value.batch.versions[0]) - expect(() => decodeLifecycleContinuation(value)).toThrow('at most 480 entries') - }) }) describe('lifecycle continuation transitions', () => { - test('rejects empty staging and advances pages without eligible candidates', () => { + test('advances pages without eligible candidates and rejects pages with candidates', () => { const initial = newContinuation(randomUUID(), { snapshotAt: '2026-08-15T00:00:00.000Z' }) const page = { rawRowsExamined: 1, @@ -243,14 +73,6 @@ describe('lifecycle continuation transitions', () => { exhausted: false, } - expect(() => - stageLifecycleBatch(initial, { - pageEnd: page.pageEnd, - rawRowsExamined: page.rawRowsExamined, - versions: [], - }) - ).toThrow('Lifecycle batch cannot be empty') - expect(advanceEmptyLifecyclePage(initial, page)).toEqual({ ...initial, cursor: page.pageEnd, @@ -271,10 +93,10 @@ describe('lifecycle continuation transitions', () => { }, ], }) - ).toThrow('Lifecycle pages with eligible candidates must be staged before advancing') + ).toThrow('Lifecycle pages with eligible candidates must be armed before advancing') }) - test('stages a batch with distinct examined and eligible counts, rejects restaging, and drops pageEnd on advance', () => { + test('advances a batch to its page end with examined, eligible and deleted counts', () => { const initial = newContinuation(randomUUID(), { snapshotAt: '2026-08-15T00:00:00.000Z' }) const versions = [0, 1].map(() => freezeLifecycleBatchVersion({ @@ -285,73 +107,39 @@ describe('lifecycle continuation transitions', () => { }) ) const pageEnd = { name: 'logs/example.json', archivedAt: '2026-08-02T00:00:00.000Z' } - const staged = stageLifecycleBatch(initial, { pageEnd, rawRowsExamined: 5, versions }) + const batch = { pageEnd, rawRowsExamined: 5, versions } - expect(staged.counters).toMatchObject({ versionsExamined: 5, versionsEligible: 2 }) - expect(() => stageLifecycleBatch(staged, { pageEnd, rawRowsExamined: 5, versions })).toThrow( - 'already has a staged batch' - ) - - const armed = armLifecycleAttempt(staged, { - attemptId: randomUUID(), - startedAt: '2026-08-15T00:01:00.000Z', + expect( + advanceLifecycleBatch(initial, batch, { objectVersions: 1, deleteMarkers: 0, bytes: 7 }) + ).toEqual({ + ...initial, + cursor: pageEnd, + counters: { + versionsExamined: 5, + versionsEligible: 2, + objectVersionsDeleted: 1, + deleteMarkersDeleted: 0, + bytesDeleted: 7, + batchesCompleted: 1, + }, }) - const resolved = recordLifecycleArtifactOutcomes( - armed, - new Map( - versions.flatMap((v) => v.artifacts.map((a) => [a.key, { outcome: 'DELETED' as const }])) - ) - ) - const advanced = commitLifecycleBatchResults(resolved) - expect(advanced).not.toHaveProperty('pageEnd') - expect(advanced).not.toHaveProperty('batch') - expect(advanced.cursor).toEqual(pageEnd) - }) - - test('skips a staged batch to its page end without counting it as completed', () => { - const pageEnd = { name: 'logs/example.json', archivedAt: '2026-08-02T00:00:00.000Z' } - const staged = stagedContinuation( - randomUUID(), - [{ id: randomUUID(), name: pageEnd.name, version: randomUUID(), isDeleteMarker: false }], - pageEnd - ) - const { batch: _batch, pageEnd: _pageEnd, ...rest } = staged - - expect(skipLifecycleBatch(staged)).toEqual({ ...rest, cursor: pageEnd }) + expect( + advanceLifecycleBatch(initial, batch, { objectVersions: 0, deleteMarkers: 0, bytes: 0 }) + .counters.batchesCompleted + ).toBe(0) }) - test('keeps null and literal null-string identities separate through recovery', () => { + test('saturates the cumulative byte counter at the safe integer limit', () => { const initial = validContinuation() - const legacy = freezeLifecycleBatchVersion({ - id: randomUUID(), - name: 'logs/example.json', - version: null, - isDeleteMarker: false, - }) - const named = freezeLifecycleBatchVersion({ - id: randomUUID(), - name: legacy.name, - version: 'null', - isDeleteMarker: false, - }) - const staged = decodeLifecycleContinuation({ - ...initial, - counters: { ...initial.counters, versionsExamined: 2, versionsEligible: 2 }, - batch: { keySeparator: '/', versions: [legacy, named] }, - }) - - const armed = armLifecycleAttempt(staged, initial.batch.inFlight) - const recorded = recordLifecycleArtifactOutcomes( - armed, - new Map(named.artifacts.map(({ key }) => [key, { outcome: 'ABSENT' as const }])) - ) - const recovered = commitLifecycleBatchResults(recorded) - expect(recovered.batch).toEqual({ - keySeparator: '/', - versions: [legacy], - inFlight: initial.batch.inFlight, - }) - expect(recovered.cursor).toBeUndefined() + initial.counters.bytesDeleted = Number.MAX_SAFE_INTEGER - 1 + const pageEnd = { name: 'a', archivedAt: '2026-08-02T00:00:00.000Z' } + expect( + advanceLifecycleBatch( + initial, + { pageEnd, rawRowsExamined: 1, versions: [] }, + { objectVersions: 1, deleteMarkers: 0, bytes: 10 } + ).counters.bytesDeleted + ).toBe(Number.MAX_SAFE_INTEGER) }) test('freezes exact physical keys using the backend locator helper', () => { @@ -368,10 +156,7 @@ describe('lifecycle continuation transitions', () => { id, name: 'logs/example.json', version: 'version-a', - artifacts: [ - { key: physicalKey, outcome: 'UNRESOLVED' }, - { key: `${physicalKey}.info`, outcome: 'UNRESOLVED' }, - ], + artifacts: [physicalKey, `${physicalKey}.info`], }) expect( @@ -390,176 +175,6 @@ describe('lifecycle continuation transitions', () => { version: null, isDeleteMarker: false, }).artifacts - ).toEqual([{ key: 'logs/example.json', outcome: 'UNRESOLVED' }]) - }) - - test('requires an armed attempt to complete a delete-marker batch', () => { - const initial = validContinuation() - const marker = freezeLifecycleBatchVersion({ - id: randomUUID(), - name: initial.pageEnd.name, - version: randomUUID(), - isDeleteMarker: true, - }) - const staged = decodeLifecycleContinuation({ - ...initial, - batch: { keySeparator: '/', versions: [marker] }, - }) - - expect(() => commitLifecycleBatchResults(staged)).toThrow('must be armed before committing') - expect(() => recordLifecycleArtifactOutcomes(staged, new Map())).toThrow( - 'must be armed before recording' - ) - - const armed = armLifecycleAttempt(staged, initial.batch.inFlight) - const completed = commitLifecycleBatchResults(armed) - expect(completed.cursor).toEqual(staged.pageEnd) - expect(completed.batch).toBeUndefined() - expect(completed.counters).toEqual({ ...staged.counters, batchesCompleted: 1 }) - }) - - test('keeps one stable attempt as compensation authority across generation replacement', () => { - const generation = randomUUID() - const initial = newContinuation(generation, { snapshotAt: '2026-08-15T00:00:00.000Z' }) - const candidate = freezeLifecycleBatchVersion({ - id: randomUUID(), - name: 'logs/example.json', - version: randomUUID(), - isDeleteMarker: false, - }) - const staged = stageLifecycleBatch(initial, { - pageEnd: { name: candidate.name, archivedAt: '2026-08-02T00:00:00.000Z' }, - rawRowsExamined: 1, - versions: [candidate], - }) - const attempt = { - attemptId: randomUUID(), - startedAt: '2026-08-15T00:01:00.000Z', - } - const armed = armLifecycleAttempt(staged, attempt) - - expect(commitLifecycleBatchResults(armed).batch?.versions).toHaveLength(1) - - const partiallyResolved = recordLifecycleArtifactOutcomes( - armed, - new Map([[candidate.artifacts[1].key, { outcome: 'DELETED' }]]) - ) - expect(partiallyResolved.batch?.inFlight).toEqual(attempt) - expect(commitLifecycleBatchResults(partiallyResolved).batch?.versions).toHaveLength(1) - - const resolved = recordLifecycleArtifactOutcomes( - partiallyResolved, - new Map([[candidate.artifacts[0].key, { outcome: 'ABSENT' }]]) - ) - expect(commitLifecycleBatchResults(resolved)).toMatchObject({ - cursor: staged.pageEnd, - counters: { versionsExamined: 1, versionsEligible: 1, batchesCompleted: 1 }, - }) - }) - - test('rejects replacing an armed attempt', () => { - const continuation = decodeLifecycleContinuation(validContinuation()) - expect(() => - armLifecycleAttempt(continuation, { - attemptId: randomUUID(), - startedAt: '2026-08-15T00:01:00.000Z', - }) - ).toThrow('already has an armed attempt') - }) - - test('retains a version whose data artifact is unresolved even if the info sibling was deleted', () => { - const continuation = decodeLifecycleContinuation(validContinuation()) - const version = continuation.batch!.versions[0] - const partial = recordLifecycleArtifactOutcomes( - continuation, - new Map([ - [version.artifacts[0].key, { outcome: 'UNRESOLVED' as const, error: 'retry me' }], - [version.artifacts[1].key, { outcome: 'DELETED' as const }], - ]) - ) - - expect(commitLifecycleBatchResults(partial)).toMatchObject({ - batch: { - inFlight: partial.batch!.inFlight, - versions: [ - { - name: version.name, - version: version.version, - artifacts: [ - expect.objectContaining({ outcome: 'UNRESOLVED' }), - expect.objectContaining({ outcome: 'DELETED' }), - ], - }, - ], - }, - counters: { batchesCompleted: 0 }, - }) - }) - - test.each([ - ['data FAILED with info DELETED', 'FAILED', 'DELETED'], - ['data DELETED with info FAILED', 'DELETED', 'FAILED'], - ['data ABSENT with info FAILED', 'ABSENT', 'FAILED'], - ] as const)('retains a failed artifact for retry: %s', (_label, data, info) => { - const continuation = decodeLifecycleContinuation(validContinuation()) - const version = continuation.batch!.versions[0] - const recorded = recordLifecycleArtifactOutcomes( - continuation, - new Map([ - [version.artifacts[0].key, { outcome: data, error: 'x' }], - [version.artifacts[1].key, { outcome: info, error: 'x' }], - ]) - ) - - expect(commitLifecycleBatchResults(recorded)).toMatchObject({ - batch: { versions: [expect.objectContaining({ id: version.id })] }, - counters: { batchesCompleted: 0 }, - }) - }) - - test('saturates the cumulative byte counter at the safe integer limit', () => { - const input = validContinuation() - input.counters.bytesDeleted = Number.MAX_SAFE_INTEGER - 1 - const continuation = decodeLifecycleContinuation(input) - expect( - recordLifecycleMetadataDeletions(continuation, { - objectVersions: 1, - deleteMarkers: 0, - bytes: 10, - }).counters.bytesDeleted - ).toBe(Number.MAX_SAFE_INTEGER) - }) - - test('abandonPending drains FAILED and UNRESOLVED artifacts and advances the cursor', () => { - const armed = decodeLifecycleContinuation(validContinuation()) - const [data, info] = armed.batch!.versions[0].artifacts - const recorded = recordLifecycleArtifactOutcomes( - armed, - new Map([ - [data.key, { outcome: 'FAILED' as const, error: 'not deleted' }], - [info.key, { outcome: 'UNRESOLVED' as const }], - ]) - ) - const abandoned = commitLifecycleBatchResults(recorded, true) - expect(abandoned.batch).toBeUndefined() - expect(abandoned.cursor).toEqual(armed.pageEnd) - expect(abandoned.counters.batchesCompleted).toBe(armed.counters.batchesCompleted + 1) - }) - - test('retains definitive all-failed versions for retry', () => { - const armed = decodeLifecycleContinuation(validContinuation()) - const failed = recordLifecycleArtifactOutcomes( - armed, - new Map( - armed.batch!.versions[0].artifacts.map((artifact) => [ - artifact.key, - { outcome: 'FAILED' as const, error: 'not deleted' }, - ]) - ) - ) - expect(commitLifecycleBatchResults(failed)).toMatchObject({ - batch: { versions: [expect.objectContaining({ id: armed.batch!.versions[0].id })] }, - counters: { batchesCompleted: 0 }, - }) + ).toEqual(['logs/example.json']) }) }) diff --git a/src/storage/lifecycle/continuation.ts b/src/storage/lifecycle/continuation.ts index 4b08ad0ad..84c610955 100644 --- a/src/storage/lifecycle/continuation.ts +++ b/src/storage/lifecycle/continuation.ts @@ -1,14 +1,11 @@ -import { FILE_VERSION_SEPARATOR, PATH_SEPARATOR, SEPARATOR } from '../backend/adapter' +import { SEPARATOR } from '../backend/adapter' import { LIFECYCLE_CONTINUATION_VERSION, - LIFECYCLE_MAX_PAGE_SIZE, - type LifecycleArtifact, + type LifecycleBatch, type LifecycleBatchVersion, type LifecycleCandidateIdentity, type LifecycleContinuation, - type LifecycleContinuationBatch, type LifecycleEvaluationPage, - type LifecycleInFlightAttempt, type LifecycleRunCounters, type NoncurrentLifecycleCursor, } from '../schemas/lifecycle' @@ -56,72 +53,21 @@ export function freezeLifecycleBatchVersion( version: candidate.version, artifacts: candidate.isDeleteMarker ? [] - : lifecycleArtifactKeys(candidate.name, candidate.version).map((key) => ({ - key, - outcome: 'UNRESOLVED' as const, - })), + : lifecycleArtifactKeys(candidate.name, candidate.version), } } -function lifecycleArtifactKeys( - name: string, - version: string | null, - separator = SEPARATOR -): string[] { - const key = version ? `${name}${separator}${version}` : name +export function lifecycleArtifactKeys(name: string, version: string | null): string[] { + const key = version ? `${name}${SEPARATOR}${version}` : name return version ? [key, `${key}.info`] : [key] } -export function encodeLifecycleContinuation(continuation: LifecycleContinuation): unknown { - if (!continuation.batch) return continuation - return { - ...continuation, - batch: { - ...continuation.batch, - versions: continuation.batch.versions.map((version) => ({ - ...version, - artifacts: version.artifacts.map(({ key: _key, ...artifact }) => artifact), - })), - }, - } -} - -export function stageLifecycleBatch( - continuation: LifecycleContinuation, - input: { - pageEnd: NoncurrentLifecycleCursor - rawRowsExamined: number - versions: LifecycleBatchVersion[] - } -): LifecycleContinuation { - if (continuation.batch !== undefined) { - throw new Error('Lifecycle continuation already has a staged batch') - } - if (input.versions.length === 0) { - throw new Error('Lifecycle batch cannot be empty') - } - - return decodeLifecycleContinuation({ - ...continuation, - pageEnd: input.pageEnd, - counters: { - ...continuation.counters, - versionsExamined: continuation.counters.versionsExamined + input.rawRowsExamined, - versionsEligible: continuation.counters.versionsEligible + input.versions.length, - }, - batch: { keySeparator: SEPARATOR, versions: input.versions }, - }) -} - export function advanceEmptyLifecyclePage( continuation: LifecycleContinuation, page: LifecycleEvaluationPage ): LifecycleContinuation { - if (continuation.batch !== undefined) { - throw new Error('Lifecycle continuation already has a staged batch') - } if (page.candidates.length !== 0) { - throw new Error('Lifecycle pages with eligible candidates must be staged before advancing') + throw new Error('Lifecycle pages with eligible candidates must be armed before advancing') } return decodeLifecycleContinuation({ @@ -134,98 +80,25 @@ export function advanceEmptyLifecyclePage( }) } -export function armLifecycleAttempt( - continuation: LifecycleContinuation, - attempt: LifecycleInFlightAttempt -): LifecycleContinuation { - if (continuation.batch?.inFlight) { - throw new Error('Lifecycle batch already has an armed attempt') - } - - return decodeLifecycleContinuation({ - ...continuation, - batch: { ...continuation.batch, inFlight: attempt }, - }) -} - -export function recordLifecycleArtifactOutcomes( - continuation: LifecycleContinuation, - outcomes: ReadonlyMap> -): LifecycleContinuation { - if (!continuation.batch?.inFlight) { - throw new Error('Lifecycle batch must be armed before recording artifact outcomes') - } - - const versions = continuation.batch.versions.map((version) => ({ - ...version, - artifacts: version.artifacts.map((artifact) => { - const outcome = outcomes.get(artifact.key) - return outcome ? { key: artifact.key, ...outcome } : artifact - }), - })) - - return decodeLifecycleContinuation({ - ...continuation, - batch: { ...continuation.batch, versions }, - }) -} - -export function commitLifecycleBatchResults( +export function advanceLifecycleBatch( continuation: LifecycleContinuation, - abandonPending = false + batch: LifecycleBatch, + deleted: LifecycleDeletionCounts ): LifecycleContinuation { - if (!continuation.batch?.inFlight) { - throw new Error('Lifecycle batch must be armed before committing results') - } - - const versions = abandonPending - ? [] - : continuation.batch.versions.filter(lifecycleVersionNeedsCompensation) - if (versions.length === 0) return advanceLifecycleBatch(continuation) - + const counters = addDeletionCounters(continuation.counters, deleted) return decodeLifecycleContinuation({ ...continuation, - batch: { ...continuation.batch, versions }, - }) -} - -export function skipLifecycleBatch(continuation: LifecycleContinuation): LifecycleContinuation { - const { batch: _batch, pageEnd: _pageEnd, ...rest } = continuation - return decodeLifecycleContinuation({ ...rest, cursor: continuation.pageEnd }) -} - -function advanceLifecycleBatch(continuation: LifecycleContinuation): LifecycleContinuation { - const skipped = skipLifecycleBatch(continuation) - return decodeLifecycleContinuation({ - ...skipped, + cursor: batch.pageEnd, counters: { - ...skipped.counters, - batchesCompleted: skipped.counters.batchesCompleted + 1, + ...counters, + versionsExamined: counters.versionsExamined + batch.rawRowsExamined, + versionsEligible: counters.versionsEligible + batch.versions.length, + batchesCompleted: + counters.batchesCompleted + (deleted.objectVersions + deleted.deleteMarkers > 0 ? 1 : 0), }, }) } -export function pendingArtifacts( - continuation: LifecycleContinuation -): Array> { - return continuation.batch!.versions.flatMap(({ name, version, artifacts }) => - artifacts - .filter((artifact) => artifact.outcome === 'UNRESOLVED' || artifact.outcome === 'FAILED') - .map((artifact) => ({ ...artifact, name, version })) - ) -} - -function lifecycleVersionNeedsCompensation(version: LifecycleBatchVersion): boolean { - return version.artifacts.some(({ outcome }) => outcome === 'UNRESOLVED' || outcome === 'FAILED') -} - -export function recordLifecycleMetadataDeletions( - continuation: LifecycleContinuation, - deleted: LifecycleDeletionCounts -): LifecycleContinuation { - return { ...continuation, counters: addDeletionCounters(continuation.counters, deleted) } -} - function addDeletionCounters( counters: LifecycleRunCounters, deleted: LifecycleDeletionCounts @@ -255,13 +128,8 @@ export function decodeLifecycleContinuation(value: unknown): LifecycleContinuati if (continuation.cursor !== undefined) decoded.cursor = decodeCursor(continuation.cursor, 'cursor') - if (continuation.pageEnd !== undefined) { - decoded.pageEnd = decodeCursor(continuation.pageEnd, 'pageEnd') - } - if (continuation.batch !== undefined) decoded.batch = decodeBatch(continuation.batch) - - if (decoded.batch?.versions.length && decoded.pageEnd === undefined) { - throw decodeError('A lifecycle continuation batch requires pageEnd') + if (continuation.batch !== undefined) { + throw decodeError('A lifecycle continuation cannot carry a batch') } return decoded @@ -293,73 +161,6 @@ function decodeCounters(value: unknown): LifecycleRunCounters { } } -function decodeBatch(value: unknown): LifecycleContinuationBatch { - const batch = record(value, 'batch') - const keySeparator = oneOf( - batch.keySeparator, - [PATH_SEPARATOR, FILE_VERSION_SEPARATOR] as const, - 'batch.keySeparator' - ) - if (!Array.isArray(batch.versions) || batch.versions.length > LIFECYCLE_MAX_PAGE_SIZE) { - throw decodeError(`batch.versions must contain at most ${LIFECYCLE_MAX_PAGE_SIZE} entries`) - } - - const versions = batch.versions.map((entry, index): LifecycleBatchVersion => { - const version = record(entry, `batch.versions[${index}]`) - const artifactCount = version.version ? 2 : 1 - if ( - !Array.isArray(version.artifacts) || - (version.artifacts.length !== 0 && version.artifacts.length !== artifactCount) - ) { - throw decodeError( - `batch.versions[${index}].artifacts must contain 0 or ${artifactCount} entries` - ) - } - const name = string(version.name, `batch.versions[${index}].name`) - const versionId = - version.version === null ? null : string(version.version, `batch.versions[${index}].version`) - const keys = lifecycleArtifactKeys(name, versionId, keySeparator) - return { - id: uuid(version.id, `batch.versions[${index}].id`), - name, - version: versionId, - artifacts: version.artifacts.map((artifact, artifactIndex) => - decodeArtifact( - artifact, - keys[artifactIndex], - `batch.versions[${index}].artifacts[${artifactIndex}]` - ) - ), - } - }) - - const decoded: LifecycleContinuationBatch = { keySeparator, versions } - if (batch.inFlight !== undefined) decoded.inFlight = decodeAttempt(batch.inFlight) - return decoded -} - -function decodeArtifact(value: unknown, key: string, label: string): LifecycleArtifact { - const artifact = record(value, label) - const decoded: LifecycleArtifact = { - key, - outcome: oneOf( - artifact.outcome, - ['UNRESOLVED', 'DELETED', 'ABSENT', 'FAILED'] as const, - `${label}.outcome` - ), - } - if (artifact.error !== undefined) decoded.error = string(artifact.error, `${label}.error`) - return decoded -} - -function decodeAttempt(value: unknown): LifecycleInFlightAttempt { - const attempt = record(value, 'batch.inFlight') - return { - attemptId: uuid(attempt.attemptId, 'batch.inFlight.attemptId'), - startedAt: timestamp(attempt.startedAt, 'batch.inFlight.startedAt'), - } -} - function record(value: unknown, label: string): Record { if (value === null || typeof value !== 'object' || Array.isArray(value)) { throw decodeError(`${label} must be an object`) @@ -392,13 +193,6 @@ function nonnegativeInteger(value: unknown, label: string): number { return value as number } -function oneOf(value: unknown, values: T, label: string): T[number] { - if (typeof value !== 'string' || !values.includes(value)) { - throw decodeError(`${label} must be one of ${values.join(', ')}`) - } - return value as T[number] -} - function decodeError(message: string) { return new LifecycleContinuationDecodeError(message) } diff --git a/src/storage/schemas/lifecycle.ts b/src/storage/schemas/lifecycle.ts index 1eaa2fc8e..a36328fea 100644 --- a/src/storage/schemas/lifecycle.ts +++ b/src/storage/schemas/lifecycle.ts @@ -118,30 +118,17 @@ export interface LifecycleRunCounters { batchesCompleted: number } -export type LifecycleArtifactOutcome = 'UNRESOLVED' | 'DELETED' | 'ABSENT' | 'FAILED' - -export interface LifecycleArtifact { - key: string - outcome: LifecycleArtifactOutcome - error?: string -} - export interface LifecycleBatchVersion { id: string name: string version: string | null - artifacts: LifecycleArtifact[] -} - -export interface LifecycleInFlightAttempt { - attemptId: string - startedAt: string + artifacts: string[] } -export interface LifecycleContinuationBatch { - keySeparator: '/' | '-$v-' +export interface LifecycleBatch { + pageEnd: NoncurrentLifecycleCursor + rawRowsExamined: number versions: LifecycleBatchVersion[] - inFlight?: LifecycleInFlightAttempt } export interface LifecycleContinuation { @@ -150,9 +137,7 @@ export interface LifecycleContinuation { generation: string snapshotAt: string cursor?: NoncurrentLifecycleCursor - pageEnd?: NoncurrentLifecycleCursor counters: LifecycleRunCounters - batch?: LifecycleContinuationBatch } export interface LifecycleClaimIdentity { @@ -171,26 +156,25 @@ export interface LifecycleClaimInput extends LifecycleClaimIdentity { export interface LifecycleArmAttemptInput extends LifecycleClaimIdentity { leaseMs: number continuation: LifecycleContinuation + batch: LifecycleBatch +} + +export interface LifecycleOutboxArtifact { + bucketId: string + key: string + name: string + version: string | null } export type LifecycleArmAttemptResult = - | { status: 'armed'; state: LifecycleState } + | { status: 'armed'; state: LifecycleState; artifacts: LifecycleOutboxArtifact[] } | { status: 'ineligible' } | { status: 'fenced' } -export interface LifecycleCommitAttemptInput extends LifecycleClaimIdentity { - attemptId: string - continuation: LifecycleContinuation - abandonPending?: boolean -} - export interface LifecycleReleaseClaimInput extends LifecycleClaimIdentity { - // Omit after an ambiguous write. - // Omit when an armed or unknown journal version must be preserved for recovery. + // Omit after an ambiguous write or to preserve an unknown continuation version. continuation?: LifecycleContinuation | null nextRunAt: string | null - // Cap a retry only when preserving a persisted journal in flight. - maxInFlightRetryAt?: string error?: Record } @@ -208,13 +192,6 @@ export interface LifecycleState { failureCount: number } -export type LifecycleClaimedState = LifecycleState & { - bucket: Pick< - LifecycleBucket, - 'versioning_status' | 'lifecycle_configuration' | 'lifecycle_configuration_generation' - > -} - export type LifecycleBucket = Pick & { type: 'STANDARD' | 'ANALYTICS' versioning_status: BucketVersioningStatus diff --git a/src/test/bucket-lifecycle.test.ts b/src/test/bucket-lifecycle.test.ts index 235e45958..14b30f027 100644 --- a/src/test/bucket-lifecycle.test.ts +++ b/src/test/bucket-lifecycle.test.ts @@ -1,6 +1,6 @@ import { randomUUID } from 'node:crypto' import type { BucketLifecycleConfiguration } from '@storage/schemas' -import { armedContinuation, newContinuation } from './utils/lifecycle-continuation' +import { newContinuation } from './utils/lifecycle-continuation' import { useLifecycleStorage } from './utils/lifecycle-versioning' import { withDeleteEnabled } from './utils/storage' @@ -131,12 +131,22 @@ describe('bucket lifecycle controls', () => { function armContinuation(generation: string) { const now = new Date().toISOString() - return armedContinuation( - generation, - [{ id: randomUUID(), name: 'object', version: randomUUID(), isDeleteMarker: false }], - { name: 'object', archivedAt: now }, - { snapshotAt: now, startedAt: now } - ) + return { + ...newContinuation(generation, { snapshotAt: now }), + pageEnd: { name: 'object', archivedAt: now }, + batch: { + keySeparator: '/', + versions: [ + { + id: randomUUID(), + name: 'object', + version: randomUUID(), + artifacts: [{ outcome: 'UNRESOLVED' }, { outcome: 'UNRESOLVED' }], + }, + ], + inFlight: { attemptId: randomUUID(), startedAt: now }, + }, + } } it('blocks bucket deletion while an armed attempt is in flight', async () => { @@ -540,7 +550,7 @@ describe('bucket lifecycle controls', () => { const generation = configured.lifecycle_configuration_generation! const armed = armContinuation(generation) const continuation = staged - ? (({ inFlight: _inFlight, ...batch }) => ({ ...armed, batch }))(armed.batch!) + ? (({ inFlight: _inFlight, ...batch }) => ({ ...armed, batch }))(armed.batch) : newContinuation(generation) await helper.database.connection.query( 'UPDATE storage.bucket_lifecycle_states SET continuation = $2::jsonb WHERE bucket_id = $1', diff --git a/src/test/noncurrent-lifecycle-database.test.ts b/src/test/noncurrent-lifecycle-database.test.ts index fd4b16a23..b1ec69707 100644 --- a/src/test/noncurrent-lifecycle-database.test.ts +++ b/src/test/noncurrent-lifecycle-database.test.ts @@ -1,20 +1,16 @@ import { randomUUID } from 'node:crypto' -import { Readable } from 'node:stream' +import { withOptionalVersion } from '@storage/backend/adapter' import { buildEvaluateNoncurrentLifecyclePageStatement, mapLifecycleEvaluationPage, } from '@storage/database/lifecycle' import { compileLifecycleEvaluationRules } from '@storage/lifecycle/configuration' -import { - armLifecycleAttempt, - encodeLifecycleContinuation, - recordLifecycleArtifactOutcomes, -} from '@storage/lifecycle/continuation' +import { advanceLifecycleBatch } from '@storage/lifecycle/continuation' import { lifecycleEvaluationBatches } from '@storage/lifecycle/evaluation' import type { NoncurrentLifecycleCursor } from '@storage/schemas' import { Client } from 'pg' import { getConfig } from '../config' -import { armedContinuation, stagedContinuation } from './utils/lifecycle-continuation' +import { batchOf, newContinuation } from './utils/lifecycle-continuation' import { useLifecycleStorage } from './utils/lifecycle-versioning' import { withDeleteEnabled } from './utils/storage' @@ -464,19 +460,16 @@ describe('noncurrent lifecycle database API', () => { isDeleteMarker: options.marker ?? false, }), ]) - const armed = armedContinuation(generation, page.candidates, page.pageEnd!, { - snapshotAt, - }) const arm = () => tHelper.database.armLifecycleAttempt({ ...identity, leaseMs: 60_000, - continuation: armed, + continuation: newContinuation(generation, { snapshotAt }), + batch: batchOf(page.candidates, page.pageEnd!), }) return { identity, olderId, - armed, arm, evaluation, name, @@ -494,14 +487,7 @@ describe('noncurrent lifecycle database API', () => { 'arms $label using the compatible row identity and policy', async ({ label: _label, ...options }) => { const fixture = await stageCountProtectedVersion(options) - await expect(fixture.arm()).resolves.toMatchObject({ - status: 'armed', - state: { - continuation: { - batch: { inFlight: { attemptId: fixture.armed.batch!.inFlight!.attemptId } }, - }, - }, - }) + await expect(fixture.arm()).resolves.toMatchObject({ status: 'armed' }) await expect( tHelper.database.connection.query('SELECT 1 FROM storage.objects WHERE id = $1', [ fixture.olderId, @@ -534,114 +520,6 @@ describe('noncurrent lifecycle database API', () => { ).resolves.toMatchObject({ candidates: [] }) }) }) - it('records artifact outcomes and advances only once every artifact is resolved', async () => { - const generation = await putConfiguration() - const name = 'history/a-expired.txt' - const version = randomUUID() - const id = randomUUID() - const completedName = 'history/b-completed.txt' - const completedVersion = randomUUID() - const completedId = randomUUID() - await fixtures.withServiceOperation('storage.s3.bucket.put_lifecycle', async (transaction) => { - for (const [objectId, objectName, objectVersion, size] of [ - [id, name, version, 42], - [completedId, completedName, completedVersion, 8], - ]) { - await transaction.query( - `INSERT INTO storage.objects (id, bucket_id, name, version, metadata, archived_at, is_versioned) - VALUES ($1, $2, $3, $4, $5::jsonb, '2026-07-02', true)`, - [objectId, bucketId, objectName, objectVersion, JSON.stringify({ size })] - ) - } - }) - - const claimId = randomUUID() - const claimed = await tHelper.database.claimLifecycleBucket({ - bucketId, - claimId, - leaseMs: 120_000, - }) - expect(claimed).toMatchObject({ claimId }) - const candidate = { - id, - name, - version, - isDeleteMarker: false, - archivedAt: '2026-07-02T00:00:00.000Z', - } - const completedCandidate = { - ...candidate, - id: completedId, - name: completedName, - version: completedVersion, - } - const armed = armedContinuation( - generation, - [candidate, completedCandidate], - { name: completedName, archivedAt: completedCandidate.archivedAt }, - { snapshotAt: '2026-09-01T00:00:00.000Z' } - ) - const armedState = await tHelper.database.armLifecycleAttempt({ - bucketId, - claimId, - leaseMs: 120_000, - continuation: armed, - }) - expect(armedState.status).toBe('armed') - if (armedState.status !== 'armed') throw new Error('Expected armed result') - expect(armedState.state.continuation).toMatchObject({ - batch: { inFlight: { attemptId: armed.batch!.inFlight!.attemptId } }, - }) - - const partiallyRecorded = recordLifecycleArtifactOutcomes( - { ...armed, counters: armedState.state.continuation!.counters }, - new Map([ - [ - armed.batch!.versions[0].artifacts[0].key, - { outcome: 'UNRESOLVED' as const, error: 'retryable backend failure' }, - ], - [armed.batch!.versions[0].artifacts[1].key, { outcome: 'DELETED' as const }], - ...armed.batch!.versions[1].artifacts.map( - (artifact) => [artifact.key, { outcome: 'DELETED' as const }] as const - ), - ]) - ) - const partial = await tHelper.database.commitLifecycleAttempt({ - bucketId, - claimId, - attemptId: armed.batch!.inFlight!.attemptId, - continuation: partiallyRecorded, - }) - expect(partial?.continuation).toMatchObject({ - batch: { - inFlight: { attemptId: armed.batch!.inFlight!.attemptId }, - versions: [ - { - artifacts: [ - expect.objectContaining({ outcome: 'UNRESOLVED' }), - expect.objectContaining({ outcome: 'DELETED' }), - ], - }, - ], - }, - counters: { objectVersionsDeleted: 2, bytesDeleted: 50, batchesCompleted: 0 }, - }) - - const recorded = recordLifecycleArtifactOutcomes( - partial!.continuation!, - new Map([[armed.batch!.versions[0].artifacts[0].key, { outcome: 'DELETED' as const }]]) - ) - const committed = await tHelper.database.commitLifecycleAttempt({ - bucketId, - claimId, - attemptId: armed.batch!.inFlight!.attemptId, - continuation: recorded, - }) - expect(committed?.continuation).toMatchObject({ - cursor: { name: completedName, archivedAt: candidate.archivedAt }, - counters: { objectVersionsDeleted: 2, bytesDeleted: 50, batchesCompleted: 1 }, - }) - }) describe('bucket queries', () => { const claim = (id = bucketId, leaseMs = 60_000) => tHelper.database.claimLifecycleBucket({ bucketId: id, claimId: randomUUID(), leaseMs }) @@ -650,8 +528,7 @@ describe('noncurrent lifecycle database API', () => { `UPDATE storage.bucket_lifecycle_states SET ${sets} WHERE bucket_id = $1`, [id, ...values] ) - const dispatchState = (options: { inFlightOnly?: boolean } = {}) => - tHelper.database.findLifecycleDispatchState({ runBudgetMs: 20_000, ...options }) + const dispatchState = () => tHelper.database.findLifecycleDispatchState({ runBudgetMs: 20_000 }) const listed = async (id: string) => (await tHelper.database.listDueLifecycleBuckets(1000)).some((bucketId) => bucketId === id) @@ -845,14 +722,21 @@ describe('noncurrent lifecycle database API', () => { ) .toBe(true) await writer.query('COMMIT') - expect((await claim)?.bucket.lifecycle_configuration_generation).toBe(stale) + await expect(claim).resolves.toMatchObject({ claimId: identity.claimId }) + expect( + (await tHelper.database.findLifecycleBucket(bucketId)).lifecycle_configuration_generation + ).not.toBe(stale) - const continuation = armedContinuation(stale!, [{ ...row, isDeleteMarker: false }], { - name: row.name, - archivedAt: row.archivedAt, - }) await expect( - tHelper.database.armLifecycleAttempt({ ...identity, leaseMs: 60_000, continuation }) + tHelper.database.armLifecycleAttempt({ + ...identity, + leaseMs: 60_000, + continuation: newContinuation(stale!), + batch: batchOf([{ ...row, isDeleteMarker: false }], { + name: row.name, + archivedAt: row.archivedAt, + }), + }) ).resolves.toEqual({ status: 'fenced' }) await expect(tHelper.database.revalidateLifecycleClaim(identity)).resolves.toMatchObject({ claimId: identity.claimId, @@ -878,83 +762,40 @@ describe('noncurrent lifecycle database API', () => { await expect(dispatchState()).resolves.toMatchObject({ nextDispatchAt: null }) }) - - test('restricts due buckets and the next dispatch time to journals in flight on request', async () => { - const other = await fixtures.createBucket() - for (const id of [bucketId, other]) { - await fixtures.setStatus(id, 'ENABLED') - await tHelper.database.putLifecycleConfiguration(id, { - rules: [ - { status: 'Enabled', filter: {}, noncurrentVersionExpiration: { noncurrentDays: 30 } }, - ], - }) - } - await setState("next_run_at = '2000-01-01'", other) - await setState("next_run_at = '2000-01-02'", bucketId) - const generation = (await tHelper.database.findLifecycleBucket(bucketId)) - .lifecycle_configuration_generation! - const continuation = armedContinuation( - generation, - [{ id: randomUUID(), name: 'a', version: randomUUID(), isDeleteMarker: false }], - { name: 'a', archivedAt: new Date().toISOString() } - ) - await setState('continuation = $2::jsonb', bucketId, [JSON.stringify(continuation)]) - - const all = await tHelper.database.listDueLifecycleBuckets(1000) - expect(all).toEqual(expect.arrayContaining([bucketId, other])) - expect(all.indexOf(other)).toBeLessThan(all.indexOf(bucketId)) - const filtered = await tHelper.database.listDueLifecycleBuckets(1000, { inFlightOnly: true }) - expect(filtered).toContain(bucketId) - expect(filtered).not.toContain(other) - - const { nextDispatchAt: next } = await dispatchState({ inFlightOnly: true }) - expect(next).toEqual(new Date('2000-01-02')) - await expect(dispatchState()).resolves.toMatchObject({ - nextDispatchAt: new Date('2000-01-01'), - }) - }) }) - test('reports parked failures and in-flight age without decoding journals', async () => { + test('reports parked failures and the oldest pending deletion attempt', async () => { await putConfiguration() - const other = await fixtures.createBucket() - await tHelper.database.putLifecycleConfiguration(other, { - rules: [ - { status: 'Enabled', filter: {}, noncurrentVersionExpiration: { noncurrentDays: 1 } }, - ], - }) await tHelper.database.connection.query( `UPDATE storage.bucket_lifecycle_states SET next_run_at = NULL, failure_count = 9 WHERE bucket_id = $1`, [bucketId] ) await tHelper.database.connection.query( - `UPDATE storage.bucket_lifecycle_states SET next_run_at = '2026-01-01', failure_count = 3, - continuation = '{"continuationVersion":999,"batch":{"inFlight":{"startedAt":"2025-12-01T00:00:00.000Z"}}}' - WHERE bucket_id = $1`, - [other] + `INSERT INTO storage.lifecycle_abandoned_artifacts (bucket_id, key, name, attempted_at) + VALUES ($1, 'a', 'a', '2025-12-05T00:00:00Z'), ($1, 'b', 'b', '2025-12-01T00:00:00Z')`, + [bucketId] ) await expect( - tHelper.database.findLifecycleDispatchState({ inFlightOnly: true, runBudgetMs: 20_000 }) + tHelper.database.findLifecycleDispatchState({ runBudgetMs: 20_000 }) ).resolves.toEqual({ - nextDispatchAt: new Date('2026-01-01'), + nextDispatchAt: null, maxFailureCount: 9, oldestInFlightStartedAt: '2025-12-01T00:00:00.000Z', - hasAbandonedArtifacts: false, + hasAbandonedArtifacts: true, }) - await tHelper.database.connection.query( - `UPDATE storage.bucket_lifecycle_states SET continuation = jsonb_set(continuation, '{batch,inFlight,startedAt}', '"invalid"') WHERE bucket_id = $1`, - [other] + await tHelper.database.resolveLifecycleAbandonedArtifacts( + [ + { bucketId, key: 'a' }, + { bucketId, key: 'b' }, + ], + [] ) await expect( tHelper.database.findLifecycleDispatchState({ runBudgetMs: 20_000 }) - ).resolves.toMatchObject({ - nextDispatchAt: new Date('2026-01-01'), - maxFailureCount: 9, - oldestInFlightStartedAt: null, - }) + ).resolves.toMatchObject({ oldestInFlightStartedAt: null, hasAbandonedArtifacts: false }) }) - describe('arm and commit coordination', () => { + describe('arm coordination', () => { async function armBatch( names: string[], options: { @@ -980,49 +821,26 @@ describe('noncurrent lifecycle database API', () => { ) const identity = { bucketId, claimId: randomUUID() } await tHelper.database.claimLifecycleBucket({ ...identity, leaseMs: 60_000 }) - const staged = stagedContinuation( - generation, + const batch = batchOf( rows.map(({ id, name, version }) => ({ id, name, version, isDeleteMarker: false })), - { name: rows.at(-1)!.name, archivedAt: rows.at(-1)!.archivedAt }, - options + { name: rows.at(-1)!.name, archivedAt: rows.at(-1)!.archivedAt } ) - const armed = armLifecycleAttempt(staged, { - attemptId: randomUUID(), - startedAt: new Date().toISOString(), + const continuation = newContinuation(generation, { + ...(options.snapshotAt && { snapshotAt: options.snapshotAt }), }) - const input = { - ...identity, - leaseMs: 60_000, - continuation: armed, - } - const recorded = () => - recordLifecycleArtifactOutcomes( - armed, - new Map( - armed.batch!.versions.flatMap((version) => - version.artifacts.map((artifact) => [artifact.key, { outcome: 'DELETED' as const }]) - ) - ) - ) + const input = { ...identity, leaseMs: 60_000, continuation, batch } return { identity, rows, generation, - staged, - armed, + continuation, + batch, input, arm: () => tHelper.database.armLifecycleAttempt(input), - commit: (continuation = recorded()) => - tHelper.database.commitLifecycleAttempt({ - ...identity, - attemptId: armed.batch!.inFlight!.attemptId, - continuation, - }), - recorded, } } - test('rejects oversized new arms before deleting metadata or persisting a journal', async () => { + test('rejects oversized new arms before deleting metadata or persisting a continuation', async () => { const fixture = await armBatch(Array.from({ length: 49 }, (_, index) => `oversized/${index}`)) await expect(fixture.arm()).rejects.toMatchObject({ code: 'InvalidParameter' }) @@ -1036,114 +854,24 @@ describe('noncurrent lifecycle database API', () => { ]) }) - test('rejects arm and commit inputs whose journal does not describe an armed attempt', async () => { + test('rejects an empty batch', async () => { const fixture = await armBatch(['invalid-input']) await expect( - tHelper.database.armLifecycleAttempt({ ...fixture.input, continuation: fixture.staged }) - ).rejects.toMatchObject({ code: 'InvalidParameter' }) - await expect( - tHelper.database.commitLifecycleAttempt({ - ...fixture.identity, - attemptId: randomUUID(), - continuation: fixture.recorded(), + tHelper.database.armLifecycleAttempt({ + ...fixture.input, + batch: { ...fixture.batch, versions: [] }, }) ).rejects.toMatchObject({ code: 'InvalidParameter' }) }) test.each([ - { state: 'armed', capped: true }, - { state: 'unreadable armed', capped: true }, - { state: 'unreadable idle', capped: false }, - { state: 'empty', capped: false }, - ])('caps only preserved journals on release: $state', async ({ state, capped }) => { - const fixture = await armBatch(['retry-cap']) - await expect(fixture.arm()).resolves.toMatchObject({ status: 'armed' }) - const continuation = - state === 'armed' - ? encodeLifecycleContinuation(fixture.armed) - : state === 'empty' - ? null - : { - continuationVersion: 999, - ...(state === 'unreadable armed' && { - batch: { inFlight: { attemptId: randomUUID() } }, - }), - } - await tHelper.database.connection.query( - `UPDATE storage.bucket_lifecycle_states - SET continuation = $2::jsonb, claim_until = '2000-01-01' WHERE bucket_id = $1`, - [bucketId, continuation === null ? null : JSON.stringify(continuation)] - ) - const nextRunAt = '2099-01-01T01:00:00.000Z' - const maxInFlightRetryAt = '2099-01-01T00:15:00.000Z' - await expect( - tHelper.database.releaseLifecycleClaim({ - ...fixture.identity, - nextRunAt, - maxInFlightRetryAt, - error: { message: 'lost acknowledgement' }, - }) - ).resolves.toBe(true) - await expect( - fixtures.readState(bucketId, 'continuation, claim_id, next_run_at') - ).resolves.toEqual([ - { - continuation, - claim_id: null, - next_run_at: new Date(capped ? maxInFlightRetryAt : nextRunAt), - }, - ]) - }) - - test.each([ - { continuation: 'idle', error: false, expected: 0 }, - { continuation: 'armed', error: false, expected: 7 }, - { continuation: 'idle', error: true, expected: 8 }, + { continuation: undefined, outcome: 'InvalidParameter' }, + { continuation: null, outcome: true }, ])( - 'release with $continuation journal, error=$error leaves failure_count $expected', - async ({ continuation, error, expected }) => { - const fixture = await armBatch(['failure-count']) - const armed = await fixture.arm() - expect(armed).toBeDefined() - if (continuation === 'idle') await expect(fixture.commit()).resolves.toBeDefined() - await tHelper.database.connection.query( - 'UPDATE storage.bucket_lifecycle_states SET failure_count = 7 WHERE bucket_id = $1', - [bucketId] - ) - await expect( - tHelper.database.releaseLifecycleClaim({ - ...fixture.identity, - ...(continuation === 'idle' && { continuation: fixture.staged }), - nextRunAt: '2099-01-01T01:00:00.000Z', - ...(error && { error: { message: 'boom' } }), - }) - ).resolves.toBe(true) - await expect(fixtures.readState(bucketId, 'failure_count')).resolves.toEqual([ - { failure_count: expected }, - ]) - } - ) - - test.each([ - { journal: 'preserved', outcome: 'InvalidParameter' }, - { journal: 'armed', outcome: 'InvalidParameter' }, - { journal: 'staged', outcome: true }, - { journal: 'empty', outcome: true }, - ] as const)( - 'release without a schedule with $journal journal: $outcome', - async ({ journal, outcome }) => { + 'release without a schedule with continuation $continuation: $outcome', + async ({ continuation, outcome }) => { const fixture = await armBatch(['release/unscheduled']) - await expect(fixture.arm()).resolves.toMatchObject({ status: 'armed' }) - if (journal === 'staged' || journal === 'empty') { - await expect(fixture.commit()).resolves.toBeDefined() - } - const continuation = { - preserved: undefined, - armed: fixture.armed, - staged: fixture.staged, - empty: null, - }[journal] await expect( tHelper.database @@ -1153,21 +881,6 @@ describe('noncurrent lifecycle database API', () => { } ) - test('rejects an explicit armed journal that was never armed', async () => { - const fixture = await armBatch(['release/unarmed']) - await expect( - tHelper.database.releaseLifecycleClaim({ - ...fixture.identity, - continuation: fixture.armed, - nextRunAt: '2099-01-01T00:00:00.000Z', - }) - ).rejects.toMatchObject({ code: 'InvalidParameter' }) - await expect(fixtures.readState(bucketId, 'continuation, claim_id')).resolves.toEqual([ - { continuation: null, claim_id: fixture.identity.claimId }, - ]) - expect(await rowExists(fixture.rows[0])).toBe(true) - }) - const expireClaim = () => tHelper.database.connection.query( "UPDATE storage.bucket_lifecycle_states SET claim_until = now() - interval '1 second' WHERE bucket_id = $1", @@ -1184,16 +897,6 @@ describe('noncurrent lifecycle database API', () => { ]) }) - test('rejects arming over a journal in flight even when its rows are eligible again', async () => { - const fixture = await armBatch(['arm/rearmed']) - await expect(fixture.arm()).resolves.toMatchObject({ status: 'armed' }) - await insertArchivedObjects(fixture.rows) - const before = await fixtures.readState(bucketId, 'continuation') - - await expect(fixture.arm()).resolves.toEqual({ status: 'fenced' }) - await expect(fixtures.readState(bucketId, 'continuation')).resolves.toEqual(before) - }) - test.each([ { change: 'versioning is disabled', @@ -1213,7 +916,7 @@ describe('noncurrent lifecycle database API', () => { WHERE id = $1`, claimKept: true, }, - ])('does not arm a staged batch after $change', async ({ sql, claimKept }) => { + ])('does not arm a batch after $change', async ({ sql, claimKept }) => { const fixture = await armBatch(['arm/recheck']) const writer = await holdWriter( sql, @@ -1245,16 +948,102 @@ describe('noncurrent lifecycle database API', () => { ]) }) - test('persists the armed journal without artifact keys', async () => { - const fixture = await armBatch(['arm/no-keys']) - await expect(fixture.arm()).resolves.toMatchObject({ status: 'armed' }) + test('derives outbox keys from the deleted version instead of caller-supplied keys', async () => { + const fixture = await armBatch(['outbox/derived']) + fixture.batch.versions[0].artifacts = ['unrelated/object'] - await expect( - fixtures.readState( - bucketId, - "jsonb_path_exists(continuation, '$.batch.versions[*].artifacts[*].key') AS has_key, jsonb_array_length(continuation #> '{batch,versions,0,artifacts}') AS artifacts" + const result = await fixture.arm() + if (result.status !== 'armed') throw new Error('Expected armed result') + + const { name, version } = fixture.rows[0] + const key = withOptionalVersion(name, version) + const expected = [key, `${key}.info`] + const outbox = await tHelper.database.connection.query<{ key: string }>( + 'SELECT key FROM storage.lifecycle_abandoned_artifacts WHERE bucket_id = $1 ORDER BY key', + [bucketId] + ) + expect(outbox.rows.map((row) => row.key)).toEqual(expected) + expect(result.artifacts.map((artifact) => artifact.key)).toEqual(expected) + }) + + test('enqueues outbox rows for exactly the deleted versions with the metadata delete', async () => { + const generation = await putConfiguration() + const live = { id: randomUUID(), name: 'outbox/live', version: randomUUID() } + const vetoed = { id: randomUUID(), name: 'outbox/vetoed', version: randomUUID() } + const marker = { id: randomUUID(), name: 'outbox/marker', version: randomUUID() } + const archivedAt = '2026-06-01T00:00:00.000Z' + await insertArchivedObjects([ + { ...live, archivedAt }, + { ...vetoed, archivedAt }, + ]) + await tHelper.database.connection.query( + `INSERT INTO storage.objects (id, bucket_id, name, version, archived_at, is_versioned, is_delete_marker) + VALUES ($1, $2, $3, $4, $5, true, true)`, + [marker.id, bucketId, marker.name, marker.version, archivedAt] + ) + const identity = { bucketId, claimId: randomUUID() } + await tHelper.database.claimLifecycleBucket({ ...identity, leaseMs: 60_000 }) + const input = { + ...identity, + leaseMs: 60_000, + continuation: newContinuation(generation), + batch: batchOf( + [ + { ...live, isDeleteMarker: false }, + { ...vetoed, isDeleteMarker: false }, + { ...marker, isDeleteMarker: true }, + ], + { name: marker.name, archivedAt } + ), + } + const outbox = async () => + ( + await tHelper.database.connection.query<{ key: string }>( + 'SELECT key FROM storage.lifecycle_abandoned_artifacts WHERE bucket_id = $1 ORDER BY key', + [bucketId] + ) + ).rows.map((row) => row.key) + try { + await reference(vetoed.id) + const rolledBack = new Error('rollback after arm') + await expect( + tHelper.database.withTransaction(async (db) => { + await expect(db.armLifecycleAttempt(input)).resolves.toMatchObject({ status: 'armed' }) + throw rolledBack + }) + ).rejects.toBe(rolledBack) + expect(await outbox()).toEqual([]) + expect(await rowExists(live)).toBe(true) + + const result = await tHelper.database.armLifecycleAttempt(input) + if (result.status !== 'armed') throw new Error('Expected armed result') + + const keys = [`${live.name}/${live.version}`, `${live.name}/${live.version}.info`] + expect(await outbox()).toEqual(keys) + expect(result.artifacts).toEqual( + keys.map((key) => ({ bucketId, key, name: live.name, version: live.version })) ) - ).resolves.toEqual([{ has_key: false, artifacts: 2 }]) + expect(await rowExists(live)).toBe(false) + expect(await rowExists(vetoed)).toBe(true) + expect(await rowExists(marker)).toBe(false) + expect(result.state.continuation).toEqual({ + ...input.continuation, + cursor: { name: marker.name, archivedAt }, + counters: { + ...input.continuation.counters, + versionsExamined: 3, + versionsEligible: 3, + objectVersionsDeleted: 1, + deleteMarkersDeleted: 1, + bytesDeleted: 10, + batchesCompleted: 1, + }, + }) + const [stored] = await fixtures.readState(bucketId, 'continuation') + expect(stored.continuation).not.toHaveProperty('batch') + } finally { + await dropReferences() + } }) async function reference(id: string) { @@ -1268,46 +1057,22 @@ describe('noncurrent lifecycle database API', () => { const dropReferences = () => tHelper.database.connection.query('DROP TABLE IF EXISTS public.lifecycle_arm_refs') - test('arms only the versions whose metadata delete is not vetoed', async () => { - const fixture = await armBatch(['arm/vetoed', 'arm/kept']) - try { - await reference(fixture.rows[0].id) - const result = await fixture.arm() - expect(result.status).toBe('armed') - if (result.status !== 'armed') throw new Error('Expected armed result') - const state = result.state - - expect(state.claimId).toBe(fixture.identity.claimId) - expect(state.continuation?.batch?.versions.map(({ id }) => id)).toEqual([ - fixture.rows[1].id, - ]) - expect(state.continuation?.counters).toMatchObject({ objectVersionsDeleted: 1 }) - await expect( - fixtures.readState( - bucketId, - "jsonb_path_exists(continuation, '$.batch.versions[*].artifacts[*].key') AS has_key" - ) - ).resolves.toEqual([{ has_key: false }]) - expect(await rowExists(fixture.rows[0])).toBe(true) - expect(await rowExists(fixture.rows[1])).toBe(false) - } finally { - await dropReferences() - } - }) - - test('advances the cursor past a page whose every version is vetoed without arming', async () => { + test('advances the cursor past a page whose every version is vetoed without enqueueing', async () => { const fixture = await armBatch(['arm/all-vetoed']) - const { batch: _batch, pageEnd, ...rest } = fixture.staged - const skipped = { ...rest, cursor: pageEnd } + const skipped = advanceLifecycleBatch(fixture.continuation, fixture.batch, { + objectVersions: 0, + deleteMarkers: 0, + bytes: 0, + }) try { await reference(fixture.rows[0].id) const result = await fixture.arm() expect(result.status).toBe('armed') if (result.status !== 'armed') throw new Error('Expected armed result') - const state = result.state - expect(state.claimId).toBe(fixture.identity.claimId) - expect(state.continuation).toEqual(skipped) + expect(result.artifacts).toEqual([]) + expect(result.state.claimId).toBe(fixture.identity.claimId) + expect(result.state.continuation).toEqual(skipped) await expect( fixtures.readState( bucketId, @@ -1316,6 +1081,9 @@ describe('noncurrent lifecycle database API', () => { ).resolves.toEqual([ { continuation: skipped, claim_id: fixture.identity.claimId, claimed: true, due: true }, ]) + await expect( + tHelper.database.findLifecycleDispatchState({ runBudgetMs: 1 }) + ).resolves.toMatchObject({ hasAbandonedArtifacts: false }) } finally { await dropReferences() } @@ -1336,51 +1104,6 @@ describe('noncurrent lifecycle database API', () => { } }) - test('persists the committed results without artifact keys', async () => { - const fixture = await armBatch(['commit/no-keys']) - await expect(fixture.arm()).resolves.toMatchObject({ status: 'armed' }) - await expect(fixture.commit(fixture.armed)).resolves.toBeDefined() - - await expect( - fixtures.readState( - bucketId, - "jsonb_path_exists(continuation, '$.batch.versions[*].artifacts[*].key') AS has_key, jsonb_array_length(continuation #> '{batch,versions}') AS versions" - ) - ).resolves.toEqual([{ has_key: false, versions: 1 }]) - }) - - test('keeps the arm counters when the commit proposal predates the arm', async () => { - const fixture = await armBatch(['commit/stale-counters']) - const armed = await fixture.arm() - expect(armed.status).toBe('armed') - if (armed.status !== 'armed') throw new Error('Expected armed result') - expect(armed.state.continuation?.counters).toMatchObject({ - objectVersionsDeleted: 1, - bytesDeleted: 10, - batchesCompleted: 0, - }) - const committed = await fixture.commit() - expect(committed?.continuation?.counters).toMatchObject({ - objectVersionsDeleted: 1, - bytesDeleted: 10, - batchesCompleted: 1, - }) - }) - - test('revalidates a recovery attempt only while the claim lease is live', async () => { - const fixture = await armBatch(['arm/recovery']) - await expect(fixture.arm()).resolves.toMatchObject({ status: 'armed' }) - const attemptId = fixture.armed.batch!.inFlight!.attemptId - - await expect( - tHelper.database.revalidateLifecycleRecoveryAttempt(fixture.identity, attemptId, 60_000) - ).resolves.toBeDefined() - await expireClaim() - await expect( - tHelper.database.revalidateLifecycleRecoveryAttempt(fixture.identity, attemptId, 60_000) - ).resolves.toBeUndefined() - }) - const SNAPSHOT_AT = '2026-08-15T00:00:00.000Z' test.each([ { label: 'exactly at the rule cutoff', archivedAt: '2026-07-16T00:00:00.000Z', arms: false }, @@ -1479,80 +1202,6 @@ describe('noncurrent lifecycle database API', () => { } } - test('preserves an armed journal until its backend artifacts are resolved', async () => { - const fixture = await armBatch(['completion/pending']) - const row = fixture.rows[0] - const { adapter, storage } = tHelper - const root = storage.location.getRootLocation() - const key = storage.location.getKeyLocation({ - tenantId: getConfig().tenantId, - bucketId, - objectName: row.name, - }) - await adapter.uploadObject( - root, - key, - row.version, - Readable.from(['pending bytes']), - 'text/plain', - 'no-cache' - ) - - const armed = await fixture.arm() - expect(armed.status).toBe('armed') - if (armed.status !== 'armed') throw new Error('Expected armed result') - expect(armed.state.continuation?.batch?.inFlight).toBeDefined() - expect(await rowExists(row)).toBe(false) - const before = await fixtures.readState(bucketId, 'continuation, claim_id, next_run_at') - const nextRunAt = '2099-01-01T00:00:00.000Z' - await expect( - tHelper.database.completeLifecycleRun( - { ...fixture.identity, generation: fixture.generation }, - {}, - nextRunAt - ) - ).resolves.toBe(false) - const changed = recordLifecycleArtifactOutcomes( - armed.state.continuation!, - new Map([ - [armed.state.continuation!.batch!.versions[0].artifacts[0].key, { outcome: 'DELETED' }], - ]) - ) - for (const continuation of [null, fixture.staged]) { - await expect( - tHelper.database.releaseLifecycleClaim({ ...fixture.identity, continuation, nextRunAt }) - ).resolves.toBe(false) - } - for (const continuation of [changed, armed.state.continuation!]) { - await expect( - tHelper.database.releaseLifecycleClaim({ ...fixture.identity, continuation, nextRunAt }) - ).rejects.toMatchObject({ code: 'InvalidParameter' }) - } - await expect( - fixtures.readState(bucketId, 'continuation, claim_id, next_run_at') - ).resolves.toEqual(before) - await expect(adapter.headObject(root, key, row.version)).resolves.toMatchObject({ size: 13 }) - - await adapter.deleteObject(root, key, row.version) - const [data, info] = armed.state.continuation!.batch!.versions[0].artifacts - const recorded = recordLifecycleArtifactOutcomes( - armed.state.continuation!, - new Map([ - [data.key, { outcome: 'DELETED' as const }], - [info.key, { outcome: 'ABSENT' as const }], - ]) - ) - await expect(fixture.commit(recorded)).resolves.toBeDefined() - await expect( - tHelper.database.completeLifecycleRun( - { ...fixture.identity, generation: fixture.generation }, - {}, - nextRunAt - ) - ).resolves.toBe(true) - await expect(adapter.headObject(root, key, row.version)).rejects.toThrow() - }) - test('preserves an unknown journal on completion or explicit release', async () => { const fixture = await armBatch(['completion/unknown']) const continuation = { continuationVersion: 999 } @@ -1583,7 +1232,7 @@ describe('noncurrent lifecycle database API', () => { ]) }) - test('preserves an in-flight journal with newer fields when releasing the claim', async () => { + test('preserves a continuation with newer fields when releasing the claim', async () => { const fixture = await armBatch(['release/future-fields']) await expect(fixture.arm()).resolves.toMatchObject({ status: 'armed' }) const nextRunAt = '2099-01-01T01:00:00.000Z' @@ -1612,11 +1261,10 @@ describe('noncurrent lifecycle database API', () => { ]) }) - test('keeps a policy change made during an armed batch due after the stale run completes', async () => { + test('keeps a policy change made during a batch due and ends the stale run', async () => { const fixture = await armBatch(['policy/armed']) await expect(fixture.arm()).resolves.toMatchObject({ status: 'armed' }) await tHelper.database.putLifecycleConfiguration(bucketId, configuration(2)) - await expect(fixture.commit()).resolves.toBeDefined() await expect( tHelper.database.completeLifecycleRun( @@ -1624,76 +1272,10 @@ describe('noncurrent lifecycle database API', () => { {}, new Date(Date.now() + 86_400_000).toISOString() ) - ).resolves.toBe(true) + ).resolves.toBe(false) await expect(dueNow()).resolves.toBe(true) }) - test('rejects a commit for a stale attempt', async () => { - const fixture = await armBatch(['commit/stale-attempt']) - await expect(fixture.arm()).resolves.toMatchObject({ status: 'armed' }) - const continuation = fixture.recorded() - const attemptId = randomUUID() - continuation.batch!.inFlight!.attemptId = attemptId - await expect( - tHelper.database.commitLifecycleAttempt({ - ...fixture.identity, - attemptId, - continuation, - }) - ).resolves.toBeUndefined() - await expect( - fixtures.readState(bucketId, "continuation #>> '{batch,inFlight,attemptId}' AS attempt") - ).resolves.toEqual([{ attempt: fixture.armed.batch!.inFlight!.attemptId }]) - }) - - test('abandons pending artifacts only for the claim holder and advances the cursor', async () => { - const fixture = await armBatch(['abandon/persist']) - await expect(fixture.arm()).resolves.toMatchObject({ status: 'armed' }) - const keys = fixture.armed.batch!.versions[0].artifacts.map((artifact) => artifact.key) - const failed = recordLifecycleArtifactOutcomes( - fixture.armed, - new Map([ - [keys[0], { outcome: 'FAILED' as const, error: 'denied' }], - [keys[1], { outcome: 'DELETED' as const }], - ]) - ) - const commit = (claimId: string, abandonPending = false) => - tHelper.database.commitLifecycleAttempt({ - ...fixture.identity, - claimId, - attemptId: fixture.armed.batch!.inFlight!.attemptId, - continuation: failed, - abandonPending, - }) - const abandoned = async () => - ( - await tHelper.database.connection.query( - 'SELECT key, name, version FROM storage.lifecycle_abandoned_artifacts WHERE bucket_id = $1', - [bucketId] - ) - ).rows - - const retained = await commit(fixture.identity.claimId) - expect(retained?.continuation?.batch?.versions).toHaveLength(1) - await expect(abandoned()).resolves.toEqual([]) - - await expect(commit(randomUUID(), true)).resolves.toBeUndefined() - expect( - (await fixtures.readState(bucketId, 'continuation'))[0].continuation.batch - ).toBeDefined() - await expect(abandoned()).resolves.toEqual([]) - - const result = await commit(fixture.identity.claimId, true) - expect(result?.continuation?.batch).toBeUndefined() - expect(result?.continuation).toMatchObject({ - cursor: fixture.staged.pageEnd, - counters: { batchesCompleted: 1 }, - }) - await expect(abandoned()).resolves.toEqual([ - { key: keys[0], name: 'abandon/persist', version: fixture.rows[0].version }, - ]) - }) - const mutations = { renamed: () => ['UPDATE storage.objects SET name = name || $2 WHERE id = $1', '-moved'], reversioned: () => ['UPDATE storage.objects SET version = $2 WHERE id = $1', randomUUID()], @@ -1714,7 +1296,7 @@ describe('noncurrent lifecycle database API', () => { } test.each(Object.keys(mutations) as Array)( - 'does not arm a row that was %s after staging', + 'does not arm a row that was %s after evaluation', async (kind) => { const fixture = await armBatch(['identity/pre-arm']) await mutateRow(kind, fixture.rows[0].id) @@ -1723,49 +1305,6 @@ describe('noncurrent lifecycle database API', () => { } ) - test('rejects a commit whose version or artifact set differs from the armed batch', async () => { - const fixture = await armBatch(['commit/a', 'commit/b']) - await expect(fixture.arm()).resolves.toMatchObject({ status: 'armed' }) - const recorded = fixture.recorded() - - const dropped = { - ...recorded, - batch: { ...recorded.batch!, versions: recorded.batch!.versions.slice(0, 1) }, - } - const withoutInfo = { - ...recorded, - batch: { - ...recorded.batch!, - versions: recorded.batch!.versions.map((version, index) => - index === 0 ? { ...version, artifacts: version.artifacts.slice(0, 1) } : version - ), - }, - } - const retargeted = (changes: Record) => ({ - ...recorded, - batch: { - ...recorded.batch!, - versions: recorded.batch!.versions.map((version, index) => - index === 0 ? { ...version, ...changes } : version - ), - }, - }) - for (const proposed of [ - dropped, - withoutInfo, - { ...recorded, batch: { ...recorded.batch!, keySeparator: '-$v-' as const } }, - retargeted({ id: randomUUID() }), - retargeted({ name: 'commit/other' }), - retargeted({ version: randomUUID() }), - ]) { - await expect(fixture.commit(proposed)).rejects.toMatchObject({ - code: 'InvalidRequest', - message: expect.stringContaining('artifact set changed'), - }) - } - await expect(fixture.commit()).resolves.toBeDefined() - }) - test.each(['name', 'version', 'bucket_id'] as const)( 'does not arm a row whose %s changes while arm waits for its row lock', async (column) => { @@ -1927,6 +1466,27 @@ describe('noncurrent lifecycle database API', () => { } }) + test('starts the claim lease after slow delete triggers finish', async () => { + const fixture = await armBatch(['lease/slow']) + const query = (text: string) => tHelper.database.connection.query(text) + await query(`CREATE FUNCTION public.lifecycle_slow_delete() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN PERFORM pg_sleep(1); RETURN OLD; END $$`) + await query(`CREATE TRIGGER lifecycle_slow_delete BEFORE DELETE ON storage.objects + FOR EACH ROW EXECUTE FUNCTION public.lifecycle_slow_delete()`) + try { + await expect(fixture.arm()).resolves.toMatchObject({ status: 'armed' }) + } finally { + await query('DROP TRIGGER IF EXISTS lifecycle_slow_delete ON storage.objects') + await query('DROP FUNCTION IF EXISTS public.lifecycle_slow_delete()') + } + const { rows } = await tHelper.database.connection.query( + `SELECT extract(epoch FROM claim_until - clock_timestamp()) * 1000 AS remaining_ms + FROM storage.bucket_lifecycle_states WHERE bucket_id = $1`, + [bucketId] + ) + expect(Number(rows[0].remaining_ms)).toBeGreaterThan(fixture.input.leaseMs - 500) + }) + test('holds a bucket share lock during arming that blocks bucket deletion', async () => { const fixture = await armBatch(['lock/share']) const deleter = new Client({ connectionString: getConfig().databaseURL }) diff --git a/src/test/utils/lifecycle-continuation.ts b/src/test/utils/lifecycle-continuation.ts index 6e0e86404..8b8e876f3 100644 --- a/src/test/utils/lifecycle-continuation.ts +++ b/src/test/utils/lifecycle-continuation.ts @@ -1,9 +1,7 @@ import { randomUUID } from 'node:crypto' import { - armLifecycleAttempt, createLifecycleContinuation, freezeLifecycleBatchVersion, - stageLifecycleBatch, } from '@storage/lifecycle/continuation' import type { LifecycleCandidateIdentity, NoncurrentLifecycleCursor } from '@storage/schemas' @@ -19,30 +17,13 @@ export function newContinuation( }) } -export function stagedContinuation( - generation: string, +export function batchOf( versions: LifecycleCandidateIdentity[], - pageEnd: NoncurrentLifecycleCursor, - options: { snapshotAt?: string } = {} + pageEnd: NoncurrentLifecycleCursor ) { - return stageLifecycleBatch( - newContinuation(generation, options.snapshotAt ? { snapshotAt: options.snapshotAt } : {}), - { - rawRowsExamined: versions.length, - pageEnd, - versions: versions.map((version) => freezeLifecycleBatchVersion(version)), - } - ) -} - -export function armedContinuation( - generation: string, - versions: LifecycleCandidateIdentity[], - pageEnd: NoncurrentLifecycleCursor, - options: { snapshotAt?: string; startedAt?: string } = {} -) { - return armLifecycleAttempt(stagedContinuation(generation, versions, pageEnd, options), { - attemptId: randomUUID(), - startedAt: options.startedAt ?? new Date().toISOString(), - }) + return { + rawRowsExamined: versions.length, + pageEnd, + versions: versions.map((version) => freezeLifecycleBatchVersion(version)), + } }