diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,15 @@ # Changelog +## 0.3.2.0 — 2026-08-13 + +### New Features + +* Added forward migration `0010`, which installs durable replay-history + retention leases, a per-schema singleton coordinator, a partial active-lease + index, and statement-level `DELETE`/`TRUNCATE` guards for `events`, + `stream_events`, and `streams`. Active retention is refused with SQLSTATE + `KR001` even when the existing hard-delete GUC is enabled. + ## 0.3.1.0 — 2026-08-13 ### New Features diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -2,9 +2,9 @@ `kiroku-store-migrations` owns Kiroku's PostgreSQL schema as one native `pg-migrate` component named `kiroku`. The component embeds an ordered manifest -and nine SQL payloads, so applications can compose it with other libraries +and ten SQL payloads, so applications can compose it with other libraries without copying Kiroku SQL. The first seven payloads are immutable historical -Codd bytes; `0008` and `0009` are native-only forward migrations. +Codd bytes; `0008`, `0009`, and `0010` are native-only forward migrations. ## Public API @@ -108,6 +108,10 @@ view survival, and indexed query plan. `cabal test kiroku-store:kiroku-store-test` consumes the same native plan through `kiroku-test-support` and proves the complete store behavior, including append and read scenarios. + +Migration `0010` adds replay-history retention leases, the per-schema +coordinator, an indexed active-lease predicate, and statement-level +`DELETE`/`TRUNCATE` guards on the three event-store data tables. ## Recovery diff --git a/kiroku-store-migrations.cabal b/kiroku-store-migrations.cabal --- a/kiroku-store-migrations.cabal +++ b/kiroku-store-migrations.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: kiroku-store-migrations -version: 0.3.1.0 +version: 0.3.2.0 synopsis: Schema migrations for kiroku-store description: Native pg-migrate component, Codd history mapping, and migration executable diff --git a/migrations/0010.sql b/migrations/0010.sql new file mode 100644 --- /dev/null +++ b/migrations/0010.sql @@ -0,0 +1,112 @@ +-- add replay history retention + +-- A schema-local singleton row serializes lease lifecycle changes with every +-- destructive statement without adding work to append or ordinary read paths. +CREATE TABLE kiroku.history_retention_coordinator ( + singleton BOOLEAN PRIMARY KEY DEFAULT TRUE, + CONSTRAINT chk_history_retention_coordinator_singleton CHECK (singleton) +); + +INSERT INTO kiroku.history_retention_coordinator (singleton) +VALUES (TRUE) +ON CONFLICT DO NOTHING; + +-- Durable operational evidence for rebuilds that require the retained event +-- set to remain stable. Active state is derived from released_at and database +-- time; expiry needs no worker or cleanup mutation. +CREATE TABLE kiroku.history_retention_leases ( + lease_id UUID PRIMARY KEY DEFAULT uuidv7(), + owner TEXT NOT NULL, + reason TEXT NOT NULL, + protected_through BIGINT NOT NULL, + created_at TIMESTAMPTZ NOT NULL, + renewed_at TIMESTAMPTZ NOT NULL, + expires_at TIMESTAMPTZ NOT NULL, + released_at TIMESTAMPTZ, + CONSTRAINT chk_history_retention_lease_owner_bytes + CHECK (octet_length(owner) BETWEEN 1 AND 512), + CONSTRAINT chk_history_retention_lease_reason_bytes + CHECK (octet_length(reason) BETWEEN 1 AND 2048), + CONSTRAINT chk_history_retention_lease_frontier + CHECK (protected_through >= 0), + CONSTRAINT chk_history_retention_lease_renewal_time + CHECK (renewed_at >= created_at), + CONSTRAINT chk_history_retention_lease_expiry + CHECK (expires_at > created_at), + CONSTRAINT chk_history_retention_lease_release_time + CHECK (released_at IS NULL OR released_at >= created_at) +); + +CREATE INDEX ix_history_retention_leases_unreleased_expiry + ON kiroku.history_retention_leases (expires_at) + WHERE released_at IS NULL; + +-- This function is attached only to DELETE and TRUNCATE. TG_TABLE_SCHEMA is +-- quoted into each dynamic statement so a schema-qualified maintenance command +-- cannot be redirected to a coordinator selected through the caller's +-- search_path. +CREATE FUNCTION kiroku.protect_replay_history_from_destruction() +RETURNS TRIGGER +LANGUAGE plpgsql +AS $$ +DECLARE + active_count BIGINT; +BEGIN + EXECUTE format( + 'SELECT singleton FROM %I.history_retention_coordinator WHERE singleton FOR UPDATE', + TG_TABLE_SCHEMA + ); + + EXECUTE format( + 'SELECT count(*) FROM %I.history_retention_leases WHERE released_at IS NULL AND expires_at > clock_timestamp()', + TG_TABLE_SCHEMA + ) INTO active_count; + + IF active_count > 0 THEN + RAISE EXCEPTION USING + ERRCODE = 'KR001', + MESSAGE = format( + 'replay history is protected by %s active retention lease(s)', + active_count + ); + END IF; + + RETURN NULL; +END; +$$; + +CREATE TRIGGER protect_replay_history_delete + BEFORE DELETE ON kiroku.events + FOR EACH STATEMENT + EXECUTE FUNCTION kiroku.protect_replay_history_from_destruction(); + +CREATE TRIGGER protect_replay_history_truncate + BEFORE TRUNCATE ON kiroku.events + FOR EACH STATEMENT + EXECUTE FUNCTION kiroku.protect_replay_history_from_destruction(); + +CREATE TRIGGER protect_replay_history_delete + BEFORE DELETE ON kiroku.stream_events + FOR EACH STATEMENT + EXECUTE FUNCTION kiroku.protect_replay_history_from_destruction(); + +CREATE TRIGGER protect_replay_history_truncate + BEFORE TRUNCATE ON kiroku.stream_events + FOR EACH STATEMENT + EXECUTE FUNCTION kiroku.protect_replay_history_from_destruction(); + +CREATE TRIGGER protect_replay_history_delete + BEFORE DELETE ON kiroku.streams + FOR EACH STATEMENT + EXECUTE FUNCTION kiroku.protect_replay_history_from_destruction(); + +CREATE TRIGGER protect_replay_history_truncate + BEFORE TRUNCATE ON kiroku.streams + FOR EACH STATEMENT + EXECUTE FUNCTION kiroku.protect_replay_history_from_destruction(); + +COMMENT ON TABLE kiroku.history_retention_leases IS + 'Durable, expiring replay-history retention leases; rows are operational evidence and remain until explicitly pruned.'; + +COMMENT ON SCHEMA kiroku IS + 'Managed by pg-migrate component kiroku through 0010'; diff --git a/migrations/manifest b/migrations/manifest --- a/migrations/manifest +++ b/migrations/manifest @@ -7,3 +7,4 @@ 0007-stream-truncate-before.sql 0008-schema-management-comment.sql 0009.sql +0010.sql diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -42,7 +42,7 @@ main :: IO () main = hspec $ do describe "native Kiroku migration definition" $ do - it "tracks the nine native files in manifest order" $ do + it "tracks the ten native files in manifest order" $ do directory <- findMigrationsDirectory manifest <- Text.lines <$> Text.IO.readFile (directory "manifest") manifest `shouldBe` Text.pack <$> nativeMigrationFiles @@ -55,7 +55,7 @@ bytes <- ByteString.readFile (directory nativeName) lookup legacyName lockEntries `shouldBe` Just (checksumText bytes) - it "builds component kiroku and a nine-migration plan" $ do + it "builds component kiroku and a ten-migration plan" $ do component <- requireRight kirokuMigrations component `seq` pure () plan <- requirePlan @@ -91,7 +91,7 @@ `shouldReturn` "0007-existing.sql\n" describe "fresh native databases" $ do - it "applies all nine, verifies strictly, and reports AlreadyApplied on rerun" $ do + it "applies all ten, verifies strictly, and reports AlreadyApplied on rerun" $ do plan <- requirePlan result <- withMigratedDatabase plan $ \connection -> do assertSchema connection @@ -248,6 +248,31 @@ Text.isInfixOf "CTE Scan" planText `shouldBe` False either (expectationFailure . show) pure result + describe "history retention schema" $ do + it "publishes the exact coordinator and lease table columns" $ do + plan <- requirePlan + result <- withMigratedDatabase plan $ \connection -> do + columns <- useSession connection (Session.statement () historyRetentionColumnsStatement) + columns + `shouldBe` [ ("history_retention_coordinator", "singleton", "boolean", True) + , ("history_retention_leases", "lease_id", "uuid", True) + , ("history_retention_leases", "owner", "text", True) + , ("history_retention_leases", "reason", "text", True) + , ("history_retention_leases", "protected_through", "bigint", True) + , ("history_retention_leases", "created_at", "timestamp with time zone", True) + , ("history_retention_leases", "renewed_at", "timestamp with time zone", True) + , ("history_retention_leases", "expires_at", "timestamp with time zone", True) + , ("history_retention_leases", "released_at", "timestamp with time zone", False) + ] + either (expectationFailure . show) pure result + + it "installs the singleton, checks, partial index, function, and six triggers" $ do + plan <- requirePlan + result <- withMigratedDatabase plan $ \connection -> do + facts <- useSession connection (Session.statement () historyRetentionFactsStatement) + facts `shouldBe` (1, 9, True, True, 6) + either (expectationFailure . show) pure result + describe "Codd history import" $ do it "imports a current codd V5 ledger, verifies, and never replays SQL" $ importFixture "codd" @@ -293,14 +318,14 @@ pendingIds <- traverse (requireRight . migrationId "kiroku") - ["0008-schema-management-comment", "0009"] + ["0008-schema-management-comment", "0009", "0010"] verifiedBeforeCanary <- verifyMigrationPlan defaultRunOptions settings plan >>= requireMigration case verifiedBeforeCanary of VerificationReport verificationIssues _ _ _ -> verificationIssues `shouldBe` (PendingMigration <$> pendingIds) up <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration - reportOutcomes up `shouldBe` replicate 7 AlreadyApplied <> replicate 2 AppliedNow + reportOutcomes up `shouldBe` replicate 7 AlreadyApplied <> replicate 3 AppliedNow verifiedAfterCanary <- verifyMigrationPlan defaultRunOptions settings plan >>= requireMigration case verifiedAfterCanary of VerificationReport verificationIssues _ _ _ -> @@ -329,6 +354,7 @@ , "0007-stream-truncate-before.sql" , "0008-schema-management-comment.sql" , "0009.sql" + , "0010.sql" ] findMigrationsDirectory :: IO FilePath @@ -446,11 +472,95 @@ (EXISTS (SELECT 1 FROM pg_catalog.pg_indexes WHERE schemaname = 'kiroku' AND indexname = 'ix_dead_letters_event_id')), (EXISTS (SELECT 1 FROM pg_catalog.pg_constraint WHERE conname = 'chk_streams_stream_name_length')), (EXISTS (SELECT 1 FROM pg_catalog.pg_attribute a JOIN pg_catalog.pg_class c ON c.oid = a.attrelid JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace WHERE n.nspname = 'kiroku' AND c.relname = 'streams' AND a.attname = 'truncate_before' AND NOT a.attisdropped)), - (obj_description(to_regnamespace('kiroku'), 'pg_namespace') = 'Managed by pg-migrate component kiroku through 0008-schema-management-comment') + (obj_description(to_regnamespace('kiroku'), 'pg_namespace') = 'Managed by pg-migrate component kiroku through 0010') ) AS checks(ok) """ Encoders.noParams (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.bool))) + +historyRetentionColumnsStatement :: Statement () [(Text, Text, Text, Bool)] +historyRetentionColumnsStatement = + Statement.preparable + """ + SELECT relation.relname::text, + attribute.attname::text, + pg_catalog.format_type(attribute.atttypid, attribute.atttypmod), + attribute.attnotnull + FROM pg_catalog.pg_class AS relation + JOIN pg_catalog.pg_namespace AS namespace + ON namespace.oid = relation.relnamespace + JOIN pg_catalog.pg_attribute AS attribute + ON attribute.attrelid = relation.oid + WHERE namespace.nspname = 'kiroku' + AND relation.relname IN ('history_retention_coordinator', 'history_retention_leases') + AND attribute.attnum > 0 + AND NOT attribute.attisdropped + ORDER BY relation.relname, attribute.attnum + """ + Encoders.noParams + ( Decoders.rowList + ( (,,,) + <$> column Decoders.text + <*> column Decoders.text + <*> column Decoders.text + <*> column Decoders.bool + ) + ) + where + column = Decoders.column . Decoders.nonNullable + +historyRetentionFactsStatement :: Statement () (Int64, Int64, Bool, Bool, Int64) +historyRetentionFactsStatement = + Statement.preparable + """ + SELECT + (SELECT count(*) FROM kiroku.history_retention_coordinator WHERE singleton), + (SELECT count(*) + FROM pg_catalog.pg_constraint AS con_record + JOIN pg_catalog.pg_class AS relation ON relation.oid = con_record.conrelid + JOIN pg_catalog.pg_namespace AS namespace ON namespace.oid = relation.relnamespace + WHERE namespace.nspname = 'kiroku' + AND relation.relname IN ('history_retention_coordinator', 'history_retention_leases') + AND con_record.conname IN ( + 'history_retention_coordinator_pkey', + 'chk_history_retention_coordinator_singleton', + 'history_retention_leases_pkey', + 'chk_history_retention_lease_owner_bytes', + 'chk_history_retention_lease_reason_bytes', + 'chk_history_retention_lease_frontier', + 'chk_history_retention_lease_renewal_time', + 'chk_history_retention_lease_expiry', + 'chk_history_retention_lease_release_time' + )), + EXISTS ( + SELECT 1 + FROM pg_catalog.pg_indexes + WHERE schemaname = 'kiroku' + AND indexname = 'ix_history_retention_leases_unreleased_expiry' + AND indexdef LIKE '%WHERE (released_at IS NULL)' + ), + to_regprocedure('kiroku.protect_replay_history_from_destruction()') IS NOT NULL, + (SELECT count(*) + FROM pg_catalog.pg_trigger AS trigger_record + JOIN pg_catalog.pg_class AS relation ON relation.oid = trigger_record.tgrelid + JOIN pg_catalog.pg_namespace AS namespace ON namespace.oid = relation.relnamespace + WHERE namespace.nspname = 'kiroku' + AND relation.relname IN ('events', 'stream_events', 'streams') + AND trigger_record.tgname IN ('protect_replay_history_delete', 'protect_replay_history_truncate') + AND NOT trigger_record.tgisinternal) + """ + Encoders.noParams + ( Decoders.singleRow + ( (,,,,) + <$> column Decoders.int8 + <*> column Decoders.int8 + <*> column Decoders.bool + <*> column Decoders.bool + <*> column Decoders.int8 + ) + ) + where + column = Decoders.column . Decoders.nonNullable oversizedStreamStatement :: Statement Text () oversizedStreamStatement =