kiroku-store 0.4.0.0 → 0.5.0.0

raw patch · 28 files changed

+1636/−200 lines, 28 filesdep +tastyPVP ok

version bump matches the API change (PVP)

Dependencies added: tasty

API changes (from Hackage documentation)

Files

• CHANGELOG.md +36/−1
• bench/Main.hs +4/−148
• bench/RegressionGate.hs +136/−0
• bench/ShibuyaOverhead.hs +3/−0
• kiroku-store.cabal +37/−3
• src/Kiroku/Store.hs +2/−0
• src/Kiroku/Store/Effect.hs +22/−1
• src/Kiroku/Store/Observability.hs +21/−4
• src/Kiroku/Store/Subscription.hs +29/−4
• src/Kiroku/Store/Subscription/Checkpoint.hs +70/−0
• src/Kiroku/Store/Subscription/Checkpoint/SQL.hs +160/−0
• src/Kiroku/Store/Subscription/Types.hs +66/−1
• src/Kiroku/Store/Subscription/Worker.hs +35/−14
• test/Main.hs +34/−1
• test/Test/CatchupDbErrorNoPrematureSwitch.hs +1/−0
• test/Test/ConsumerGroup.hs +1/−0
• test/Test/FailureInjection.hs +1/−0
• test/Test/PerformanceStructure.hs +257/−0
• test/Test/PublisherRestartNoRebroadcast.hs +3/−0
• test/Test/StartupFailureSurfacing.hs +43/−0
• test/Test/StreamNameLookup.hs +26/−23
• test/Test/SubscriptionCheckpointInitialization.hs +137/−0
• test/Test/SubscriptionCheckpointInitializationMock.hs +44/−0
• test/Test/SubscriptionCheckpointReset.hs +213/−0
• test/Test/SubscriptionCheckpointWorker.hs +250/−0
• test/Test/SubscriptionPauseResume.hs +2/−0
• test/Test/SubscriptionReconnect.hs +1/−0
• test/Test/SubscriptionState.hs +2/−0

@@ -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 

@@ -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'                         ]

@@ -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 ()

@@ -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

@@ -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

@@ -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

@@ -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 ->

@@ -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

@@ -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

@@ -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

@@ -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)+        )

@@ -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

@@ -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

@@ -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

@@ -49,6 +49,7 @@                         , overflowPolicy = DropSubscription                         , consumerGroup = Nothing                         , consumerGroupGuard = False+                        , missingCheckpointPolicy = FromBeginning                         , retryPolicy = defaultRetryPolicy                         , eventTypeFilter = AllEventTypes                         , selector = Nothing

@@ -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

@@ -65,6 +65,7 @@                     , overflowPolicy = DropSubscription                     , consumerGroup = Nothing                     , consumerGroupGuard = False+                    , missingCheckpointPolicy = FromBeginning                     , retryPolicy = defaultRetryPolicy                     , eventTypeFilter = AllEventTypes                     , selector = Nothing

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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 ()

@@ -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"

@@ -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)))

@@ -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

@@ -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

@@ -98,6 +98,7 @@                     , overflowPolicy = PauseAndResume                     , consumerGroup = Nothing                     , consumerGroupGuard = False+                    , missingCheckpointPolicy = FromBeginning                     , retryPolicy = defaultRetryPolicy                     , eventTypeFilter = AllEventTypes                     , selector = Nothing

@@ -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