diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,41 @@ # Changelog -## Unreleased +## 0.5.0.0 — 2026-08-11 + +### Breaking Changes + +* `SubscriptionConfigM` gains `missingCheckpointPolicy`, defaulting to + `FromBeginning` in `defaultSubscriptionConfig`. Exhaustive record literals + must choose a policy or switch to the smart constructor. +* The exported `Store` effect gains `InitializeSubscriptionCheckpoint`. + Exhaustive custom and mock interpreters must handle the new constructor. +* `KirokuEvent` gains `KirokuEventSubscriptionCheckpointResolved` and + `KirokuEventSubscriptionCheckpointMissing`. Exhaustive event handlers must + handle both startup lifecycle events. + +### New Features + +* Subscription startup now resolves an absent exact `(name, member)` checkpoint + through a closed `MissingCheckpointPolicy`: `FromBeginning` durably seeds + zero, `FromCurrentHead` atomically seeds the current `$all` position, and + `FailIfMissing` refuses startup with `SubscriptionCheckpointMissing` before + handler delivery. Existing rows always win and concurrent initializers + converge on the first committed row. +* `initializeSubscriptionCheckpoint` exposes the same closed initialization + contract through the mockable `Store` effect, returning an + `ExistingCheckpoint` or `InitializedCheckpoint` result with its exact key and + durable position. +* `Kiroku.Store.Subscription.Checkpoint` exposes + `resetSubscriptionCheckpointsTx`, an explicitly non-monotonic transaction + combinator that assigns one exact position to every persisted member of a + non-empty name set and returns sorted affected keys plus names with no rows. + Missing rows are never created, and ordinary worker saves remain monotonic. + +### Other Changes + +* Added native, Effectful, Streamly, consumer-group, Shibuya, mock-interpreter, + startup-race, rollback, and ordinary-save monotonicity coverage for the + checkpoint lifecycle contract. ## 0.4.0.0 — 2026-08-09 diff --git a/bench/Main.hs b/bench/Main.hs --- a/bench/Main.hs +++ b/bench/Main.hs @@ -3,18 +3,13 @@ module Main where -import Control.Concurrent (threadDelay) import Control.Concurrent.Async (mapConcurrently_) -import Control.Concurrent.Async qualified as Async -import Control.Exception (finally) import Control.Lens ((^.)) -import Control.Monad (replicateM_) import Data.Aeson qualified as Aeson import Data.Functor.Contravariant ((>$<)) import Data.Generics.Labels () import Data.IORef import Data.Int (Int32, Int64) -import Data.Maybe (isNothing) import Data.Text (Text) import Data.Text qualified as T import Data.Time.Clock (UTCTime, diffUTCTime, getCurrentTime) @@ -25,14 +20,12 @@ import GHC.Generics (Generic) import Hasql.Decoders qualified as D import Hasql.Encoders qualified as E -import Hasql.Pipeline qualified as Pipeline import Hasql.Pool qualified as Pool import Hasql.Session qualified as Session -import Hasql.Statement (Statement, preparable, unpreparable) +import Hasql.Statement (Statement, preparable) import Hasql.Transaction qualified as Tx import Hasql.Transaction.Sessions qualified as TxSessions import Kiroku.Store -import Kiroku.Store.Effect (buildAppendParams, prepareEvents) import Kiroku.Store.SQL qualified as SQL import Kiroku.Test.Postgres (migrateTestDatabase, withMigratedTestDatabase, withSharedMigratedPostgres) import Test.Tasty.Bench @@ -107,45 +100,6 @@ rawProductionAppendParamsEncoder rawAppendResultDecoder -beginStmt :: Statement () () -beginStmt = unpreparable "BEGIN" E.noParams D.noResult - -commitStmt :: Statement () () -commitStmt = unpreparable "COMMIT" E.noParams D.noResult - -rollbackStmt :: Statement () () -rollbackStmt = unpreparable "ROLLBACK" E.noParams D.noResult - -pipelinedMultiAppend4Streams :: [(StreamName, Text)] -pipelinedMultiAppend4Streams = - namedBenchStreams "pipe4" 4 - -pipelinedMultiAppend8Streams :: [(StreamName, Text)] -pipelinedMultiAppend8Streams = - namedBenchStreams "pipe8" 8 - -pipelinedContentionCurrentGroups :: [[(StreamName, Text)]] -pipelinedContentionCurrentGroups = - namedBenchGroups "pipe-current-writer" 4 4 - -pipelinedContentionPipelinedGroups :: [[(StreamName, Text)]] -pipelinedContentionPipelinedGroups = - namedBenchGroups "pipe-pipelined-writer" 4 4 - -namedBenchGroups :: Text -> Int -> Int -> [[(StreamName, Text)]] -namedBenchGroups prefix groups streamsPerGroup = - [ namedBenchStreams (prefix <> "-" <> T.pack (show groupId)) streamsPerGroup - | groupId <- [1 .. groups] - ] - -namedBenchStreams :: Text -> Int -> [(StreamName, Text)] -namedBenchStreams prefix count = - [ ( StreamName (prefix <> "-" <> T.pack (show i)) - , prefix <> "-event-" <> T.pack (show i) - ) - | i <- [1 .. count] - ] - rawScalarAppendAnyVersionSQL :: Text rawScalarAppendAnyVersionSQL = """ @@ -645,58 +599,6 @@ ] forceAppendList r -runCurrentMultiAppend :: KirokuStore -> [(StreamName, Text)] -> IO () -runCurrentMultiAppend store streams = do - r <- - runStoreIO store $ - appendMultiStream - [ (sn, AnyVersion, [makeEvent typ]) - | (sn, typ) <- streams - ] - forceAppendList r - -runPipelinedMultiAppend :: KirokuStore -> [(StreamName, Text)] -> IO () -runPipelinedMultiAppend store streams = do - now <- getCurrentTime - params <- - mapM - ( \(StreamName name, typ) -> do - prepared <- prepareEvents [makeEvent typ] - pure (buildAppendParams name now prepared) - ) - streams - let names = V.fromList [name | (StreamName name, _) <- streams] - result <- - Pool.use (store ^. #pool) $ do - results <- - Session.pipeline $ - Pipeline.statement () beginStmt - *> Pipeline.statement names SQL.lockStreamsForMultiStmt - *> traverse (`Pipeline.statement` SQL.appendAnyVersion) params - Session.statement () $ - if any isNothing results - then rollbackStmt - else commitStmt - pure results - forcePipelinedAppendList result - -runSingleAppendUnderMultiWriters :: - KirokuStore -> - IORef Int -> - (KirokuStore -> [(StreamName, Text)] -> IO ()) -> - Text -> - [[(StreamName, Text)]] -> - IO () -runSingleAppendUnderMultiWriters store runCounter multiAppendRunner prefix groups = do - runId <- atomicModifyIORef' runCounter (\m -> (m + 1, m)) - workers <- mapM (Async.async . replicateM_ 20 . multiAppendRunner store) groups - let cleanup = mapM_ Async.cancel workers - appendMeasured i = do - let sn = StreamName (prefix <> "-single-" <> T.pack (show runId) <> "-" <> T.pack (show i)) - r <- runStoreIO store $ appendToStream sn AnyVersion [makeEvent (prefix <> "-single")] - forceAppend r - (threadDelay 20_000 >> mapM_ appendMeasured [1 .. 10 :: Int] >> mapM_ Async.wait workers) `finally` cleanup - -- | Exercise subscription catch-up over a compact category-local backlog. runSubscriptionCatchup :: KirokuStore -> IORef Int -> IO () runSubscriptionCatchup store runCounter = do @@ -721,6 +623,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -741,16 +644,6 @@ forceAppendList (Right rs) = mapM_ (\r -> (r ^. #streamVersion) `seq` (r ^. #globalPosition) `seq` pure ()) rs forceAppendList (Left e) = error ("Benchmark appendMultiStream failed: " <> show e) -forcePipelinedAppendList :: Either Pool.UsageError [Maybe AppendResult] -> IO () -forcePipelinedAppendList (Right rs) = - mapM_ - ( \mResult -> case mResult of - Just r -> (r ^. #streamVersion) `seq` (r ^. #globalPosition) `seq` pure () - Nothing -> error "Pipelined appendMultiStream benchmark produced no result" - ) - rs -forcePipelinedAppendList (Left e) = error ("Pipelined appendMultiStream benchmark failed: " <> show e) - -- | Force evaluation of a raw append shape result or fail the benchmark. forceRawAppend :: Either Pool.UsageError (Maybe RawAppendResult) -> IO () forceRawAppend (Right (Just (!streamId, !streamVersion, !globalPosition))) = @@ -874,12 +767,7 @@ r' <- runStoreIO store $ appendToStream sn NoStream [makeEvent "Init"] forceAppend r' ) - ( [StreamName "bench-multi-a", StreamName "bench-multi-b", StreamName "bench-multi-c"] - <> map fst pipelinedMultiAppend4Streams - <> map fst pipelinedMultiAppend8Streams - <> map fst (concat pipelinedContentionCurrentGroups) - <> map fst (concat pipelinedContentionPipelinedGroups) - ) + [StreamName "bench-multi-a", StreamName "bench-multi-b", StreamName "bench-multi-c"] -- Pre-create the hot stream targeted by the two-round-trip raw -- shape variant. rawAppendUpdateExisting requires the row to @@ -927,7 +815,6 @@ concCounter <- newIORef (0 :: Int) rawCounter <- newIORef (0 :: Int) subCounter <- newIORef (0 :: Int) - pipelinedContentionCounter <- newIORef (0 :: Int) withInventoryBenchmarkStores $ \inventory100Store inventory10000Store -> defaultMain @@ -1007,37 +894,6 @@ ] ] , 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 @@ -1057,7 +913,7 @@ -- 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 + , bench "$all forward (100-event page baseline)" $ whnfIO $ do r' <- runStoreIO store $ readAllForward (GlobalPosition 0) 100 forceRead r' ] diff --git a/bench/RegressionGate.hs b/bench/RegressionGate.hs new file mode 100644 --- /dev/null +++ b/bench/RegressionGate.hs @@ -0,0 +1,136 @@ +module Main where + +import Control.Lens ((^.)) +import Control.Monad (forM, unless) +import Data.Aeson qualified as Aeson +import Data.Generics.Labels () +import Data.Maybe (isNothing) +import Data.Text (Text) +import Data.Text qualified as T +import Data.Time.Clock (getCurrentTime) +import Data.Vector qualified as V +import Hasql.Pool qualified as Pool +import Hasql.Transaction qualified as Tx +import Hasql.Transaction.Sessions qualified as TxSessions +import Kiroku.Store +import Kiroku.Store.Effect (appendDispatchTx, buildAppendParams, prepareEvents) +import Kiroku.Store.SQL qualified as SQL +import Kiroku.Store.Settings (enrichEvents) +import Kiroku.Test.Postgres (withMigratedTestDatabase, withSharedMigratedPostgres) +import Test.Tasty (localOption) +import Test.Tasty.Bench + +main :: IO () +main = + withSharedMigratedPostgres $ + withMigratedTestDatabase $ \controlConnectionString -> + withMigratedTestDatabase $ \candidateConnectionString -> + withStore (defaultConnectionSettings controlConnectionString) $ \controlStore -> + withStore (defaultConnectionSettings candidateConnectionString) $ \candidateStore -> do + let fourStreams = namedStreams "workload-gate-4" 4 + eightStreams = namedStreams "workload-gate-8" 8 + seedStreams controlStore (fourStreams <> eightStreams) + seedStreams candidateStore (fourStreams <> eightStreams) + + runSequentialMultiAppend controlStore fourStreams + runProductionMultiAppend candidateStore fourStreams + runSequentialMultiAppend controlStore eightStreams + runProductionMultiAppend candidateStore eightStreams + + defaultMain + [ localOption WallTime $ + bgroup + "append-multi-stream" + [ bench "sequential-control-4" $ + whnfIO (runSequentialMultiAppend controlStore fourStreams) + , bcompareWithin 0 0.90 "sequential-control-4" $ + bench "production-pipeline-4" $ + whnfIO (runProductionMultiAppend candidateStore fourStreams) + , bench "sequential-control-8" $ + whnfIO (runSequentialMultiAppend controlStore eightStreams) + , bcompareWithin 0 0.90 "sequential-control-8" $ + bench "production-pipeline-8" $ + whnfIO (runProductionMultiAppend candidateStore eightStreams) + ] + ] + +namedStreams :: Text -> Int -> [(StreamName, Text)] +namedStreams prefix count = + [ ( StreamName (prefix <> "-" <> T.pack (show index)) + , "WorkloadGate" <> T.pack (show index) + ) + | index <- [1 .. count] + ] + +seedStreams :: KirokuStore -> [(StreamName, Text)] -> IO () +seedStreams store streams = + mapM_ + ( \(streamName, eventType) -> do + result <- runStoreIO store $ appendToStream streamName NoStream [makeEvent (eventType <> "Seed")] + forceStoreResults "seed append" (fmap pure result) + ) + streams + +runProductionMultiAppend :: KirokuStore -> [(StreamName, Text)] -> IO () +runProductionMultiAppend store streams = do + result <- + runStoreIO store $ + appendMultiStream + [ (streamName, AnyVersion, [makeEvent eventType]) + | (streamName, eventType) <- streams + ] + forceStoreResults "production appendMultiStream" result + +runSequentialMultiAppend :: KirokuStore -> [(StreamName, Text)] -> IO () +runSequentialMultiAppend store streams = do + now <- getCurrentTime + preparedOps <- + forM streams $ \(streamName@(StreamName name), eventType) -> do + enriched <- enrichEvents (store ^. #storeSettings) [makeEvent eventType] + prepared <- prepareEvents enriched + pure (streamName, name, buildAppendParams name now prepared) + let names = V.fromList [name | (_, name, _) <- preparedOps] + transaction = do + Tx.statement names SQL.lockStreamsForMultiStmt + results <- + forM preparedOps $ \(_, _, params) -> + appendDispatchTx AnyVersion params + if any isNothing results + then Tx.condemn >> pure results + else pure results + result <- + Pool.use (store ^. #pool) $ + TxSessions.transaction TxSessions.ReadCommitted TxSessions.Write transaction + case result of + Left err -> error ("sequential appendMultiStream control failed: " <> show err) + Right maybeResults -> do + unless (all isJustAppend maybeResults) $ + error "sequential appendMultiStream control returned an empty append result" + forceAppendResults [appendResult | Just appendResult <- maybeResults] + +makeEvent :: Text -> EventData +makeEvent eventType = + EventData + { eventId = Nothing + , eventType = EventType eventType + , payload = Aeson.object [("workloadGate", Aeson.Bool True)] + , metadata = Nothing + , causationId = Nothing + , correlationId = Nothing + } + +isJustAppend :: Maybe AppendResult -> Bool +isJustAppend (Just _) = True +isJustAppend Nothing = False + +forceStoreResults :: String -> Either StoreError [AppendResult] -> IO () +forceStoreResults _ (Right results) = forceAppendResults results +forceStoreResults label (Left err) = error (label <> " failed: " <> show err) + +forceAppendResults :: [AppendResult] -> IO () +forceAppendResults = + mapM_ $ \result -> + (result ^. #streamId) `seq` + (result ^. #streamVersion) `seq` + (result ^. #globalPosition) `seq` + pure () diff --git a/bench/ShibuyaOverhead.hs b/bench/ShibuyaOverhead.hs --- a/bench/ShibuyaOverhead.hs +++ b/bench/ShibuyaOverhead.hs @@ -125,6 +125,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -151,6 +152,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -182,6 +184,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing 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.4.0.0 +version: 0.5.0.0 synopsis: High-performance PostgreSQL event store description: Kiroku is a PostgreSQL-backed event store for Haskell applications. It @@ -46,6 +46,7 @@ Kiroku.Store.Settings Kiroku.Store.SQL Kiroku.Store.Subscription + Kiroku.Store.Subscription.Checkpoint Kiroku.Store.Subscription.Effect Kiroku.Store.Subscription.EventPublisher Kiroku.Store.Subscription.Fsm @@ -55,7 +56,10 @@ Kiroku.Store.Transaction Kiroku.Store.Types - other-modules: Kiroku.Store.Subscription.CheckpointInventory.SQL + other-modules: + Kiroku.Store.Subscription.Checkpoint.SQL + Kiroku.Store.Subscription.CheckpointInventory.SQL + build-depends: , aeson >=2.1 && <2.3 , async >=2.2 && <2.3 @@ -101,6 +105,7 @@ Test.Helpers Test.InterpreterHooks Test.NotifyGuard + Test.PerformanceStructure Test.Properties Test.PublisherCallbackResilience Test.PublisherIdleAdvance @@ -109,8 +114,12 @@ Test.StartupFailureSurfacing Test.StreamBridgeTermination Test.StreamNameLookup + Test.SubscriptionCheckpointInitialization + Test.SubscriptionCheckpointInitializationMock Test.SubscriptionCheckpointInventory Test.SubscriptionCheckpointInventoryMock + Test.SubscriptionCheckpointReset + Test.SubscriptionCheckpointWorker Test.SubscriptionPauseResume Test.SubscriptionReconnect Test.SubscriptionRegistry @@ -154,7 +163,9 @@ type: exitcode-stdio-1.0 main-is: Main.hs hs-source-dirs: bench - ghc-options: -threaded -rtsopts "-with-rtsopts=-N -A32m" + ghc-options: + -threaded -rtsopts "-with-rtsopts=-N -A32m" -fproc-alignment=64 + build-depends: , aeson >=2.1 && <2.3 , async >=2.2 && <2.3 @@ -173,6 +184,29 @@ , text >=2.0 && <2.2 , time >=1.12 && <1.15 , uuid >=1.3 && <1.4 + , vector >=0.13 && <0.14 + +benchmark kiroku-store-bench-workload-gate + import: common + type: exitcode-stdio-1.0 + main-is: RegressionGate.hs + hs-source-dirs: bench + ghc-options: + -threaded -rtsopts "-with-rtsopts=-N -A32m" -fproc-alignment=64 + + build-depends: + , aeson >=2.1 && <2.3 + , base >=4.18 && <5 + , generic-lens >=2.2 && <2.4 + , hasql-pool >=1.2 && <1.5 + , hasql-transaction >=1.1 && <1.3 + , kiroku-store + , kiroku-test-support + , lens >=5.2 && <5.4 + , tasty >=1.4 && <1.6 + , tasty-bench >=0.4 + , text >=2.0 && <2.2 + , time >=1.12 && <1.15 , vector >=0.13 && <0.14 benchmark kiroku-shibuya-overhead diff --git a/src/Kiroku/Store.hs b/src/Kiroku/Store.hs --- a/src/Kiroku/Store.hs +++ b/src/Kiroku/Store.hs @@ -20,6 +20,7 @@ module Kiroku.Store.Read, module Kiroku.Store.Settings, module Kiroku.Store.Subscription, + module Kiroku.Store.Subscription.Checkpoint, module Kiroku.Store.Transaction, -- * Subscription effect (interpreter only — import Effect module for @subscribe@) @@ -61,6 +62,7 @@ import Kiroku.Store.Read import Kiroku.Store.Settings import Kiroku.Store.Subscription +import Kiroku.Store.Subscription.Checkpoint import Kiroku.Store.Subscription.Effect (Subscription, runSubscription, runSubscriptionResource) import Kiroku.Store.Subscription.Fsm (stateCursor, stateName) import Kiroku.Store.Transaction 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,8 +56,15 @@ import Kiroku.Store.Observability (KirokuEvent (..)) import Kiroku.Store.SQL qualified as SQL import Kiroku.Store.Settings (decodeEvents, enrichEvents) +import Kiroku.Store.Subscription.Checkpoint.SQL qualified as CheckpointSQL import Kiroku.Store.Subscription.CheckpointInventory.SQL qualified as CheckpointInventorySQL -import Kiroku.Store.Subscription.Types (SubscriptionCheckpointInventory) +import Kiroku.Store.Subscription.Types ( + CheckpointInitialization, + MissingCheckpointPolicy, + SubscriptionCheckpointInventory, + SubscriptionCheckpointMissing, + SubscriptionName, + ) import Kiroku.Store.Types -- --------------------------------------------------------------------------- @@ -132,6 +139,17 @@ 'Kiroku.Store.Subscription.subscriptionCheckpointInventory'. -} GetSubscriptionCheckpointInventory :: Store m SubscriptionCheckpointInventory + {- | Resolve one exact subscription checkpoint key according to its + missing-row policy. Existing rows always take precedence. + + Surfaced as + 'Kiroku.Store.Subscription.initializeSubscriptionCheckpoint'. + -} + InitializeSubscriptionCheckpoint :: + SubscriptionName -> + Int32 -> + MissingCheckpointPolicy -> + Store m (Either SubscriptionCheckpointMissing CheckpointInitialization) {- | 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 @@ -351,6 +369,9 @@ GetSubscriptionCheckpointInventory -> usePool (store ^. #pool) $ Session.statement () CheckpointInventorySQL.getSubscriptionCheckpointInventoryStmt + InitializeSubscriptionCheckpoint subscriptionName member policy -> + usePool (store ^. #pool) $ + CheckpointSQL.initializeSubscriptionCheckpointSession subscriptionName member policy RunTransaction tx -> runTxOnPool (store ^. #pool) TxSessions.transaction tx RunTransactionNoRetry tx -> diff --git a/src/Kiroku/Store/Observability.hs b/src/Kiroku/Store/Observability.hs --- a/src/Kiroku/Store/Observability.hs +++ b/src/Kiroku/Store/Observability.hs @@ -49,7 +49,11 @@ import Data.Int (Int32) import Hasql.Pool (UsageError) import Kiroku.Store.Subscription.Fsm (DeadLetterReason (..), SubscriptionStopReason (..)) -import Kiroku.Store.Subscription.Types (SubscriptionName) +import Kiroku.Store.Subscription.Types ( + CheckpointInitialization, + SubscriptionCheckpointMissing, + SubscriptionName, + ) import Kiroku.Store.Types (GlobalPosition, StreamId, StreamName) {- | A structured operational event emitted by 'Kiroku.Store' itself. @@ -96,10 +100,23 @@ (if any) emitted it. -} KirokuEventSubscriptionDbError !SubscriptionName !SubscriptionDbPhase !UsageError !SubscriptionGroupContext + | {- | Startup resolved an exact checkpoint key before the worker emitted + 'KirokuEventSubscriptionStarted'. 'ExistingCheckpoint' means the worker + resumed durable progress; 'InitializedCheckpoint' identifies whether it + seeded zero or the current store head. The event contains no payload data. + -} + KirokuEventSubscriptionCheckpointResolved !CheckpointInitialization !SubscriptionGroupContext + | {- | Startup applied 'Kiroku.Store.Subscription.Types.FailIfMissing' to + an absent exact checkpoint key. The worker emits this event and throws the + carried typed exception before emitting 'KirokuEventSubscriptionStarted' + or invoking the handler. + -} + KirokuEventSubscriptionCheckpointMissing !SubscriptionCheckpointMissing !SubscriptionGroupContext | {- | A subscription's worker thread has just started; the worker - will begin from the recorded 'GlobalPosition' (zero only when no - checkpoint exists). The trailing 'SubscriptionGroupContext' identifies - which consumer-group member (if any) started. + will begin from the durable 'GlobalPosition' reported by the immediately + preceding checkpoint-resolution event. The trailing + 'SubscriptionGroupContext' identifies which consumer-group member (if any) + started. -} KirokuEventSubscriptionStarted !SubscriptionName !GlobalPosition !SubscriptionGroupContext | {- | The subscription has reached the EventPublisher's 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 + initializeSubscriptionCheckpoint, subscriptionCheckpointInventory, subscriptionStates, SubscriptionStateView (..), @@ -31,7 +32,7 @@ import GHC.Generics (Generic) import GHC.Stack (HasCallStack) import Kiroku.Store.Connection (KirokuStore (..)) -import Kiroku.Store.Effect (Store (GetSubscriptionCheckpointInventory)) +import Kiroku.Store.Effect (Store (GetSubscriptionCheckpointInventory, InitializeSubscriptionCheckpoint)) import Kiroku.Store.Notification qualified as Notifier import Kiroku.Store.Subscription.EventPublisher qualified as Pub import Kiroku.Store.Subscription.Fsm (SubscriptionState (..), stateCursor, stateName) @@ -43,9 +44,11 @@ The subscription spawns a worker thread that: -1. Reads the checkpoint from the database (or starts from global position 0 - for a fresh subscription name). A database error while loading the checkpoint - fails the worker loudly through 'wait'; it does not fall back to 0. +1. Resolves the exact @(subscription name, consumer-group member)@ checkpoint. + An existing row always wins. If absent, 'missingCheckpointPolicy' durably + seeds zero, atomically seeds the current @$all@ head, or fails before the + handler runs. A database error fails the worker loudly through 'wait'; it + does not fall back to 0. 2. Catches up by querying the database directly until it reaches the 'Kiroku.Store.Subscription.EventPublisher.lastPublished' cursor. 3. Switches to live mode. For 'Kiroku.Store.Subscription.Types.AllStreams' @@ -109,6 +112,10 @@ * @Left e@ where @e@ is a 'Hasql.Pool.UsageError' from checkpoint load — startup could not read the saved checkpoint. The worker stops rather than silently replaying from global position 0. +* @Left e@ where @e@ is + 'Kiroku.Store.Subscription.Types.SubscriptionCheckpointMissing' — + 'Kiroku.Store.Subscription.Types.FailIfMissing' refused an absent exact key. + No checkpoint row was inserted and the handler did not run. * @Left e@ for any exception thrown by the handler — handler exceptions are not caught; the worker thread dies and the original exception propagates to the consumer. This is intentional: a handler that @@ -245,6 +252,24 @@ (HasCallStack, Store :> es) => Eff es SubscriptionCheckpointInventory subscriptionCheckpointInventory = send GetSubscriptionCheckpointInventory + +{- | Resolve the durable checkpoint for one exact @(subscription name, +consumer-group member)@ key. + +An existing row is returned unchanged for every policy. When no row exists, +'FromBeginning' atomically inserts position zero, 'FromCurrentHead' atomically +inserts the current @$all@ store position, and 'FailIfMissing' returns +'SubscriptionCheckpointMissing' without inserting. Concurrent initializers +converge on the first committed row. +-} +initializeSubscriptionCheckpoint :: + (HasCallStack, Store :> es) => + SubscriptionName -> + Int32 -> + MissingCheckpointPolicy -> + Eff es (Either SubscriptionCheckpointMissing CheckpointInitialization) +initializeSubscriptionCheckpoint subscriptionName member policy = + send (InitializeSubscriptionCheckpoint subscriptionName member policy) {- | 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/Checkpoint.hs b/src/Kiroku/Store/Subscription/Checkpoint.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Store/Subscription/Checkpoint.hs @@ -0,0 +1,70 @@ +{- | Explicit mutation operations for durable subscription checkpoints. + +Ordinary subscription checkpoint saves are monotonic. This module owns the +separate, deliberately named reset operation for callers that need to move +persisted progress backward or forward as part of a larger transaction. +-} +module Kiroku.Store.Subscription.Checkpoint ( + SubscriptionCheckpointResetReport (..), + resetSubscriptionCheckpointsTx, +) where + +import Data.List.NonEmpty (NonEmpty) +import Data.List.NonEmpty qualified as NonEmpty +import Data.Vector (Vector) +import Data.Vector qualified as Vector +import GHC.Generics (Generic) +import Hasql.Transaction qualified as Tx +import Kiroku.Store.Subscription.Checkpoint.SQL qualified as SQL +import Kiroku.Store.Subscription.Types ( + SubscriptionCheckpointKey (..), + SubscriptionName (..), + ) +import Kiroku.Store.Types (GlobalPosition (..)) + +{- | Exact result of resetting a non-empty set of subscription names. + +'resetCheckpointKeys' contains every persisted @(name, member)@ row that was +updated. 'missingSubscriptionNames' contains requested names for which no row +existed. Both vectors are sorted by subscription name (and then member for +keys); duplicate requested names appear only once in the report. +-} +data SubscriptionCheckpointResetReport = SubscriptionCheckpointResetReport + { resetCheckpointKeys :: !(Vector SubscriptionCheckpointKey) + , missingSubscriptionNames :: !(Vector SubscriptionName) + } + deriving stock (Eq, Show, Generic) + +{- | Set every existing checkpoint member for the requested subscription names +to the exact target position and return complete deterministic evidence. + +The operation treats duplicate requested names as one name, updates all +persisted members for each name, and never creates checkpoint rows for missing +names. Unlike ordinary worker saves, this operation can move a checkpoint +backward. It is a 'Tx.Transaction' combinator so a caller can atomically compose +the reset with its own projection fence and target preparation; condemning that +surrounding transaction rolls back all of those writes together. +-} +resetSubscriptionCheckpointsTx :: + NonEmpty SubscriptionName -> + GlobalPosition -> + Tx.Transaction SubscriptionCheckpointResetReport +resetSubscriptionCheckpointsTx names (GlobalPosition position) = do + rows <- + Tx.statement + ( Vector.fromList + [name | SubscriptionName name <- NonEmpty.toList names] + , position + ) + SQL.resetSubscriptionCheckpointsStmt + pure + SubscriptionCheckpointResetReport + { resetCheckpointKeys = Vector.mapMaybe resetKey rows + , missingSubscriptionNames = Vector.mapMaybe missingName rows + } + where + resetKey (name, Just member) = + Just (SubscriptionCheckpointKey (SubscriptionName name) member) + resetKey (_, Nothing) = Nothing + missingName (name, Nothing) = Just (SubscriptionName name) + missingName (_, Just _) = Nothing diff --git a/src/Kiroku/Store/Subscription/Checkpoint/SQL.hs b/src/Kiroku/Store/Subscription/Checkpoint/SQL.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Store/Subscription/Checkpoint/SQL.hs @@ -0,0 +1,160 @@ +{-# LANGUAGE MultilineStrings #-} + +-- | Package-internal SQL for subscription checkpoint lifecycle operations. +module Kiroku.Store.Subscription.Checkpoint.SQL ( + initializeSubscriptionCheckpointSession, + resetSubscriptionCheckpointsStmt, +) where + +import Contravariant.Extras (contrazip2, contrazip3) +import Data.Int (Int32, Int64) +import Data.Text (Text) +import Data.Vector (Vector) +import Hasql.Decoders qualified as D +import Hasql.Encoders qualified as E +import Hasql.Session qualified as Session +import Hasql.Statement (Statement, preparable) +import Kiroku.Store.Subscription.Types ( + CheckpointInitialization (..), + MissingCheckpointPolicy (..), + SubscriptionCheckpointKey (..), + SubscriptionCheckpointMissing (..), + SubscriptionName (..), + ) +import Kiroku.Store.Types (GlobalPosition (..)) + +{- | Resolve one checkpoint key in a Hasql session. + +The first statement inserts the policy-selected position with @ON CONFLICT DO +NOTHING@ and then reads the winning row. PostgreSQL can report no row to that +final read when another transaction committed the conflicting insert after the +statement snapshot was taken. For an initializing policy, a second statement +therefore reads the now-committed winner on a fresh snapshot. 'FailIfMissing' +does not retry because it never attempts an insert. +-} +initializeSubscriptionCheckpointSession :: + SubscriptionName -> + Int32 -> + MissingCheckpointPolicy -> + Session.Session (Either SubscriptionCheckpointMissing CheckpointInitialization) +initializeSubscriptionCheckpointSession subscriptionName@(SubscriptionName name) member policy = do + first <- Session.statement (name, member, policyCode policy) initializeSubscriptionCheckpointStmt + case first of + Just result -> pure (Right (decodeResult result)) + Nothing -> case policy of + FailIfMissing -> pure (Left missing) + _ -> do + -- The insert lost a concurrent unique-key race after this + -- statement's snapshot. A fresh statement snapshot observes + -- the committed winner; singleRow turns a violated invariant + -- into a structured Hasql session error. + position <- Session.statement (name, member) readInitializedCheckpointStmt + pure (Right (ExistingCheckpoint key (GlobalPosition position))) + where + key = SubscriptionCheckpointKey subscriptionName member + missing = SubscriptionCheckpointMissing key + decodeResult (position, inserted) + | inserted = InitializedCheckpoint policy key (GlobalPosition position) + | otherwise = ExistingCheckpoint key (GlobalPosition position) + +policyCode :: MissingCheckpointPolicy -> Text +policyCode = \case + FromBeginning -> "from_beginning" + FromCurrentHead -> "from_current_head" + FailIfMissing -> "fail_if_missing" + +initializeSubscriptionCheckpointStmt :: Statement (Text, Int32, Text) (Maybe (Int64, Bool)) +initializeSubscriptionCheckpointStmt = + preparable + """ + WITH desired AS ( + SELECT CASE $3 + WHEN 'from_beginning' THEN 0::bigint + WHEN 'from_current_head' THEN ( + SELECT stream_version FROM streams WHERE stream_id = 0 + ) + ELSE NULL::bigint + END AS last_seen + ), + inserted AS ( + INSERT INTO subscriptions + (subscription_name, consumer_group_member, last_seen, updated_at) + SELECT $1, $2, desired.last_seen, now() + FROM desired + WHERE desired.last_seen IS NOT NULL + ON CONFLICT (subscription_name, consumer_group_member) DO NOTHING + RETURNING last_seen + ) + SELECT inserted.last_seen, TRUE AS initialized + FROM inserted + UNION ALL + SELECT subscriptions.last_seen, FALSE AS initialized + FROM subscriptions + WHERE subscription_name = $1 + AND consumer_group_member = $2 + LIMIT 1 + """ + ( contrazip3 + (E.param (E.nonNullable E.text)) + (E.param (E.nonNullable E.int4)) + (E.param (E.nonNullable E.text)) + ) + ( D.rowMaybe $ + (,) + <$> D.column (D.nonNullable D.int8) + <*> D.column (D.nonNullable D.bool) + ) + +readInitializedCheckpointStmt :: Statement (Text, Int32) Int64 +readInitializedCheckpointStmt = + preparable + """ + SELECT last_seen + FROM subscriptions + WHERE subscription_name = $1 + AND consumer_group_member = $2 + """ + ( contrazip2 + (E.param (E.nonNullable E.text)) + (E.param (E.nonNullable E.int4)) + ) + (D.singleRow (D.column (D.nonNullable D.int8))) + +{- | Reset every persisted member belonging to the requested subscription +names. The input is treated as a set by PostgreSQL. Each returned row contains +either one reset member or a requested name with no persisted rows, and the +result is deterministically ordered by name and member. + +This statement deliberately assigns @last_seen@ directly. Ordinary worker +saves retain their separate @GREATEST(...)@ monotonicity contract. +-} +resetSubscriptionCheckpointsStmt :: + Statement (Vector Text, Int64) (Vector (Text, Maybe Int32)) +resetSubscriptionCheckpointsStmt = + preparable + """ + WITH requested AS ( + SELECT DISTINCT requested_name AS subscription_name + FROM unnest($1::text[]) AS requested_name + ), + updated AS ( + UPDATE subscriptions AS checkpoint + SET last_seen = $2, updated_at = now() + FROM requested + WHERE checkpoint.subscription_name = requested.subscription_name + RETURNING checkpoint.subscription_name, checkpoint.consumer_group_member + ) + SELECT requested.subscription_name, updated.consumer_group_member + FROM requested + LEFT JOIN updated USING (subscription_name) + ORDER BY requested.subscription_name, updated.consumer_group_member + """ + ( contrazip2 + (E.param (E.nonNullable (E.foldableArray (E.nonNullable E.text)))) + (E.param (E.nonNullable E.int8)) + ) + ( D.rowVector $ + (,) + <$> D.column (D.nonNullable D.text) + <*> D.column (D.nullable D.int4) + ) 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,11 @@ -} module Kiroku.Store.Subscription.Types ( SubscriptionName (..), + MissingCheckpointPolicy (..), + SubscriptionCheckpointKey (..), + CheckpointInitialization (..), + checkpointInitializationPosition, + SubscriptionCheckpointMissing (..), SubscriptionCheckpoint (..), SubscriptionCheckpointInventory (..), SubscriptionTarget (..), @@ -141,6 +146,52 @@ newtype SubscriptionName = SubscriptionName Text deriving newtype (Eq, Ord, Show) +{- | What a subscription does when its exact +@('SubscriptionName', consumer-group member)@ checkpoint key has no durable +row. The policy is consulted only for absence; an existing row always wins and +is never moved by initialization. +-} +data MissingCheckpointPolicy + = -- | Materialize position zero and replay retained history. + FromBeginning + | -- | Atomically materialize the current @$all@ store head. + FromCurrentHead + | -- | Refuse startup without inserting a row. + FailIfMissing + deriving stock (Eq, Show, Generic) + +-- | The durable identity of one subscription checkpoint row. +data SubscriptionCheckpointKey = SubscriptionCheckpointKey + { subscriptionName :: !SubscriptionName + , consumerGroupMember :: !Int32 + } + deriving stock (Eq, Ord, Show, Generic) + +{- | The successful result of resolving one subscription checkpoint at +startup. 'ExistingCheckpoint' means a durable row already existed; +'InitializedCheckpoint' records the policy that created the row. In both +cases the position is the durable position returned by PostgreSQL. +-} +data CheckpointInitialization + = ExistingCheckpoint !SubscriptionCheckpointKey !GlobalPosition + | InitializedCheckpoint !MissingCheckpointPolicy !SubscriptionCheckpointKey !GlobalPosition + deriving stock (Eq, Show, Generic) + +-- | Extract the durable position from either successful initialization result. +checkpointInitializationPosition :: CheckpointInitialization -> GlobalPosition +checkpointInitializationPosition = \case + ExistingCheckpoint _ position -> position + InitializedCheckpoint _ _ position -> position + +{- | A typed startup refusal produced by 'FailIfMissing'. No checkpoint row is +inserted and a worker throws this exception before invoking its handler. +-} +newtype SubscriptionCheckpointMissing = SubscriptionCheckpointMissing + { checkpointKey :: SubscriptionCheckpointKey + } + deriving stock (Eq, Show, Generic) + deriving anyclass (Exception) + -- | One checkpoint row that has been durably persisted by a subscription. data SubscriptionCheckpoint = SubscriptionCheckpoint { subscriptionName :: !SubscriptionName @@ -302,6 +353,14 @@ (a startup detection probe, not a lifetime-held lock). Ignored when 'consumerGroup' is 'Nothing'. -} + , missingCheckpointPolicy :: !MissingCheckpointPolicy + {- ^ What startup does only when the exact @(name, member)@ checkpoint row + is absent. 'FromBeginning' is the compatibility default and materializes + position zero. 'FromCurrentHead' atomically seeds the current store head, + and 'FailIfMissing' refuses startup before the handler runs. Existing rows + always take precedence, so changing this field never rewinds or advances + durable progress. + -} , retryPolicy :: !RetryPolicy {- ^ Bounds redelivery of an event for which the handler returned 'Retry' before the worker dead-letters it. Default: 'defaultRetryPolicy' @@ -348,8 +407,13 @@ for long. Override the 'batchSize' field on the returned record if a different value suits the workload. +The missing-checkpoint policy defaults to 'FromBeginning' for source +compatibility. New call sites should override 'missingCheckpointPolicy' +deliberately when a future-only or pre-provisioned worker is intended. + @ -let cfg = defaultSubscriptionConfig "my-projection" AllStreams handler +let cfg = (defaultSubscriptionConfig "my-projection" AllStreams handler) + { missingCheckpointPolicy = FromBeginning } withSubscription store cfg $ \\h -> wait h @ -} @@ -368,6 +432,7 @@ , overflowPolicy = PauseAndResume , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing diff --git a/src/Kiroku/Store/Subscription/Worker.hs b/src/Kiroku/Store/Subscription/Worker.hs --- a/src/Kiroku/Store/Subscription/Worker.hs +++ b/src/Kiroku/Store/Subscription/Worker.hs @@ -59,6 +59,7 @@ ) import Kiroku.Store.SQL qualified as SQL import Kiroku.Store.Settings (StoreSettings, decodeEvents) +import Kiroku.Store.Subscription.Checkpoint.SQL qualified as CheckpointSQL import Kiroku.Store.Subscription.EventPublisher (SubscriberStatus) import Kiroku.Store.Subscription.EventPublisher qualified as Pub import Kiroku.Store.Subscription.Fsm ( @@ -86,7 +87,13 @@ type LoadCheckpointHook = SubscriptionConfig -> - IO (Maybe (Either Pool.UsageError (Maybe Int64))) + IO + ( Maybe + ( Either + Pool.UsageError + (Either SubscriptionCheckpointMissing CheckpointInitialization) + ) + ) {-# NOINLINE fetchBatchHookRef #-} fetchBatchHookRef :: IORef (Maybe FetchBatchHook) @@ -163,8 +170,12 @@ If an 'eventHandler' callback is supplied, the worker emits: -* 'Kiroku.Store.Observability.KirokuEventSubscriptionStarted' once at - startup, after the checkpoint has been read. +* 'Kiroku.Store.Observability.KirokuEventSubscriptionCheckpointResolved' once + when startup resumes or initializes a checkpoint, followed by + 'Kiroku.Store.Observability.KirokuEventSubscriptionStarted'. A refused + missing checkpoint emits + 'Kiroku.Store.Observability.KirokuEventSubscriptionCheckpointMissing' + instead and fails before delivery. * 'Kiroku.Store.Observability.KirokuEventSubscriptionCaughtUp' when catch-up completes and the worker switches to live mode. * 'Kiroku.Store.Observability.KirokuEventSubscriptionDbError' in the @@ -209,7 +220,14 @@ case (consumerGroupGuard config, consumerGroup config) of (True, Just (ConsumerGroup m _)) -> guardMember pool subName m _ -> pure () - checkpoint <- loadCheckpoint pool config emit + resolution <- loadCheckpoint pool config emit + checkpoint <- case resolution of + Left missing -> do + emit (KirokuEventSubscriptionCheckpointMissing missing groupCtx) + throwIO missing + Right initialization -> do + emit (KirokuEventSubscriptionCheckpointResolved initialization groupCtx) + pure (checkpointInitializationPosition initialization) writeIORef posRef checkpoint emit (KirokuEventSubscriptionStarted subName checkpoint groupCtx) -- Drive the explicit FSM from the catch-up state. The pure 'step' @@ -435,30 +453,33 @@ configMember :: SubscriptionConfig -> Int32 configMember config = maybe 0 member (consumerGroup config) --- Load the checkpoint from the database, defaulting to 0 only when no checkpoint --- row exists. A database error is emitted and rethrown so startup fails loudly --- instead of silently re-processing from position 0. --- Keyed by (subscription_name, member) so each group member resumes from its --- own saved position. +-- Resolve the exact checkpoint key through the shared initializer. A database +-- error is emitted and rethrown so startup fails loudly. A semantic +-- 'FailIfMissing' result remains typed so the caller can emit the distinct +-- refusal event before throwing it. Each group member resolves its own key. loadCheckpoint :: Pool -> SubscriptionConfig -> (KirokuEvent -> IO ()) -> - IO GlobalPosition + IO (Either SubscriptionCheckpointMissing CheckpointInitialization) loadCheckpoint pool config emit = do - let subName@(SubscriptionName name') = name config + let subName = name config mem = configMember config mHook <- readIORef loadCheckpointHookRef injected <- maybe (pure Nothing) (\hook -> hook config) mHook result <- case injected of Just hooked -> pure hooked - Nothing -> Pool.use pool (Session.statement (name', mem) SQL.getCheckpointMemberStmt) + Nothing -> + Pool.use pool $ + CheckpointSQL.initializeSubscriptionCheckpointSession + subName + mem + (missingCheckpointPolicy config) case result of Left err -> do emit (KirokuEventSubscriptionDbError subName LoadCheckpoint err (groupCtxOf config)) throwIO err - Right Nothing -> pure (GlobalPosition 0) - Right (Just pos) -> pure (GlobalPosition pos) + Right resolution -> pure resolution -- How a DB-driven live loop ('liveLoopCategoryNotify' / 'liveLoopDbDriven') -- exited. The driver maps these onto FSM inputs: a clean handler stop becomes diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -41,6 +41,7 @@ import Test.Hspec import Test.InterpreterHooks qualified as InterpreterHooks import Test.NotifyGuard qualified as NotifyGuard +import Test.PerformanceStructure qualified as PerformanceStructure import Test.Properties qualified as Properties import Test.PublisherCallbackResilience qualified as PublisherCallbackResilience import Test.PublisherIdleAdvance qualified as PublisherIdleAdvance @@ -49,8 +50,12 @@ import Test.StartupFailureSurfacing qualified as StartupFailureSurfacing import Test.StreamBridgeTermination qualified as StreamBridgeTermination import Test.StreamNameLookup qualified as StreamNameLookup +import Test.SubscriptionCheckpointInitialization qualified as SubscriptionCheckpointInitialization +import Test.SubscriptionCheckpointInitializationMock qualified as SubscriptionCheckpointInitializationMock import Test.SubscriptionCheckpointInventory qualified as SubscriptionCheckpointInventory import Test.SubscriptionCheckpointInventoryMock qualified as SubscriptionCheckpointInventoryMock +import Test.SubscriptionCheckpointReset qualified as SubscriptionCheckpointReset +import Test.SubscriptionCheckpointWorker qualified as SubscriptionCheckpointWorker import Test.SubscriptionPauseResume qualified as SubscriptionPauseResume import Test.SubscriptionReconnect qualified as SubscriptionReconnect import Test.SubscriptionRegistry qualified as SubscriptionRegistry @@ -75,15 +80,22 @@ ConsumerGroupSql.spec ConsumerGroup.spec ConsumerGroupEffect.spec - NotifyGuard.spec + describe "performance structure" $ do + PerformanceStructure.spec + NotifyGuard.spec + StreamNameLookup.noOpSpec CategoryIdleNoSpin.spec PublisherCallbackResilience.spec PublisherIdleAdvance.spec PublisherRestartNoRebroadcast.spec CatchupDbErrorNoPrematureSwitch.spec SubscriptionPauseResume.spec + SubscriptionCheckpointInitialization.spec + SubscriptionCheckpointInitializationMock.spec + SubscriptionCheckpointWorker.spec SubscriptionCheckpointInventory.spec SubscriptionCheckpointInventoryMock.spec + SubscriptionCheckpointReset.spec SubscriptionReconnect.spec StartupFailureSurfacing.spec SubscriptionState.spec @@ -1070,6 +1082,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1106,6 +1119,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1151,6 +1165,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1180,6 +1195,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1225,6 +1241,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1278,6 +1295,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1301,6 +1319,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1337,6 +1356,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1364,6 +1384,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1417,6 +1438,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1471,6 +1493,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1517,6 +1540,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1548,6 +1572,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1578,6 +1603,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1627,6 +1653,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1682,6 +1709,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1742,6 +1770,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1785,6 +1814,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1819,6 +1849,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -1846,6 +1877,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -2003,6 +2035,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing diff --git a/test/Test/CatchupDbErrorNoPrematureSwitch.hs b/test/Test/CatchupDbErrorNoPrematureSwitch.hs --- a/test/Test/CatchupDbErrorNoPrematureSwitch.hs +++ b/test/Test/CatchupDbErrorNoPrematureSwitch.hs @@ -49,6 +49,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing diff --git a/test/Test/ConsumerGroup.hs b/test/Test/ConsumerGroup.hs --- a/test/Test/ConsumerGroup.hs +++ b/test/Test/ConsumerGroup.hs @@ -295,6 +295,7 @@ (defaultSubscriptionConfig (SubscriptionName "guard-sub") (Category (CategoryName "guardcat")) (\_ -> pure Continue)) { consumerGroup = Just (ConsumerGroup{member = 3, size = 4}) , consumerGroupGuard = True + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing diff --git a/test/Test/FailureInjection.hs b/test/Test/FailureInjection.hs --- a/test/Test/FailureInjection.hs +++ b/test/Test/FailureInjection.hs @@ -65,6 +65,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing diff --git a/test/Test/PerformanceStructure.hs b/test/Test/PerformanceStructure.hs new file mode 100644 --- /dev/null +++ b/test/Test/PerformanceStructure.hs @@ -0,0 +1,257 @@ +{-# LANGUAGE MultilineStrings #-} + +module Test.PerformanceStructure (spec) where + +import Control.Lens ((^.)) +import Control.Monad (unless) +import Data.Aeson (Value (..)) +import Data.Aeson qualified as Aeson +import Data.Aeson.KeyMap qualified as KeyMap +import Data.ByteString (ByteString) +import Data.Foldable (foldl') +import Data.Generics.Labels () +import Data.IORef (IORef, modifyIORef', newIORef, readIORef) +import Data.Text (Text) +import Data.Text qualified as T +import Hasql.Decoders qualified as D +import Hasql.Encoders qualified as E +import Hasql.Pool qualified as Pool +import Hasql.Session qualified as Session +import Hasql.Statement (Statement, unpreparable) +import Hasql.Statement qualified as Statement +import Kiroku.Store +import Kiroku.Store.SQL qualified as SQL +import Test.Helpers (withTestStore, withTestStoreSettings) +import Test.Hspec + +spec :: Spec +spec = do + noOpAppendSpec + queryPlanSpec + +noOpAppendSpec :: Spec +noOpAppendSpec = + describe "no-op paths use no pooled connection" $ do + it "rejects an empty appendToStream batch before pool checkout" $ do + checkouts <- newIORef (0 :: Int) + withObservedStore checkouts $ \store -> do + before <- readIORef checkouts + result <- runStoreIO store $ appendToStream (StreamName "performance-empty-append") AnyVersion [] + after <- readIORef checkouts + result `shouldBe` Left (EmptyAppendBatch (StreamName "performance-empty-append")) + after - before `shouldBe` 0 + + it "returns an empty appendMultiStream result before pool checkout" $ do + checkouts <- newIORef (0 :: Int) + withObservedStore checkouts $ \store -> do + before <- readIORef checkouts + result <- runStoreIO store $ appendMultiStream [] + after <- readIORef checkouts + result `shouldBe` Right [] + after - before `shouldBe` 0 +queryPlanSpec :: Spec +queryPlanSpec = + describe "production query plans" $ + aroundAll withQueryPlanStore $ do + it "category high-cursor reads use ix_stream_events_all_by_origin" $ \store -> do + plan <- + explainProductionStatement + store + SQL.readCategoryForwardStmt + [ ("$3", "100::int4") + , ("$2", "'performance'::text") + , ("$1", "15000::bigint") + ] + expectIndex "ix_stream_events_all_by_origin" plan + + it "dead-letter reads use ix_dead_letters_subscription_position without Sort" $ \store -> do + plan <- + explainProductionStatement + store + SQL.readDeadLettersStmt + [ ("$2", "0::int4") + , ("$1", "'performance-read'::text") + ] + expectIndex "ix_dead_letters_subscription_position" plan + expectNoNodeType "Sort" plan + + it "orphan dead-letter cleanup uses ix_dead_letters_event_id" $ \store -> do + plan <- + explainProductionStatement + store + SQL.deleteDeadLettersForOrphanedEventsStmt + [("$1", "ARRAY['00000000-0000-0000-0000-000000000001'::uuid]::uuid[]")] + expectIndex "ix_dead_letters_event_id" plan + +withObservedStore :: IORef Int -> (KirokuStore -> IO ()) -> IO () +withObservedStore checkouts = + withTestStoreSettings $ \settings -> + settings + { observationHandler = + Just $ \case + ConnectionObservation _ InUseConnectionStatus -> modifyIORef' checkouts (+ 1) + _ -> pure () + } + +withQueryPlanStore :: (KirokuStore -> IO ()) -> IO () +withQueryPlanStore action = + withTestStore $ \store -> do + result <- Pool.use (store ^. #pool) (Session.script queryPlanFixture) + case result of + Left err -> expectationFailure ("failed to seed performance query-plan fixture: " <> show err) + Right () -> action store + +queryPlanFixture :: Text +queryPlanFixture = + """ + BEGIN; + + WITH new_streams AS ( + INSERT INTO streams (stream_name, stream_version) + SELECT 'performance-' || n::text, 100 + FROM generate_series(1, 200) AS n + RETURNING stream_id + ), fixture_events AS MATERIALIZED ( + SELECT uuidv7() AS event_id, + s.stream_id, + per_stream_position::bigint AS stream_version, + row_number() OVER (ORDER BY per_stream_position, s.stream_id)::bigint AS global_position + FROM new_streams AS s + CROSS JOIN generate_series(1, 100) AS per_stream_position + ), inserted_events AS ( + INSERT INTO events (event_id, event_type, data) + SELECT event_id, 'PerformanceFixture', '{}'::jsonb + FROM fixture_events + RETURNING event_id + ), source_links AS ( + INSERT INTO stream_events + (event_id, stream_id, stream_version, original_stream_id, original_stream_version) + SELECT fixture.event_id, + fixture.stream_id, + fixture.stream_version, + fixture.stream_id, + fixture.stream_version + FROM fixture_events AS fixture + JOIN inserted_events USING (event_id) + RETURNING event_id + ), all_links AS ( + INSERT INTO stream_events + (event_id, stream_id, stream_version, original_stream_id, original_stream_version) + SELECT fixture.event_id, + 0, + fixture.global_position, + fixture.stream_id, + fixture.stream_version + FROM fixture_events AS fixture + JOIN inserted_events USING (event_id) + RETURNING event_id + ), advanced_all AS ( + UPDATE streams + SET stream_version = (SELECT max(global_position) FROM fixture_events) + WHERE stream_id = 0 + RETURNING stream_id + ), inserted_dead_letters AS ( + INSERT INTO dead_letters + (subscription_name, consumer_group_member, global_position, event_id, + reason, reason_summary, attempt_count) + SELECT CASE + WHEN fixture.global_position % 10 = 0 THEN 'performance-read' + ELSE 'performance-other-' || (fixture.global_position % 9)::text + END, + 0, + fixture.global_position, + fixture.event_id, + '{}'::jsonb, + 'performance fixture', + 1 + FROM fixture_events AS fixture + JOIN inserted_events USING (event_id) + RETURNING dead_letter_id + ) + SELECT (SELECT count(*) FROM source_links), + (SELECT count(*) FROM all_links), + (SELECT count(*) FROM advanced_all), + (SELECT count(*) FROM inserted_dead_letters); + + COMMIT; + ANALYZE streams; + ANALYZE events; + ANALYZE stream_events; + ANALYZE dead_letters; + """ + +explainProductionStatement :: + KirokuStore -> + Statement params result -> + [(Text, Text)] -> + IO Value +explainProductionStatement store productionStatement replacements = do + let productionSql = Statement.toSql productionStatement + explainedSql = + "EXPLAIN (FORMAT JSON, COSTS OFF)\n" + <> foldl' (\sql (placeholder, literal) -> T.replace placeholder literal sql) productionSql replacements + explainStatement :: Statement () ByteString + explainStatement = + unpreparable + explainedSql + E.noParams + (D.singleRow (D.column (D.nonNullable (D.jsonBytes Right)))) + result <- Pool.use (store ^. #pool) (Session.statement () explainStatement) + bytes <- case result of + Left err -> expectationFailure ("EXPLAIN failed: " <> show err) >> fail "unreachable" + Right value -> pure value + case Aeson.eitherDecodeStrict' bytes of + Left err -> expectationFailure ("could not decode EXPLAIN JSON: " <> err) >> fail "unreachable" + Right value -> pure value + +data PlanFacts = PlanFacts + { nodeTypes :: [Text] + , indexNames :: [Text] + } + deriving stock (Show) + +instance Semigroup PlanFacts where + PlanFacts nodeTypesA indexNamesA <> PlanFacts nodeTypesB indexNamesB = + PlanFacts (nodeTypesA <> nodeTypesB) (indexNamesA <> indexNamesB) + +instance Monoid PlanFacts where + mempty = PlanFacts [] [] + +collectPlanFacts :: Value -> PlanFacts +collectPlanFacts (Object object) = + PlanFacts + { nodeTypes = maybe [] pure (textField "Node Type" object) + , indexNames = maybe [] pure (textField "Index Name" object) + } + <> foldMap collectPlanFacts (KeyMap.elems object) +collectPlanFacts (Array values) = foldMap collectPlanFacts values +collectPlanFacts _ = mempty + +textField :: Aeson.Key -> Aeson.Object -> Maybe Text +textField key object = case KeyMap.lookup key object of + Just (String value) -> Just value + _ -> Nothing + +expectIndex :: Text -> Value -> Expectation +expectIndex expected plan = do + let facts = collectPlanFacts plan + unless (expected `elem` indexNames facts) $ + expectationFailure $ + "expected plan to use index " + <> T.unpack expected + <> ", but collected " + <> show facts + <> " from:\n" + <> show plan + +expectNoNodeType :: Text -> Value -> Expectation +expectNoNodeType forbidden plan = do + let facts = collectPlanFacts plan + unless (forbidden `notElem` nodeTypes facts) $ + expectationFailure $ + "expected plan not to contain node type " + <> T.unpack forbidden + <> ", but collected " + <> show facts + <> " from:\n" + <> show plan diff --git a/test/Test/PublisherRestartNoRebroadcast.hs b/test/Test/PublisherRestartNoRebroadcast.hs --- a/test/Test/PublisherRestartNoRebroadcast.hs +++ b/test/Test/PublisherRestartNoRebroadcast.hs @@ -55,6 +55,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -117,6 +118,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -196,6 +198,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing diff --git a/test/Test/StartupFailureSurfacing.hs b/test/Test/StartupFailureSurfacing.hs --- a/test/Test/StartupFailureSurfacing.hs +++ b/test/Test/StartupFailureSurfacing.hs @@ -48,6 +48,39 @@ any (isLoadCheckpointError subName) observed `shouldBe` True any (isCrashedStop subName) observed `shouldBe` True + it "refuses a missing checkpoint before Started or handler delivery" $ do + handlerCalled <- newTVarIO False + observedRef <- newIORef ([] :: [KirokuEvent]) + let subName = SubscriptionName "fail-if-checkpoint-missing" + key = SubscriptionCheckpointKey subName 0 + cfg = + ( defaultSubscriptionConfig subName AllStreams $ \_ -> do + atomically (writeTVar handlerCalled True) + pure Continue + ) + { missingCheckpointPolicy = FailIfMissing + } + observe evt = modifyIORef' observedRef (evt :) + tweak settings = settings & #eventHandler .~ Just observe + + withTestStoreSettings tweak $ \store -> do + handle <- subscribe store cfg + result <- waitWithTimeout 5_000_000 handle + case result of + Right (Left e) + | Just (SubscriptionCheckpointMissing actualKey) <- fromException e -> + actualKey `shouldBe` key + Left timeout -> expectationFailure timeout + Right other -> expectationFailure ("expected missing-checkpoint refusal, got: " <> show other) + + runStoreIO store subscriptionCheckpointInventory + `shouldReturn` Right (SubscriptionCheckpointInventory (GlobalPosition 0) mempty) + + readTVarIO handlerCalled `shouldReturn` False + observed <- reverse <$> readIORef observedRef + observed `shouldSatisfy` any (isMissingCheckpoint key) + observed `shouldSatisfy` all (not . isStarted subName) + it "leaves no publisher or subscription registry entries after a subscribe/cancel storm" $ withTestStore $ \store -> do let cfg = defaultSubscriptionConfig (SubscriptionName "subscribe-cancel-storm") AllStreams (\_ -> pure Continue) @@ -78,4 +111,14 @@ | actual == expected , Just Pool.AcquisitionTimeoutUsageError <- fromException e -> True + _ -> False + +isMissingCheckpoint :: SubscriptionCheckpointKey -> KirokuEvent -> Bool +isMissingCheckpoint expected = \case + KirokuEventSubscriptionCheckpointMissing (SubscriptionCheckpointMissing actual) _ -> actual == expected + _ -> False + +isStarted :: SubscriptionName -> KirokuEvent -> Bool +isStarted expected = \case + KirokuEventSubscriptionStarted actual _ _ -> actual == expected _ -> False diff --git a/test/Test/StreamNameLookup.hs b/test/Test/StreamNameLookup.hs --- a/test/Test/StreamNameLookup.hs +++ b/test/Test/StreamNameLookup.hs @@ -3,7 +3,7 @@ by fan-in reads back to a human-readable 'StreamName', without every read having to return the name (which a benchmark showed costs ~13% on @$all@ pages). -} -module Test.StreamNameLookup (spec) where +module Test.StreamNameLookup (spec, noOpSpec) where import Control.Lens ((^.)) import Data.Aeson qualified as Aeson @@ -52,27 +52,30 @@ Right missing <- runStoreIO store $ lookupStreamName (StreamId 888888) missing `shouldBe` Nothing - it "short-circuits empty input without a pool checkout" $ do - ref <- newIORef (0 :: Int) - let handler (ConnectionObservation _ InUseConnectionStatus) = - modifyIORef' ref (+ 1) - handler _ = - pure () - withTestStoreSettings (\settings -> settings{observationHandler = Just handler}) $ \store -> do - Right appendResult <- - runStoreIO store $ - appendToStream (StreamName "lookup-count-1") NoStream [makeEvent "LookupCounted" (Aeson.object [])] - waitForPublisher store (appendResult ^. #globalPosition) - Right (Just sid) <- runStoreIO store $ lookupStreamId (StreamName "lookup-count-1") +noOpSpec :: Spec +noOpSpec = + describe "empty stream-name lookup" $ + it "short-circuits empty input without a pool checkout" $ do + ref <- newIORef (0 :: Int) + let handler (ConnectionObservation _ InUseConnectionStatus) = + modifyIORef' ref (+ 1) + handler _ = + pure () + withTestStoreSettings (\settings -> settings{observationHandler = Just handler}) $ \store -> do + Right appendResult <- + runStoreIO store $ + appendToStream (StreamName "lookup-count-1") NoStream [makeEvent "LookupCounted" (Aeson.object [])] + waitForPublisher store (appendResult ^. #globalPosition) + Right (Just sid) <- runStoreIO store $ lookupStreamId (StreamName "lookup-count-1") - beforeEmpty <- readIORef ref - Right emptyNames <- runStoreIO store $ lookupStreamNames [] - afterEmpty <- readIORef ref - emptyNames `shouldBe` Map.empty - afterEmpty - beforeEmpty `shouldBe` 0 + beforeEmpty <- readIORef ref + Right emptyNames <- runStoreIO store $ lookupStreamNames [] + afterEmpty <- readIORef ref + emptyNames `shouldBe` Map.empty + afterEmpty - beforeEmpty `shouldBe` 0 - beforeReal <- readIORef ref - Right names <- runStoreIO store $ lookupStreamNames [sid] - afterReal <- readIORef ref - names `shouldBe` Map.singleton sid (StreamName "lookup-count-1") - afterReal - beforeReal `shouldBe` 1 + beforeReal <- readIORef ref + Right names <- runStoreIO store $ lookupStreamNames [sid] + afterReal <- readIORef ref + names `shouldBe` Map.singleton sid (StreamName "lookup-count-1") + afterReal - beforeReal `shouldBe` 1 diff --git a/test/Test/SubscriptionCheckpointInitialization.hs b/test/Test/SubscriptionCheckpointInitialization.hs new file mode 100644 --- /dev/null +++ b/test/Test/SubscriptionCheckpointInitialization.hs @@ -0,0 +1,137 @@ +{-# LANGUAGE NumericUnderscores #-} +{-# LANGUAGE TypeApplications #-} + +module Test.SubscriptionCheckpointInitialization (spec) where + +import Control.Concurrent.Async qualified as Async +import Control.Lens ((&), (.~)) +import Data.Aeson qualified as Aeson +import Data.Int (Int32) +import Data.Text (Text) +import Data.Text qualified as T +import Data.Vector qualified as V +import Effectful (runEff) +import Effectful.Error.Static (runErrorNoCallStack) +import Kiroku.Store +import Test.Helpers (makeEvent, withTestStore) +import Test.Hspec + +spec :: Spec +spec = describe "subscription checkpoint initialization" $ do + it "materializes position zero for FromBeginning" $ + withTestStore $ \store -> do + let name = SubscriptionName "initialize-from-beginning" + key = SubscriptionCheckpointKey name 0 + result <- initialize store name 0 FromBeginning + result `shouldBe` Right (InitializedCheckpoint FromBeginning key (GlobalPosition 0)) + inventoryKeys <$> inventory store `shouldReturn` [(name, 0, GlobalPosition 0)] + + it "atomically materializes the current store head for FromCurrentHead" $ + withTestStore $ \store -> do + appendEvents store "initialize-current-head-events" 3 + let name = SubscriptionName "initialize-current-head" + key = SubscriptionCheckpointKey name 0 + result <- initialize store name 0 FromCurrentHead + result `shouldBe` Right (InitializedCheckpoint FromCurrentHead key (GlobalPosition 3)) + inventoryKeys <$> inventory store `shouldReturn` [(name, 0, GlobalPosition 3)] + + it "returns a typed missing result without creating a row for FailIfMissing" $ + withTestStore $ \store -> do + let name = SubscriptionName "initialize-fail-if-missing" + key = SubscriptionCheckpointKey name 0 + result <- initialize store name 0 FailIfMissing + result `shouldBe` Left (SubscriptionCheckpointMissing key) + inventoryKeys <$> inventory store `shouldReturn` [] + + it "preserves an existing row for every configured policy" $ + withTestStore $ \store -> do + appendEvents store "initialize-existing-events" 5 + let name = SubscriptionName "initialize-existing" + key = SubscriptionCheckpointKey name 0 + initialize store name 0 FromCurrentHead + `shouldReturn` Right (InitializedCheckpoint FromCurrentHead key (GlobalPosition 5)) + mapM_ (assertExisting store key) [FromBeginning, FromCurrentHead, FailIfMissing] + inventoryKeys <$> inventory store `shouldReturn` [(name, 0, GlobalPosition 5)] + + it "isolates checkpoint initialization by consumer-group member" $ + withTestStore $ \store -> do + appendEvents store "initialize-member-events" 4 + let name = SubscriptionName "initialize-members" + initialize store name 0 FromBeginning + `shouldReturn` Right (InitializedCheckpoint FromBeginning (SubscriptionCheckpointKey name 0) (GlobalPosition 0)) + initialize store name 1 FromCurrentHead + `shouldReturn` Right (InitializedCheckpoint FromCurrentHead (SubscriptionCheckpointKey name 1) (GlobalPosition 4)) + inventoryKeys <$> inventory store + `shouldReturn` [ (name, 0, GlobalPosition 0) + , (name, 1, GlobalPosition 4) + ] + + it "converges concurrent initializers on one durable winner" $ + withTestStore $ \store -> do + appendEvents store "initialize-race-events" 7 + let name = SubscriptionName "initialize-race" + policies = take 20 (cycle [FromBeginning, FromCurrentHead]) + results <- Async.mapConcurrently (initialize store name 3) policies + let successes = [initialization | Right initialization <- results] + positions = fmap checkpointInitializationPosition successes + initializedCount = length [() | InitializedCheckpoint{} <- successes] + length successes `shouldBe` length policies + initializedCount `shouldBe` 1 + positions `shouldSatisfy` \case + [] -> False + first : rest -> all (== first) rest + case positions of + [] -> expectationFailure "expected concurrent initialization results" + winner : _ -> + inventoryKeys <$> inventory store + `shouldReturn` [(name, 3, winner)] + + it "runs through the resource-backed Store interpreter" $ + withTestStore $ \store -> do + let name = SubscriptionName "initialize-resource" + key = SubscriptionCheckpointKey name 2 + result <- + runEff + . runErrorNoCallStack @StoreError + . runKirokuStoreWith store + . runStoreResource + $ initializeSubscriptionCheckpoint name 2 FromBeginning + result `shouldBe` Right (Right (InitializedCheckpoint FromBeginning key (GlobalPosition 0))) + +assertExisting :: KirokuStore -> SubscriptionCheckpointKey -> MissingCheckpointPolicy -> IO () +assertExisting store key@(SubscriptionCheckpointKey name member) policy = + initialize store name member policy + `shouldReturn` Right (ExistingCheckpoint key (GlobalPosition 5)) + +initialize :: + KirokuStore -> + SubscriptionName -> + Int32 -> + MissingCheckpointPolicy -> + IO (Either SubscriptionCheckpointMissing CheckpointInitialization) +initialize store name member policy = do + result <- runStoreIO store (initializeSubscriptionCheckpoint name member policy) + case result of + Left err -> expectationFailure ("checkpoint initialization failed: " <> show err) >> error "unreachable" + Right initialized -> pure initialized + +inventory :: KirokuStore -> IO SubscriptionCheckpointInventory +inventory store = do + result <- runStoreIO store subscriptionCheckpointInventory + case result of + Left err -> expectationFailure ("checkpoint inventory failed: " <> show err) >> error "unreachable" + Right rows -> pure rows + +inventoryKeys :: SubscriptionCheckpointInventory -> [(SubscriptionName, Int32, GlobalPosition)] +inventoryKeys (SubscriptionCheckpointInventory _ rows) = + [ (name, member, position) + | SubscriptionCheckpoint name member position _ <- V.toList rows + ] + +appendEvents :: KirokuStore -> Text -> Int -> IO () +appendEvents store stream count = do + let events = [makeEvent ("Initialize" <> T.pack (show i)) (Aeson.object []) | i <- [1 .. count]] + result <- runStoreIO store $ appendToStream (StreamName stream) NoStream events + case result of + Left err -> expectationFailure ("append failed: " <> show err) + Right _ -> pure () diff --git a/test/Test/SubscriptionCheckpointInitializationMock.hs b/test/Test/SubscriptionCheckpointInitializationMock.hs new file mode 100644 --- /dev/null +++ b/test/Test/SubscriptionCheckpointInitializationMock.hs @@ -0,0 +1,44 @@ +module Test.SubscriptionCheckpointInitializationMock (spec) where + +import Control.Monad.IO.Class (liftIO) +import Data.IORef (IORef, modifyIORef', newIORef, readIORef) +import Effectful (Eff, IOE, runEff, (:>)) +import Effectful.Dispatch.Dynamic (interpret_) +import Kiroku.Store.Effect (Store (..)) +import Kiroku.Store.Subscription (initializeSubscriptionCheckpoint) +import Kiroku.Store.Subscription.Types +import Kiroku.Store.Types (GlobalPosition (..)) +import Test.Hspec + +spec :: Spec +spec = describe "subscription checkpoint initialization mock interpreter" $ do + it "carries the closed policy and typed result through one Store effect call" $ do + calls <- newIORef (0 :: Int) + let name = SubscriptionName "mock-initialization" + key = SubscriptionCheckpointKey name 4 + expected = Right (InitializedCheckpoint FromCurrentHead key (GlobalPosition 23)) + actual <- + runEff $ + runInitializationMock calls name 4 FromCurrentHead expected $ + initializeSubscriptionCheckpoint name 4 FromCurrentHead + actual `shouldBe` expected + readIORef calls `shouldReturn` 1 + +runInitializationMock :: + (IOE :> es) => + IORef Int -> + SubscriptionName -> + Int -> + MissingCheckpointPolicy -> + Either SubscriptionCheckpointMissing CheckpointInitialization -> + Eff (Store : es) a -> + Eff es a +runInitializationMock calls expectedName expectedMember expectedPolicy expected = interpret_ $ \case + InitializeSubscriptionCheckpoint actualName actualMember actualPolicy -> do + liftIO $ do + actualName `shouldBe` expectedName + fromIntegral actualMember `shouldBe` expectedMember + actualPolicy `shouldBe` expectedPolicy + modifyIORef' calls (+ 1) + pure expected + _ -> error "unexpected Store operation in checkpoint initialization mock" diff --git a/test/Test/SubscriptionCheckpointReset.hs b/test/Test/SubscriptionCheckpointReset.hs new file mode 100644 --- /dev/null +++ b/test/Test/SubscriptionCheckpointReset.hs @@ -0,0 +1,213 @@ +{-# LANGUAGE NumericUnderscores #-} + +module Test.SubscriptionCheckpointReset (spec) where + +import Contravariant.Extras (contrazip2) +import Control.Lens ((^.)) +import Data.Aeson qualified as Aeson +import Data.Generics.Labels () +import Data.Int (Int32, Int64) +import Data.List.NonEmpty (NonEmpty (..)) +import Data.Text (Text) +import Data.Text qualified as Text +import Data.Vector qualified as Vector +import Hasql.Decoders qualified as D +import Hasql.Encoders qualified as E +import Hasql.Pool qualified as Pool +import Hasql.Session qualified as Session +import Hasql.Statement (Statement, preparable, unpreparable) +import Hasql.Transaction qualified as Tx +import Kiroku.Store +import Kiroku.Store.SQL qualified as SQL +import Test.Helpers (makeEvent, withTestStore) +import Test.Hspec + +spec :: Spec +spec = describe "subscription checkpoint reset" $ do + it "commits every persisted member and reports sorted affected and missing names exactly" $ + withTestStore $ \store -> do + appendEvents store "reset-commit-events" 10 + initializeRows store [("zeta", 1), ("alpha", 2), ("alpha", 0)] + createSentinelTable store + + result <- runStoreIO store $ runTransaction $ do + Tx.statement (1, "committed") insertSentinelStmt + resetSubscriptionCheckpointsTx + ( SubscriptionName "zeta" + :| [ SubscriptionName "missing" + , SubscriptionName "alpha" + , SubscriptionName "alpha" + ] + ) + (GlobalPosition 7) + + result + `shouldBe` Right + SubscriptionCheckpointResetReport + { resetCheckpointKeys = + Vector.fromList + [ SubscriptionCheckpointKey (SubscriptionName "alpha") 0 + , SubscriptionCheckpointKey (SubscriptionName "alpha") 2 + , SubscriptionCheckpointKey (SubscriptionName "zeta") 1 + ] + , missingSubscriptionNames = Vector.singleton (SubscriptionName "missing") + } + countSentinels store `shouldReturn` 1 + inventoryKeys <$> inventory store + `shouldReturn` [ ("alpha", 0, 7) + , ("alpha", 2, 7) + , ("zeta", 1, 7) + ] + + it "rolls back the reset and an application-table write when the transaction is condemned" $ + withTestStore $ \store -> do + appendEvents store "reset-rollback-events" 10 + initializeRows store [("rollback", 0)] + reset store (SubscriptionName "rollback" :| []) (GlobalPosition 9) + createSentinelTable store + + result <- runStoreIO store $ runTransaction $ do + Tx.statement (2, "rolled back") insertSentinelStmt + report <- + resetSubscriptionCheckpointsTx + (SubscriptionName "rollback" :| []) + (GlobalPosition 3) + Tx.condemn + pure report + + result + `shouldBe` Right + SubscriptionCheckpointResetReport + { resetCheckpointKeys = + Vector.singleton + (SubscriptionCheckpointKey (SubscriptionName "rollback") 0) + , missingSubscriptionNames = Vector.empty + } + countSentinels store `shouldReturn` 0 + inventoryKeys <$> inventory store `shouldReturn` [("rollback", 0, 9)] + + it "can rewind while later ordinary saves remain monotonic" $ + withTestStore $ \store -> do + appendEvents store "reset-rewind-events" 10 + initializeRows store [("rewind", 0)] + reset store (SubscriptionName "rewind" :| []) (GlobalPosition 8) + reset store (SubscriptionName "rewind" :| []) (GlobalPosition 4) + inventoryKeys <$> inventory store `shouldReturn` [("rewind", 0, 4)] + + saveCheckpoint store "rewind" 0 2 + inventoryKeys <$> inventory store `shouldReturn` [("rewind", 0, 4)] + saveCheckpoint store "rewind" 0 6 + inventoryKeys <$> inventory store `shouldReturn` [("rewind", 0, 6)] + + it "reports missing names without manufacturing checkpoint rows" $ + withTestStore $ \store -> do + result <- + reset + store + (SubscriptionName "absent-b" :| [SubscriptionName "absent-a"]) + (GlobalPosition 5) + result + `shouldBe` SubscriptionCheckpointResetReport + { resetCheckpointKeys = Vector.empty + , missingSubscriptionNames = + Vector.fromList + [SubscriptionName "absent-a", SubscriptionName "absent-b"] + } + inventoryKeys <$> inventory store `shouldReturn` [] + +initializeRows :: KirokuStore -> [(Text, Int32)] -> IO () +initializeRows store rows = + mapM_ initialize rows + where + initialize (name, member) = do + result <- + runStoreIO store $ + initializeSubscriptionCheckpoint + (SubscriptionName name) + member + FromBeginning + case result of + Right (Right _) -> pure () + other -> expectationFailure ("checkpoint initialization failed: " <> show other) + +reset :: + KirokuStore -> + NonEmpty SubscriptionName -> + GlobalPosition -> + IO SubscriptionCheckpointResetReport +reset store names position = do + result <- runStoreIO store $ runTransaction $ resetSubscriptionCheckpointsTx names position + case result of + Left err -> expectationFailure ("checkpoint reset failed: " <> show err) >> error "unreachable" + Right report -> pure report + +inventory :: KirokuStore -> IO SubscriptionCheckpointInventory +inventory store = do + result <- runStoreIO store subscriptionCheckpointInventory + case result of + Left err -> expectationFailure ("checkpoint inventory failed: " <> show err) >> error "unreachable" + Right rows -> pure rows + +inventoryKeys :: SubscriptionCheckpointInventory -> [(Text, Int32, Int64)] +inventoryKeys (SubscriptionCheckpointInventory _ rows) = + [ (name, member, position) + | SubscriptionCheckpoint (SubscriptionName name) member (GlobalPosition position) _ <- + Vector.toList rows + ] + +appendEvents :: KirokuStore -> Text -> Int -> IO () +appendEvents store stream count = do + let events = + [makeEvent ("Reset" <> Text.pack (show i)) (Aeson.object []) | i <- [1 .. count]] + result <- runStoreIO store $ appendToStream (StreamName stream) NoStream events + case result of + Left err -> expectationFailure ("append failed: " <> show err) + Right _ -> pure () + +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 -> expectationFailure ("ordinary checkpoint save failed: " <> show err) + Right () -> pure () + +createSentinelTable :: KirokuStore -> IO () +createSentinelTable store = do + result <- Pool.use (store ^. #pool) (Session.statement () createSentinelTableStmt) + case result of + Left err -> expectationFailure ("sentinel table creation failed: " <> show err) + Right () -> pure () + +countSentinels :: KirokuStore -> IO Int64 +countSentinels store = do + result <- Pool.use (store ^. #pool) (Session.statement () countSentinelsStmt) + case result of + Left err -> expectationFailure ("sentinel count failed: " <> show err) >> error "unreachable" + Right count -> pure count + +createSentinelTableStmt :: Statement () () +createSentinelTableStmt = + unpreparable + "CREATE TABLE public.checkpoint_reset_sentinel \ + \(id BIGINT PRIMARY KEY, value TEXT NOT NULL)" + E.noParams + D.noResult + +insertSentinelStmt :: Statement (Int64, Text) () +insertSentinelStmt = + preparable + "INSERT INTO public.checkpoint_reset_sentinel (id, value) VALUES ($1, $2)" + ( contrazip2 + (E.param (E.nonNullable E.int8)) + (E.param (E.nonNullable E.text)) + ) + D.noResult + +countSentinelsStmt :: Statement () Int64 +countSentinelsStmt = + preparable + "SELECT COUNT(*) FROM public.checkpoint_reset_sentinel" + E.noParams + (D.singleRow (D.column (D.nonNullable D.int8))) diff --git a/test/Test/SubscriptionCheckpointWorker.hs b/test/Test/SubscriptionCheckpointWorker.hs new file mode 100644 --- /dev/null +++ b/test/Test/SubscriptionCheckpointWorker.hs @@ -0,0 +1,250 @@ +{-# LANGUAGE NumericUnderscores #-} + +module Test.SubscriptionCheckpointWorker (spec) where + +import Control.Concurrent (threadDelay) +import Control.Concurrent.Async qualified as Async +import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar, tryPutMVar) +import Control.Concurrent.STM (atomically, check, modifyTVar', newTVarIO, readTVar, readTVarIO, writeTVar) +import Control.Exception (SomeException, finally, fromException, try) +import Control.Lens ((&), (.~), (^.)) +import Control.Monad (forM, void) +import Control.Monad.IO.Class (liftIO) +import Data.Aeson qualified as Aeson +import Data.Generics.Labels () +import Data.IORef (modifyIORef', newIORef, readIORef) +import Data.List (sortOn) +import Data.Text (Text) +import Data.Text qualified as T +import Data.Vector qualified as V +import Effectful (runEff) +import Kiroku.Store +import Kiroku.Store.Subscription.Effect qualified as SubEff +import Kiroku.Store.Subscription.Stream (subscriptionAckStream) +import Streamly.Data.Stream qualified as Stream +import Test.Helpers (makeEvent, waitForPublisher, waitWithTimeout, withTestStore, withTestStoreSettings) +import Test.Hspec + +spec :: Spec +spec = describe "subscription checkpoint worker policies" $ do + it "starts a non-group worker FromBeginning and reports the durable seed" $ do + eventsRef <- newIORef [] + resolutionsRef <- newIORef [] + let name = SubscriptionName "worker-from-beginning" + observe evt = case evt of + KirokuEventSubscriptionCheckpointResolved initialization NonGroup -> + modifyIORef' resolutionsRef (initialization :) + _ -> pure () + tweak settings = settings & #eventHandler .~ Just observe + withTestStoreSettings tweak $ \store -> do + appendBatch store "worker-from-beginning-events" 3 + waitForPublisher store (GlobalPosition 3) + let handler event = do + modifyIORef' eventsRef (event ^. #globalPosition :) + seen <- length <$> readIORef eventsRef + pure (if seen >= 3 then Stop else Continue) + config = + (defaultSubscriptionConfig name AllStreams handler) + { missingCheckpointPolicy = FromBeginning + } + handle <- subscribe store config + expectClean handle + + reverse <$> readIORef eventsRef + `shouldReturn` fmap GlobalPosition [1, 2, 3] + inventoryPositions store + `shouldReturn` [(SubscriptionCheckpointKey name 0, GlobalPosition 3)] + + readIORef resolutionsRef + `shouldReturn` [InitializedCheckpoint FromBeginning (SubscriptionCheckpointKey name 0) (GlobalPosition 0)] + + it "forms a clean FromCurrentHead cut while appends race startup" $ do + initializationReady <- newEmptyMVar + deliveredRef <- newIORef [] + let name = SubscriptionName "worker-current-head-race" + key = SubscriptionCheckpointKey name 0 + observe evt = case evt of + KirokuEventSubscriptionCheckpointResolved initialization NonGroup + | initializationKey initialization == key -> + void (tryPutMVar initializationReady initialization) + _ -> pure () + tweak settings = settings & #eventHandler .~ Just observe + withTestStoreSettings tweak $ \store -> do + appendBatch store "worker-current-head-race-events" 10 + waitForPublisher store (GlobalPosition 10) + gate <- newEmptyMVar + let handler event = do + modifyIORef' deliveredRef (event ^. #globalPosition :) + pure $ if event ^. #eventType == EventType "RaceSentinel" then Stop else Continue + config = + (defaultSubscriptionConfig name AllStreams handler) + { missingCheckpointPolicy = FromCurrentHead + , batchSize = 3 + } + subscribeThread <- Async.async (takeMVar gate >> subscribe store config) + appendThread <- Async.async $ do + takeMVar gate + forM [1 .. 20 :: Int] $ \i -> + appendExisting store "worker-current-head-race-events" ("Racing" <> T.pack (show i)) + putMVar gate () + putMVar gate () + handle <- Async.wait subscribeThread + initialization <- waitMVar "checkpoint resolution" initializationReady + racePositions <- Async.wait appendThread + raceTail <- case reverse racePositions of + [] -> expectationFailure "expected racing appends" >> error "unreachable" + position : _ -> pure position + let seed = checkpointInitializationPosition initialization + waitForPublisher store raceTail + finalPosition <- appendExisting store "worker-current-head-race-events" "RaceSentinel" + waitForPublisher store finalPosition + expectClean handle + + initialization `shouldBe` InitializedCheckpoint FromCurrentHead key seed + delivered <- reverse <$> readIORef deliveredRef + delivered `shouldBe` positionsAfter seed finalPosition + + it "initializes each consumer-group member independently at the current head" $ do + resolutionsVar <- newTVarIO [] + handlerCalled <- newTVarIO False + let name = SubscriptionName "worker-current-head-members" + observe evt = case evt of + KirokuEventSubscriptionCheckpointResolved initialization GroupMember{} -> + atomically (modifyTVar' resolutionsVar (initialization :)) + _ -> pure () + tweak settings = settings & #eventHandler .~ Just observe + withTestStoreSettings tweak $ \store -> do + appendBatch store "worker-current-head-members-events" 6 + waitForPublisher store (GlobalPosition 6) + let config member = + ( defaultSubscriptionConfig name AllStreams $ \_ -> do + atomically (writeTVar handlerCalled True) + pure Continue + ) + { consumerGroup = Just (ConsumerGroup member 2) + , missingCheckpointPolicy = FromCurrentHead + } + handles <- mapM (subscribe store . config) [0, 1] + atomically $ do + resolutions <- readTVar resolutionsVar + check (length resolutions >= 2) + mapM_ cancel handles + mapM_ wait handles + + resolutions <- sortOn initializationKey <$> readTVarIO resolutionsVar + resolutions + `shouldBe` [ InitializedCheckpoint FromCurrentHead (SubscriptionCheckpointKey name 0) (GlobalPosition 6) + , InitializedCheckpoint FromCurrentHead (SubscriptionCheckpointKey name 1) (GlobalPosition 6) + ] + readTVarIO handlerCalled `shouldReturn` False + inventoryPositions store + `shouldReturn` [ (SubscriptionCheckpointKey name 0, GlobalPosition 6) + , (SubscriptionCheckpointKey name 1, GlobalPosition 6) + ] + + it "preserves FailIfMissing through the bracketed plain-IO entry point" $ + withTestStore $ \store -> do + let name = SubscriptionName "worker-bracketed-missing" + key = SubscriptionCheckpointKey name 0 + config = + (defaultSubscriptionConfig name AllStreams (\_ -> pure Continue)) + { missingCheckpointPolicy = FailIfMissing + } + outcome <- withSubscription store config wait + case outcome of + Left err + | Just (SubscriptionCheckpointMissing actual) <- fromException err -> + actual `shouldBe` key + other -> expectationFailure ("expected bracketed missing refusal, got: " <> show other) + + it "preserves FromCurrentHead through the higher-order effect entry point" $ do + initializationReady <- newEmptyMVar + deliveredRef <- newIORef [] + let name = SubscriptionName "worker-effect-current-head" + key = SubscriptionCheckpointKey name 0 + observe evt = case evt of + KirokuEventSubscriptionCheckpointResolved initialization NonGroup + | initializationKey initialization == key -> + void (tryPutMVar initializationReady initialization) + _ -> pure () + tweak settings = settings & #eventHandler .~ Just observe + withTestStoreSettings tweak $ \store -> do + appendBatch store "worker-effect-current-head-events" 5 + waitForPublisher store (GlobalPosition 5) + runEff $ SubEff.runSubscription store $ do + let config = + ( defaultSubscriptionConfig name AllStreams $ \event -> do + liftIO (modifyIORef' deliveredRef (event ^. #globalPosition :)) + pure Stop + ) + { missingCheckpointPolicy = FromCurrentHead + } + handle <- SubEff.subscribe config + initialization <- liftIO (waitMVar "effect checkpoint resolution" initializationReady) + liftIO $ initialization `shouldBe` InitializedCheckpoint FromCurrentHead key (GlobalPosition 5) + position <- liftIO $ appendExisting store "worker-effect-current-head-events" "EffectFuture" + liftIO (waitForPublisher store position) + liftIO (expectClean handle) + reverse <$> readIORef deliveredRef `shouldReturn` [GlobalPosition 6] + + it "preserves FailIfMissing through the Streamly bridge" $ + withTestStore $ \store -> do + let name = SubscriptionName "worker-streamly-missing" + key = SubscriptionCheckpointKey name 0 + config = + (defaultSubscriptionConfig name AllStreams (\_ -> pure Continue)) + { missingCheckpointPolicy = FailIfMissing + } + (stream, cancelStream) <- subscriptionAckStream store config 1 + pulled <- finally (try (Stream.uncons stream)) cancelStream + case pulled of + Left err + | Just (SubscriptionCheckpointMissing actual) <- fromException (err :: SomeException) -> + actual `shouldBe` key + Left err -> expectationFailure ("expected typed Streamly refusal, got: " <> show err) + Right _ -> expectationFailure "expected Streamly bridge startup to fail" + +initializationKey :: CheckpointInitialization -> SubscriptionCheckpointKey +initializationKey = \case + ExistingCheckpoint key _ -> key + InitializedCheckpoint _ key _ -> key + +positionsAfter :: GlobalPosition -> GlobalPosition -> [GlobalPosition] +positionsAfter (GlobalPosition start) (GlobalPosition end) = + fmap GlobalPosition [start + 1 .. end] + +inventoryPositions :: KirokuStore -> IO [(SubscriptionCheckpointKey, GlobalPosition)] +inventoryPositions store = do + Right (SubscriptionCheckpointInventory _ rows) <- runStoreIO store subscriptionCheckpointInventory + pure + [ (SubscriptionCheckpointKey name member, position) + | SubscriptionCheckpoint name member position _ <- V.toList rows + ] + +appendBatch :: KirokuStore -> Text -> Int -> IO GlobalPosition +appendBatch store stream count = do + let events = [makeEvent ("History" <> T.pack (show i)) (Aeson.object []) | i <- [1 .. count]] + Right result <- runStoreIO store $ appendToStream (StreamName stream) NoStream events + pure (result ^. #globalPosition) + +appendExisting :: KirokuStore -> Text -> Text -> IO GlobalPosition +appendExisting store stream typ = do + Right result <- + runStoreIO store $ + appendToStream (StreamName stream) StreamExists [makeEvent typ (Aeson.object [])] + pure (result ^. #globalPosition) + +expectClean :: SubscriptionHandle -> IO () +expectClean handle = do + result <- waitWithTimeout 15_000_000 handle + case result of + Left message -> expectationFailure message + Right (Left err) -> expectationFailure ("subscription failed: " <> show err) + Right (Right ()) -> pure () + +waitMVar :: String -> MVar a -> IO a +waitMVar label var = do + result <- Async.race (threadDelay 10_000_000) (takeMVar var) + case result of + Left () -> expectationFailure ("timed out waiting for " <> label) >> error "unreachable" + Right value -> pure value diff --git a/test/Test/SubscriptionPauseResume.hs b/test/Test/SubscriptionPauseResume.hs --- a/test/Test/SubscriptionPauseResume.hs +++ b/test/Test/SubscriptionPauseResume.hs @@ -87,6 +87,7 @@ , overflowPolicy = PauseAndResume , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -151,6 +152,7 @@ , overflowPolicy = DropSubscription , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing diff --git a/test/Test/SubscriptionReconnect.hs b/test/Test/SubscriptionReconnect.hs --- a/test/Test/SubscriptionReconnect.hs +++ b/test/Test/SubscriptionReconnect.hs @@ -98,6 +98,7 @@ , overflowPolicy = PauseAndResume , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing diff --git a/test/Test/SubscriptionState.hs b/test/Test/SubscriptionState.hs --- a/test/Test/SubscriptionState.hs +++ b/test/Test/SubscriptionState.hs @@ -90,6 +90,7 @@ , overflowPolicy = PauseAndResume , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing @@ -132,6 +133,7 @@ , overflowPolicy = PauseAndResume , consumerGroup = Nothing , consumerGroupGuard = False + , missingCheckpointPolicy = FromBeginning , retryPolicy = defaultRetryPolicy , eventTypeFilter = AllEventTypes , selector = Nothing