diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/bench/Main.hs b/bench/Main.hs --- a/bench/Main.hs +++ b/bench/Main.hs @@ -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 () diff --git a/kiroku-store.cabal b/kiroku-store.cabal --- a/kiroku-store.cabal +++ b/kiroku-store.cabal @@ -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 diff --git a/src/Kiroku/Store/Effect.hs b/src/Kiroku/Store/Effect.hs --- a/src/Kiroku/Store/Effect.hs +++ b/src/Kiroku/Store/Effect.hs @@ -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 -> diff --git a/src/Kiroku/Store/Subscription.hs b/src/Kiroku/Store/Subscription.hs --- a/src/Kiroku/Store/Subscription.hs +++ b/src/Kiroku/Store/Subscription.hs @@ -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 diff --git a/src/Kiroku/Store/Subscription/CheckpointInventory/SQL.hs b/src/Kiroku/Store/Subscription/CheckpointInventory/SQL.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Store/Subscription/CheckpointInventory/SQL.hs @@ -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" diff --git a/src/Kiroku/Store/Subscription/Types.hs b/src/Kiroku/Store/Subscription/Types.hs --- a/src/Kiroku/Store/Subscription/Types.hs +++ b/src/Kiroku/Store/Subscription/Types.hs @@ -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 diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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 diff --git a/test/Test/SubscriptionCheckpointInventory.hs b/test/Test/SubscriptionCheckpointInventory.hs new file mode 100644 --- /dev/null +++ b/test/Test/SubscriptionCheckpointInventory.hs @@ -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 + ] diff --git a/test/Test/SubscriptionCheckpointInventoryMock.hs b/test/Test/SubscriptionCheckpointInventoryMock.hs new file mode 100644 --- /dev/null +++ b/test/Test/SubscriptionCheckpointInventoryMock.hs @@ -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"