diff --git a/migrations/multitenant/0031-lifecycle-tenants.sql b/migrations/multitenant/0031-lifecycle-tenants.sql new file mode 100644 index 000000000..4120dd94c --- /dev/null +++ b/migrations/multitenant/0031-lifecycle-tenants.sql @@ -0,0 +1,71 @@ +CREATE TABLE IF NOT EXISTS public.lifecycle_tenants ( + tenant_id text PRIMARY KEY + REFERENCES public.tenants(id) + ON DELETE CASCADE, + + next_dispatch_at timestamptz NOT NULL DEFAULT now(), + + claim_id uuid, + claim_until timestamptz, + + last_dispatched_at timestamptz, + last_completed_at timestamptz, + last_success_at timestamptz, + + failure_count integer NOT NULL DEFAULT 0, + last_error jsonb, + + -- Advisory snapshots collected by tenant dispatchers. + -- Timestamps expose stale or unavailable tenant observations. + -- Scheduling continues to use the tenant state. + max_bucket_failure_count integer, + oldest_in_flight_started_at timestamptz, + bucket_status_observed_at timestamptz, + + CONSTRAINT lifecycle_tenants_failure_count_check + CHECK (failure_count >= 0) +); + +CREATE INDEX IF NOT EXISTS lifecycle_tenants_due_idx +ON public.lifecycle_tenants ( + next_dispatch_at, + tenant_id +); + +CREATE INDEX IF NOT EXISTS lifecycle_tenants_expired_claim_idx +ON public.lifecycle_tenants ( + claim_until, + tenant_id +) +WHERE claim_until IS NOT NULL; + +-- Existing policies, restored tenants and freshly migrated tenant DBs need dispatch without another policy PUT. +CREATE OR REPLACE FUNCTION public.enroll_lifecycle_tenant() +RETURNS trigger +LANGUAGE plpgsql +SET search_path = pg_catalog +AS $$ +BEGIN + IF TG_OP = 'UPDATE' + AND OLD.feature_object_versioning IS NOT DISTINCT FROM NEW.feature_object_versioning + AND OLD.migrations_version IS NOT DISTINCT FROM NEW.migrations_version THEN + RETURN NEW; + END IF; + INSERT INTO public.lifecycle_tenants (tenant_id, next_dispatch_at) + VALUES (NEW.id, clock_timestamp()) + ON CONFLICT (tenant_id) DO UPDATE + SET next_dispatch_at = LEAST(lifecycle_tenants.next_dispatch_at, EXCLUDED.next_dispatch_at); + RETURN NEW; +END; +$$; + +DROP TRIGGER IF EXISTS enroll_lifecycle_tenant ON public.tenants; +CREATE TRIGGER enroll_lifecycle_tenant +AFTER INSERT OR UPDATE OF feature_object_versioning, migrations_version ON public.tenants +FOR EACH ROW WHEN (NEW.feature_object_versioning) +EXECUTE FUNCTION public.enroll_lifecycle_tenant(); + +INSERT INTO public.lifecycle_tenants (tenant_id, next_dispatch_at) +SELECT id, clock_timestamp() FROM public.tenants WHERE feature_object_versioning +ON CONFLICT (tenant_id) DO UPDATE +SET next_dispatch_at = LEAST(lifecycle_tenants.next_dispatch_at, EXCLUDED.next_dispatch_at); diff --git a/migrations/tenant/0076-noncurrent-lifecycle.sql b/migrations/tenant/0076-noncurrent-lifecycle.sql new file mode 100644 index 000000000..3d105303a --- /dev/null +++ b/migrations/tenant/0076-noncurrent-lifecycle.sql @@ -0,0 +1,86 @@ +-- Add durable lifecycle execution state with one row per bucket. + +CREATE TABLE IF NOT EXISTS storage.bucket_lifecycle_states ( + bucket_id text PRIMARY KEY, + generation uuid, + next_run_at timestamptz, + claim_id uuid, + claim_until timestamptz, + continuation jsonb, + last_started_at timestamptz, + last_completed_at timestamptz, + last_success_at timestamptz, + last_result jsonb, + failure_count integer NOT NULL DEFAULT 0, + last_error jsonb, + updated_at timestamptz NOT NULL DEFAULT now(), + + CONSTRAINT bucket_lifecycle_states_bucket_id_fkey + FOREIGN KEY (bucket_id) + REFERENCES storage.buckets(id) + ON DELETE CASCADE, + + CONSTRAINT bucket_lifecycle_failure_count_check + CHECK (failure_count >= 0), + + CONSTRAINT bucket_lifecycle_continuation_object_check + CHECK ( + continuation IS NULL + OR jsonb_typeof(continuation) = 'object' + ), + + CONSTRAINT bucket_lifecycle_recovery_due_check + CHECK ( + next_run_at IS NOT NULL + OR NOT jsonb_path_exists( + COALESCE(continuation, '{}'::jsonb), + '$.batch.inFlight' + ) + ), + + CONSTRAINT bucket_lifecycle_last_result_object_check + CHECK ( + last_result IS NULL + OR jsonb_typeof(last_result) = 'object' + ) +); + +DO $$ +BEGIN + IF NOT (SELECT relrowsecurity FROM pg_catalog.pg_class + WHERE oid = 'storage.bucket_lifecycle_states'::regclass) THEN + ALTER TABLE storage.bucket_lifecycle_states ENABLE ROW LEVEL SECURITY; + END IF; + IF to_regclass('storage.bucket_lifecycle_states_due_idx') IS NULL THEN + CREATE INDEX IF NOT EXISTS bucket_lifecycle_states_due_idx + ON storage.bucket_lifecycle_states (next_run_at, bucket_id) + WHERE next_run_at IS NOT NULL; + END IF; +END; +$$; + +DO $$ +DECLARE + anon_role text = COALESCE(current_setting('storage.anon_role', true), 'anon'); + authenticated_role text = COALESCE(current_setting('storage.authenticated_role', true), 'authenticated'); + service_role text = COALESCE(current_setting('storage.service_role', true), 'service_role'); + super_user text = COALESCE(current_setting('storage.super_user', true), 'postgres'); + role_name text; +BEGIN + REVOKE ALL ON TABLE storage.bucket_lifecycle_states FROM PUBLIC; + + FOREACH role_name IN ARRAY ARRAY[anon_role, authenticated_role] + LOOP + IF EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = role_name) THEN + EXECUTE format('REVOKE ALL ON TABLE storage.bucket_lifecycle_states FROM %I', role_name); + END IF; + END LOOP; + + FOREACH role_name IN ARRAY ARRAY['postgres', super_user, service_role] + LOOP + IF EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = role_name) THEN + EXECUTE format('GRANT ALL ON TABLE storage.bucket_lifecycle_states TO %I', role_name); + END IF; + END LOOP; +END; +$$; diff --git a/migrations/tenant/0077-noncurrent-lifecycle-archived-order-index.sql b/migrations/tenant/0077-noncurrent-lifecycle-archived-order-index.sql new file mode 100644 index 000000000..c6ba36d94 --- /dev/null +++ b/migrations/tenant/0077-noncurrent-lifecycle-archived-order-index.sql @@ -0,0 +1,5 @@ +-- postgres-migrations disable-transaction + +CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS objects_archived_order_uq +ON storage.objects (bucket_id, name COLLATE "C", archived_at) +WHERE archived_at IS NOT NULL; diff --git a/migrations/tenant/0078-noncurrent-lifecycle-archived-due-index.sql b/migrations/tenant/0078-noncurrent-lifecycle-archived-due-index.sql new file mode 100644 index 000000000..45cc0fb54 --- /dev/null +++ b/migrations/tenant/0078-noncurrent-lifecycle-archived-due-index.sql @@ -0,0 +1,10 @@ +-- postgres-migrations disable-transaction + +CREATE INDEX CONCURRENTLY IF NOT EXISTS objects_archived_due_idx +ON storage.objects ( + bucket_id, + archived_at, + name COLLATE "C" +) +INCLUDE (id, version, is_delete_marker) +WHERE archived_at IS NOT NULL; diff --git a/migrations/tenant/0079-lifecycle-writer-coordination.sql b/migrations/tenant/0079-lifecycle-writer-coordination.sql new file mode 100644 index 000000000..ed4364a91 --- /dev/null +++ b/migrations/tenant/0079-lifecycle-writer-coordination.sql @@ -0,0 +1,371 @@ +-- Lock buckets before installing triggers and backfilling lifecycle state, +-- so bucket writes cannot race with the backfill. This migration does not +-- touch objects avoiding a lock cycle with object writers. +-- Try briefly and pause between attempts so writers queued behind the lock +-- request wait for one attempt at most while another transaction blocks it. +-- Give up after the caller's lock_timeout, or a minute when unset, so the +-- runner can retry instead of leaving callers waiting indefinitely. +DO $$ +DECLARE + previous_lock_timeout text := current_setting('lock_timeout'); + deadline timestamptz := clock_timestamp() + CASE previous_lock_timeout + WHEN '0' THEN interval '1 minute' + ELSE previous_lock_timeout::interval + END; +BEGIN + LOOP + -- PostgreSQL lock_timeout uses milliseconds. + -- Never set it to zero because that would disable the timeout. + PERFORM set_config( + 'lock_timeout', + greatest(1, least(250, floor(extract(epoch FROM deadline - clock_timestamp()) * 1000)))::text || 'ms', + true + ); + BEGIN + LOCK TABLE storage.buckets IN EXCLUSIVE MODE; + EXIT; + EXCEPTION WHEN lock_not_available THEN + IF clock_timestamp() >= deadline THEN + RAISE; + END IF; + PERFORM pg_sleep(greatest(0, least(2, extract(epoch FROM deadline - clock_timestamp())))); + IF clock_timestamp() >= deadline THEN + RAISE; + END IF; + END; + END LOOP; + PERFORM set_config('lock_timeout', previous_lock_timeout, true); +END; +$$; + +-- Older builds rely on the foreign key to delete lifecycle state with the bucket. +-- Allow this only when there is no in-flight batch and the continuation is understood. +-- The check does not change state so a later trigger can still cancel the deletion. +CREATE OR REPLACE FUNCTION storage.protect_active_lifecycle_states() +RETURNS trigger +LANGUAGE plpgsql +SECURITY DEFINER +SET search_path = pg_catalog +AS $$ +BEGIN + IF EXISTS ( + SELECT 1 FROM storage.bucket_lifecycle_states + WHERE bucket_id = OLD.id + AND NOT ( + continuation IS NULL + OR (continuation @> '{"continuationVersion": 1}' AND continuation #> '{batch,inFlight}' IS NULL) + ) + ) THEN + RAISE EXCEPTION 'update or delete on table "buckets" violates foreign key constraint "bucket_lifecycle_states_bucket_id_fkey" on table "bucket_lifecycle_states"' + USING ERRCODE = 'foreign_key_violation', + DETAIL = format('Key (id)=(%s) is still referenced from table "bucket_lifecycle_states".', OLD.id), + SCHEMA = 'storage', + TABLE = 'bucket_lifecycle_states', + CONSTRAINT = 'bucket_lifecycle_states_bucket_id_fkey'; + END IF; + RETURN OLD; +END; +$$; + +REVOKE ALL ON FUNCTION storage.protect_active_lifecycle_states() FROM PUBLIC; + +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_catalog.pg_trigger + WHERE tgrelid = 'storage.buckets'::regclass AND tgname = 'protect_active_lifecycle_states' + ) THEN + CREATE TRIGGER protect_active_lifecycle_states + BEFORE DELETE ON storage.buckets + FOR EACH ROW + EXECUTE FUNCTION storage.protect_active_lifecycle_states(); + END IF; +END; +$$; + +-- Policy changes reset pending work but an in-flight batch must finish +-- recovery before its state can be cleared. +CREATE OR REPLACE FUNCTION storage.reconcile_lifecycle_state(target_bucket_id text, policy_changed boolean) +RETURNS void +LANGUAGE plpgsql +SECURITY DEFINER +SET search_path = pg_catalog +AS $$ +DECLARE + bucket storage.buckets%ROWTYPE; + active boolean; + idle boolean; + state_generation uuid; +BEGIN + SELECT * INTO bucket FROM storage.buckets WHERE id = target_bucket_id FOR NO KEY UPDATE; + IF NOT FOUND THEN + RETURN; + END IF; + + SELECT continuation IS NULL OR ( + continuation @> '{"continuationVersion": 1}' + AND continuation #> '{batch,inFlight}' IS NULL + ), generation + INTO idle, state_generation + FROM storage.bucket_lifecycle_states + WHERE bucket_id = target_bucket_id + FOR UPDATE; + + IF bucket.lifecycle_configuration IS NULL THEN + DELETE FROM storage.bucket_lifecycle_states + WHERE bucket_id = target_bucket_id + AND idle IS TRUE; + + UPDATE storage.bucket_lifecycle_states + SET next_run_at = clock_timestamp(), updated_at = clock_timestamp() + WHERE bucket_id = target_bucket_id + AND idle IS NOT TRUE; + RETURN; + END IF; + + active := bucket.versioning_status <> 'DISABLED' + AND bucket.lifecycle_configuration @> '{"rules":[{"status":"Enabled"}]}'::jsonb; + + -- The state row records the generation it was reconciled with so callers need not detect + -- policy changes. Rows without a recorded generation keep their work. + policy_changed := policy_changed OR ( + state_generation IS NOT NULL AND state_generation <> bucket.lifecycle_configuration_generation + ); + + INSERT INTO storage.bucket_lifecycle_states AS state (bucket_id, generation, next_run_at) + VALUES (bucket.id, bucket.lifecycle_configuration_generation, CASE WHEN active THEN clock_timestamp() END) + ON CONFLICT (bucket_id) DO UPDATE + SET next_run_at = CASE + WHEN policy_changed THEN + CASE WHEN active OR idle IS NOT TRUE THEN clock_timestamp() END + ELSE EXCLUDED.next_run_at + END, + generation = EXCLUDED.generation, + claim_id = CASE WHEN policy_changed AND idle IS TRUE + THEN NULL ELSE state.claim_id END, + claim_until = CASE WHEN policy_changed AND idle IS TRUE + THEN NULL ELSE state.claim_until END, + continuation = CASE WHEN policy_changed AND idle IS TRUE + THEN NULL ELSE state.continuation END, + failure_count = CASE WHEN policy_changed AND idle IS TRUE + THEN 0 ELSE state.failure_count END, + last_error = CASE WHEN policy_changed AND idle IS TRUE + THEN NULL ELSE state.last_error END, + updated_at = clock_timestamp() + WHERE policy_changed OR (state.next_run_at IS NULL AND EXCLUDED.next_run_at IS NOT NULL); +END; +$$; + +REVOKE ALL ON FUNCTION storage.reconcile_lifecycle_state(text, boolean) FROM PUBLIC; + +DO $$ +DECLARE + role_name text; +BEGIN + FOR role_name IN + SELECT DISTINCT acl.grantee::regrole::text + FROM pg_catalog.pg_proc AS proc, pg_catalog.aclexplode(proc.proacl) AS acl + WHERE proc.oid = 'storage.reconcile_lifecycle_state(text, boolean)'::regprocedure + AND acl.grantee NOT IN (0, proc.proowner) + LOOP + EXECUTE format('REVOKE ALL ON FUNCTION storage.reconcile_lifecycle_state(text, boolean) FROM %s', role_name); + END LOOP; + + FOREACH role_name IN ARRAY ARRAY[ + 'postgres', + COALESCE(current_setting('storage.super_user', true), 'postgres'), + COALESCE(current_setting('storage.service_role', true), 'service_role') + ] LOOP + IF EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = role_name) THEN + EXECUTE format('GRANT EXECUTE ON FUNCTION storage.reconcile_lifecycle_state(text, boolean) TO %I', role_name); + END IF; + END LOOP; +END; +$$; + +-- OrioleDB can give AFTER triggers identical OLD/NEW images for a changed row so waking never compares them. +CREATE OR REPLACE FUNCTION storage.wake_lifecycle_state() +RETURNS trigger +LANGUAGE plpgsql +SECURITY DEFINER +SET search_path = pg_catalog +AS $$ +BEGIN + IF TG_OP = 'INSERT' AND NEW.lifecycle_configuration IS NULL THEN + RETURN NEW; + END IF; + + PERFORM storage.reconcile_lifecycle_state(NEW.id, false); + RETURN NEW; +END; +$$; + +REVOKE ALL ON FUNCTION storage.wake_lifecycle_state() FROM PUBLIC; + +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_catalog.pg_trigger + WHERE tgrelid = 'storage.buckets'::regclass AND tgname = 'wake_lifecycle_on_versioning_enable' + ) THEN + CREATE TRIGGER wake_lifecycle_on_versioning_enable + AFTER UPDATE OF versioning_status ON storage.buckets + FOR EACH ROW + WHEN (NEW.lifecycle_configuration IS NOT NULL) + EXECUTE FUNCTION storage.wake_lifecycle_state(); + END IF; + + IF NOT EXISTS ( + SELECT 1 FROM pg_catalog.pg_trigger + WHERE tgrelid = 'storage.buckets'::regclass AND tgname = 'wake_lifecycle_on_policy_write' + ) THEN + CREATE TRIGGER wake_lifecycle_on_policy_write + AFTER INSERT OR UPDATE OF lifecycle_configuration_generation ON storage.buckets + FOR EACH ROW + EXECUTE FUNCTION storage.wake_lifecycle_state(); + END IF; +END; +$$; + +-- Delete metadata and record deleted IDs in the same transaction before +-- deleting backend bytes. Blocked rows keep both their metadata and bytes. +-- Deleting metadata also prevents new foreign-key references during recovery. +CREATE OR REPLACE FUNCTION storage.delete_lifecycle_objects(object_ids uuid[]) +RETURNS TABLE(id uuid, is_delete_marker boolean, metadata jsonb) +LANGUAGE plpgsql +AS $$ +DECLARE + object_id uuid; + deleted_row storage.objects%ROWTYPE; + deleted_ids uuid[]; + deleted_markers boolean[]; + deleted_metadata jsonb[]; + bulk_deleted boolean := false; +BEGIN + -- Surface pre-existing deferred violations instead of treating them as blocked rows. + SET CONSTRAINTS ALL IMMEDIATE; + PERFORM set_config('storage.allow_delete_query', 'true', true); + + -- One subxact keeps the subxid cache from overflowing; rows return only once it commits. + BEGIN + WITH deleted AS ( + DELETE FROM storage.objects AS object_row + WHERE object_row.id = ANY(object_ids) AND object_row.archived_at IS NOT NULL + RETURNING object_row.id, object_row.is_delete_marker, object_row.metadata + ) + SELECT array_agg(deleted.id), array_agg(deleted.is_delete_marker), array_agg(deleted.metadata) + INTO deleted_ids, deleted_markers, deleted_metadata + FROM deleted; + bulk_deleted := true; + EXCEPTION + WHEN OTHERS THEN + -- User triggers may use insufficient_privilege to block a row. + IF SQLSTATE <> '42501' AND left(SQLSTATE, 2) IN ('08', '40', '42', '53', '55', '57', '58', 'XX') THEN + RAISE; + END IF; + END; + IF bulk_deleted THEN + RETURN QUERY SELECT * FROM unnest(deleted_ids, deleted_markers, deleted_metadata); + RETURN; + END IF; + + FOREACH object_id IN ARRAY object_ids LOOP + BEGIN + DELETE FROM storage.objects AS object_row + WHERE object_row.id = object_id AND object_row.archived_at IS NOT NULL + RETURNING object_row.* INTO deleted_row; + IF FOUND THEN + id := deleted_row.id; + is_delete_marker := deleted_row.is_delete_marker; + metadata := deleted_row.metadata; + RETURN NEXT; + END IF; + EXCEPTION + WHEN OTHERS THEN + IF SQLSTATE <> '42501' AND left(SQLSTATE, 2) IN ('08', '40', '42', '53', '55', '57', '58', 'XX') THEN + RAISE; + END IF; + END; + END LOOP; +END; +$$; + +REVOKE ALL ON FUNCTION storage.delete_lifecycle_objects(uuid[]) FROM PUBLIC; + +DO $$ +DECLARE + role_name text; +BEGIN + FOR role_name IN + SELECT DISTINCT acl.grantee::regrole::text + FROM pg_catalog.pg_proc AS proc, pg_catalog.aclexplode(proc.proacl) AS acl + WHERE proc.oid = 'storage.delete_lifecycle_objects(uuid[])'::regprocedure + AND acl.grantee NOT IN (0, proc.proowner) + LOOP + EXECUTE format('REVOKE ALL ON FUNCTION storage.delete_lifecycle_objects(uuid[]) FROM %s', role_name); + END LOOP; + + FOREACH role_name IN ARRAY ARRAY[ + 'postgres', + COALESCE(current_setting('storage.super_user', true), 'postgres'), + COALESCE(current_setting('storage.service_role', true), 'service_role') + ] LOOP + IF EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = role_name) THEN + EXECUTE format('GRANT EXECUTE ON FUNCTION storage.delete_lifecycle_objects(uuid[]) TO %I', role_name); + END IF; + END LOOP; +END; +$$; + +CREATE TABLE IF NOT EXISTS storage.lifecycle_abandoned_artifacts ( + bucket_id text NOT NULL, + key text NOT NULL, + name text NOT NULL, + version text, + attempted_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (bucket_id, key) +); + +DO $$ +BEGIN + IF NOT (SELECT relrowsecurity FROM pg_catalog.pg_class + WHERE oid = 'storage.lifecycle_abandoned_artifacts'::regclass) THEN + ALTER TABLE storage.lifecycle_abandoned_artifacts ENABLE ROW LEVEL SECURITY; + END IF; + IF to_regclass('storage.lifecycle_abandoned_artifacts_attempted_at_idx') IS NULL THEN + CREATE INDEX IF NOT EXISTS lifecycle_abandoned_artifacts_attempted_at_idx + ON storage.lifecycle_abandoned_artifacts (attempted_at); + END IF; +END; +$$; + +DO $$ +DECLARE + anon_role text = COALESCE(current_setting('storage.anon_role', true), 'anon'); + authenticated_role text = COALESCE(current_setting('storage.authenticated_role', true), 'authenticated'); + service_role text = COALESCE(current_setting('storage.service_role', true), 'service_role'); + super_user text = COALESCE(current_setting('storage.super_user', true), 'postgres'); + role_name text; +BEGIN + REVOKE ALL ON TABLE storage.lifecycle_abandoned_artifacts FROM PUBLIC; + + FOREACH role_name IN ARRAY ARRAY[anon_role, authenticated_role] + LOOP + IF EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = role_name) THEN + EXECUTE format('REVOKE ALL ON TABLE storage.lifecycle_abandoned_artifacts FROM %I', role_name); + END IF; + END LOOP; + + FOREACH role_name IN ARRAY ARRAY['postgres', super_user, service_role] + LOOP + IF EXISTS (SELECT 1 FROM pg_catalog.pg_roles WHERE rolname = role_name) THEN + EXECUTE format('GRANT ALL ON TABLE storage.lifecycle_abandoned_artifacts TO %I', role_name); + END IF; + END LOOP; +END; +$$; + +-- Repair missing/dormant state without resetting existing schedules or journals. +SELECT storage.reconcile_lifecycle_state(id, false) +FROM storage.buckets +WHERE lifecycle_configuration IS NOT NULL; diff --git a/src/internal/database/migrations/concurrent-index-guard.test.ts b/src/internal/database/migrations/concurrent-index-guard.test.ts index 250c2098c..60b040b1e 100644 --- a/src/internal/database/migrations/concurrent-index-guard.test.ts +++ b/src/internal/database/migrations/concurrent-index-guard.test.ts @@ -236,6 +236,6 @@ CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_objects_test ON storage.objects (nam } // Migration 65 generates its statement; the runner tests cover its resolved SQL. - expect(query).toHaveBeenCalledTimes(8) + expect(query).toHaveBeenCalledTimes(10) }) }) diff --git a/src/internal/database/migrations/types.ts b/src/internal/database/migrations/types.ts index f21abd43e..cb59341a8 100644 --- a/src/internal/database/migrations/types.ts +++ b/src/internal/database/migrations/types.ts @@ -75,4 +75,8 @@ export const DBMigration = { 'revoke-grants-to-unused-operations': 73, 'unlock-object-versioning': 74, 'enforce-bucket-lifecycle-update-role': 75, + 'noncurrent-lifecycle': 76, + 'noncurrent-lifecycle-archived-order-index': 77, + 'noncurrent-lifecycle-archived-due-index': 78, + 'lifecycle-writer-coordination': 79, } as const diff --git a/src/test/bucket-lifecycle.test.ts b/src/test/bucket-lifecycle.test.ts new file mode 100644 index 000000000..48885f399 --- /dev/null +++ b/src/test/bucket-lifecycle.test.ts @@ -0,0 +1,533 @@ +import { randomUUID } from 'node:crypto' +import type { BucketLifecycleConfiguration } from '@storage/schemas' +import { armedContinuation, newContinuation } from './utils/lifecycle-continuation' +import { useLifecycleStorage } from './utils/lifecycle-versioning' +import { withDeleteEnabled } from './utils/storage' + +describe('bucket lifecycle controls', () => { + const fixtures = useLifecycleStorage('bucket-lifecycle') + const helper = fixtures.helper + let bucketId: string + + beforeEach(async () => { + bucketId = fixtures.bucketId + }) + + const configuration = { + rules: [ + { + id: 'expire-history', + status: 'Enabled' as const, + filter: {}, + noncurrentVersionExpiration: { + noncurrentDays: 30, + newerNoncurrentVersions: 2, + }, + }, + ], + } + + async function writePolicy(writer: 'API' | 'SQL', policy: BucketLifecycleConfiguration | null) { + if (writer === 'API') { + return policy === null + ? helper.database.deleteLifecycleConfiguration(bucketId) + : helper.database.putLifecycleConfiguration(bucketId, policy) + } + await fixtures.withServiceOperation('storage.s3.bucket.put_lifecycle', (transaction) => + transaction.query( + `UPDATE storage.buckets + SET lifecycle_configuration = $2::jsonb, lifecycle_configuration_generation = $3::uuid + WHERE id = $1`, + [ + bucketId, + policy === null ? null : JSON.stringify(policy), + policy === null ? null : randomUUID(), + ] + ) + ) + return helper.database.findLifecycleBucket(bucketId) + } + + it.each([ + { status: 'Enabled', due: true }, + { status: 'Disabled', due: null }, + { status: null, due: null }, + ] as const)( + 'handles a direct SQL bucket INSERT with policy status $status', + async ({ status, due }) => { + await withDeleteEnabled(helper.database.connection, (transaction) => + transaction.query('DELETE FROM storage.buckets WHERE id = $1', [bucketId]) + ) + const policy = + status === null + ? null + : { + rules: [{ ...configuration.rules[0], status }], + } + await fixtures.withServiceOperation('storage.s3.bucket.put_lifecycle', (transaction) => + transaction.query( + `INSERT INTO storage.buckets (id, name, versioning_status, + lifecycle_configuration, lifecycle_configuration_generation) + VALUES ($1, $1, 'ENABLED', $2::jsonb, $3::uuid)`, + [ + bucketId, + policy === null ? null : JSON.stringify(policy), + policy === null ? null : randomUUID(), + ] + ) + ) + + await expect( + fixtures.readState(bucketId, 'next_run_at <= clock_timestamp() AS due') + ).resolves.toEqual(policy === null ? [] : [{ due }]) + } + ) + + it.each(['API', 'SQL'] as const)( + 'keeps state dormant when configured on a versioning-disabled bucket via %s', + async (writer) => { + await writePolicy(writer, configuration) + await expect(fixtures.readState(bucketId, 'next_run_at')).resolves.toEqual([ + { next_run_at: null }, + ]) + } + ) + + it('wakes an existing policy when versioning is enabled without another policy PUT', async () => { + const configured = await helper.database.putLifecycleConfiguration(bucketId, configuration) + await helper.database.updateBucket(bucketId, { versioning_status: 'ENABLED' }) + + await expect( + fixtures.readState(bucketId, 'next_run_at <= clock_timestamp() AS due') + ).resolves.toEqual([{ due: true }]) + const retried = await helper.database.putLifecycleConfiguration(bucketId, configuration) + expect(retried.lifecycle_configuration_generation).toBe( + configured.lifecycle_configuration_generation + ) + await expect( + fixtures.readState(bucketId, 'next_run_at <= clock_timestamp() AS due') + ).resolves.toEqual([{ due: true }]) + }) + + it.each(['DISABLED', 'ENABLED'] as const)( + 'deletes lifecycle state with an empty %s bucket', + async (status) => { + await fixtures.setStatus(bucketId, status) + await helper.database.putLifecycleConfiguration(bucketId, configuration) + + await expect(helper.storage.deleteBucket(bucketId)).resolves.toBe(1) + await expect(fixtures.readState(bucketId, '1')).resolves.toHaveLength(0) + await expect( + helper.database.connection.query('SELECT 1 FROM storage.buckets WHERE id = $1', [bucketId]) + ).resolves.toMatchObject({ rowCount: 0 }) + } + ) + + function armContinuation(generation: string) { + const now = new Date().toISOString() + return armedContinuation( + generation, + [{ name: 'object', version: randomUUID(), isDeleteMarker: false }], + { name: 'object', archivedAt: now }, + { snapshotAt: now, startedAt: now } + ) + } + + it('blocks bucket deletion while an armed attempt is in flight', async () => { + const configured = await helper.database.putLifecycleConfiguration(bucketId, configuration) + const continuation = armContinuation(configured.lifecycle_configuration_generation!) + await helper.database.connection.query( + `UPDATE storage.bucket_lifecycle_states + SET continuation = $2::jsonb, next_run_at = clock_timestamp() + WHERE bucket_id = $1`, + [bucketId, JSON.stringify(continuation)] + ) + + await expect(helper.storage.deleteBucket(bucketId)).rejects.toMatchObject({ + code: 'ResourceReferenced', + }) + await expect(fixtures.readState(bucketId, 'continuation')).resolves.toEqual([{ continuation }]) + }) + + it.each(['API', 'SQL'] as const)( + 'retains and wakes an armed attempt when the lifecycle configuration is deleted via %s', + async (writer) => { + const configured = await helper.database.putLifecycleConfiguration(bucketId, configuration) + const continuation = armContinuation(configured.lifecycle_configuration_generation!) + await helper.database.connection.query( + `UPDATE storage.bucket_lifecycle_states + SET continuation = $2::jsonb, next_run_at = '2099-01-01T00:00:00Z' + WHERE bucket_id = $1`, + [bucketId, JSON.stringify(continuation)] + ) + + const deleted = await writePolicy(writer, null) + expect(deleted).toMatchObject({ lifecycle_configuration: null }) + + const [state] = await fixtures.readState( + bucketId, + 'continuation, next_run_at <= clock_timestamp() AS woken' + ) + expect(state).toEqual({ continuation, woken: true }) + + const retry = await helper.database.deleteLifecycleConfiguration(bucketId) + expect(retry.lifecycle_configuration).toBeNull() + expect(await fixtures.readState(bucketId, '1')).toHaveLength(1) + } + ) + + it.each(['API', 'SQL'] as const)( + 'retains an armed attempt and wakes it when the policy is replaced via %s', + async (writer) => { + const configured = await helper.database.putLifecycleConfiguration(bucketId, configuration) + const continuation = armContinuation(configured.lifecycle_configuration_generation!) + await helper.database.connection.query( + `UPDATE storage.bucket_lifecycle_states + SET continuation = $2::jsonb, claim_id = gen_random_uuid(), claim_until = '2099-01-01', + failure_count = 3, last_error = '"boom"', next_run_at = '2099-01-01' + WHERE bucket_id = $1`, + [bucketId, JSON.stringify(continuation)] + ) + + await writePolicy(writer, { + rules: [{ ...configuration.rules[0], status: 'Disabled' }], + }) + + await expect( + fixtures.readState( + bucketId, + `continuation, claim_id IS NOT NULL AS claimed, claim_until = '2099-01-01' AS leased, + failure_count, last_error, next_run_at <= clock_timestamp() AS woken` + ) + ).resolves.toEqual([ + { + continuation, + claimed: true, + leased: true, + failure_count: 3, + last_error: 'boom', + woken: true, + }, + ]) + } + ) + + it.each( + (['API', 'SQL'] as const).flatMap((writer) => + [ + { + format: 'version 2', + continuation: { continuationVersion: 2, recovery: { pending: true } }, + }, + { format: 'missing version', continuation: {} }, + ].flatMap(({ format, continuation }) => + (['delete', 'disable'] as const).map((action) => ({ writer, format, continuation, action })) + ) + ) + )( + 'preserves $format continuation on $action via $writer', + async ({ writer, continuation, action }) => { + await fixtures.setStatus(bucketId, 'ENABLED') + await helper.database.putLifecycleConfiguration(bucketId, configuration) + const claimId = randomUUID() + await helper.database.connection.query( + `UPDATE storage.bucket_lifecycle_states + SET continuation = $2::jsonb, claim_id = $3::uuid, claim_until = '2099-01-01', + failure_count = 3, last_error = '"boom"', next_run_at = '2099-01-01' + WHERE bucket_id = $1`, + [bucketId, JSON.stringify(continuation), claimId] + ) + + const policy = + action === 'delete' + ? null + : { rules: [{ ...configuration.rules[0], status: 'Disabled' as const }] } + await writePolicy(writer, policy) + await expect( + fixtures.readState( + bucketId, + `continuation, claim_id, claim_until = '2099-01-01' AS leased, + failure_count, last_error, next_run_at <= clock_timestamp() AS woken` + ) + ).resolves.toEqual([ + { + continuation, + claim_id: claimId, + leased: true, + failure_count: 3, + last_error: 'boom', + woken: true, + }, + ]) + } + ) + + it.each(['API', 'SQL'] as const)( + 'clears an idle claimed run and wakes it when the policy is replaced via %s', + async (writer) => { + await fixtures.setStatus(bucketId, 'ENABLED') + const configured = await helper.database.putLifecycleConfiguration(bucketId, configuration) + await helper.database.connection.query( + `UPDATE storage.bucket_lifecycle_states + SET continuation = $2::jsonb, claim_id = gen_random_uuid(), claim_until = '2099-01-01', + failure_count = 3, last_error = '"boom"', next_run_at = '2099-01-01' + WHERE bucket_id = $1`, + [bucketId, JSON.stringify(newContinuation(configured.lifecycle_configuration_generation!))] + ) + + await writePolicy(writer, { + rules: [ + { + ...configuration.rules[0], + noncurrentVersionExpiration: { noncurrentDays: 7 }, + }, + ], + }) + + await expect( + fixtures.readState( + bucketId, + `continuation, claim_id, claim_until, failure_count, last_error, + next_run_at <= clock_timestamp() AS woken` + ) + ).resolves.toEqual([ + { + continuation: null, + claim_id: null, + claim_until: null, + failure_count: 0, + last_error: null, + woken: true, + }, + ]) + } + ) + + it.each(['API', 'SQL'] as const)( + 'removes idle state when a policy is deleted via %s', + async (writer) => { + await helper.database.putLifecycleConfiguration(bucketId, configuration) + await writePolicy(writer, null) + await expect(fixtures.readState(bucketId, '1')).resolves.toEqual([]) + } + ) + + it('preserves healthy state on a SQL update that does not change the generation', async () => { + await fixtures.setStatus(bucketId, 'ENABLED') + await helper.database.putLifecycleConfiguration(bucketId, configuration) + await helper.database.connection.query( + `UPDATE storage.bucket_lifecycle_states SET next_run_at = '2099-01-01', + claim_id = gen_random_uuid(), claim_until = '2099-01-01', failure_count = 3, + last_error = '{"message":"preserved"}' WHERE bucket_id = $1`, + [bucketId] + ) + const before = await fixtures.readState(bucketId, '*') + await fixtures.withServiceOperation('storage.s3.bucket.put_lifecycle', (transaction) => + transaction.query( + 'UPDATE storage.buckets SET lifecycle_configuration_generation = lifecycle_configuration_generation WHERE id = $1', + [bucketId] + ) + ) + expect(await fixtures.readState(bucketId, '*')).toEqual(before) + }) + + it('resets run state when the state row has an older generation', async () => { + await fixtures.setStatus(bucketId, 'ENABLED') + const configured = await helper.database.putLifecycleConfiguration(bucketId, configuration) + await helper.database.connection.query( + `UPDATE storage.bucket_lifecycle_states SET generation = gen_random_uuid(), + next_run_at = '2099-01-01', claim_id = gen_random_uuid(), claim_until = '2099-01-01', + failure_count = 3, last_error = '{"message":"stale"}' WHERE bucket_id = $1`, + [bucketId] + ) + + await helper.database.connection.query('SELECT storage.reconcile_lifecycle_state($1, false)', [ + bucketId, + ]) + await expect( + fixtures.readState( + bucketId, + `generation, claim_id, claim_until, failure_count, last_error, + next_run_at <= clock_timestamp() AS woken` + ) + ).resolves.toEqual([ + { + generation: configured.lifecycle_configuration_generation, + claim_id: null, + claim_until: null, + failure_count: 0, + last_error: null, + woken: true, + }, + ]) + }) + + describe('wake trigger on versioning enable', () => { + const state = () => fixtures.readState(bucketId, 'next_run_at, continuation') + const enable = () => helper.database.updateBucket(bucketId, { versioning_status: 'ENABLED' }) + const setState = (sets: string, values: unknown[] = []) => + helper.database.connection.query( + `UPDATE storage.bucket_lifecycle_states SET ${sets} WHERE bucket_id = $1`, + [bucketId, ...values] + ) + + it('leaves a healthy future schedule untouched', async () => { + await helper.database.putLifecycleConfiguration(bucketId, configuration) + await setState("next_run_at = '2099-01-01T00:00:00Z'") + const before = await state() + + await enable() + expect(await state()).toEqual(before) + }) + + it('leaves a row with a journal in flight untouched', async () => { + const configured = await helper.database.putLifecycleConfiguration(bucketId, configuration) + const continuation = armContinuation(configured.lifecycle_configuration_generation!) + await setState("continuation = $2::jsonb, next_run_at = '2099-01-01T00:00:00Z'", [ + JSON.stringify(continuation), + ]) + const before = await state() + + await enable() + expect(await state()).toEqual(before) + }) + + it.each(['SUSPENDED', 'ENABLED'] as const)( + 'wakes a dormant state when a %s bucket is enabled', + async (status) => { + await fixtures.setStatus(bucketId, status) + await helper.database.putLifecycleConfiguration(bucketId, configuration) + await setState('next_run_at = NULL') + + await enable() + await expect( + fixtures.readState(bucketId, 'next_run_at <= clock_timestamp() AS due') + ).resolves.toEqual([{ due: true }]) + } + ) + + it('keeps a policy with only disabled rules dormant', async () => { + await helper.database.putLifecycleConfiguration(bucketId, { + rules: [{ ...configuration.rules[0], status: 'Disabled' as const }], + }) + + await enable() + await expect(state()).resolves.toMatchObject([{ next_run_at: null }]) + }) + }) + + describe('direct bucket deletion by builds without lifecycle awareness', () => { + const deleteBucketRow = () => + withDeleteEnabled(helper.database.connection, (transaction) => + transaction.query('DELETE FROM storage.buckets WHERE id = $1', [bucketId]) + ) + + it('removes an idle state row with the bucket', async () => { + await helper.database.putLifecycleConfiguration(bucketId, configuration) + + await expect(deleteBucketRow()).resolves.toMatchObject({ rowCount: 1 }) + await expect(fixtures.readState(bucketId, '1')).resolves.toHaveLength(0) + }) + + it('keeps other buckets idle state rows', async () => { + const otherId = await fixtures.createBucket() + await helper.database.putLifecycleConfiguration(bucketId, configuration) + await helper.database.putLifecycleConfiguration(otherId, configuration) + + await expect(deleteBucketRow()).resolves.toMatchObject({ rowCount: 1 }) + await expect(fixtures.readState(bucketId, '1')).resolves.toHaveLength(0) + await expect(fixtures.readState(otherId, '1')).resolves.toHaveLength(1) + }) + + it('keeps idle state when a later trigger skips the delete', async () => { + await helper.database.putLifecycleConfiguration(bucketId, configuration) + const name = `skip_bucket_delete_${randomUUID().replaceAll('-', '')}` + const { connection } = helper.database + await connection.query( + `CREATE FUNCTION public.${name}() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + IF OLD.id = TG_ARGV[0] THEN RETURN NULL; END IF; + RETURN OLD; + END; + $$` + ) + try { + const { rows } = await connection.query<{ literal: string }>( + 'SELECT quote_literal($1::text) AS literal', + [bucketId] + ) + await connection.query( + `CREATE TRIGGER ${name} BEFORE DELETE ON storage.buckets + FOR EACH ROW EXECUTE FUNCTION public.${name}(${rows[0].literal})` + ) + + await expect(deleteBucketRow()).resolves.toMatchObject({ rowCount: 0 }) + await expect(fixtures.readState(bucketId, '1')).resolves.toHaveLength(1) + await expect( + connection.query('SELECT 1 FROM storage.buckets WHERE id = $1', [bucketId]) + ).resolves.toMatchObject({ rowCount: 1 }) + } finally { + await connection.query(`DROP TRIGGER IF EXISTS ${name} ON storage.buckets`) + await connection.query(`DROP FUNCTION IF EXISTS public.${name}()`) + } + }) + + it('blocks deletion and preserves an armed attempt', async () => { + const configured = await helper.database.putLifecycleConfiguration(bucketId, configuration) + const continuation = armContinuation(configured.lifecycle_configuration_generation!) + await helper.database.connection.query( + `UPDATE storage.bucket_lifecycle_states + SET continuation = $2::jsonb, next_run_at = clock_timestamp() WHERE bucket_id = $1`, + [bucketId, JSON.stringify(continuation)] + ) + + await expect(deleteBucketRow()).rejects.toMatchObject({ code: '23503' }) + await expect(fixtures.readState(bucketId, 'continuation')).resolves.toEqual([ + { continuation }, + ]) + await expect( + helper.database.connection.query('SELECT 1 FROM storage.buckets WHERE id = $1', [bucketId]) + ).resolves.toMatchObject({ rowCount: 1 }) + }) + + it.each([{ continuationVersion: 999 }, { continuationVersion: 2, batch: {} }, {}])( + 'preserves unreadable continuation %j', + async (continuation) => { + await helper.database.putLifecycleConfiguration(bucketId, configuration) + await helper.database.connection.query( + 'UPDATE storage.bucket_lifecycle_states SET continuation = $2::jsonb WHERE bucket_id = $1', + [bucketId, JSON.stringify(continuation)] + ) + + await expect(deleteBucketRow()).rejects.toMatchObject({ code: '23503' }) + await expect(fixtures.readState(bucketId, 'continuation')).resolves.toEqual([ + { continuation }, + ]) + await expect( + helper.database.connection.query('SELECT 1 FROM storage.buckets WHERE id = $1', [ + bucketId, + ]) + ).resolves.toMatchObject({ rowCount: 1 }) + } + ) + + it.each([ + { name: 'cursor-only', staged: false }, + { name: 'staged unarmed batch', staged: true }, + ])('removes version 1 state with a $name continuation', async ({ staged }) => { + const configured = await helper.database.putLifecycleConfiguration(bucketId, configuration) + const generation = configured.lifecycle_configuration_generation! + const armed = armContinuation(generation) + const continuation = staged + ? (({ 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', + [bucketId, JSON.stringify(continuation)] + ) + + await expect(deleteBucketRow()).resolves.toMatchObject({ rowCount: 1 }) + await expect(fixtures.readState(bucketId, '1')).resolves.toHaveLength(0) + }) + }) +}) diff --git a/src/test/lifecycle-migration-upgrade.test.ts b/src/test/lifecycle-migration-upgrade.test.ts new file mode 100644 index 000000000..5c2aec998 --- /dev/null +++ b/src/test/lifecycle-migration-upgrade.test.ts @@ -0,0 +1,810 @@ +import { randomUUID } from 'node:crypto' +import { readdir, readFile } from 'node:fs/promises' +import { setTimeout } from 'node:timers/promises' +import { runMigrationsOnTenant } from '@internal/database/migrations' +import { Client } from 'pg' +import { getConfig } from '../config' + +const rerunCases = [ + { + file: '0076-noncurrent-lifecycle.sql', + table: 'bucket_lifecycle_states', + index: 'bucket_lifecycle_states_due_idx', + }, + { + file: '0079-lifecycle-writer-coordination.sql', + table: 'lifecycle_abandoned_artifacts', + index: 'lifecycle_abandoned_artifacts_attempted_at_idx', + }, +] as const + +// Upgrade fixtures use disposable databases but Multigres pooler doesn't support creating them. +describe.skipIf(getConfig().databaseEngine === 'multigres')('lifecycle migration upgrades', () => { + const databaseName = `lifecycle_upgrade_${randomUUID().replaceAll('-', '')}` + const databaseUrl = new URL(getConfig().databaseURL) + databaseUrl.pathname = `/${databaseName}` + const admin = new Client({ connectionString: getConfig().databaseURL }) + let client: Client + const policy = { + rules: [{ status: 'Enabled', filter: {}, noncurrentVersionExpiration: { noncurrentDays: 1 } }], + } + + beforeAll(async () => { + await admin.connect() + }) + + beforeEach(async () => { + await admin.query(`CREATE DATABASE ${databaseName}`) + client = new Client({ connectionString: databaseUrl.toString() }) + await client.connect() + }) + + afterEach(async () => { + await client.end() + await admin.query(`DROP DATABASE IF EXISTS ${databaseName} WITH (FORCE)`) + }) + + afterAll(async () => { + await admin.end() + }) + + it.each(['backfill', 'policy write', 'versioning enable'] as const)( + 'creates state for a legacy long-named bucket during %s', + async (entryPoint) => { + await runMigrationsOnTenant({ + databaseUrl: databaseUrl.toString(), + ...(entryPoint === 'backfill' + ? { upToMigration: 'noncurrent-lifecycle-archived-due-index' as const } + : {}), + }) + const name = 'x'.repeat(120) + const hasPolicy = entryPoint !== 'policy write' + await client.query('BEGIN') + let updatedAt: Date + try { + // Model a bucket retained from before the name-length trigger in migration 37. + await client.query('SET LOCAL session_replication_role = replica') + const inserted = await client.query( + `INSERT INTO storage.buckets (id, name, versioning_status, + lifecycle_configuration, lifecycle_configuration_generation) + VALUES ($1, $1, $2, $3, $4) RETURNING updated_at`, + [ + name, + entryPoint === 'versioning enable' ? 'DISABLED' : 'ENABLED', + hasPolicy ? JSON.stringify(policy) : null, + hasPolicy ? randomUUID() : null, + ] + ) + updatedAt = inserted.rows[0].updated_at + await client.query('COMMIT') + } catch (error) { + await client.query('ROLLBACK') + throw error + } + expect((await client.query('SELECT 1 FROM storage.bucket_lifecycle_states')).rows).toEqual([]) + + if (entryPoint === 'backfill') { + await runMigrationsOnTenant({ databaseUrl: databaseUrl.toString() }) + } else if (entryPoint === 'policy write') { + await client.query('BEGIN') + try { + await client.query("SELECT set_config('role', $1, true)", [getConfig().dbServiceRole]) + await client.query( + `UPDATE storage.buckets SET lifecycle_configuration = $2, + lifecycle_configuration_generation = gen_random_uuid() WHERE id = $1`, + [name, JSON.stringify(policy)] + ) + await client.query('COMMIT') + } catch (error) { + await client.query('ROLLBACK') + throw error + } + } else { + await client.query( + "UPDATE storage.buckets SET versioning_status = 'ENABLED' WHERE id = $1", + [name] + ) + } + + expect( + ( + await client.query( + 'SELECT bucket_id, next_run_at <= clock_timestamp() AS due FROM storage.bucket_lifecycle_states' + ) + ).rows + ).toEqual([{ bucket_id: name, due: true }]) + expect( + (await client.query('SELECT name, updated_at FROM storage.buckets WHERE id = $1', [name])) + .rows + ).toEqual([{ name, updated_at: updatedAt }]) + } + ) + + it('enrolls pre-execution policies and repairs dormant schedules without losing existing work', async () => { + await runMigrationsOnTenant({ + databaseUrl: databaseUrl.toString(), + upToMigration: 'unlock-object-versioning', + }) + for (const [id, status, enabled] of [ + ['enabled', 'ENABLED', true], + ['suspended', 'SUSPENDED', true], + ['disabled', 'DISABLED', true], + ['disabled-rule', 'ENABLED', false], + ['stranded', 'ENABLED', true], + ['claimed', 'ENABLED', true], + ]) { + await client.query( + `INSERT INTO storage.buckets (id, name, versioning_status, + lifecycle_configuration, lifecycle_configuration_generation) + VALUES ($1, $1, $2, $3, gen_random_uuid())`, + [ + id, + status, + JSON.stringify({ + rules: [{ ...policy.rules[0], status: enabled ? 'Enabled' : 'Disabled' }], + }), + ] + ) + } + await client.query("INSERT INTO storage.buckets (id, name) VALUES ('no-policy', 'no-policy')") + await runMigrationsOnTenant({ + databaseUrl: databaseUrl.toString(), + upToMigration: 'noncurrent-lifecycle-archived-due-index', + }) + expect((await client.query('SELECT * FROM storage.bucket_lifecycle_states')).rows).toEqual([]) + await client.query(`INSERT INTO storage.bucket_lifecycle_states ( + bucket_id, + next_run_at, continuation, claim_id, claim_until + ) SELECT id, + CASE WHEN id = 'claimed' THEN '2099-01-01'::timestamptz END, + CASE WHEN id = 'claimed' THEN '{"continuationVersion":999,"preserve":true}'::jsonb END, + CASE WHEN id = 'claimed' THEN gen_random_uuid() END, + CASE WHEN id = 'claimed' THEN '2099-01-01'::timestamptz END + FROM storage.buckets WHERE id IN ('stranded', 'claimed')`) + const before = ( + await client.query( + "SELECT * FROM storage.bucket_lifecycle_states WHERE bucket_id = 'claimed'" + ) + ).rows + await runMigrationsOnTenant({ databaseUrl: databaseUrl.toString() }) + expect( + ( + await client.query(`SELECT bucket_id, next_run_at <= clock_timestamp() AS due + FROM storage.bucket_lifecycle_states ORDER BY bucket_id`) + ).rows + ).toEqual([ + { bucket_id: 'claimed', due: false }, + { bucket_id: 'disabled', due: null }, + { bucket_id: 'disabled-rule', due: null }, + { bucket_id: 'enabled', due: true }, + { bucket_id: 'stranded', due: true }, + { bucket_id: 'suspended', due: true }, + ]) + const read = () => + client.query('SELECT * FROM storage.bucket_lifecycle_states ORDER BY bucket_id') + expect( + ( + await client.query( + "SELECT * FROM storage.bucket_lifecycle_states WHERE bucket_id = 'claimed'" + ) + ).rows + ).toEqual(before) + const first = (await read()).rows + for (const file of (await readdir('migrations/tenant')) + .filter((name) => name >= '0079') + .sort()) { + await client.query(await readFile(`migrations/tenant/${file}`, 'utf8')) + } + expect((await read()).rows).toEqual(first) + await client.query( + "UPDATE storage.buckets SET versioning_status = 'ENABLED' WHERE id = 'disabled'" + ) + expect( + ( + await client.query( + "SELECT next_run_at <= clock_timestamp() AS due FROM storage.bucket_lifecycle_states WHERE bucket_id = 'disabled'" + ) + ).rows + ).toEqual([{ due: true }]) + }) + + it('creates the central registry in public with a shadow storage schema', async () => { + const insertTenant = await prepareCentralRegistry(client) + await insertTenant('existing', true) + await client.query('CREATE SCHEMA storage') + await client.query('CREATE TABLE storage.tenants (id text PRIMARY KEY)') + await client.query('SET search_path = storage, public, extensions') + + await applyEnrollment(client) + await applyEnrollment(client) + await insertTenant('later', true) + + expect( + (await client.query('SELECT tenant_id FROM public.lifecycle_tenants ORDER BY tenant_id')).rows + ).toEqual([{ tenant_id: 'existing' }, { tenant_id: 'later' }]) + expect( + ( + await client.query(`SELECT referenced_namespace.nspname AS schema_name, + referenced_table.relname AS table_name + FROM pg_catalog.pg_constraint AS constraint_row + JOIN pg_catalog.pg_class AS referenced_table ON referenced_table.oid = constraint_row.confrelid + JOIN pg_catalog.pg_namespace AS referenced_namespace ON referenced_namespace.oid = referenced_table.relnamespace + WHERE constraint_row.conrelid = 'public.lifecycle_tenants'::regclass + AND constraint_row.contype = 'f'`) + ).rows + ).toEqual([{ schema_name: 'public', table_name: 'tenants' }]) + expect( + ( + await client.query(`SELECT schemaname, indexname FROM pg_catalog.pg_indexes + WHERE tablename = 'lifecycle_tenants' + AND indexname IN ('lifecycle_tenants_due_idx', 'lifecycle_tenants_expired_claim_idx') + ORDER BY indexname`) + ).rows + ).toEqual([ + { schemaname: 'public', indexname: 'lifecycle_tenants_due_idx' }, + { schemaname: 'public', indexname: 'lifecycle_tenants_expired_claim_idx' }, + ]) + }) + + it('backfills the central registry and enrolls later feature activation and restored tenants', async () => { + const insertTenant = await prepareCentralRegistry(client) + await insertTenant('existing-enabled', true) + await insertTenant('enable-later', false) + await applyEnrollment(client) + expect((await client.query('SELECT tenant_id FROM public.lifecycle_tenants')).rows).toEqual([ + { tenant_id: 'existing-enabled' }, + ]) + await client.query( + "UPDATE public.lifecycle_tenants SET claim_id = gen_random_uuid(), claim_until = '2099-01-01'" + ) + const before = (await client.query('SELECT * FROM public.lifecycle_tenants')).rows + await applyEnrollment(client) + expect((await client.query('SELECT * FROM public.lifecycle_tenants')).rows).toEqual(before) + await client.query( + "UPDATE public.tenants SET feature_object_versioning = true WHERE id = 'enable-later'" + ) + await insertTenant('restored-enabled', true) + await insertTenant('disabled', false) + expect( + ( + await client.query(`SELECT tenant_id, next_dispatch_at <= clock_timestamp() AS due + FROM public.lifecycle_tenants ORDER BY tenant_id`) + ).rows + ).toEqual([ + { tenant_id: 'enable-later', due: true }, + { tenant_id: 'existing-enabled', due: true }, + { tenant_id: 'restored-enabled', due: true }, + ]) + await client.query( + "UPDATE public.lifecycle_tenants SET next_dispatch_at = '2099-01-01' WHERE tenant_id = 'enable-later'" + ) + await client.query( + "UPDATE public.tenants SET migrations_version = migrations_version, feature_object_versioning = feature_object_versioning WHERE id = 'enable-later'" + ) + const dispatchDue = async () => + ( + await client.query( + "SELECT next_dispatch_at <= clock_timestamp() AS due FROM public.lifecycle_tenants WHERE tenant_id = 'enable-later'" + ) + ).rows[0].due + expect(await dispatchDue()).toBe(false) + await client.query( + "UPDATE public.tenants SET migrations_version = 'changed' WHERE id = 'enable-later'" + ) + expect(await dispatchDue()).toBe(true) + }) + + it('enforces registry identity and failure count, indexes dispatch, and cascades tenant deletion', async () => { + const insertTenant = await prepareCentralRegistry(client) + await insertTenant('delete-me', true) + await insertTenant('keep-me', true) + await applyEnrollment(client) + await expect( + client.query("INSERT INTO public.lifecycle_tenants (tenant_id) VALUES ('delete-me')") + ).rejects.toMatchObject({ code: '23505' }) + await expect( + client.query( + "UPDATE public.lifecycle_tenants SET failure_count = -1 WHERE tenant_id = 'delete-me'" + ) + ).rejects.toMatchObject({ code: '23514' }) + await client.query( + "UPDATE public.lifecycle_tenants SET failure_count = 0 WHERE tenant_id = 'delete-me'" + ) + const indexes = await client.query<{ indexname: string; indexdef: string }>( + `SELECT indexname, indexdef FROM pg_catalog.pg_indexes + WHERE schemaname = 'public' AND tablename = 'lifecycle_tenants' + AND indexname IN ('lifecycle_tenants_due_idx', 'lifecycle_tenants_expired_claim_idx') + ORDER BY indexname` + ) + expect(indexes.rows).toEqual([ + { + indexname: 'lifecycle_tenants_due_idx', + indexdef: + 'CREATE INDEX lifecycle_tenants_due_idx ON public.lifecycle_tenants USING btree (next_dispatch_at, tenant_id)', + }, + { + indexname: 'lifecycle_tenants_expired_claim_idx', + indexdef: + 'CREATE INDEX lifecycle_tenants_expired_claim_idx ON public.lifecycle_tenants USING btree (claim_until, tenant_id) WHERE (claim_until IS NOT NULL)', + }, + ]) + await client.query("DELETE FROM public.tenants WHERE id = 'delete-me'") + expect((await client.query('SELECT tenant_id FROM public.lifecycle_tenants')).rows).toEqual([ + { tenant_id: 'keep-me' }, + ]) + }) + + it('migrates without waiting on a writer that holds objects and then needs its bucket', async () => { + await runMigrationsOnTenant({ + databaseUrl: databaseUrl.toString(), + upToMigration: 'noncurrent-lifecycle-archived-due-index', + }) + await client.query( + "INSERT INTO storage.buckets (id, name, versioning_status) VALUES ('busy', 'busy', 'ENABLED')" + ) + await client.query( + `INSERT INTO storage.objects (id, bucket_id, name, version, archived_at, is_versioned) + VALUES ($1, 'busy', 'history', $2, now(), true)`, + [randomUUID(), randomUUID()] + ) + const writer = new Client({ connectionString: databaseUrl.toString() }) + await writer.connect() + try { + await writer.query('BEGIN') + await writer.query("UPDATE storage.objects SET metadata = '{}' WHERE bucket_id = 'busy'") + await expect( + runMigrationsOnTenant({ databaseUrl: databaseUrl.toString() }) + ).resolves.toBeUndefined() + await expect( + writer.query("SELECT 1 FROM storage.buckets WHERE id = 'busy' FOR SHARE") + ).resolves.toMatchObject({ rowCount: 1 }) + await writer.query('COMMIT') + } finally { + await writer.query('ROLLBACK').catch(() => undefined) + await writer.end() + } + }) + + it('does not deadlock with a writer that locks a policy bucket and then updates it', async () => { + await runMigrationsOnTenant({ + databaseUrl: databaseUrl.toString(), + upToMigration: 'noncurrent-lifecycle-archived-due-index', + }) + await client.query( + `INSERT INTO storage.buckets (id, name, versioning_status, + lifecycle_configuration, lifecycle_configuration_generation) + VALUES ('locked', 'locked', 'ENABLED', $1, gen_random_uuid())`, + [JSON.stringify(policy)] + ) + const writer = new Client({ connectionString: databaseUrl.toString() }) + await writer.connect() + try { + await writer.query('BEGIN') + await writer.query("SELECT 1 FROM storage.buckets WHERE id = 'locked' FOR UPDATE") + const migration = runMigrationsOnTenant({ databaseUrl: databaseUrl.toString() }) + migration.catch(() => undefined) + await waitUntilRows( + client, + "SELECT 1 FROM pg_stat_activity WHERE datname = current_database() AND wait_event_type = 'Lock'" + ) + await expect( + writer.query("UPDATE storage.buckets SET public = public WHERE id = 'locked'") + ).resolves.toMatchObject({ rowCount: 1 }) + await writer.query('COMMIT') + await expect(migration).resolves.toBeUndefined() + } finally { + await writer.query('ROLLBACK').catch(() => undefined) + await writer.end() + } + }) + + it('retries the bucket lock and lets queued writers proceed between attempts', async () => { + await runMigrationsOnTenant({ + databaseUrl: databaseUrl.toString(), + upToMigration: 'noncurrent-lifecycle-archived-due-index', + }) + await client.query("INSERT INTO storage.buckets (id, name) VALUES ('held', 'held')") + const writer = new Client({ connectionString: databaseUrl.toString() }) + const queuedWriter = new Client({ connectionString: databaseUrl.toString() }) + await Promise.all([writer.connect(), queuedWriter.connect()]) + try { + await writer.query('BEGIN') + await writer.query("INSERT INTO storage.objects (bucket_id, name) VALUES ('held', 'open')") + let migrationFinished = false + const migration = runMigrationsOnTenant({ databaseUrl: databaseUrl.toString() }).finally( + () => { + migrationFinished = true + } + ) + migration.catch(() => undefined) + await waitUntilRows( + client, + `SELECT 1 FROM pg_catalog.pg_locks + WHERE database = (SELECT oid FROM pg_catalog.pg_database WHERE datname = current_database()) + AND relation = 'storage.buckets'::regclass AND mode = 'ExclusiveLock' AND NOT granted` + ) + const queuedAt = performance.now() + await expect( + queuedWriter.query( + "INSERT INTO storage.objects (bucket_id, name) VALUES ('held', 'queued')" + ) + ).resolves.toMatchObject({ rowCount: 1 }) + expect(performance.now() - queuedAt).toBeLessThan(2_000) + expect(migrationFinished).toBe(false) + await writer.query('ROLLBACK') + await expect(migration).resolves.toBeUndefined() + } finally { + await writer.query('ROLLBACK').catch(() => undefined) + await Promise.all([writer.end(), queuedWriter.end()]) + } + }, 20_000) + + it('propagates statement cancellation while waiting for the bucket lock', async () => { + await runMigrationsOnTenant({ + databaseUrl: databaseUrl.toString(), + upToMigration: 'noncurrent-lifecycle-archived-due-index', + }) + await client.query("INSERT INTO storage.buckets (id, name) VALUES ('held', 'held')") + const writer = new Client({ connectionString: databaseUrl.toString() }) + const migration = new Client({ connectionString: databaseUrl.toString() }) + await Promise.all([writer.connect(), migration.connect()]) + try { + await writer.query('BEGIN') + await writer.query("INSERT INTO storage.objects (bucket_id, name) VALUES ('held', 'open')") + await migration.query('BEGIN') + await migration.query("SET LOCAL statement_timeout = '250ms'") + await expect( + migration.query( + await readFile('migrations/tenant/0079-lifecycle-writer-coordination.sql', 'utf8') + ) + ).rejects.toMatchObject({ code: '57014' }) + await migration.query('ROLLBACK') + await writer.query('ROLLBACK') + await expect( + runMigrationsOnTenant({ databaseUrl: databaseUrl.toString() }) + ).resolves.toBeUndefined() + } finally { + await migration.query('ROLLBACK').catch(() => undefined) + await writer.query('ROLLBACK').catch(() => undefined) + await Promise.all([migration.end(), writer.end()]) + } + }) + + it.each([ + { lockTimeout: '1s', statementTimeout: '1800ms' }, + { lockTimeout: '50ms', statementTimeout: '200ms' }, + ])( + 'gives up within the caller lock_timeout of $lockTimeout', + async ({ lockTimeout, statementTimeout }) => { + await runMigrationsOnTenant({ + databaseUrl: databaseUrl.toString(), + upToMigration: 'noncurrent-lifecycle-archived-due-index', + }) + await client.query("INSERT INTO storage.buckets (id, name) VALUES ('held', 'held')") + const writer = new Client({ connectionString: databaseUrl.toString() }) + const migration = new Client({ connectionString: databaseUrl.toString() }) + await Promise.all([writer.connect(), migration.connect()]) + try { + await writer.query('BEGIN') + await writer.query("INSERT INTO storage.objects (bucket_id, name) VALUES ('held', 'open')") + await migration.query('BEGIN') + // A separate statement deadline catches either an unclamped retry sleep or + // a lock attempt that ignores a caller timeout shorter than 250ms. + await migration.query( + "SELECT set_config('lock_timeout', $1, true), set_config('statement_timeout', $2, true)", + [lockTimeout, statementTimeout] + ) + await expect( + migration.query( + await readFile('migrations/tenant/0079-lifecycle-writer-coordination.sql', 'utf8') + ) + ).rejects.toMatchObject({ code: '55P03' }) + } finally { + await migration.query('ROLLBACK').catch(() => undefined) + await writer.query('ROLLBACK').catch(() => undefined) + await Promise.all([migration.end(), writer.end()]) + } + } + ) + + it('reruns while a bucket reader holds its table lock, without upgrading the migration lock', async () => { + await runMigrationsOnTenant({ databaseUrl: databaseUrl.toString() }) + await client.query("INSERT INTO storage.buckets (id, name) VALUES ('busy', 'busy')") + const writer = new Client({ connectionString: databaseUrl.toString() }) + const migration = new Client({ connectionString: databaseUrl.toString() }) + await writer.connect() + await migration.connect() + try { + await writer.query('BEGIN') + const writerPid = (await writer.query<{ pid: number }>('SELECT pg_backend_pid() AS pid')) + .rows[0].pid + await writer.query("SELECT 1 FROM storage.buckets WHERE id = 'busy'") + await migration.query('BEGIN') + await migration.query("SET LOCAL lock_timeout = '2s'") + await migration.query( + await readFile('migrations/tenant/0079-lifecycle-writer-coordination.sql', 'utf8') + ) + expect((await migration.query('SHOW lock_timeout')).rows[0].lock_timeout).toBe('2s') + + // The migration retains its EXCLUSIVE lock until commit. A row-locking + // request waits for it but cannot form a cycle with a trigger drop. + const rowLock = writer.query("SELECT 1 FROM storage.buckets WHERE id = 'busy' FOR UPDATE") + rowLock.catch(() => undefined) + await waitUntilRows( + client, + `SELECT 1 FROM pg_catalog.pg_locks + WHERE pid = ${writerPid} AND relation = 'storage.buckets'::regclass + AND mode = 'RowShareLock' AND NOT granted` + ) + await migration.query('COMMIT') + await expect(rowLock).resolves.toMatchObject({ rowCount: 1 }) + await writer.query('COMMIT') + } finally { + await migration.query('ROLLBACK').catch(() => undefined) + await writer.query('ROLLBACK').catch(() => undefined) + await migration.end() + await writer.end() + } + }) + + it.each(rerunCases)( + 'reruns $file without blocking a lifecycle table reader and writer', + async ({ file, table }) => { + await runMigrationsOnTenant({ databaseUrl: databaseUrl.toString() }) + const worker = new Client({ connectionString: databaseUrl.toString() }) + const migration = new Client({ connectionString: databaseUrl.toString() }) + await worker.connect() + await migration.connect() + try { + await worker.query('BEGIN') + await worker.query(`SELECT 1 FROM storage.${table} LIMIT 1`) + await migration.query('BEGIN') + await migration.query("SET LOCAL lock_timeout = '2s'") + await migration.query(await readFile(`migrations/tenant/${file}`, 'utf8')) + await worker.query("SET LOCAL lock_timeout = '2s'") + await worker.query(`DELETE FROM storage.${table} WHERE false`) + await worker.query('COMMIT') + await migration.query('COMMIT') + } finally { + await worker.query('ROLLBACK').catch(() => undefined) + await migration.query('ROLLBACK').catch(() => undefined) + await worker.end() + await migration.end() + } + } + ) + + it.each(rerunCases)('keeps or restores index/RLS in $file', async ({ file, table, index }) => { + await runMigrationsOnTenant({ databaseUrl: databaseUrl.toString() }) + const sql = await readFile(`migrations/tenant/${file}`, 'utf8') + const state = async () => + ( + await client.query<{ index_oid: number | null; rls: boolean }>( + `SELECT to_regclass($1)::oid AS index_oid, relrowsecurity AS rls + FROM pg_catalog.pg_class WHERE oid = $2::regclass`, + [`storage.${index}`, `storage.${table}`] + ) + ).rows[0] + const original = await state() + expect(original.index_oid).not.toBeNull() + expect(original.rls).toBe(true) + await client.query(sql) + expect(await state()).toEqual(original) + await client.query(`DROP INDEX storage.${index}`) + await client.query(`ALTER TABLE storage.${table} DISABLE ROW LEVEL SECURITY`) + await client.query(sql) + expect(await state()).toMatchObject({ index_oid: expect.any(Number), rls: true }) + }) + + it.each(rerunCases)('takes RLS lock before index in $file', async ({ file, table, index }) => { + await runMigrationsOnTenant({ databaseUrl: databaseUrl.toString() }) + await client.query(`DROP INDEX storage.${index}`) + await client.query(`ALTER TABLE storage.${table} DISABLE ROW LEVEL SECURITY`) + const reader = new Client({ connectionString: databaseUrl.toString() }) + const migration = new Client({ connectionString: databaseUrl.toString() }) + await Promise.all([reader.connect(), migration.connect()]) + try { + await reader.query('BEGIN') + await reader.query(`SELECT 1 FROM storage.${table} LIMIT 1`) + await migration.query('BEGIN') + await migration.query("SET LOCAL lock_timeout = '5s'") + const pid = (await migration.query<{ pid: number }>('SELECT pg_backend_pid() AS pid')).rows[0] + .pid + const replay = migration.query(await readFile(`migrations/tenant/${file}`, 'utf8')) + replay.catch(() => undefined) + await waitUntilRows( + client, + `SELECT 1 FROM pg_catalog.pg_locks + WHERE pid = ${pid} AND relation = 'storage.${table}'::regclass + AND mode = 'AccessExclusiveLock' AND NOT granted` + ) + expect( + ( + await client.query( + `SELECT 1 FROM pg_catalog.pg_locks + WHERE pid = $1 AND relation = $2::regclass AND mode = 'ShareLock' AND granted`, + [pid, `storage.${table}`] + ) + ).rowCount + ).toBe(0) + await reader.query('COMMIT') + await replay + await migration.query('COMMIT') + } finally { + await reader.query('ROLLBACK').catch(() => undefined) + await migration.query('ROLLBACK').catch(() => undefined) + await Promise.all([reader.end(), migration.end()]) + } + }) + + it('preserves existing lifecycle triggers and recreates each missing one', async () => { + await runMigrationsOnTenant({ databaseUrl: databaseUrl.toString() }) + const sql = await readFile('migrations/tenant/0079-lifecycle-writer-coordination.sql', 'utf8') + const names = [ + 'protect_active_lifecycle_states', + 'wake_lifecycle_on_versioning_enable', + 'wake_lifecycle_on_policy_write', + ] + const triggers = async () => + ( + await client.query<{ oid: number; tgname: string; definition: string }>( + `SELECT oid, tgname, pg_get_triggerdef(oid) AS definition + FROM pg_catalog.pg_trigger + WHERE tgrelid = 'storage.buckets'::regclass AND tgname = ANY($1::text[]) + ORDER BY tgname`, + [names] + ) + ).rows + const original = await triggers() + expect(original).toHaveLength(names.length) + await client.query(sql) + expect(await triggers()).toEqual(original) + + for (const name of names) { + await client.query(`DROP TRIGGER ${name} ON storage.buckets`) + const remaining = await triggers() + await client.query(sql) + const restored = await triggers() + expect(restored.filter((trigger) => trigger.tgname !== name)).toEqual(remaining) + expect(restored.find((trigger) => trigger.tgname === name)?.definition).toEqual( + original.find((trigger) => trigger.tgname === name)?.definition + ) + } + }) + + it('revokes execute from roles no longer configured when upgrading', async () => { + const role = `lifecycle_former_${randomUUID().replaceAll('-', '')}` + await runMigrationsOnTenant({ + databaseUrl: databaseUrl.toString(), + upToMigration: 'noncurrent-lifecycle-archived-due-index', + }) + await admin.query(`CREATE ROLE ${role} NOLOGIN`) + try { + await client.query( + `ALTER DEFAULT PRIVILEGES IN SCHEMA storage GRANT EXECUTE ON FUNCTIONS TO ${role}` + ) + await runMigrationsOnTenant({ databaseUrl: databaseUrl.toString() }) + + const { rows } = await client.query( + `SELECT + has_function_privilege($1, 'storage.reconcile_lifecycle_state(text,boolean)', 'EXECUTE') AS former_reconcile, + has_function_privilege($1, 'storage.delete_lifecycle_objects(uuid[])', 'EXECUTE') AS former_delete, + has_function_privilege($2, 'storage.reconcile_lifecycle_state(text,boolean)', 'EXECUTE') AS service_reconcile`, + [role, getConfig().dbServiceRole] + ) + expect(rows[0]).toEqual({ + former_reconcile: false, + former_delete: false, + service_reconcile: true, + }) + } finally { + await client.query(`DROP OWNED BY ${role}`) + await admin.query(`DROP ROLE ${role}`) + } + }) + + it('wakes flagged tenants when their migrations_version changes, without postponing or enrolling others', async () => { + const insertTenant = await prepareCentralRegistry(client) + await insertTenant('late', true) + await insertTenant('early', true) + await insertTenant('unflagged', false) + await applyEnrollment(client) + await setNextDispatch(client, 'late', FUTURE) + await setNextDispatch(client, 'early', PAST) + + await client.query( + "UPDATE public.tenants SET migrations_version = 'lifecycle-writer-coordination'" + ) + + const state = await nextDispatches(client) + expect([...state.keys()]).toEqual(['early', 'late']) + expect(state.get('late')).toMatchObject({ due: true }) + expect(state.get('early')).toEqual({ at: Date.parse(PAST), due: true }) + }) + + it('wakes with LEAST on feature activation and backfill re-runs', async () => { + const insertTenant = await prepareCentralRegistry(client) + await applyEnrollment(client) + for (const id of ['flip-future', 'flip-past']) { + await insertTenant(id, false) + } + for (const id of ['rerun-future', 'rerun-past']) { + await insertTenant(id, true) + } + await setNextDispatch(client, 'flip-future', FUTURE) + await setNextDispatch(client, 'flip-past', PAST) + await setNextDispatch(client, 'rerun-future', FUTURE) + await setNextDispatch(client, 'rerun-past', PAST) + + await applyEnrollment(client) + + const rerun = await nextDispatches(client) + expect(rerun.get('rerun-future')).toMatchObject({ due: true }) + expect(rerun.get('rerun-past')).toEqual({ at: Date.parse(PAST), due: true }) + expect(rerun.get('flip-future')).toEqual({ at: Date.parse(FUTURE), due: false }) + + await client.query( + "UPDATE public.tenants SET feature_object_versioning = true WHERE id LIKE 'flip-%'" + ) + + const flipped = await nextDispatches(client) + expect(flipped.get('flip-future')).toMatchObject({ due: true }) + expect(flipped.get('flip-past')).toEqual({ at: Date.parse(PAST), due: true }) + }) +}) + +const FUTURE = '2099-01-01T00:00:00Z' +const PAST = '2000-01-01T00:00:00Z' + +async function waitUntilRows(client: Client, sql: string) { + const timeoutMs = 3_000 + const deadline = performance.now() + timeoutMs + while (!(await client.query(sql)).rowCount) { + if (performance.now() >= deadline) { + throw new Error(`Timed out after ${timeoutMs}ms waiting for rows: ${sql}`) + } + await setTimeout(20) + } +} + +async function prepareCentralRegistry(client: Client) { + const directory = 'migrations/multitenant' + await client.query("SET storage.iceberg_shards = '{}'") + for (const file of (await readdir(directory)) + .filter((name) => name.endsWith('.sql') && name < '0031') + .sort()) { + await client.query(await readFile(`${directory}/${file}`, 'utf8')) + } + return (id: string, enabled: boolean) => + client.query( + `INSERT INTO public.tenants (id, anon_key, database_url, jwt_secret, service_key, feature_object_versioning) + VALUES ($1, 'fixture', 'fixture', 'fixture', 'fixture', $2)`, + [id, enabled] + ) +} + +async function applyEnrollment(client: Client) { + await client.query(await readFile('migrations/multitenant/0031-lifecycle-tenants.sql', 'utf8')) +} + +function setNextDispatch(client: Client, tenantId: string, at: string) { + return client.query( + `INSERT INTO public.lifecycle_tenants (tenant_id, next_dispatch_at) VALUES ($1, $2) + ON CONFLICT (tenant_id) DO UPDATE SET next_dispatch_at = EXCLUDED.next_dispatch_at`, + [tenantId, at] + ) +} + +async function nextDispatches(client: Client) { + const result = await client.query<{ + tenant_id: string + next_dispatch_at: Date + due: boolean + }>( + `SELECT tenant_id, next_dispatch_at, next_dispatch_at <= clock_timestamp() AS due + FROM public.lifecycle_tenants ORDER BY tenant_id` + ) + return new Map( + result.rows.map((row) => [row.tenant_id, { at: row.next_dispatch_at.getTime(), due: row.due }]) + ) +} diff --git a/src/test/noncurrent-lifecycle-schema.test.ts b/src/test/noncurrent-lifecycle-schema.test.ts new file mode 100644 index 000000000..873c43830 --- /dev/null +++ b/src/test/noncurrent-lifecycle-schema.test.ts @@ -0,0 +1,724 @@ +import { randomUUID } from 'node:crypto' +import { buildEvaluateNoncurrentLifecyclePageStatement } from '@storage/database/lifecycle' +import { Client } from 'pg' +import { getConfig } from '../config' +import { useLifecycleStorage } from './utils/lifecycle-versioning' + +function findPlanNode( + value: unknown, + predicate: (node: Record) => boolean +): Record | undefined { + if (value === null || typeof value !== 'object' || Array.isArray(value)) return undefined + const node = value as Record + if (predicate(node)) return node + if (!Array.isArray(node.Plans)) return undefined + for (const child of node.Plans) { + const match = findPlanNode(child, predicate) + if (match) return match + } + return undefined +} + +async function withRollback(fn: (client: Client) => Promise): Promise { + const client = new Client({ connectionString: getConfig().databaseURL }) + await client.connect() + try { + await client.query('BEGIN') + await fn(client) + } finally { + await client.query('ROLLBACK').catch(() => undefined) + await client.end() + } +} + +describe('noncurrent lifecycle schema foundation', () => { + const fixtures = useLifecycleStorage('lifecycle-schema') + const tHelper = fixtures.helper + let bucketId: string + + beforeEach(() => { + bucketId = fixtures.bucketId + }) + + it('installs inert bucket defaults and the archived-history indexes', async () => { + const bucket = await tHelper.database.connection.query<{ + versioning_status: string + lifecycle_configuration: unknown | null + lifecycle_configuration_generation: string | null + }>( + `SELECT versioning_status, + lifecycle_configuration, + lifecycle_configuration_generation + FROM storage.buckets + WHERE id = $1`, + [bucketId] + ) + + expect(bucket.rows[0]).toMatchObject({ + versioning_status: 'DISABLED', + lifecycle_configuration: null, + lifecycle_configuration_generation: null, + }) + + const indexes = await tHelper.database.connection.query<{ + indexname: string + indexdef: string + indisvalid: boolean + }>( + ` + SELECT indexes.indexname, indexes.indexdef, pg_index.indisvalid + FROM pg_catalog.pg_indexes AS indexes + JOIN pg_catalog.pg_class AS index_relation + ON index_relation.relname = indexes.indexname + AND index_relation.relnamespace = 'storage'::regnamespace + JOIN pg_catalog.pg_index + ON pg_index.indexrelid = index_relation.oid + WHERE indexes.schemaname = 'storage' + AND indexes.tablename = 'objects' + AND indexes.indexname = ANY($1::text[]) + ORDER BY indexes.indexname + `, + [['objects_archived_due_idx', 'objects_archived_order_uq']] + ) + + expect(indexes.rows.map((row) => row.indexname)).toEqual([ + 'objects_archived_due_idx', + 'objects_archived_order_uq', + ]) + expect(indexes.rows.every((row) => row.indisvalid)).toBe(true) + expect( + indexes.rows.find((row) => row.indexname === 'objects_archived_order_uq')?.indexdef + ).toContain('CREATE UNIQUE INDEX objects_archived_order_uq') + expect( + indexes.rows.find((row) => row.indexname === 'objects_archived_order_uq')?.indexdef + ).toContain('(bucket_id, name COLLATE "C", archived_at) WHERE') + + expect( + indexes.rows.find((row) => row.indexname === 'objects_archived_due_idx')?.indexdef + ).toContain( + '(bucket_id, archived_at, name COLLATE "C") INCLUDE (id, version, is_delete_marker)' + ) + }) + + it('uses the archived due index for explicit history fixtures', async () => { + const objectName = 'history/object.txt' + const version = randomUUID() + const rowId = randomUUID() + + const transaction = await tHelper.database.connection.transaction() + try { + await transaction.query( + `INSERT INTO storage.objects (id, bucket_id, name, version) + VALUES ($1, $2, $3, $4)`, + [rowId, bucketId, objectName, version] + ) + await transaction.query( + `UPDATE storage.objects + SET archived_at = clock_timestamp() + WHERE bucket_id = $1 AND name COLLATE "C" = $2 AND version = $3`, + [bucketId, objectName, version] + ) + // Keep most archived rows outside the due range. + await transaction.query( + `INSERT INTO storage.objects (bucket_id, name, version, archived_at) + SELECT $1, 'future/' || n, gen_random_uuid()::text, '2099-01-01'::timestamptz + FROM generate_series(1, 1000) AS n`, + [bucketId] + ) + await transaction.query('ANALYZE storage.objects') + await transaction.query('SET LOCAL enable_seqscan = off') + await transaction.query('SET LOCAL enable_bitmapscan = off') + // Check that the index supplies the ordering. A one-row sort can otherwise + // be cheaper after other tests change the shared table's index statistics. + await transaction.query('SET LOCAL enable_sort = off') + + const evaluation = buildEvaluateNoncurrentLifecyclePageStatement({ + bucketId, + snapshotAt: '2030-01-01T00:00:00.000Z', + rules: [ + { + cutoffAt: '2030-01-01T00:00:00.000Z', + newerNoncurrentVersions: 1, + }, + ], + pageSize: 500, + }) + const plan = await transaction.query<{ 'QUERY PLAN': unknown }>({ + text: `EXPLAIN (FORMAT JSON) ${evaluation.text}`, + values: evaluation.values, + }) + const explain = plan.rows[0]?.['QUERY PLAN'] + const root = Array.isArray(explain) + ? (explain[0] as { Plan?: unknown } | undefined)?.Plan + : undefined + const rawPage = findPlanNode(root, (node) => node['Subplan Name'] === 'CTE raw_page') + expect(rawPage).toBeDefined() + const indexScan = findPlanNode( + rawPage, + (node) => node['Index Name'] === 'objects_archived_due_idx' + ) + expect(indexScan, JSON.stringify(rawPage, null, 2)).toBeDefined() + expect(['Index Only Scan', 'Custom Scan'], JSON.stringify(rawPage, null, 2)).toContain( + indexScan?.['Node Type'] + ) + await transaction.commit() + } catch (error) { + await transaction.rollback() + throw error + } + }) + + it('enforces lifecycle state constraints and indexes', async () => { + await withRollback(async (client) => { + await client.query('INSERT INTO storage.bucket_lifecycle_states (bucket_id) VALUES ($1)', [ + bucketId, + ]) + for (const [sql, code, constraint] of [ + ['INSERT INTO storage.bucket_lifecycle_states (bucket_id) VALUES ($1)', '23505', undefined], + [ + 'UPDATE storage.bucket_lifecycle_states SET failure_count = -1 WHERE bucket_id = $1', + '23514', + 'bucket_lifecycle_failure_count_check', + ], + [ + "UPDATE storage.bucket_lifecycle_states SET continuation = '[]' WHERE bucket_id = $1", + '23514', + 'bucket_lifecycle_continuation_object_check', + ], + [ + "UPDATE storage.bucket_lifecycle_states SET last_result = '[]' WHERE bucket_id = $1", + '23514', + 'bucket_lifecycle_last_result_object_check', + ], + [ + 'UPDATE storage.bucket_lifecycle_states SET continuation = \'{"batch":{"inFlight":{}}}\', next_run_at = NULL WHERE bucket_id = $1', + '23514', + 'bucket_lifecycle_recovery_due_check', + ], + ] as const) { + await client.query('SAVEPOINT invalid_state') + await expect(client.query(sql, [bucketId])).rejects.toMatchObject({ + code, + ...(constraint && { constraint }), + }) + await client.query('ROLLBACK TO SAVEPOINT invalid_state') + } + await client.query( + `UPDATE storage.bucket_lifecycle_states + SET failure_count = 0, continuation = '{"batch":{"inFlight":{}}}', + next_run_at = now(), last_result = '{}' WHERE bucket_id = $1`, + [bucketId] + ) + await client.query( + `INSERT INTO storage.lifecycle_abandoned_artifacts (bucket_id, key, name) + VALUES ($1, 'artifact', 'name')`, + [bucketId] + ) + await client.query( + `INSERT INTO storage.lifecycle_abandoned_artifacts (bucket_id, key, name) + VALUES ($1, 'other-artifact', 'name'), ($2, 'artifact', 'name')`, + [bucketId, `${bucketId}-other`] + ) + await client.query('SAVEPOINT duplicate_artifact') + await expect( + client.query( + `INSERT INTO storage.lifecycle_abandoned_artifacts (bucket_id, key, name) + VALUES ($1, 'artifact', 'other')`, + [bucketId] + ) + ).rejects.toMatchObject({ code: '23505' }) + await client.query('ROLLBACK TO SAVEPOINT duplicate_artifact') + const indexes = await client.query<{ indexname: string; indexdef: string }>( + `SELECT indexname, indexdef FROM pg_catalog.pg_indexes + WHERE schemaname = 'storage' AND indexname IN + ('bucket_lifecycle_states_due_idx', 'lifecycle_abandoned_artifacts_attempted_at_idx') + ORDER BY indexname` + ) + expect(indexes.rows).toEqual([ + { + indexname: 'bucket_lifecycle_states_due_idx', + indexdef: + 'CREATE INDEX bucket_lifecycle_states_due_idx ON storage.bucket_lifecycle_states USING btree (next_run_at, bucket_id) WHERE (next_run_at IS NOT NULL)', + }, + { + indexname: 'lifecycle_abandoned_artifacts_attempted_at_idx', + indexdef: + 'CREATE INDEX lifecycle_abandoned_artifacts_attempted_at_idx ON storage.lifecycle_abandoned_artifacts USING btree (attempted_at)', + }, + ]) + }) + }) + + it('keeps bucket state service-owned', async () => { + await tHelper.database.putLifecycleConfiguration(bucketId, { + rules: [ + { + id: 'expire', + status: 'Enabled', + filter: {}, + noncurrentVersionExpiration: { noncurrentDays: 30 }, + }, + ], + }) + await expect(fixtures.readState(bucketId, '1')).resolves.toHaveLength(1) + + const security = await tHelper.database.connection.query<{ + relrowsecurity: boolean + anon_select: boolean + authenticated_select: boolean + service_select: boolean + }>(` + SELECT relation.relrowsecurity, + has_table_privilege('anon', 'storage.bucket_lifecycle_states', 'SELECT') AS anon_select, + has_table_privilege('authenticated', 'storage.bucket_lifecycle_states', 'SELECT') AS authenticated_select, + has_table_privilege('service_role', 'storage.bucket_lifecycle_states', 'SELECT') AS service_select + FROM pg_catalog.pg_class AS relation + WHERE relation.oid = 'storage.bucket_lifecycle_states'::regclass + `) + + expect(security.rows[0]).toEqual({ + relrowsecurity: true, + anon_select: false, + authenticated_select: false, + service_select: true, + }) + const permissions = await tHelper.database.connection.query(` + SELECT has_function_privilege('anon', 'storage.reconcile_lifecycle_state(text,boolean)', 'EXECUTE') AS anon_execute, + has_function_privilege('authenticated', 'storage.reconcile_lifecycle_state(text,boolean)', 'EXECUTE') AS authenticated_execute, + has_function_privilege('service_role', 'storage.reconcile_lifecycle_state(text,boolean)', 'EXECUTE') AS service_execute + `) + expect(permissions.rows).toEqual([ + { anon_execute: false, authenticated_execute: false, service_execute: true }, + ]) + const probePermissions = await tHelper.database.connection.query(` + SELECT has_function_privilege('anon', 'storage.delete_lifecycle_objects(uuid[])', 'EXECUTE') AS anon_execute, + has_function_privilege('authenticated', 'storage.delete_lifecycle_objects(uuid[])', 'EXECUTE') AS authenticated_execute, + has_function_privilege('service_role', 'storage.delete_lifecycle_objects(uuid[])', 'EXECUTE') AS service_execute + `) + expect(probePermissions.rows).toEqual(permissions.rows) + }) + + it('keeps abandoned artifacts service-owned', async () => { + const security = await tHelper.database.connection.query(` + SELECT relation.relrowsecurity, + has_table_privilege('anon', 'storage.lifecycle_abandoned_artifacts', 'SELECT') AS anon_select, + has_table_privilege('authenticated', 'storage.lifecycle_abandoned_artifacts', 'SELECT') AS authenticated_select, + has_table_privilege('service_role', 'storage.lifecycle_abandoned_artifacts', 'SELECT') AS service_select + FROM pg_catalog.pg_class AS relation + WHERE relation.oid = 'storage.lifecycle_abandoned_artifacts'::regclass + `) + + expect(security.rows).toEqual([ + { + relrowsecurity: true, + anon_select: false, + authenticated_select: false, + service_select: true, + }, + ]) + }) + + it.each(['unset', 'false'] as const)( + 'deletes lifecycle objects when the delete guard setting is %s', + async (setting) => { + await withRollback(async (client) => { + if (setting === 'false') { + await client.query("SELECT set_config('storage.allow_delete_query', 'false', true)") + } + const id = ( + await client.query<{ id: string }>( + `INSERT INTO storage.objects (bucket_id, name, version, archived_at, is_versioned) + VALUES ($1, 'guard/object', gen_random_uuid()::text, now(), true) RETURNING id`, + [bucketId] + ) + ).rows[0].id + const deleted = await client.query<{ id: string }>( + 'SELECT * FROM storage.delete_lifecycle_objects($1::uuid[])', + [[id]] + ) + expect(deleted.rows.map((row) => row.id)).toEqual([id]) + expect( + (await client.query('SELECT 1 FROM storage.objects WHERE id = $1', [id])).rowCount + ).toBe(0) + expect( + ( + await client.query( + "SELECT current_setting('storage.allow_delete_query', true) AS value" + ) + ).rows + ).toEqual([{ value: 'true' }]) + }) + } + ) + + it.each([ + { code: '42501', vetoAll: false }, + { code: '42501', vetoAll: true }, + ])( + 'keeps rows vetoed with $code (all: $vetoAll) while deleting other archived rows', + async ({ code, vetoAll }) => { + await withRollback(async (client) => { + await client.query("SELECT set_config('storage.allow_delete_query', 'false', true)") + const ids = ( + await client.query<{ id: string }>( + `INSERT INTO storage.objects (bucket_id, name, version, archived_at, is_versioned) + SELECT $1, 'veto/' || n, gen_random_uuid()::text, now(), true + FROM generate_series(1, 2) AS n RETURNING id`, + [bucketId] + ) + ).rows.map((row) => row.id) + await client.query(`CREATE FUNCTION public.lifecycle_veto_probe() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + IF ${vetoAll ? 'true' : "OLD.name = 'veto/1'"} THEN + RAISE EXCEPTION 'protected' USING ERRCODE = '${code}'; + END IF; + RETURN OLD; + END $$`) + await client.query(`CREATE TRIGGER lifecycle_veto_probe BEFORE DELETE ON storage.objects + FOR EACH ROW EXECUTE FUNCTION public.lifecycle_veto_probe()`) + + const deleted = await client.query<{ id: string }>( + 'SELECT * FROM storage.delete_lifecycle_objects($1::uuid[])', + [ids] + ) + expect(deleted.rows.map((row) => row.id)).toEqual(vetoAll ? [] : [ids[1]]) + expect( + ( + await client.query( + 'SELECT id FROM storage.objects WHERE id = ANY($1::uuid[]) ORDER BY id', + [ids] + ) + ).rows + ).toEqual((vetoAll ? ids : [ids[0]]).sort().map((id) => ({ id }))) + }) + } + ) + + it('deletes only non-vetoed metadata and propagates non-veto errors', async () => { + await withRollback(async (client) => { + await client.query("SELECT set_config('storage.allow_delete_query', 'true', true)") + const ids = ( + await client.query<{ id: string }>( + `INSERT INTO storage.objects (bucket_id, name, version, archived_at, is_versioned) + SELECT $1, 'probe/' || n, gen_random_uuid()::text, now(), true + FROM generate_series(1, 3) AS n RETURNING id`, + [bucketId] + ) + ).rows.map((row) => row.id) + const current = ( + await client.query<{ id: string }>( + `INSERT INTO storage.objects (bucket_id, name, version, is_versioned) + VALUES ($1, 'probe/current', gen_random_uuid()::text, true) RETURNING id`, + [bucketId] + ) + ).rows[0].id + await client.query( + `CREATE TABLE public.lifecycle_probe_refs ( + object_id uuid REFERENCES storage.objects(id) DEFERRABLE INITIALLY DEFERRED + )` + ) + await client.query('INSERT INTO public.lifecycle_probe_refs VALUES ($1)', [ids[1]]) + await client.query('CREATE SEQUENCE public.lifecycle_probe_calls') + const codes = [ + '08006', + '40001', + '42P01', + '42883', + '53300', + '55P03', + '57P03', + '58030', + 'XX001', + ] + // The Multigres pooler discards the transaction for connection/startup errors. + for (const code of codes.filter( + (value) => getConfig().databaseEngine !== 'multigres' || !['08006', '57P03'].includes(value) + )) { + // The first error is one-shot; the second occurs only after the FK forces fallback. + for (const raiseAfter of [0, 3]) { + await client.query('SAVEPOINT transient') + await client.query("SELECT setval('public.lifecycle_probe_calls', 1, false)") + await client.query(`CREATE FUNCTION public.lifecycle_probe_busy() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + IF nextval('public.lifecycle_probe_calls') ${raiseAfter === 0 ? '= 1' : '> 3'} THEN + RAISE EXCEPTION 'busy' USING ERRCODE = '${code}'; + END IF; + RETURN OLD; + END $$`) + await client.query(`CREATE TRIGGER lifecycle_probe_busy BEFORE DELETE ON storage.objects + FOR EACH ROW EXECUTE FUNCTION public.lifecycle_probe_busy()`) + await expect( + client.query('SELECT * FROM storage.delete_lifecycle_objects($1::uuid[])', [ids]) + ).rejects.toMatchObject({ code }) + await client.query('ROLLBACK TO SAVEPOINT transient') + } + } + const currentOnly = await client.query( + 'SELECT * FROM storage.delete_lifecycle_objects($1::uuid[])', + [[current]] + ) + expect(currentOnly.rows).toEqual([]) + await client.query('SET CONSTRAINTS ALL DEFERRED') + const deleted = await client.query<{ id: string }>( + 'SELECT * FROM storage.delete_lifecycle_objects($1::uuid[])', + [[...ids, current]] + ) + expect(deleted.rows.map((row) => row.id).sort()).toEqual([ids[0], ids[2]].sort()) + const remaining = await client.query( + 'SELECT id FROM storage.objects WHERE id = ANY($1::uuid[]) ORDER BY id', + [[...ids, current]] + ) + expect(remaining.rows).toEqual( + [{ id: ids[1] }, { id: current }].sort((x, y) => x.id.localeCompare(y.id)) + ) + }) + }) + + it.each(['bulk', 'fallback'] as const)( + 'returns marker and metadata values aligned with deleted ids in %s', + async (mode) => { + await withRollback(async (client) => { + const inserted = await client.query<{ + id: string + is_delete_marker: boolean + metadata: unknown + }>( + `INSERT INTO storage.objects (bucket_id, name, version, archived_at, is_versioned, is_delete_marker, metadata) + VALUES ($1, 'return/marker', gen_random_uuid()::text, now(), true, true, NULL), + ($1, 'return/veto', gen_random_uuid()::text, now(), true, false, '{"size":17}'), + ($1, 'return/object', gen_random_uuid()::text, now(), true, false, '{"size":43}'), + ($1, 'return/no-metadata', gen_random_uuid()::text, now(), true, false, NULL) + RETURNING id, is_delete_marker, metadata`, + [bucketId] + ) + if (mode === 'fallback') { + await client.query( + 'CREATE TABLE public.lifecycle_return_ref (id uuid REFERENCES storage.objects(id))' + ) + await client.query('INSERT INTO lifecycle_return_ref VALUES ($1)', [inserted.rows[1].id]) + } + const deleted = await client.query( + 'SELECT * FROM storage.delete_lifecycle_objects($1::uuid[]) ORDER BY id', + [inserted.rows.map((row) => row.id)] + ) + const expected = inserted.rows.filter((_, index) => mode === 'bulk' || index !== 1) + expect(deleted.rows).toEqual(expected.sort((a, b) => a.id.localeCompare(b.id))) + }) + } + ) + + it('propagates pending deferred violations before deleting lifecycle objects', async () => { + await withRollback(async (client) => { + await client.query("SELECT set_config('storage.allow_delete_query', 'true', true)") + const id = ( + await client.query<{ id: string }>( + `INSERT INTO storage.objects (bucket_id, name, version, archived_at, is_versioned) + VALUES ($1, 'deferred/object', gen_random_uuid()::text, now(), true) RETURNING id`, + [bucketId] + ) + ).rows[0].id + await client.query('CREATE TEMP TABLE lifecycle_deferred_parent (id integer PRIMARY KEY)') + await client.query(`CREATE TEMP TABLE lifecycle_deferred_child ( + parent_id integer REFERENCES lifecycle_deferred_parent(id) DEFERRABLE INITIALLY DEFERRED + )`) + await client.query('INSERT INTO lifecycle_deferred_child VALUES (1)') + await client.query('SAVEPOINT before_delete') + await expect( + client.query('SELECT * FROM storage.delete_lifecycle_objects($1::uuid[])', [[id]]) + ).rejects.toMatchObject({ code: '23503' }) + await client.query('ROLLBACK TO SAVEPOINT before_delete') + expect( + (await client.query('SELECT id FROM storage.objects WHERE id = $1', [id])).rows + ).toEqual([{ id }]) + + await client.query('INSERT INTO lifecycle_deferred_parent VALUES (1)') + const deleted = await client.query( + 'SELECT * FROM storage.delete_lifecycle_objects($1::uuid[])', + [[id]] + ) + expect(deleted.rows.map((row) => row.id)).toEqual([id]) + await expect( + client.query('INSERT INTO lifecycle_deferred_child VALUES (2)') + ).rejects.toMatchObject({ + code: '23503', + }) + }) + }) + + it('runs user delete triggers with the caller search_path', async () => { + await withRollback(async (client) => { + await client.query("SELECT set_config('storage.allow_delete_query', 'true', true)") + const id = ( + await client.query<{ id: string }>( + `INSERT INTO storage.objects (bucket_id, name, version, archived_at, is_versioned) + VALUES ($1, 'search-path/k', gen_random_uuid()::text, now(), true) RETURNING id`, + [bucketId] + ) + ).rows[0].id + await client.query('CREATE TABLE public.lifecycle_probe_audit (id uuid)') + await client.query(`CREATE FUNCTION public.lifecycle_probe_audit_fn() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN INSERT INTO lifecycle_probe_audit VALUES (OLD.id); RETURN OLD; END $$`) + await client.query(`CREATE TRIGGER lifecycle_probe_audit AFTER DELETE ON storage.objects + FOR EACH ROW EXECUTE FUNCTION public.lifecycle_probe_audit_fn()`) + const deleted = await client.query( + 'SELECT * FROM storage.delete_lifecycle_objects($1::uuid[])', + [[id]] + ) + expect(deleted.rows.map((row) => row.id)).toEqual([id]) + const audit = await client.query('SELECT id FROM public.lifecycle_probe_audit') + expect(audit.rows).toEqual([{ id }]) + }) + }) + + it('deletes a full batch without overflowing the subtransaction cache', async (context) => { + const client = new Client({ connectionString: getConfig().databaseURL }) + await client.connect() + try { + const server = await client.query<{ version: string }>( + `SELECT current_setting('server_version_num') AS version` + ) + if (Number(server.rows[0].version) < 160000) context.skip() + await client.query('BEGIN') + await client.query("SELECT set_config('storage.allow_delete_query', 'true', true)") + const ids = ( + await client.query<{ id: string }>( + `INSERT INTO storage.objects (bucket_id, name, version, archived_at, is_versioned) + SELECT $1, 'batch/' || n, gen_random_uuid()::text, now(), true + FROM generate_series(1, 100) AS n RETURNING id`, + [bucketId] + ) + ).rows.map((row) => row.id) + const deleted = await client.query( + 'SELECT * FROM storage.delete_lifecycle_objects($1::uuid[])', + [ids] + ) + expect(deleted.rowCount).toBe(ids.length) + const subxact = await client.query( + `SELECT subxact.subxact_overflowed + FROM pg_stat_get_backend_idset() AS backend, + LATERAL pg_stat_get_backend_subxact(backend) AS subxact + WHERE pg_stat_get_backend_pid(backend) = pg_backend_pid()` + ) + expect(subxact.rows).toEqual([{ subxact_overflowed: false }]) + } finally { + await client.query('ROLLBACK').catch(() => undefined) + await client.end() + } + }) + + it('takes the bucket row lock first so reconcile excludes arm', async () => { + await tHelper.database.putLifecycleConfiguration(bucketId, { + rules: [ + { status: 'Enabled', filter: {}, noncurrentVersionExpiration: { noncurrentDays: 1 } }, + ], + }) + const owner = new Client({ connectionString: getConfig().databaseURL }) + const prober = new Client({ connectionString: getConfig().databaseURL }) + await Promise.all([owner.connect(), prober.connect()]) + try { + await owner.query('BEGIN') + await owner.query('SELECT storage.reconcile_lifecycle_state($1, true)', [bucketId]) + await prober.query('BEGIN') + await expect( + prober.query('SELECT 1 FROM storage.buckets WHERE id = $1 FOR SHARE NOWAIT', [bucketId]) + ).rejects.toMatchObject({ code: '55P03' }) + } finally { + await prober.query('ROLLBACK').catch(() => undefined) + await owner.query('ROLLBACK').catch(() => undefined) + await Promise.all([owner.end(), prober.end()]) + } + }) + + it('allows a bucket foreign-key insert during versioning enable while excluding arm', async () => { + await tHelper.database.putLifecycleConfiguration(bucketId, { + rules: [ + { status: 'Enabled', filter: {}, noncurrentVersionExpiration: { noncurrentDays: 1 } }, + ], + }) + const owner = new Client({ connectionString: getConfig().databaseURL }) + const prober = new Client({ connectionString: getConfig().databaseURL }) + await Promise.all([owner.connect(), prober.connect()]) + try { + await owner.query('BEGIN') + await owner.query("UPDATE storage.buckets SET versioning_status = 'ENABLED' WHERE id = $1", [ + bucketId, + ]) + await prober.query('BEGIN') + await prober.query("SET LOCAL lock_timeout = '2s'") + await prober.query( + "INSERT INTO storage.objects (bucket_id, name) VALUES ($1, 'concurrent-insert')", + [bucketId] + ) + await prober.query('SAVEPOINT before_arm_lock') + await expect( + prober.query('SELECT 1 FROM storage.buckets WHERE id = $1 FOR SHARE NOWAIT', [bucketId]) + ).rejects.toMatchObject({ code: '55P03' }) + await prober.query('ROLLBACK TO SAVEPOINT before_arm_lock') + } finally { + await prober.query('ROLLBACK').catch(() => undefined) + await owner.query('ROLLBACK').catch(() => undefined) + await Promise.all([owner.end(), prober.end()]) + } + }) + + it('preserves a state row inserted after reconciliation sees no row', async () => { + await fixtures.setStatus(bucketId, 'ENABLED') + await tHelper.database.putLifecycleConfiguration(bucketId, { + rules: [ + { status: 'Enabled', filter: {}, noncurrentVersionExpiration: { noncurrentDays: 1 } }, + ], + }) + await tHelper.database.connection.query( + 'DELETE FROM storage.bucket_lifecycle_states WHERE bucket_id = $1', + [bucketId] + ) + const claimId = randomUUID() + const writer = new Client({ connectionString: getConfig().databaseURL }) + const reconciler = new Client({ connectionString: getConfig().databaseURL }) + await Promise.all([writer.connect(), reconciler.connect()]) + try { + await writer.query('BEGIN') + const writerPid = (await writer.query<{ pid: number }>('SELECT pg_backend_pid() AS pid')) + .rows[0].pid + const reconcilerPid = ( + await reconciler.query<{ pid: number }>('SELECT pg_backend_pid() AS pid') + ).rows[0].pid + await writer.query( + `INSERT INTO storage.bucket_lifecycle_states (bucket_id, generation, continuation, claim_id, next_run_at) + SELECT id, lifecycle_configuration_generation, '{"continuationVersion": 2}', $2, '2099-01-01' + FROM storage.buckets WHERE id = $1`, + [bucketId, claimId] + ) + const reconciliation = reconciler.query( + 'SELECT storage.reconcile_lifecycle_state($1, true)', + [bucketId] + ) + reconciliation.catch(() => undefined) + await vi.waitFor( + async () => { + const waiting = await tHelper.database.connection.query( + `SELECT 1 FROM pg_stat_activity + WHERE pid = $2::int AND wait_event IN ('transactionid', 'virtualxid') + AND $1::int = ANY(pg_blocking_pids(pid))`, + [writerPid, reconcilerPid] + ) + expect(waiting.rowCount).toBe(1) + }, + { timeout: 3000 } + ) + await writer.query('COMMIT') + await reconciliation + await expect( + fixtures.readState( + bucketId, + 'continuation, claim_id, next_run_at <= clock_timestamp() AS due' + ) + ).resolves.toEqual([ + { continuation: { continuationVersion: 2 }, claim_id: claimId, due: true }, + ]) + } finally { + await writer.query('ROLLBACK').catch(() => undefined) + await Promise.all([writer.end(), reconciler.end()]) + } + }) +}) diff --git a/src/test/utils/lifecycle-continuation.ts b/src/test/utils/lifecycle-continuation.ts new file mode 100644 index 000000000..c53b621b0 --- /dev/null +++ b/src/test/utils/lifecycle-continuation.ts @@ -0,0 +1,52 @@ +import { randomUUID } from 'node:crypto' +import { + armLifecycleAttempt, + createLifecycleContinuation, + freezeLifecycleBatchVersion, + stageLifecycleBatch, +} from '@storage/lifecycle/continuation' +import type { LifecycleCandidateIdentity, NoncurrentLifecycleShardCursor } from '@storage/schemas' + +export function newContinuation( + generation: string, + overrides: Partial[0]> = {} +) { + return createLifecycleContinuation({ + runId: randomUUID(), + generation, + snapshotAt: new Date().toISOString(), + trigger: 'scheduled', + mode: 'DELETE', + topology: { scanKind: 'NONCURRENT', epoch: '1', shardId: 0, shardCount: 1 }, + ...overrides, + }) +} + +export function stagedContinuation( + generation: string, + versions: LifecycleCandidateIdentity[], + pageEnd: NoncurrentLifecycleShardCursor, + options: { snapshotAt?: string } = {} +) { + 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: NoncurrentLifecycleShardCursor, + options: { snapshotAt?: string; startedAt?: string } = {} +) { + return armLifecycleAttempt(stagedContinuation(generation, versions, pageEnd, options), { + attemptId: randomUUID(), + authorizedGeneration: generation, + startedAt: options.startedAt ?? new Date().toISOString(), + }) +} diff --git a/src/test/utils/lifecycle-versioning.ts b/src/test/utils/lifecycle-versioning.ts new file mode 100644 index 000000000..1fa8507c1 --- /dev/null +++ b/src/test/utils/lifecycle-versioning.ts @@ -0,0 +1,86 @@ +import { randomUUID } from 'node:crypto' +import type { DatabaseTransaction } from '@internal/database' +import { useStorage, withDeleteEnabled } from './storage' + +/** Shared bucket setup, scoped cleanup and service-operation transactions. */ +export function useLifecycleStorage( + prefix: string, + options: { operation?: () => string | undefined } = {} +) { + const helper = useStorage({ operation: options.operation }) + const buckets = new Set() + let bucketId: string + + async function createBucket(type: 'STANDARD' | 'ANALYTICS' = 'STANDARD') { + const id = `${prefix}-${randomUUID()}` + buckets.add(id) + await helper.database.createBucket({ id, name: id }) + if (type === 'ANALYTICS') { + await helper.database.connection.query( + "UPDATE storage.buckets SET type = 'ANALYTICS' WHERE id = $1", + [id] + ) + } + return id + } + + beforeEach(async () => { + bucketId = await createBucket() + }) + + afterEach(async () => { + const ids = [...buckets] + await withDeleteEnabled(helper.database.connection, async (transaction) => { + await transaction.query( + 'DELETE FROM storage.bucket_lifecycle_states WHERE bucket_id = ANY($1::text[])', + [ids] + ) + await transaction.query( + 'DELETE FROM storage.lifecycle_abandoned_artifacts WHERE bucket_id = ANY($1::text[])', + [ids] + ) + await transaction.query('DELETE FROM storage.objects WHERE bucket_id = ANY($1::text[])', [ + ids, + ]) + await transaction.query('DELETE FROM storage.buckets WHERE id = ANY($1::text[])', [ids]) + }) + buckets.clear() + }) + + return { + helper, + get bucketId() { + return bucketId + }, + createBucket, + async readState(id: string, columns: string) { + const result = await helper.database.connection.query( + `SELECT ${columns} FROM storage.bucket_lifecycle_states WHERE bucket_id = $1`, + [id] + ) + return result.rows + }, + async setStatus(id: string, status: 'DISABLED' | 'ENABLED' | 'SUSPENDED') { + if (status === 'SUSPENDED') { + await helper.database.updateBucket(id, { versioning_status: 'ENABLED' }) + } + await helper.database.updateBucket(id, { versioning_status: status }) + }, + async withServiceOperation( + operation: string, + fn: (transaction: DatabaseTransaction) => Promise + ): Promise { + const transaction = await helper.database.connection.transaction() + try { + await helper.database.connection.setScope(transaction) + await transaction.query("SELECT set_config('storage.operation', $1, true)", [operation]) + const result = await fn(transaction) + await transaction.commit() + return result + } catch (error) { + await transaction.rollback() + throw error + } + }, + } +}