diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,14 @@ ## Unreleased +## 0.14.0.0 — 2026-08-21 + +### New Features + +- Migration `0031.sql` adds the terminal `rejected` outbox status audit columns, + validates the bounded code/detail contract in PostgreSQL, and updates per-key and + per-source head-of-line indexes so rejected rows release ordered successors. + ## 0.13.0.0 — 2026-08-17 ### Breaking Changes diff --git a/expected-schema/native/keiro-v18.txt b/expected-schema/native/keiro-v18.txt --- a/expected-schema/native/keiro-v18.txt +++ b/expected-schema/native/keiro-v18.txt @@ -98,6 +98,9 @@ column keiro_outbox.outbox_id uuid not null column keiro_outbox.payload_bytes bytea not null column keiro_outbox.published_at timestamp with time zone +column keiro_outbox.rejected_at timestamp with time zone +column keiro_outbox.rejection_code text +column keiro_outbox.rejection_detail text column keiro_outbox.schema_fingerprint text column keiro_outbox.schema_id bigint column keiro_outbox.schema_registry text @@ -376,6 +379,9 @@ constraint keiro_outbox.keiro_outbox_outbox_id_not_null NOT NULL outbox_id constraint keiro_outbox.keiro_outbox_payload_bytes_not_null NOT NULL payload_bytes constraint keiro_outbox.keiro_outbox_pkey PRIMARY KEY (outbox_id) +constraint keiro_outbox.keiro_outbox_rejection_audit_check CHECK ((((status = 'rejected'::text) AND (rejected_at IS NOT NULL) AND (rejection_code IS NOT NULL)) OR ((status <> 'rejected'::text) AND (rejected_at IS NULL) AND (rejection_code IS NULL) AND (rejection_detail IS NULL)))) +constraint keiro_outbox.keiro_outbox_rejection_code_check CHECK (((rejection_code IS NULL) OR (rejection_code ~ '^[a-z][a-z0-9._-]{0,63}$'::text))) +constraint keiro_outbox.keiro_outbox_rejection_detail_check CHECK (((rejection_detail IS NULL) OR ((octet_length(rejection_detail) >= 1) AND (octet_length(rejection_detail) <= 1024)))) constraint keiro_outbox.keiro_outbox_schema_version_not_null NOT NULL schema_version constraint keiro_outbox.keiro_outbox_source_message_id_key UNIQUE (source, message_id) constraint keiro_outbox.keiro_outbox_source_not_null NOT NULL source @@ -598,12 +604,12 @@ index keiro_managed_read_objects_contract_idx CREATE INDEX keiro_managed_read_objects_contract_idx ON keiro.keiro_managed_read_objects USING btree (contract_id, contract_version) index keiro_managed_read_objects_pkey CREATE UNIQUE INDEX keiro_managed_read_objects_pkey ON keiro.keiro_managed_read_objects USING btree (object_schema, object_name, object_kind, object_signature) index keiro_outbox_claim_order_idx CREATE INDEX keiro_outbox_claim_order_idx ON keiro.keiro_outbox USING btree (created_at, outbox_id) WHERE (status = ANY (ARRAY['pending'::text, 'failed'::text])) -index keiro_outbox_head_of_line_idx CREATE INDEX keiro_outbox_head_of_line_idx ON keiro.keiro_outbox USING btree (source, message_key, created_at) WHERE ((status <> ALL (ARRAY['sent'::text, 'dead'::text])) AND (message_key IS NOT NULL)) +index keiro_outbox_head_of_line_idx CREATE INDEX keiro_outbox_head_of_line_idx ON keiro.keiro_outbox USING btree (source, message_key, created_at) WHERE ((status <> ALL (ARRAY['sent'::text, 'dead'::text, 'rejected'::text])) AND (message_key IS NOT NULL)) index keiro_outbox_pending_idx CREATE INDEX keiro_outbox_pending_idx ON keiro.keiro_outbox USING btree (status, next_attempt_at, created_at) index keiro_outbox_pkey CREATE UNIQUE INDEX keiro_outbox_pkey ON keiro.keiro_outbox USING btree (outbox_id) index keiro_outbox_sent_gc_idx CREATE INDEX keiro_outbox_sent_gc_idx ON keiro.keiro_outbox USING btree (published_at) WHERE (status = 'sent'::text) index keiro_outbox_source_message_id_key CREATE UNIQUE INDEX keiro_outbox_source_message_id_key ON keiro.keiro_outbox USING btree (source, message_id) -index keiro_outbox_source_order_idx CREATE INDEX keiro_outbox_source_order_idx ON keiro.keiro_outbox USING btree (source, created_at, outbox_id) WHERE (status <> ALL (ARRAY['sent'::text, 'dead'::text])) +index keiro_outbox_source_order_idx CREATE INDEX keiro_outbox_source_order_idx ON keiro.keiro_outbox USING btree (source, created_at, outbox_id) WHERE (status <> ALL (ARRAY['sent'::text, 'dead'::text, 'rejected'::text])) index keiro_projection_dedup_applied_at_idx CREATE INDEX keiro_projection_dedup_applied_at_idx ON keiro.keiro_projection_dedup USING btree (applied_at) index keiro_projection_dedup_pkey CREATE UNIQUE INDEX keiro_projection_dedup_pkey ON keiro.keiro_projection_dedup USING btree (projection_name, event_id) index keiro_projection_group_cursors_pkey CREATE UNIQUE INDEX keiro_projection_group_cursors_pkey ON keiro.keiro_projection_group_cursors USING btree (group_id) diff --git a/keiro-migrations.cabal b/keiro-migrations.cabal --- a/keiro-migrations.cabal +++ b/keiro-migrations.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: keiro-migrations -version: 0.13.0.0 +version: 0.14.0.0 synopsis: Schema migrations for keiro description: Embedded PostgreSQL schema migrations and a migration runner for the Keiro diff --git a/migrations.native.lock b/migrations.native.lock --- a/migrations.native.lock +++ b/migrations.native.lock @@ -28,3 +28,4 @@ 524e90271704112722bbfe0fc54f1c8dcc06b1f9e2ce1821e54dc232d434104f 0028.sql 40f62a18adf0457fc6d35e880eca8be1922f22a2fd8c893c13d59537fe941719 0029.sql 0e0b847088d178290e77913224df82435b86c00062feb47c79351ed014a605a1 0030.sql +d78b8400638b3a9005f4e964f2cb10d3863d60d1038735ed6fa09507a912a26e 0031.sql diff --git a/migrations/0031.sql b/migrations/0031.sql new file mode 100644 --- /dev/null +++ b/migrations/0031.sql @@ -0,0 +1,39 @@ +-- add terminal outbox rejection outcome + +ALTER TABLE keiro.keiro_outbox + ADD COLUMN rejected_at TIMESTAMPTZ, + ADD COLUMN rejection_code TEXT, + ADD COLUMN rejection_detail TEXT, + ADD CONSTRAINT keiro_outbox_rejection_audit_check CHECK ( + ( + status = 'rejected' + AND rejected_at IS NOT NULL + AND rejection_code IS NOT NULL + ) + OR + ( + status <> 'rejected' + AND rejected_at IS NULL + AND rejection_code IS NULL + AND rejection_detail IS NULL + ) + ), + ADD CONSTRAINT keiro_outbox_rejection_code_check CHECK ( + rejection_code IS NULL + OR rejection_code ~ '^[a-z][a-z0-9._-]{0,63}$' + ), + ADD CONSTRAINT keiro_outbox_rejection_detail_check CHECK ( + rejection_detail IS NULL + OR octet_length(rejection_detail) BETWEEN 1 AND 1024 + ); + +-- Rejected rows are terminal and must not remain in head-of-line indexes. +DROP INDEX IF EXISTS keiro.keiro_outbox_head_of_line_idx; +CREATE INDEX keiro_outbox_head_of_line_idx + ON keiro.keiro_outbox (source, message_key, created_at) + WHERE status NOT IN ('sent', 'dead', 'rejected') AND message_key IS NOT NULL; + +DROP INDEX IF EXISTS keiro.keiro_outbox_source_order_idx; +CREATE INDEX keiro_outbox_source_order_idx + ON keiro.keiro_outbox (source, created_at, outbox_id) + WHERE status NOT IN ('sent', 'dead', 'rejected'); diff --git a/migrations/manifest b/migrations/manifest --- a/migrations/manifest +++ b/migrations/manifest @@ -28,3 +28,4 @@ 0028.sql 0029.sql 0030.sql +0031.sql diff --git a/src/Keiro/Migrations/SchemaCheck.hs b/src/Keiro/Migrations/SchemaCheck.hs --- a/src/Keiro/Migrations/SchemaCheck.hs +++ b/src/Keiro/Migrations/SchemaCheck.hs @@ -107,7 +107,7 @@ snapshotSchema schema = Text.unlines <$> Session.statement schema schemaSnapshotStatement --- | PostgreSQL 18 snapshot regenerated through migration 0030. +-- | PostgreSQL 18 snapshot regenerated through migration 0031. expectedSchemaSnapshot :: Text expectedSchemaSnapshot = $(embedTextFile "expected-schema/native/keiro-v18.txt") diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -54,7 +54,7 @@ main :: IO () main = hspec $ do describe "native Keiro migration definition" $ do - it "tracks thirty native files in manifest order" $ do + it "tracks thirty-one native files in manifest order" $ do directory <- findMigrationsDirectory manifest <- Text.lines <$> Text.IO.readFile (directory "manifest") manifest `shouldBe` Text.pack <$> nativeMigrationFiles @@ -67,7 +67,7 @@ bytes <- ByteString.readFile (directory nativeName) lookup legacyName lockEntries `shouldBe` Just (checksumText bytes) - it "builds component keiro with dependency kiroku and thirty migrations" $ do + it "builds component keiro with dependency kiroku and thirty-one migrations" $ do plan <- requirePlan let PlanDescription components = planDescription plan case toList components of @@ -80,7 +80,7 @@ componentNameText keiroName `shouldBe` "keiro" dependencyName <- requireRight (componentName "kiroku") keiroDependencies `shouldBe` Set.singleton dependencyName - length keiroEntries `shouldBe` 30 + length keiroEntries `shouldBe` 31 actual -> expectationFailure ("unexpected plan description: " <> show actual) validateHistoryMappingTargets plan frameworkCoddHistoryMappings `shouldBe` Right () @@ -169,7 +169,7 @@ [("9999-fixture.sql", "-- Never set search_path in a migration.\nSELECT 1;")] `shouldBe` [] - it "passes all 30 embedded native bodies" $ do + it "passes all 31 embedded native bodies" $ do lintViolations config (toList embeddedMigrationEntries) `shouldBe` [] describe "startup handshake" $ do @@ -183,7 +183,7 @@ plan >>= requireRight Keiro.pendingMigrations handshake `shouldBe` planMigrationIds plan - length (Keiro.pendingMigrations handshake) `shouldBe` 41 + length (Keiro.pendingMigrations handshake) `shouldBe` 42 Keiro.ledgerIssues handshake `shouldBe` [] handshakePassed handshake `shouldBe` False @@ -209,7 +209,7 @@ handshake <- missingMigrations defaultRunOptions provider plan >>= requireRight Keiro.pendingMigrations handshake `shouldBe` drop 11 (planMigrationIds plan) - length (Keiro.pendingMigrations handshake) `shouldBe` 30 + length (Keiro.pendingMigrations handshake) `shouldBe` 31 Keiro.ledgerIssues handshake `shouldBe` [] handshakePassed handshake `shouldBe` False @@ -292,12 +292,12 @@ assertSchema connection let provider = providerFor connection rerun <- runMigrationPlanWith defaultRunOptions provider plan >>= requireRight - reportOutcomes rerun `shouldBe` replicate 41 AlreadyApplied + reportOutcomes rerun `shouldBe` replicate 42 AlreadyApplied verified <- verifyMigrationPlanWith defaultRunOptions provider plan >>= requireRight case verified of VerificationReport verificationIssues applied pending unknown -> do verificationIssues `shouldBe` [] - length applied `shouldBe` 41 + length applied `shouldBe` 42 pending `shouldBe` [] unknown `shouldBe` [] either (expectationFailure . show) pure result @@ -346,7 +346,7 @@ (runMigrationPlan defaultRunOptions settings plan >>= requireRight) (runMigrationPlan defaultRunOptions settings plan >>= requireRight) sort [reportOutcomes first, reportOutcomes second] - `shouldBe` sort [replicate 41 AppliedNow, replicate 41 AlreadyApplied] + `shouldBe` sort [replicate 42 AppliedNow, replicate 42 AlreadyApplied] it "upgrades singleton read-model rows into deterministic rebuild groups" $ do fullPlan <- requirePlan @@ -365,7 +365,7 @@ withConnection settings $ \connection -> useSession connection (Session.script legacyReadModelFixtureSql) report <- runMigrationPlan defaultRunOptions settings fullPlan >>= requireRight - Prelude.drop 32 (reportOutcomes report) `shouldBe` replicate 9 AppliedNow + Prelude.drop 32 (reportOutcomes report) `shouldBe` replicate 10 AppliedNow withConnection settings $ \connection -> do rows <- useSession connection (Session.statement () legacyGroupUpgradeStatement) rows @@ -391,7 +391,7 @@ withConnection settings $ \connection -> useSession connection (Session.script preCanonicalRebuildFixtureSql) report <- runMigrationPlan defaultRunOptions settings fullPlan >>= requireRight - Prelude.drop 34 (reportOutcomes report) `shouldBe` replicate 7 AppliedNow + Prelude.drop 34 (reportOutcomes report) `shouldBe` replicate 8 AppliedNow withConnection settings $ \connection -> do rows <- useSession connection (Session.statement () preCanonicalRebuildShapeStatement) rows @@ -416,7 +416,7 @@ withConnection settings $ \connection -> useSession connection (Session.script preStatusContractFixtureSql) report <- runMigrationPlan defaultRunOptions settings fullPlan >>= requireRight - Prelude.drop 36 (reportOutcomes report) `shouldBe` replicate 5 AppliedNow + Prelude.drop 36 (reportOutcomes report) `shouldBe` replicate 6 AppliedNow withConnection settings $ \connection -> do facts <- useSession connection (Session.statement () preStatusContractFactsStatement) facts `shouldBe` ("unmanaged", 0, "unmanaged", True, True) @@ -439,7 +439,7 @@ withConnection settings $ \connection -> useSession connection (Session.statement () guardFunctionOidStatement) report <- runMigrationPlan defaultRunOptions settings fullPlan >>= requireRight - Prelude.drop 40 (reportOutcomes report) `shouldBe` [AppliedNow] + Prelude.drop 40 (reportOutcomes report) `shouldBe` replicate 2 AppliedNow withConnection settings $ \connection -> do (afterOid, epochFenced, publicRevoked) <- useSession connection (Session.statement () upgradedGuardFactsStatement) @@ -447,6 +447,35 @@ epochFenced `shouldBe` True publicRevoked `shouldBe` True + it "0031 adds bounded terminal rejection audit constraints to existing outbox rows" $ do + fullPlan <- requirePlan + kiroku <- requireRight Kiroku.kirokuMigrations + priorKeiro <- + requireRight + ( migrationComponentFromEmbeddedSql + "keiro" + (Set.singleton "kiroku") + (NonEmpty.fromList (take 30 (toList embeddedMigrationEntries))) + ) + priorPlan <- requireRight (frameworkMigrationPlan kiroku priorKeiro) + withKeiroPg $ \database -> do + let settings = Pg.connectionSettings database + _ <- runMigrationPlan defaultRunOptions settings priorPlan >>= requireRight + withConnection settings $ \connection -> + useSession connection (Session.script preRejectionOutboxFixtureSql) + report <- runMigrationPlan defaultRunOptions settings fullPlan >>= requireRight + Prelude.drop 41 (reportOutcomes report) `shouldBe` [AppliedNow] + withConnection settings $ \connection -> do + useSession connection (Session.script validRejectionAuditSql) + missingAudit <- Connection.use connection (Session.script missingRejectionAuditSql) + missingAudit `shouldSatisfy` isLeft + invalidCode <- Connection.use connection (Session.script invalidRejectionCodeSql) + invalidCode `shouldSatisfy` isLeft + oversizedDetail <- Connection.use connection (Session.script oversizedRejectionDetailSql) + oversizedDetail `shouldSatisfy` isLeft + auditOnSent <- Connection.use connection (Session.script rejectionAuditOnSentSql) + auditOnSent `shouldSatisfy` isLeft + it "enforces replay source, adapter, and verification membership constraints" $ do plan <- requirePlan result <- withMigratedDatabase plan $ \connection -> do @@ -608,7 +637,7 @@ `shouldBe` replicate 7 AlreadyApplied <> replicate 4 AppliedNow <> replicate 16 AlreadyApplied - <> replicate 14 AppliedNow + <> replicate 15 AppliedNow verifiedAfterUp <- verifyMigrationPlan defaultRunOptions settings plan >>= requireRight @@ -808,6 +837,72 @@ ('upgrade-run-failed', 'upgrade-failed', repeat('b', 64), 'contract-v2:' || repeat('d', 64), 'keiro/projection-replay/v2', 6, 100, 'failed', now(), 'operator.abandoned', 'abandoned with the old binary'); """ +preRejectionOutboxFixtureSql :: Text +preRejectionOutboxFixtureSql = + """ + INSERT INTO keiro.keiro_outbox + (outbox_id, message_id, source, destination, event_type, schema_version, + content_type, payload_bytes, occurred_at, status) + SELECT + ('018f0f18-0000-7000-8000-000000000b0' || ordinal::text)::uuid, + 'migration-rejection-' || ordinal::text, + 'migration-test', + 'sink', + 'Fixture', + 1, + 'application/json', + '{}'::bytea, + now(), + 'publishing' + FROM generate_series(1, 5) AS ordinal; + """ + +validRejectionAuditSql :: Text +validRejectionAuditSql = + """ + UPDATE keiro.keiro_outbox + SET status = 'rejected', + rejected_at = now(), + rejection_code = 'unsupported.sink', + rejection_detail = 'the sink cannot accept this event' + WHERE message_id = 'migration-rejection-1'; + """ + +missingRejectionAuditSql :: Text +missingRejectionAuditSql = + """ + UPDATE keiro.keiro_outbox + SET status = 'rejected' + WHERE message_id = 'migration-rejection-2'; + """ + +invalidRejectionCodeSql :: Text +invalidRejectionCodeSql = + """ + UPDATE keiro.keiro_outbox + SET status = 'rejected', rejected_at = now(), rejection_code = 'Invalid Code' + WHERE message_id = 'migration-rejection-3'; + """ + +oversizedRejectionDetailSql :: Text +oversizedRejectionDetailSql = + """ + UPDATE keiro.keiro_outbox + SET status = 'rejected', + rejected_at = now(), + rejection_code = 'invalid.payload', + rejection_detail = repeat('é', 513) + WHERE message_id = 'migration-rejection-4'; + """ + +rejectionAuditOnSentSql :: Text +rejectionAuditOnSentSql = + """ + UPDATE keiro.keiro_outbox + SET status = 'sent', rejected_at = now(), rejection_code = 'invalid.destination' + WHERE message_id = 'migration-rejection-5'; + """ + preStatusContractFixtureSql :: Text preStatusContractFixtureSql = """ @@ -979,12 +1074,12 @@ `shouldBe` replicate 7 AlreadyApplied <> replicate 4 AppliedNow <> replicate 16 AlreadyApplied - <> replicate 14 AppliedNow + <> replicate 15 AppliedNow verifiedAfterCanaries <- verifyMigrationPlan defaultRunOptions settings plan >>= requireRight case verifiedAfterCanaries of VerificationReport verificationIssues _ _ _ -> verificationIssues `shouldBe` [] rerun <- runMigrationPlan defaultRunOptions settings plan >>= requireRight - reportOutcomes rerun `shouldBe` replicate 41 AlreadyApplied + reportOutcomes rerun `shouldBe` replicate 42 AlreadyApplied second <- importCoddHistory defaultImportOptions config provider plan frameworkCoddHistoryMappings >>= requireRight @@ -994,7 +1089,7 @@ sourceRows <- useSession connection (Session.statement () (sourceRowCountStatement sourceSchema)) sourceRows `shouldBe` 23 facts <- useSession connection (Session.statement () importFactsStatement) - facts `shouldBe` (41, 23, True) + facts `shouldBe` (42, 23, True) postCoddImportPendingIssues :: IO [VerificationIssue] postCoddImportPendingIssues = @@ -1021,7 +1116,8 @@ ("keiro", "0027"), ("keiro", "0028"), ("keiro", "0029"), - ("keiro", "0030") + ("keiro", "0030"), + ("keiro", "0031") ] assertPoisonedLedger :: Settings.Settings -> Expectation @@ -1064,7 +1160,8 @@ "0027.sql", "0028.sql", "0029.sql", - "0030.sql" + "0030.sql", + "0031.sql" ] findMigrationsDirectory :: IO FilePath