diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,37 @@ # Changelog +## 0.6.0.0 — 2026-09-25 + +### Breaking Changes + +* New forward migration `0012` adds `kiroku.stream_events.category`, the + source stream's category on every `$all` junction row, so category reads can + range-scan an index instead of probing every stream in the category (BUG-2). + It backfills the column on existing `$all` rows from `streams.category`, + adds `ck_stream_events_all_category` (`stream_id <> 0 OR category IS NOT + NULL`), and builds the partial index `ix_stream_events_all_by_category + (category, stream_version) INCLUDE (original_stream_id) WHERE stream_id = + 0`. kiroku-store 0.9.0.0 requires it. +* Any code that inserts `$all` junction rows directly must now set `category`; + the check constraint rejects the old five-column shape with SQLSTATE `23514`. + That includes kiroku-store 0.8 and older, whose appends fail against a + migrated schema: stop old writers before applying `0012`, and start + kiroku-store 0.9 after it. No rolling deploy is possible across this + migration. + +### Other Changes + +* `0012` runs in one transaction. Its backfill rewrites every `$all` row and + temporarily disables the `no_update_stream_events` trigger inside that + transaction; appends block until it commits. On a large store apply it in a + maintenance window and run `VACUUM (ANALYZE) kiroku.stream_events` + afterwards. +* The test-suite gains an upgrade case that applies `0012` to a store + populated with pre-`0012` junction rows and checks the backfill, the + constraint, the index definition, and the re-enabled trigger. The BUG-1 + upgrade case now bootstraps through `0009` explicitly, so `0010` stays in its + pending tail. + ## 0.5.0.0 — 2026-09-18 ### Breaking Changes 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 eleven SQL payloads, so applications can compose it with other libraries +and twelve SQL payloads, so applications can compose it with other libraries without copying Kiroku SQL. The first seven payloads are immutable historical -Codd bytes; `0008` through `0011` are native-only forward migrations. +Codd bytes; `0008` through `0012` are native-only forward migrations. ## Public API @@ -141,7 +141,10 @@ coordinator, an indexed active-lease predicate, and statement-level `DELETE`/`TRUNCATE` guards on the three event-store data tables. Migration `0011` converges databases that applied the withdrawn 0.3.2.x payload of `0010` -(see below). +(see below). Migration `0012` copies each stream's category onto its `$all` +junction rows and indexes it for category reads; it rewrites every `$all` row +in one transaction, so apply it in a maintenance window on a large store (see +`docs/user/schema-migrations.md`). ## The `kiroku.uuidv7()` generator 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.5.0.0 +version: 0.6.0.0 synopsis: Schema migrations for kiroku-store description: Native pg-migrate component, Codd history mapping, and migration executable diff --git a/migrations/0012.sql b/migrations/0012.sql new file mode 100644 --- /dev/null +++ b/migrations/0012.sql @@ -0,0 +1,51 @@ +-- denormalize category onto $all junction rows for category reads + +-- Category reads (readCategoryForwardSQL and its consumer-group variant in +-- kiroku-store) used to start from every stream of the category and probe +-- stream_events once per stream, so a caught-up poll cost work proportional to +-- the number of streams ever written in the category (BUG-2). Carrying the +-- originating stream's category on each $all junction row lets both reads run +-- as one index range scan from (category, checkpoint) that stops at the limit. +-- +-- The column is populated only on $all rows (stream_id = 0). Home rows and link +-- rows keep it NULL; nothing reads them by category. The CHECK below makes any +-- inserter that forgets the column fail loudly instead of writing rows that a +-- category read cannot see. +-- +-- The whole file runs in one transaction. The backfill rewrites every $all row +-- and the index build blocks writes, so appends wait for the duration; apply it +-- in a maintenance window on a large store and VACUUM (ANALYZE) +-- kiroku.stream_events afterwards. + +ALTER TABLE kiroku.stream_events + ADD COLUMN category TEXT; + +COMMENT ON COLUMN kiroku.stream_events.category IS + 'Originating stream''s category, present on $all rows (stream_id = 0) only; equals streams.category of original_stream_id.'; + +-- Backfill every existing $all row from its originating stream. The immutability +-- trigger rejects every UPDATE on this table, so it is suspended for this one +-- statement and re-enabled before the transaction ends. Runs as the table owner. +ALTER TABLE kiroku.stream_events DISABLE TRIGGER no_update_stream_events; + +UPDATE kiroku.stream_events AS se +SET category = s.category +FROM kiroku.streams AS s +WHERE se.stream_id = 0 + AND s.stream_id = se.original_stream_id; + +ALTER TABLE kiroku.stream_events ENABLE TRIGGER no_update_stream_events; + +ALTER TABLE kiroku.stream_events + ADD CONSTRAINT ck_stream_events_all_category + CHECK (stream_id <> 0 OR category IS NOT NULL); + +-- Category read path: rows of one category in global-position order. The +-- INCLUDE column lets the consumer-group hash predicate run on index tuples. +CREATE INDEX ix_stream_events_all_by_category + ON kiroku.stream_events (category, stream_version) + INCLUDE (original_stream_id) + WHERE stream_id = 0; + +COMMENT ON SCHEMA kiroku IS + 'Managed by pg-migrate component kiroku through 0012'; diff --git a/migrations/manifest b/migrations/manifest --- a/migrations/manifest +++ b/migrations/manifest @@ -9,3 +9,4 @@ 0009.sql 0010.sql 0011.sql +0012.sql diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -44,7 +44,7 @@ main :: IO () main = hspec $ do describe "native Kiroku migration definition" $ do - it "tracks the eleven native files in manifest order" $ do + it "tracks the twelve native files in manifest order" $ do directory <- findMigrationsDirectory manifest <- Text.lines <$> Text.IO.readFile (directory "manifest") manifest `shouldBe` Text.pack <$> nativeMigrationFiles @@ -57,7 +57,7 @@ bytes <- ByteString.readFile (directory nativeName) lookup legacyName lockEntries `shouldBe` Just (checksumText bytes) - it "builds component kiroku and an eleven-migration plan" $ do + it "builds component kiroku and a twelve-migration plan" $ do component <- requireRight kirokuMigrations component `seq` pure () plan <- requirePlan @@ -93,7 +93,7 @@ `shouldReturn` "0007-existing.sql\n" describe "fresh native databases" $ do - it "applies all eleven, verifies strictly, and reports AlreadyApplied on rerun" $ do + it "applies all twelve, verifies strictly, and reports AlreadyApplied on rerun" $ do plan <- requirePlan result <- withMigratedDatabase plan $ \connection -> do assertSchema connection @@ -284,14 +284,15 @@ describe "upgrades of already-bootstrapped databases" $ do it "applies the pending tail in a session that never ran the bootstrap" $ do plan <- requirePlan - throughBootstrap <- planThrough (length nativeMigrationFiles - 2) + let bootstrapCount = 9 -- through 0009, so 0010 is in the pending tail + throughBootstrap <- planThrough bootstrapCount withKirokuPg $ \database -> do let settings = Pg.connectionSettings database bootstrapped <- runMigrationPlan defaultRunOptions settings throughBootstrap >>= requireMigration reportOutcomes bootstrapped - `shouldBe` replicate (length nativeMigrationFiles - 2) AppliedNow + `shouldBe` replicate bootstrapCount AppliedNow -- A separate session, and one that cannot reach the Kiroku -- schema through search_path. The suite connects as role @@ -305,8 +306,8 @@ runMigrationPlanWith defaultRunOptions (providerFor upgradeSession) plan >>= requireMigration reportOutcomes upgraded - `shouldBe` replicate (length nativeMigrationFiles - 2) AlreadyApplied - <> replicate 2 AppliedNow + `shouldBe` replicate bootstrapCount AlreadyApplied + <> replicate (length nativeMigrationFiles - bootstrapCount) AppliedNow verified <- verifyMigrationPlan defaultRunOptions settings plan >>= requireMigration case verified of @@ -315,6 +316,37 @@ length applied `shouldBe` length nativeMigrationFiles withConnection settings assertSchema + -- BUG-2. 0012 copies each $all row's originating-stream category onto the + -- junction row so category reads can range-scan an index. A store + -- written before 0012 has $all rows without it; the backfill must fill + -- every one, and afterwards the CHECK must refuse a new $all row that + -- omits it, the immutability trigger must be back on, and the index must + -- exist in exactly the shape the category reads rely on. + it "backfills $all-row categories when 0012 upgrades a populated store" $ do + plan <- requirePlan + let throughCount = length nativeMigrationFiles - 1 + throughPrevious <- planThrough throughCount + withKirokuPg $ \database -> do + let settings = Pg.connectionSettings database + _ <- runMigrationPlan defaultRunOptions settings throughPrevious >>= requireMigration + withConnection settings $ \connection -> + useSession connection (Session.script preCategoryFixtureSql) + upgraded <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration + reportOutcomes upgraded + `shouldBe` replicate throughCount AlreadyApplied <> [AppliedNow] + withConnection settings $ \connection -> do + facts <- useSession connection (Session.statement () categoryBackfillFactsStatement) + facts + `shouldBe` ( True + , 0 + , True + , Just "CREATE INDEX ix_stream_events_all_by_category ON kiroku.stream_events USING btree (category, stream_version) INCLUDE (original_stream_id) WHERE (stream_id = 0)" + , "O" + ) + missingCategory <- Connection.use connection (Session.statement () insertAllRowWithoutCategoryStatement) + missingCategory `shouldSatisfy` hasSqlState "23514" + assertSchema connection + -- kiroku.uuidv7() is the component's version-independent generator, but -- it arrives by a different route on each major: 0001's fallback on -- PostgreSQL 17, 0010's alias for the builtin on PostgreSQL 18. This @@ -391,14 +423,14 @@ pendingIds <- traverse (requireRight . migrationId "kiroku") - ["0008-schema-management-comment", "0009", "0010", "0011"] + ["0008-schema-management-comment", "0009", "0010", "0011", "0012"] 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 4 AppliedNow + reportOutcomes up `shouldBe` replicate 7 AlreadyApplied <> replicate 5 AppliedNow verifiedAfterCanary <- verifyMigrationPlan defaultRunOptions settings plan >>= requireMigration case verifiedAfterCanary of VerificationReport verificationIssues _ _ _ -> @@ -429,6 +461,7 @@ , "0009.sql" , "0010.sql" , "0011.sql" + , "0012.sql" ] {- | The plan truncated to its first @count@ migrations, read from the checked-in @@ -568,13 +601,88 @@ (to_regclass('kiroku.dead_letters') IS NOT NULL), (EXISTS (SELECT 1 FROM pg_catalog.pg_trigger WHERE tgname = 'stream_events_notify_insert' AND NOT tgisinternal)), (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_indexes WHERE schemaname = 'kiroku' AND indexname = 'ix_stream_events_all_by_category')), + (EXISTS (SELECT 1 FROM pg_catalog.pg_constraint WHERE conname = 'ck_stream_events_all_category')), (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 0011') + (obj_description(to_regnamespace('kiroku'), 'pg_namespace') = 'Managed by pg-migrate component kiroku through 0012') ) AS checks(ok) """ Encoders.noParams (Decoders.singleRow (Decoders.column (Decoders.nonNullable Decoders.bool))) + +{- | One stream with one event, written with the five-column junction shape +every release before 0012 used: a home row and an @$all@ row, no category. +-} +preCategoryFixtureSql :: Text +preCategoryFixtureSql = + """ + INSERT INTO kiroku.streams (stream_name, stream_version) VALUES ('upgrade-1', 1); + INSERT INTO kiroku.events (event_id, event_type, data) + VALUES ('00000000-0000-7000-8000-000000000001', 'UpgradeFixture', '{}'::jsonb); + INSERT INTO kiroku.stream_events + (event_id, stream_id, stream_version, original_stream_id, original_stream_version) + SELECT '00000000-0000-7000-8000-000000000001'::uuid, s.stream_id, 1, s.stream_id, 1 + FROM kiroku.streams AS s WHERE s.stream_name = 'upgrade-1' + UNION ALL + SELECT '00000000-0000-7000-8000-000000000001'::uuid, 0, 1, s.stream_id, 1 + FROM kiroku.streams AS s WHERE s.stream_name = 'upgrade-1'; + UPDATE kiroku.streams SET stream_version = 1 WHERE stream_id = 0; + """ + +{- | After 0012: whether the fixture's @$all@ row carries its stream's category, +how many @$all@ rows lack one, whether the home row stayed NULL, the category +index definition, and the immutability trigger's enabled state. +-} +categoryBackfillFactsStatement :: Statement () (Bool, Int64, Bool, Maybe Text, Text) +categoryBackfillFactsStatement = + Statement.preparable + """ + SELECT (SELECT se.category = s.category + FROM kiroku.stream_events AS se + JOIN kiroku.streams AS s ON s.stream_id = se.original_stream_id + WHERE se.stream_id = 0 AND s.stream_name = 'upgrade-1'), + (SELECT count(*) FROM kiroku.stream_events + WHERE stream_id = 0 AND category IS NULL), + (SELECT se.category IS NULL + FROM kiroku.stream_events AS se + JOIN kiroku.streams AS s ON s.stream_id = se.stream_id + WHERE s.stream_name = 'upgrade-1'), + (SELECT indexdef FROM pg_catalog.pg_indexes + WHERE schemaname = 'kiroku' AND indexname = 'ix_stream_events_all_by_category'), + (SELECT tgenabled::text FROM pg_catalog.pg_trigger + WHERE tgrelid = 'kiroku.stream_events'::regclass + AND tgname = 'no_update_stream_events') + """ + Encoders.noParams + ( Decoders.singleRow + ( (,,,,) + <$> Decoders.column (Decoders.nonNullable Decoders.bool) + <*> Decoders.column (Decoders.nonNullable Decoders.int8) + <*> Decoders.column (Decoders.nonNullable Decoders.bool) + <*> Decoders.column (Decoders.nullable Decoders.text) + <*> Decoders.column (Decoders.nonNullable Decoders.text) + ) + ) + +-- | An @$all@ row written the pre-0012 way, which the CHECK must now refuse. +insertAllRowWithoutCategoryStatement :: Statement () () +insertAllRowWithoutCategoryStatement = + Statement.preparable + """ + WITH new_event AS ( + INSERT INTO kiroku.events (event_id, event_type, data) + VALUES ('00000000-0000-7000-8000-000000000002', 'UpgradeFixture', '{}'::jsonb) + RETURNING event_id + ) + INSERT INTO kiroku.stream_events + (event_id, stream_id, stream_version, original_stream_id, original_stream_version) + SELECT new_event.event_id, 0, 2, s.stream_id, 2 + FROM new_event, kiroku.streams AS s + WHERE s.stream_name = 'upgrade-1' + """ + Encoders.noParams + Decoders.noResult {- | The server major, whether each UUIDv7 generator exists, and the stored @lease_id@ default. Together these pin which route published