diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,20 @@ # Changelog +## 0.7.0.0 — 2026-10-10 + +### Breaking Changes + +* Migration 0014 adds checked, unindexed subscription target-binding columns and + removes the unused `stream_name` column. Stop subscription workers before + applying the cohort's migrations and restart with kiroku-store 0.10.0.0. + Constant defaults do not rewrite legacy rows. + +### New Features + +* Migration 0013 derives persisted consumer-group size from existing member rows. + Incomplete legacy groups refuse on next startup until explicitly resized to + the intended topology. Both migrations are included in the embedded manifest. + ## 0.6.0.0 — 2026-09-25 ### Breaking Changes diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,11 @@ +Copyright (c) 2026 Nadeem Bitar. + +Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: + +1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. + +2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. + +3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. 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.6.0.0 +version: 0.7.0.0 synopsis: Schema migrations for kiroku-store description: Native pg-migrate component, Codd history mapping, and migration executable @@ -11,6 +11,7 @@ author: Nadeem Bitar maintainer: nadeem@gmail.com license: BSD-3-Clause +license-file: LICENSE build-type: Simple category: Database, Eventing extra-doc-files: CHANGELOG.md diff --git a/migrations/0013.sql b/migrations/0013.sql new file mode 100644 --- /dev/null +++ b/migrations/0013.sql @@ -0,0 +1,13 @@ +-- Derive consumer-group topology from existing member checkpoints. +-- Stop all subscription workers before applying this migration. +-- An incomplete legacy row set derives an underestimate: startup refuses the +-- configured topology until the operator explicitly equalizes it with resize. +UPDATE kiroku.subscriptions AS s +SET consumer_group_size = derived.size +FROM ( + SELECT subscription_name, max(consumer_group_member) + 1 AS size + FROM kiroku.subscriptions + GROUP BY subscription_name +) AS derived +WHERE s.subscription_name = derived.subscription_name + AND s.consumer_group_size <> derived.size; diff --git a/migrations/0014.sql b/migrations/0014.sql new file mode 100644 --- /dev/null +++ b/migrations/0014.sql @@ -0,0 +1,11 @@ +-- bind subscription checkpoints to their targets + + +-- Stop subscription workers before applying this catalog change and CHECK scan. +-- Constant defaults preserve legacy rows without a heap rewrite. +ALTER TABLE kiroku.subscriptions + ADD COLUMN target_kind TEXT NOT NULL DEFAULT 'unbound' + CHECK (target_kind IN ('unbound', 'all', 'category')), + ADD COLUMN target_category TEXT + CHECK ((target_kind = 'category') = (target_category IS NOT NULL)), + DROP COLUMN stream_name; diff --git a/migrations/manifest b/migrations/manifest --- a/migrations/manifest +++ b/migrations/manifest @@ -10,3 +10,5 @@ 0010.sql 0011.sql 0012.sql +0013.sql +0014.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 twelve native files in manifest order" $ do + it "tracks the fourteen 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 a twelve-migration plan" $ do + it "builds component kiroku and a fourteen-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 twelve, verifies strictly, and reports AlreadyApplied on rerun" $ do + it "applies all fourteen, verifies strictly, and reports AlreadyApplied on rerun" $ do plan <- requirePlan result <- withMigratedDatabase plan $ \connection -> do assertSchema connection @@ -316,6 +316,45 @@ length applied `shouldBe` length nativeMigrationFiles withConnection settings assertSchema + it "adds unbound target metadata without rewriting legacy rows and drops stream_name" $ do + plan <- requirePlan + throughPrevious <- planThrough 13 + withKirokuPg $ \database -> do + let settings = Pg.connectionSettings database + _ <- runMigrationPlan defaultRunOptions settings throughPrevious >>= requireMigration + withConnection settings $ \connection -> do + useSession connection (Session.script "INSERT INTO kiroku.subscriptions (subscription_name, last_seen) VALUES ('legacy-target', 7)") + before <- useSession connection (Session.statement () targetUpgradeFactsStatement) + _ <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration + after <- useSession connection (Session.statement () targetUpgradeFactsStatement) + -- Relation filenode and tuple ctid prove no heap rewrite. + after `shouldBe` before + facts <- useSession connection (Session.statement () targetBindingFactsStatement) + facts `shouldBe` ("unbound", Nothing, 7, False) + + it "derives group topology from legacy rows when 0013 upgrades a populated store" $ do + plan <- requirePlan + throughPrevious <- planThrough 12 + withKirokuPg $ \database -> do + let settings = Pg.connectionSettings database + _ <- runMigrationPlan defaultRunOptions settings throughPrevious >>= requireMigration + withConnection settings $ \connection -> + useSession connection (Session.script "INSERT INTO kiroku.subscriptions (subscription_name, consumer_group_member, last_seen) VALUES ('group', 0, 5), ('group', 1, 20), ('incomplete', 0, 8), ('ordinary', 0, 10)") + upgraded <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration + reportOutcomes upgraded `shouldBe` replicate 12 AlreadyApplied <> replicate 2 AppliedNow + withConnection settings $ \connection -> do + let stmt = + Statement.preparable + "SELECT subscription_name, consumer_group_member, consumer_group_size, last_seen FROM kiroku.subscriptions ORDER BY subscription_name, consumer_group_member" + Encoders.noParams + (Decoders.rowList ((,,,) <$> Decoders.column (Decoders.nonNullable Decoders.text) <*> Decoders.column (Decoders.nonNullable Decoders.int4) <*> Decoders.column (Decoders.nonNullable Decoders.int4) <*> Decoders.column (Decoders.nonNullable Decoders.int8))) + facts <- useSession connection (Session.statement () stmt) + facts `shouldBe` [("group" :: Text, 0 :: Int32, 2 :: Int32, 5 :: Int64), ("group", 1, 2, 20), ("incomplete", 0, 1, 8), ("ordinary", 0, 1, 10)] + directory <- findMigrationsDirectory + sql <- Text.IO.readFile (directory "0013.sql") + useSession connection (Session.script sql) + useSession connection (Session.statement () stmt) `shouldReturn` facts + -- 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 @@ -324,7 +363,7 @@ -- 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 + let throughCount = 11 throughPrevious <- planThrough throughCount withKirokuPg $ \database -> do let settings = Pg.connectionSettings database @@ -333,7 +372,7 @@ useSession connection (Session.script preCategoryFixtureSql) upgraded <- runMigrationPlan defaultRunOptions settings plan >>= requireMigration reportOutcomes upgraded - `shouldBe` replicate throughCount AlreadyApplied <> [AppliedNow] + `shouldBe` replicate throughCount AlreadyApplied <> replicate (length nativeMigrationFiles - throughCount) AppliedNow withConnection settings $ \connection -> do facts <- useSession connection (Session.statement () categoryBackfillFactsStatement) facts @@ -423,14 +462,14 @@ pendingIds <- traverse (requireRight . migrationId "kiroku") - ["0008-schema-management-comment", "0009", "0010", "0011", "0012"] + ["0008-schema-management-comment", "0009", "0010", "0011", "0012", "0013", "0014"] 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 5 AppliedNow + reportOutcomes up `shouldBe` replicate 7 AlreadyApplied <> replicate (length nativeMigrationFiles - 7) AppliedNow verifiedAfterCanary <- verifyMigrationPlan defaultRunOptions settings plan >>= requireMigration case verifiedAfterCanary of VerificationReport verificationIssues _ _ _ -> @@ -462,6 +501,8 @@ , "0010.sql" , "0011.sql" , "0012.sql" + , "0013.sql" + , "0014.sql" ] {- | The plan truncated to its first @count@ migrations, read from the checked-in @@ -1200,3 +1241,21 @@ ) where column = Decoders.column . Decoders.nonNullable + +targetUpgradeFactsStatement :: Statement () (Int64, Text) +targetUpgradeFactsStatement = + Statement.preparable + "SELECT pg_relation_filenode('kiroku.subscriptions')::bigint, ctid::text FROM kiroku.subscriptions WHERE subscription_name = 'legacy-target'" + Encoders.noParams + (Decoders.singleRow ((,) <$> Decoders.column (Decoders.nonNullable Decoders.int8) <*> Decoders.column (Decoders.nonNullable Decoders.text))) + +targetBindingFactsStatement :: Statement () (Text, Maybe Text, Int64, Bool) +targetBindingFactsStatement = + Statement.preparable + """ + SELECT target_kind, target_category, last_seen, + EXISTS (SELECT 1 FROM information_schema.columns WHERE table_schema = 'kiroku' AND table_name = 'subscriptions' AND column_name = 'stream_name') + FROM kiroku.subscriptions WHERE subscription_name = 'legacy-target' + """ + Encoders.noParams + (Decoders.singleRow ((,,,) <$> Decoders.column (Decoders.nonNullable Decoders.text) <*> Decoders.column (Decoders.nullable Decoders.text) <*> Decoders.column (Decoders.nonNullable Decoders.int8) <*> Decoders.column (Decoders.nonNullable Decoders.bool)))