kiroku-store 0.3.1.0 → 0.4.0.0

raw patch · 10 files changed

+651/−140 lines, 10 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +24/−0
• bench/Main.hs +198/−138
• kiroku-store.cabal +4/−1
• src/Kiroku/Store/Effect.hs +12/−0
• src/Kiroku/Store/Subscription.hs +27/−0
• src/Kiroku/Store/Subscription/CheckpointInventory/SQL.hs +114/−0
• src/Kiroku/Store/Subscription/Types.hs +25/−1
• test/Main.hs +4/−0
• test/Test/SubscriptionCheckpointInventory.hs +198/−0
• test/Test/SubscriptionCheckpointInventoryMock.hs +45/−0

@@ -2,6 +2,30 @@  ## Unreleased +## 0.4.0.0 — 2026-08-09++### Breaking Changes++* The exported `Store` effect gains `GetSubscriptionCheckpointInventory`.+  Exhaustive custom and mock interpreters must handle the new constructor.++### New Features++* `subscriptionCheckpointInventory` returns the captured global store position+  and every persisted, member-aware subscription checkpoint from one prepared+  PostgreSQL statement snapshot. Results are ordered by subscription name and+  consumer-group member and remain available after a worker stops.+* New public `SubscriptionCheckpoint` and+  `SubscriptionCheckpointInventory` records expose the durable checkpoint+  position, member, last upsert time, and captured store position without+  requiring consumers to import Hasql or query Kiroku-owned tables.++### Other Changes++* Added integration and mock-interpreter coverage, user and Haddock+  documentation, PostgreSQL query-plan evidence, and fully materialized+  100-row and 10,000-row performance benchmarks for the inventory operation.+ ## 0.3.1.0 — 2026-07-22  ### New Features

@@ -13,7 +13,7 @@ import Data.Functor.Contravariant ((>$<)) import Data.Generics.Labels () import Data.IORef-import Data.Int (Int64)+import Data.Int (Int32, Int64) import Data.Maybe (isNothing) import Data.Text (Text) import Data.Text qualified as T@@ -34,7 +34,7 @@ import Kiroku.Store import Kiroku.Store.Effect (buildAppendParams, prepareEvents) import Kiroku.Store.SQL qualified as SQL-import Kiroku.Test.Postgres (migrateTestDatabase)+import Kiroku.Test.Postgres (migrateTestDatabase, withMigratedTestDatabase, withSharedMigratedPostgres) import Test.Tasty.Bench  data RawAppendParams = RawAppendParams@@ -763,6 +763,60 @@ forceRead (Right v) = V.length v `seq` pure () forceRead (Left e) = error ("Benchmark read failed: " <> show e) +seedCheckpointInventoryStmt :: Statement Int32 ()+seedCheckpointInventoryStmt =+    preparable+        """+        INSERT INTO subscriptions (+            subscription_name,+            consumer_group_member,+            last_seen,+            updated_at+        )+        SELECT 'inventory-' || lpad(n::text, 8, '0'),+               0,+               0,+               now()+        FROM generate_series(1, $1::int4) AS n+        """+        (E.param (E.nonNullable E.int4))+        D.noResult++seedCheckpointInventory :: KirokuStore -> Int32 -> IO ()+seedCheckpointInventory store rowCount = do+    result <- Pool.use (store ^. #pool) $ Session.statement rowCount seedCheckpointInventoryStmt+    case result of+        Left err -> error ("Checkpoint inventory benchmark setup failed: " <> show err)+        Right () -> pure ()++forceCheckpointInventory :: Either StoreError SubscriptionCheckpointInventory -> IO ()+forceCheckpointInventory (Left err) = error ("Checkpoint inventory benchmark failed: " <> show err)+forceCheckpointInventory (Right (SubscriptionCheckpointInventory (GlobalPosition storePos) rows)) = do+    let !checksum = V.foldl' forceCheckpoint (fromIntegral storePos) rows+    checksum `seq` pure ()+  where+    forceCheckpoint !acc (SubscriptionCheckpoint (SubscriptionName name) member (GlobalPosition position) updatedAt) =+        updatedAt `seq`+            acc+                + T.length name+                + fromIntegral member+                + fromIntegral position++runCheckpointInventoryBenchmark :: KirokuStore -> IO ()+runCheckpointInventoryBenchmark store =+    runStoreIO store subscriptionCheckpointInventory >>= forceCheckpointInventory++withInventoryBenchmarkStores :: (KirokuStore -> KirokuStore -> IO a) -> IO a+withInventoryBenchmarkStores action =+    withSharedMigratedPostgres $+        withMigratedTestDatabase $ \inventory100Connection ->+            withMigratedTestDatabase $ \inventory10000Connection ->+                withStore (defaultConnectionSettings inventory100Connection) $ \inventory100Store ->+                    withStore (defaultConnectionSettings inventory10000Connection) $ \inventory10000Store -> do+                        seedCheckpointInventory inventory100Store 100+                        seedCheckpointInventory inventory10000Store 10_000+                        action inventory100Store inventory10000Store+ main :: IO () main = do     -- Start ephemeral PostgreSQL once for all benchmarks@@ -875,155 +929,161 @@             subCounter <- newIORef (0 :: Int)             pipelinedContentionCounter <- newIORef (0 :: Int) -            defaultMain-                [ bgroup-                    "append"+            withInventoryBenchmarkStores $ \inventory100Store inventory10000Store ->+                defaultMain                     [ bgroup-                        "single-event"-                        [ bench "NoStream (new stream)" $ whnfIO $ do-                            sn <- nextStream "bench-single"-                            r' <- runStoreIO store $ appendToStream sn NoStream [makeEvent "BenchEvent"]-                            forceAppend r'-                        , bench "AnyVersion (new stream)" $ whnfIO $ do-                            sn <- nextStream "bench-any"-                            r' <- runStoreIO store $ appendToStream sn AnyVersion [makeEvent "BenchEvent"]-                            forceAppend r'-                        ]-                    , bgroup-                        "batch-10"-                        [ bench "NoStream" $ whnfIO $ do-                            sn <- nextStream "bench-b10"-                            let events = map (\i -> makeEvent ("E" <> T.pack (show i))) [1 .. 10 :: Int]-                            r' <- runStoreIO store $ appendToStream sn NoStream events-                            forceAppend r'+                        "append"+                        [ bgroup+                            "single-event"+                            [ bench "NoStream (new stream)" $ whnfIO $ do+                                sn <- nextStream "bench-single"+                                r' <- runStoreIO store $ appendToStream sn NoStream [makeEvent "BenchEvent"]+                                forceAppend r'+                            , bench "AnyVersion (new stream)" $ whnfIO $ do+                                sn <- nextStream "bench-any"+                                r' <- runStoreIO store $ appendToStream sn AnyVersion [makeEvent "BenchEvent"]+                                forceAppend r'+                            ]+                        , bgroup+                            "batch-10"+                            [ bench "NoStream" $ whnfIO $ do+                                sn <- nextStream "bench-b10"+                                let events = map (\i -> makeEvent ("E" <> T.pack (show i))) [1 .. 10 :: Int]+                                r' <- runStoreIO store $ appendToStream sn NoStream events+                                forceAppend r'+                            ]+                        , bgroup+                            "batch-100"+                            [ bench "NoStream" $ whnfIO $ do+                                sn <- nextStream "bench-b100"+                                let events = map (\i -> makeEvent ("E" <> T.pack (show i))) [1 .. 100 :: Int]+                                r' <- runStoreIO store $ appendToStream sn NoStream events+                                forceAppend r'+                            ]+                        , bgroup+                            "sequential"+                            [ bench "10 appends to same stream" $ whnfIO $ do+                                sn <- nextStream "bench-seq"+                                r0 <- runStoreIO store $ appendToStream sn NoStream [makeEvent "Init"]+                                forceAppend r0+                                let Right res0 = r0+                                let go _ 0 = pure ()+                                    go v n = do+                                        r' <- runStoreIO store $ appendToStream sn (ExactVersion v) [makeEvent "Seq"]+                                        case r' of+                                            Right res -> go (res ^. #streamVersion) (n - 1 :: Int)+                                            Left e -> error ("Sequential append failed: " <> show e)+                                go (res0 ^. #streamVersion) 9+                            ]                         ]                     , bgroup-                        "batch-100"-                        [ bench "NoStream" $ whnfIO $ do-                            sn <- nextStream "bench-b100"-                            let events = map (\i -> makeEvent ("E" <> T.pack (show i))) [1 .. 100 :: Int]-                            r' <- runStoreIO store $ appendToStream sn NoStream events-                            forceAppend r'+                        "raw-append-shape"+                        [ bgroup+                            "AnyVersion"+                            [ bench "scalar singleton (new stream)" $+                                whnfIO $+                                    runRawScalarAppendAnyVersionNewStream store rawCounter+                            , bench "production arrays/unnest (new stream)" $+                                whnfIO $+                                    runRawProductionAppendAnyVersionNewStream store rawCounter+                            , bench "two-roundtrip (new stream)" $+                                whnfIO $+                                    runRawTwoRoundtripAppendNewStream store rawCounter+                            , bench "two-roundtrip + BEGIN/COMMIT (new stream)" $+                                whnfIO $+                                    runRawTwoRoundtripAppendNewStreamTx store rawCounter+                            , bench "scalar singleton (hot stream)" $+                                whnfIO $+                                    runRawScalarAppendAnyVersionHotStream store+                            , bench "production arrays/unnest (hot stream)" $+                                whnfIO $+                                    runRawProductionAppendAnyVersionHotStream store+                            , bench "two-roundtrip (hot stream)" $+                                whnfIO $+                                    runRawTwoRoundtripAppendExistingHotStream store+                            , bench "two-roundtrip + BEGIN/COMMIT (hot stream)" $+                                whnfIO $+                                    runRawTwoRoundtripAppendExistingHotStreamTx store+                            ]                         ]                     , bgroup-                        "sequential"-                        [ bench "10 appends to same stream" $ whnfIO $ do-                            sn <- nextStream "bench-seq"-                            r0 <- runStoreIO store $ appendToStream sn NoStream [makeEvent "Init"]-                            forceAppend r0-                            let Right res0 = r0-                            let go _ 0 = pure ()-                                go v n = do-                                    r' <- runStoreIO store $ appendToStream sn (ExactVersion v) [makeEvent "Seq"]-                                    case r' of-                                        Right res -> go (res ^. #streamVersion) (n - 1 :: Int)-                                        Left e -> error ("Sequential append failed: " <> show e)-                            go (res0 ^. #streamVersion) 9-                        ]-                    ]-                , bgroup-                    "raw-append-shape"-                    [ bgroup-                        "AnyVersion"-                        [ bench "scalar singleton (new stream)" $-                            whnfIO $-                                runRawScalarAppendAnyVersionNewStream store rawCounter-                        , bench "production arrays/unnest (new stream)" $-                            whnfIO $-                                runRawProductionAppendAnyVersionNewStream store rawCounter-                        , bench "two-roundtrip (new stream)" $+                        "pipelined-multi-append"+                        [ bench "current shape (4 streams)" $                             whnfIO $-                                runRawTwoRoundtripAppendNewStream store rawCounter-                        , bench "two-roundtrip + BEGIN/COMMIT (new stream)" $+                                runCurrentMultiAppend store pipelinedMultiAppend4Streams+                        , bench "pipelined (4 streams)" $                             whnfIO $-                                runRawTwoRoundtripAppendNewStreamTx store rawCounter-                        , bench "scalar singleton (hot stream)" $+                                runPipelinedMultiAppend store pipelinedMultiAppend4Streams+                        , bench "current shape (8 streams)" $                             whnfIO $-                                runRawScalarAppendAnyVersionHotStream store-                        , bench "production arrays/unnest (hot stream)" $+                                runCurrentMultiAppend store pipelinedMultiAppend8Streams+                        , bench "pipelined (8 streams)" $                             whnfIO $-                                runRawProductionAppendAnyVersionHotStream store-                        , bench "two-roundtrip (hot stream)" $+                                runPipelinedMultiAppend store pipelinedMultiAppend8Streams+                        , bench "single-stream under 4 current multi-stream writers" $                             whnfIO $-                                runRawTwoRoundtripAppendExistingHotStream store-                        , bench "two-roundtrip + BEGIN/COMMIT (hot stream)" $+                                runSingleAppendUnderMultiWriters+                                    store+                                    pipelinedContentionCounter+                                    runCurrentMultiAppend+                                    "pipe-current"+                                    pipelinedContentionCurrentGroups+                        , bench "single-stream under 4 pipelined multi-stream writers" $                             whnfIO $-                                runRawTwoRoundtripAppendExistingHotStreamTx store+                                runSingleAppendUnderMultiWriters+                                    store+                                    pipelinedContentionCounter+                                    runPipelinedMultiAppend+                                    "pipe-pipelined"+                                    pipelinedContentionPipelinedGroups                         ]-                    ]-                , bgroup-                    "pipelined-multi-append"-                    [ bench "current shape (4 streams)" $-                        whnfIO $-                            runCurrentMultiAppend store pipelinedMultiAppend4Streams-                    , bench "pipelined (4 streams)" $-                        whnfIO $-                            runPipelinedMultiAppend store pipelinedMultiAppend4Streams-                    , bench "current shape (8 streams)" $-                        whnfIO $-                            runCurrentMultiAppend store pipelinedMultiAppend8Streams-                    , bench "pipelined (8 streams)" $-                        whnfIO $-                            runPipelinedMultiAppend store pipelinedMultiAppend8Streams-                    , bench "single-stream under 4 current multi-stream writers" $-                        whnfIO $-                            runSingleAppendUnderMultiWriters-                                store-                                pipelinedContentionCounter-                                runCurrentMultiAppend-                                "pipe-current"-                                pipelinedContentionCurrentGroups-                    , bench "single-stream under 4 pipelined multi-stream writers" $-                        whnfIO $-                            runSingleAppendUnderMultiWriters-                                store-                                pipelinedContentionCounter-                                runPipelinedMultiAppend-                                "pipe-pipelined"-                                pipelinedContentionPipelinedGroups-                    ]-                , bgroup-                    "read"-                    [ bench "stream forward (100-event page)" $ whnfIO $ do-                        r' <- runStoreIO store $ readStreamForward readStreamName (StreamVersion 0) 100-                        forceRead r'-                    , bench "$all forward (100-event page)" $ whnfIO $ do-                        r' <- runStoreIO store $ readAllForward (GlobalPosition 0) 100-                        forceRead r'-                    ]-                , bgroup-                    "category"-                    [ bench "category forward (100-event page)" $ whnfIO $ do-                        -- Read from cat1 category (has 10 streams × 100 events = 1000 events)-                        r' <- runStoreIO store $ readCategory (CategoryName "cat1") (GlobalPosition 0) 100-                        forceRead r'-                    , bench "exhausted-category" $ whnfIO $ do-                        -- cat1 events are inserted early in setup; a high cursor proves-                        -- category reads do not scan the rest of $all looking for matches.-                        r' <- runStoreIO store $ readCategory (CategoryName "cat1") (GlobalPosition 90_000) 100-                        forceRead r'-                    , bench "$all forward (100-event page, baseline)" $ whnfIO $ do-                        r' <- runStoreIO store $ readAllForward (GlobalPosition 0) 100-                        forceRead r'-                    ]-                , -- F19 — Concurrent-writer stress as structured benchmarks.-                  -- The legacy ad-hoc B9 measurement (still present above-                  -- for historical comparability) prints throughput and-                  -- latency once; these bgroup entries surface the same-                  -- workload through tasty-bench so it participates in the-                  -- baseline-regression workflow (Justfile bench-regression).-                  bgroup-                    "concurrent"-                    [ bench "8 writers x 10 appends" $ whnfIO $ runConcurrentWriters store concCounter 8 10-                    , bench "32 writers x 10 appends" $ whnfIO $ runConcurrentWriters store concCounter 32 10-                    ]-                , bgroup-                    "reliability-audit"-                    [ bench "hot invoice-payment 10 AnyVersion appends" $ whnfIO $ runHotInvoicePayment store 10-                    , bench "appendMultiStream 3 existing streams" $ whnfIO $ runAppendMultiStream store-                    , bench "subscription category catch-up 100 events" $ whnfIO $ runSubscriptionCatchup store subCounter+                    , bgroup+                        "read"+                        [ bench "stream forward (100-event page)" $ whnfIO $ do+                            r' <- runStoreIO store $ readStreamForward readStreamName (StreamVersion 0) 100+                            forceRead r'+                        , bench "$all forward (100-event page)" $ whnfIO $ do+                            r' <- runStoreIO store $ readAllForward (GlobalPosition 0) 100+                            forceRead r'+                        ]+                    , bgroup+                        "category"+                        [ bench "category forward (100-event page)" $ whnfIO $ do+                            -- Read from cat1 category (has 10 streams × 100 events = 1000 events)+                            r' <- runStoreIO store $ readCategory (CategoryName "cat1") (GlobalPosition 0) 100+                            forceRead r'+                        , bench "exhausted-category" $ whnfIO $ do+                            -- cat1 events are inserted early in setup; a high cursor proves+                            -- category reads do not scan the rest of $all looking for matches.+                            r' <- runStoreIO store $ readCategory (CategoryName "cat1") (GlobalPosition 90_000) 100+                            forceRead r'+                        , bench "$all forward (100-event page, baseline)" $ whnfIO $ do+                            r' <- runStoreIO store $ readAllForward (GlobalPosition 0) 100+                            forceRead r'+                        ]+                    , -- F19 — Concurrent-writer stress as structured benchmarks.+                      -- The legacy ad-hoc B9 measurement (still present above+                      -- for historical comparability) prints throughput and+                      -- latency once; these bgroup entries surface the same+                      -- workload through tasty-bench so it participates in the+                      -- baseline-regression workflow (Justfile bench-regression).+                      bgroup+                        "concurrent"+                        [ bench "8 writers x 10 appends" $ whnfIO $ runConcurrentWriters store concCounter 8 10+                        , bench "32 writers x 10 appends" $ whnfIO $ runConcurrentWriters store concCounter 32 10+                        ]+                    , bgroup+                        "reliability-audit"+                        [ bench "hot invoice-payment 10 AnyVersion appends" $ whnfIO $ runHotInvoicePayment store 10+                        , bench "appendMultiStream 3 existing streams" $ whnfIO $ runAppendMultiStream store+                        , bench "subscription category catch-up 100 events" $ whnfIO $ runSubscriptionCatchup store subCounter+                        ]+                    , bgroup+                        "subscription-checkpoint-inventory"+                        [ bench "100 rows" $ whnfIO $ runCheckpointInventoryBenchmark inventory100Store+                        , bench "10000 rows" $ whnfIO $ runCheckpointInventoryBenchmark inventory10000Store+                        ]                     ]-                ]     case result of         Left err -> error ("Failed to start ephemeral PostgreSQL: " <> show err)         Right () -> pure ()

@@ -1,6 +1,6 @@ cabal-version:   3.0 name:            kiroku-store-version:         0.3.1.0+version:         0.4.0.0 synopsis:        High-performance PostgreSQL event store description:   Kiroku is a PostgreSQL-backed event store for Haskell applications. It@@ -55,6 +55,7 @@     Kiroku.Store.Transaction     Kiroku.Store.Types +  other-modules:   Kiroku.Store.Subscription.CheckpointInventory.SQL   build-depends:     , aeson                 >=2.1  && <2.3     , async                 >=2.2  && <2.3@@ -108,6 +109,8 @@     Test.StartupFailureSurfacing     Test.StreamBridgeTermination     Test.StreamNameLookup+    Test.SubscriptionCheckpointInventory+    Test.SubscriptionCheckpointInventoryMock     Test.SubscriptionPauseResume     Test.SubscriptionReconnect     Test.SubscriptionRegistry

@@ -56,6 +56,8 @@ import Kiroku.Store.Observability (KirokuEvent (..)) import Kiroku.Store.SQL qualified as SQL import Kiroku.Store.Settings (decodeEvents, enrichEvents)+import Kiroku.Store.Subscription.CheckpointInventory.SQL qualified as CheckpointInventorySQL+import Kiroku.Store.Subscription.Types (SubscriptionCheckpointInventory) import Kiroku.Store.Types  -- ---------------------------------------------------------------------------@@ -123,6 +125,13 @@     'Kiroku.Store.Lifecycle.clearStreamTruncateBefore'.     -}     SetStreamTruncateBefore :: StreamName -> StreamVersion -> Store m (Maybe StreamId)+    {- | Read the global store position and all durable subscription checkpoint+    rows from one PostgreSQL statement snapshot.++    Surfaced as+    'Kiroku.Store.Subscription.subscriptionCheckpointInventory'.+    -}+    GetSubscriptionCheckpointInventory :: Store m SubscriptionCheckpointInventory     {- | Run an arbitrary @hasql-transaction@ value in a 'BEGIN'/'COMMIT'     block on a single pool connection. Escape hatch from the abstract     'Store' effect into the underlying SQL world; mock interpreters are@@ -339,6 +348,9 @@         rejectInvalidApplicationStream name         usePool (store ^. #pool) $             Session.statement (name, v) SQL.setStreamTruncateBeforeStmt+    GetSubscriptionCheckpointInventory ->+        usePool (store ^. #pool) $+            Session.statement () CheckpointInventorySQL.getSubscriptionCheckpointInventoryStmt     RunTransaction tx ->         runTxOnPool (store ^. #pool) TxSessions.transaction tx     RunTransactionNoRetry tx ->

@@ -4,6 +4,7 @@     withSubscription,      -- * Observability+    subscriptionCheckpointInventory,     subscriptionStates,     SubscriptionStateView (..), @@ -25,8 +26,12 @@ import Data.Map.Strict qualified as Map import Data.Text (Text) import Data.Unique (newUnique)+import Effectful (Eff, (:>))+import Effectful.Dispatch.Dynamic (send) import GHC.Generics (Generic)+import GHC.Stack (HasCallStack) import Kiroku.Store.Connection (KirokuStore (..))+import Kiroku.Store.Effect (Store (GetSubscriptionCheckpointInventory)) import Kiroku.Store.Notification qualified as Notifier import Kiroku.Store.Subscription.EventPublisher qualified as Pub import Kiroku.Store.Subscription.Fsm (SubscriptionState (..), stateCursor, stateName)@@ -218,6 +223,28 @@     m a withSubscription store config action = withRunInIO $ \runInIO ->     bracket (subscribe store config) cancel (runInIO . action)++{- | Read the durable checkpoint inventory and global store position captured by+one PostgreSQL statement snapshot.++An empty 'checkpoints' vector means that no subscription checkpoint has yet+been written. Unlike 'subscriptionStates', stopped subscriptions remain in this+inventory because their committed rows are durable. Rows are sorted by+subscription name and consumer-group member, and a new call is required to+observe commits made after this snapshot.++A live worker cursor may be ahead of its durable checkpoint while work is in+flight. Member zero alone does not reveal whether a subscription is ungrouped+or is member zero of a group. 'checkpointUpdatedAt' records the last successful+checkpoint write, not proof that its position advanced. Finally, subtracting a+'checkpointPosition' from 'storePosition' yields a global position distance,+not an exact count of relevant events for filtered, category, or sharded+consumers.+-}+subscriptionCheckpointInventory ::+    (HasCallStack, Store :> es) =>+    Eff es SubscriptionCheckpointInventory+subscriptionCheckpointInventory = send GetSubscriptionCheckpointInventory  {- | A public, point-in-time view of one live subscription's state, as returned by 'subscriptionStates'. This is the committed observability surface external

@@ -0,0 +1,114 @@+{-# LANGUAGE MultilineStrings #-}++module Kiroku.Store.Subscription.CheckpointInventory.SQL (+    getSubscriptionCheckpointInventoryStmt,+) where++import Data.Int (Int32, Int64)+import Data.Text (Text)+import Data.Time.Clock (UTCTime)+import Data.Vector (Vector)+import Data.Vector qualified as V+import Hasql.Decoders qualified as D+import Hasql.Encoders qualified as E+import Hasql.Statement (Statement, preparable)+import Hasql.Statement qualified as Statement+import Kiroku.Store.Subscription.Types (+    SubscriptionCheckpoint (..),+    SubscriptionCheckpointInventory (..),+    SubscriptionName (..),+ )+import Kiroku.Store.Types (GlobalPosition (..))++data InventoryRow = InventoryRow+    { rowStorePosition :: !Int64+    , rowSubscriptionName :: !(Maybe Text)+    , rowConsumerGroupMember :: !(Maybe Int32)+    , rowCheckpointPosition :: !(Maybe Int64)+    , rowCheckpointUpdatedAt :: !(Maybe UTCTime)+    }++getSubscriptionCheckpointInventoryStmt :: Statement () SubscriptionCheckpointInventory+getSubscriptionCheckpointInventoryStmt =+    Statement.refineResult finalizeInventory $+        preparable+            """+            SELECT store_head.stream_version,+                   checkpoint.subscription_name,+                   checkpoint.consumer_group_member,+                   checkpoint.last_seen,+                   checkpoint.updated_at+            FROM streams AS store_head+            LEFT JOIN subscriptions AS checkpoint ON TRUE+            WHERE store_head.stream_id = 0+            ORDER BY checkpoint.subscription_name ASC,+                     checkpoint.consumer_group_member ASC+            """+            E.noParams+            (D.rowVector inventoryRow)++inventoryRow :: D.Row InventoryRow+inventoryRow =+    InventoryRow+        <$> D.column (D.nonNullable D.int8)+        <*> D.column (D.nullable D.text)+        <*> D.column (D.nullable D.int4)+        <*> D.column (D.nullable D.int8)+        <*> D.column (D.nullable D.timestamptz)++finalizeInventory :: Vector InventoryRow -> Either Text SubscriptionCheckpointInventory+finalizeInventory rows = case V.uncons rows of+    Nothing -> Left "subscription checkpoint inventory: missing $all stream row"+    Just (firstRow, remainingRows) ->+        let capturedPosition = rowStorePosition firstRow+         in case checkpointColumns firstRow of+                EmptyCheckpoint+                    | V.null remainingRows ->+                        Right $+                            SubscriptionCheckpointInventory+                                (GlobalPosition capturedPosition)+                                V.empty+                    | otherwise ->+                        Left "subscription checkpoint inventory: empty checkpoint row was not the only result"+                PartialCheckpoint ->+                    Left "subscription checkpoint inventory: partially null checkpoint row"+                CompleteCheckpoint ->+                    SubscriptionCheckpointInventory (GlobalPosition capturedPosition)+                        <$> V.mapM (decodeCheckpoint capturedPosition) rows++data CheckpointColumns+    = EmptyCheckpoint+    | PartialCheckpoint+    | CompleteCheckpoint++checkpointColumns :: InventoryRow -> CheckpointColumns+checkpointColumns row =+    case ( rowSubscriptionName row+         , rowConsumerGroupMember row+         , rowCheckpointPosition row+         , rowCheckpointUpdatedAt row+         ) of+        (Nothing, Nothing, Nothing, Nothing) -> EmptyCheckpoint+        (Just _, Just _, Just _, Just _) -> CompleteCheckpoint+        _ -> PartialCheckpoint++decodeCheckpoint :: Int64 -> InventoryRow -> Either Text SubscriptionCheckpoint+decodeCheckpoint capturedPosition row+    | rowStorePosition row /= capturedPosition =+        Left "subscription checkpoint inventory: inconsistent repeated store position"+    | otherwise =+        case ( rowSubscriptionName row+             , rowConsumerGroupMember row+             , rowCheckpointPosition row+             , rowCheckpointUpdatedAt row+             ) of+            (Just name, Just member, Just position, Just updatedAt) ->+                Right $+                    SubscriptionCheckpoint+                        (SubscriptionName name)+                        member+                        (GlobalPosition position)+                        updatedAt+            (Nothing, Nothing, Nothing, Nothing) ->+                Left "subscription checkpoint inventory: unexpected empty checkpoint row"+            _ -> Left "subscription checkpoint inventory: partially null checkpoint row"

@@ -18,6 +18,8 @@ -} module Kiroku.Store.Subscription.Types (     SubscriptionName (..),+    SubscriptionCheckpoint (..),+    SubscriptionCheckpointInventory (..),     SubscriptionTarget (..),     SubscriptionResult (..),     OverflowPolicy (..),@@ -55,6 +57,9 @@ import Data.Set (Set) import Data.Set qualified as Set import Data.Text (Text)+import Data.Time.Clock (UTCTime)+import Data.Vector (Vector)+import GHC.Generics (Generic) import Kiroku.Store.Subscription.Fsm (     DeadLetterReason (..),     RetryDelay (..),@@ -63,7 +68,7 @@     deadLetterSummary,     retryDelayMicros,  )-import Kiroku.Store.Types (CategoryName, EventType, RecordedEvent (..))+import Kiroku.Store.Types (CategoryName, EventType, GlobalPosition, RecordedEvent (..)) import Numeric.Natural (Natural)  {- | A declarative, closed filter over event types for a subscription.@@ -135,6 +140,25 @@ -- | Unique name for a subscription (e.g., @"inventory-projection"@). newtype SubscriptionName = SubscriptionName Text     deriving newtype (Eq, Ord, Show)++-- | One checkpoint row that has been durably persisted by a subscription.+data SubscriptionCheckpoint = SubscriptionCheckpoint+    { subscriptionName :: !SubscriptionName+    , consumerGroupMember :: !Int32+    , checkpointPosition :: !GlobalPosition+    , checkpointUpdatedAt :: !UTCTime+    }+    deriving stock (Eq, Show, Generic)++{- | A point-in-time view of the global store position and every durable+subscription checkpoint. The rows are ordered by 'subscriptionName' and then+'consumerGroupMember'.+-}+data SubscriptionCheckpointInventory = SubscriptionCheckpointInventory+    { storePosition :: !GlobalPosition+    , checkpoints :: !(Vector SubscriptionCheckpoint)+    }+    deriving stock (Eq, Show, Generic)  -- | Which stream to subscribe to. data SubscriptionTarget

@@ -49,6 +49,8 @@ import Test.StartupFailureSurfacing qualified as StartupFailureSurfacing import Test.StreamBridgeTermination qualified as StreamBridgeTermination import Test.StreamNameLookup qualified as StreamNameLookup+import Test.SubscriptionCheckpointInventory qualified as SubscriptionCheckpointInventory+import Test.SubscriptionCheckpointInventoryMock qualified as SubscriptionCheckpointInventoryMock import Test.SubscriptionPauseResume qualified as SubscriptionPauseResume import Test.SubscriptionReconnect qualified as SubscriptionReconnect import Test.SubscriptionRegistry qualified as SubscriptionRegistry@@ -80,6 +82,8 @@     PublisherRestartNoRebroadcast.spec     CatchupDbErrorNoPrematureSwitch.spec     SubscriptionPauseResume.spec+    SubscriptionCheckpointInventory.spec+    SubscriptionCheckpointInventoryMock.spec     SubscriptionReconnect.spec     StartupFailureSurfacing.spec     SubscriptionState.spec

@@ -0,0 +1,198 @@+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE TypeApplications #-}++module Test.SubscriptionCheckpointInventory (spec) where++import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)+import Control.Lens ((&), (.~), (^.))+import Data.Aeson qualified as Aeson+import Data.Generics.Labels ()+import Data.Int (Int32, Int64)+import Data.Map.Strict qualified as Map+import Data.Text (Text)+import Data.Text qualified as T+import Data.Time.Clock (getCurrentTime)+import Data.Vector qualified as V+import Effectful (runEff)+import Effectful.Error.Static (runErrorNoCallStack)+import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import Kiroku.Store+import Kiroku.Store.SQL qualified as SQL+import System.Timeout (timeout)+import Test.Helpers (caughtUpEventHandler, insertDeadLetterForEvent, makeEvent, waitForPublisher, waitForSubscriptionLive, waitWithTimeout, withTestStore, withTestStoreSettings)+import Test.Hspec++spec :: Spec+spec = describe "SubscriptionCheckpointInventory" $ do+    it "returns position zero and no rows for an empty migrated store" $+        withTestStore $ \store -> do+            SubscriptionCheckpointInventory captured rows <- readInventory store+            captured `shouldBe` GlobalPosition 0+            rows `shouldBe` V.empty++    it "runs through the resource-backed Store interpreter" $+        withTestStore $ \store -> do+            result <-+                runEff+                    . runErrorNoCallStack @StoreError+                    . runKirokuStoreWith store+                    . runStoreResource+                    $ subscriptionCheckpointInventory+            result `shouldBe` Right (SubscriptionCheckpointInventory (GlobalPosition 0) V.empty)++    it "returns the exact store position and a member-zero checkpoint" $+        withTestStore $ \store -> do+            appendEvents store "inventory-single" 3+            saveCheckpoint store "single" 0 2++            SubscriptionCheckpointInventory captured rows <- readInventory store+            captured `shouldBe` GlobalPosition 3+            checkpointKeys rows `shouldBe` [("single", 0, 2)]+            now <- getCurrentTime+            case V.toList rows of+                [SubscriptionCheckpoint _ _ _ updatedAt] -> updatedAt `shouldSatisfy` (<= now)+                _ -> expectationFailure "expected exactly one checkpoint"++    it "returns multiple names and members in deterministic key order" $+        withTestStore $ \store -> do+            appendEvents store "inventory-many" 20+            saveCheckpoint store "zeta" 2 7+            saveCheckpoint store "alpha" 10 3+            saveCheckpoint store "alpha" 2 5++            SubscriptionCheckpointInventory captured rows <- readInventory store+            captured `shouldBe` GlobalPosition 20+            checkpointKeys rows+                `shouldBe` [ ("alpha", 2, 5)+                           , ("alpha", 10, 3)+                           , ("zeta", 2, 7)+                           ]++    it "preserves monotonic checkpoints and observes later commits on a fresh read" $+        withTestStore $ \store -> do+            appendEvents store "inventory-monotonic" 10+            saveCheckpoint store "monotonic" 0 8+            saveCheckpoint store "monotonic" 0 4++            first <- readInventory store+            inventoryKeys first `shouldBe` [("monotonic", 0, 8)]++            saveCheckpoint store "monotonic" 0 9+            second <- readInventory store+            inventoryKeys second `shouldBe` [("monotonic", 0, 9)]++    it "retains a durable row after the worker stops and leaves live state" $+        withTestStore $ \store -> do+            appendEvents store "inventory-stopped" 1+            waitForPublisher store (GlobalPosition 1)+            let name = SubscriptionName "stopped"+            handle <- subscribe store (defaultSubscriptionConfig name AllStreams (\_ -> pure Stop))+            waitClean handle++            states <- subscriptionStates store+            Map.member (name, 0) states `shouldBe` False+            inventory <- readInventory store+            inventoryKeys inventory `shouldBe` [("stopped", 0, 1)]++    it "does not expose in-flight live handler progress before checkpoint commit" $ do+        caughtUp <- newEmptyMVar+        enteredHandler <- newEmptyMVar+        releaseHandler <- newEmptyMVar+        let name = SubscriptionName "in-flight"+            observe = caughtUpEventHandler name caughtUp Nothing+            handler _ = do+                putMVar enteredHandler ()+                takeMVar releaseHandler+                pure Stop+            config = defaultSubscriptionConfig name AllStreams handler+        withTestStoreSettings (& #eventHandler .~ Just observe) $ \store -> do+            appendEvents store "inventory-live" 1+            saveCheckpoint store "in-flight" 0 1+            waitForPublisher store (GlobalPosition 1)+            handle <- subscribe store config+            waitForSubscriptionLive caughtUp++            appendEventsExisting store "inventory-live" 1+            waitForMVar "live handler did not receive the event" enteredHandler+            beforeCommit <- readInventory store+            inventoryKeys beforeCommit `shouldBe` [("in-flight", 0, 1)]++            putMVar releaseHandler ()+            waitClean handle+            afterCommit <- readInventory store+            inventoryKeys afterCommit `shouldBe` [("in-flight", 0, 2)]++    it "observes the checkpoint advanced by a dead-letter transaction" $+        withTestStore $ \store -> do+            appendEvents store "inventory-dead-letter" 1+            Right events <- runStoreIO store $ readAllForward (GlobalPosition 0) 10+            let event = V.head events+            insertDeadLetterForEvent store "dead-lettered" event++            inventory <- readInventory store+            inventoryKeys inventory `shouldBe` [("dead-lettered", 0, 1)]++    it "captures a head at or beyond every normally written checkpoint" $+        withTestStore $ \store -> do+            appendEvents store "inventory-bounds" 6+            saveCheckpoint store "bounds-a" 0 2+            saveCheckpoint store "bounds-b" 1 6++            SubscriptionCheckpointInventory (GlobalPosition captured) rows <- readInventory store+            let positions = [position | SubscriptionCheckpoint _ _ (GlobalPosition position) _ <- V.toList rows]+            positions `shouldSatisfy` all (<= captured)++readInventory :: KirokuStore -> IO SubscriptionCheckpointInventory+readInventory store = do+    result <- runStoreIO store subscriptionCheckpointInventory+    case result of+        Left err -> error ("subscriptionCheckpointInventory failed: " <> show err)+        Right inventory -> pure inventory++saveCheckpoint :: KirokuStore -> Text -> Int32 -> Int64 -> IO ()+saveCheckpoint store name member position = do+    result <- Pool.use (store ^. #pool) $ Session.statement (name, member, position) SQL.saveCheckpointMemberStmt+    case result of+        Left err -> error ("saveCheckpoint failed: " <> show err)+        Right () -> pure ()++appendEvents :: KirokuStore -> Text -> Int -> IO ()+appendEvents store stream count = do+    let events = [makeEvent ("E" <> T.pack (show i)) (Aeson.object []) | i <- [1 .. count]]+    result <- runStoreIO store $ appendToStream (StreamName stream) NoStream events+    case result of+        Left err -> error ("appendEvents failed: " <> show err)+        Right _ -> pure ()++appendEventsExisting :: KirokuStore -> Text -> Int -> IO ()+appendEventsExisting store stream count = do+    let events = [makeEvent ("Live" <> T.pack (show i)) (Aeson.object []) | i <- [1 .. count]]+    result <- runStoreIO store $ appendToStream (StreamName stream) StreamExists events+    case result of+        Left err -> error ("appendEventsExisting failed: " <> show err)+        Right _ -> pure ()++waitClean :: SubscriptionHandle -> IO ()+waitClean handle = do+    result <- waitWithTimeout 20_000_000 handle+    case result of+        Left message -> expectationFailure message+        Right (Left err) -> expectationFailure ("subscription failed: " <> show err)+        Right (Right ()) -> pure ()++waitForMVar :: String -> MVar () -> IO ()+waitForMVar failureMessage var = do+    result <- timeout 5_000_000 (takeMVar var)+    case result of+        Nothing -> expectationFailure failureMessage+        Just () -> pure ()++inventoryKeys :: SubscriptionCheckpointInventory -> [(Text, Int32, Int64)]+inventoryKeys (SubscriptionCheckpointInventory _ rows) = checkpointKeys rows++checkpointKeys :: V.Vector SubscriptionCheckpoint -> [(Text, Int32, Int64)]+checkpointKeys rows =+    [ (name, member, position)+    | SubscriptionCheckpoint (SubscriptionName name) member (GlobalPosition position) _ <- V.toList rows+    ]

@@ -0,0 +1,45 @@+module Test.SubscriptionCheckpointInventoryMock (spec) where++import Control.Monad.IO.Class (liftIO)+import Data.IORef (IORef, modifyIORef', newIORef, readIORef)+import Data.Time.Calendar (fromGregorian)+import Data.Time.Clock (UTCTime (..))+import Data.Vector qualified as V+import Effectful (Eff, IOE, runEff, (:>))+import Effectful.Dispatch.Dynamic (interpret_)+import Kiroku.Store.Effect (Store (..))+import Kiroku.Store.Subscription (subscriptionCheckpointInventory)+import Kiroku.Store.Subscription.Types+import Kiroku.Store.Types (GlobalPosition (..))+import Test.Hspec++spec :: Spec+spec = describe "SubscriptionCheckpointInventory mock interpreter" $ do+    it "returns a public inventory through one Store effect call" $ do+        calls <- newIORef (0 :: Int)+        let updatedAt = UTCTime (fromGregorian 2026 8 9) 0+            expected =+                SubscriptionCheckpointInventory+                    (GlobalPosition 17)+                    ( V.singleton $+                        SubscriptionCheckpoint+                            (SubscriptionName "mock")+                            3+                            (GlobalPosition 11)+                            updatedAt+                    )+        actual <- runEff $ runInventoryMock calls expected subscriptionCheckpointInventory+        actual `shouldBe` expected+        readIORef calls `shouldReturn` 1++runInventoryMock ::+    (IOE :> es) =>+    IORef Int ->+    SubscriptionCheckpointInventory ->+    Eff (Store : es) a ->+    Eff es a+runInventoryMock calls expected = interpret_ $ \case+    GetSubscriptionCheckpointInventory -> do+        liftIO $ modifyIORef' calls (+ 1)+        pure expected+    _ -> error "unexpected Store operation in inventory mock"