kiroku-store 0.7.0.1 → 0.8.0.0

raw patch · 32 files changed

+211/−100 lines, 32 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +41/−0
• bench/Explain.hs +3/−1
• bench/Main.hs +3/−2
• bench/RegressionGate.hs +0/−2
• bench/ShibuyaOverhead.hs +1/−3
• kiroku-store.cabal +1/−1
• src/Kiroku/Store/Error.hs +39/−6
• src/Kiroku/Store/SQL.hs +21/−6
• test/Main.hs +32/−11
• test/Test/CatchupDbErrorNoPrematureSwitch.hs +1/−1
• test/Test/CategoryIdleNoSpin.hs +0/−1
• test/Test/Causation.hs +1/−1
• test/Test/Concurrency.hs +24/−7
• test/Test/ConsumerGroup.hs +2/−3
• test/Test/ConsumerGroupEffect.hs +0/−1
• test/Test/HistoryRetention.hs +8/−8
• test/Test/NotifyGuard.hs +0/−1
• test/Test/PerformanceStructure.hs +9/−10
• test/Test/Properties.hs +1/−1
• test/Test/PublisherCallbackResilience.hs +2/−4
• test/Test/PublisherIdleAdvance.hs +3/−3
• test/Test/PublisherRestartNoRebroadcast.hs +0/−1
• test/Test/ReadStream.hs +3/−3
• test/Test/StartupFailureSurfacing.hs +4/−4
• test/Test/StreamBridgeTermination.hs +2/−5
• test/Test/SubscriptionCheckpointInitialization.hs +0/−1
• test/Test/SubscriptionCheckpointReset.hs +3/−3
• test/Test/SubscriptionCheckpointWorker.hs +4/−4
• test/Test/SubscriptionPauseResume.hs +0/−1
• test/Test/SubscriptionRegistry.hs +0/−1
• test/Test/SubscriptionState.hs +0/−1
• test/Test/TruncateBefore.hs +3/−3

@@ -1,5 +1,46 @@ # Changelog +## 0.8.0.0 — 2026-08-16++### Breaking Changes++* `StoreError` gains a `TransientTransactionFailure` constructor carrying the+  `SQLSTATE` code and message. PostgreSQL's class-40 transaction-rollback codes+  — `40001` serialization_failure and `40P01` deadlock_detected — now map to it+  instead of to `UnexpectedServerError`. Consumers that matched+  `UnexpectedServerError "40P01"` or `UnexpectedServerError "40001"` must match+  the new constructor; every other constructor is unchanged, and exhaustive+  matches over `StoreError` need a new arm.++  The old mapping was actively misleading. `UnexpectedServerError` documents+  itself as "*not* generally retryable — investigate", while these two codes are+  precisely the ones PostgreSQL expects a client to retry: the transaction+  rolled back completely and nothing was committed. A caller following the+  documentation would have diagnosed a condition whose correct handling is to+  retry. `TransientTransactionFailure` documents the retry contract, including+  that the store already retried the append once before surfacing it.++  This changes error *classification* only. No lock ordering, SQL, or hot-path+  code changed, so append and read throughput are unaffected.++### Other Changes++* Corrected the design comment on `lockStreamsForMultiStmt`, which claimed that+  deadlocks between multi-stream and single-stream appends "are avoided". They+  are not, for streams that do not exist yet: the pre-lock matches only existing+  rows, so a multi-stream append over `[A, B]` with `B` fresh takes `A` → `$all`+  → `B`, while a concurrent single-stream append to `B` takes `B` → `$all`. The+  comment now describes the cycle and points at IR-7, which proposes closing it.+  PostgreSQL detects the deadlock and nothing is committed; the surviving+  behavior is a retry and, if the conflict repeats, a+  `TransientTransactionFailure`.+* `Test.Concurrency`'s transient-leak case asserted that these SQLSTATEs never+  reach the caller, which no bounded retry can guarantee — it failed roughly+  once in 24 runs under contention. It now asserts what the store does+  guarantee, and what its own comment always described: that a surfaced+  conflict arrives typed as retryable rather than as `UnexpectedServerError`.+  A deterministic unit test covers the mapping across all five error paths.+ ## 0.7.0.1 — 2026-08-15  ### Bug Fixes

@@ -303,7 +303,9 @@     -- AnyVersion append: hits the appendAnyVersionSQL CTE.     r1 <- runStoreIO store $ appendToStream sn AnyVersion [mkEvent "Created"]     forceOk "AnyVersion" r1-    let Right res1 = r1+    let res1 = case r1 of+            Right ok -> ok+            Left e -> error ("AnyVersion append failed: " <> show e)     -- ExactVersion append against the same stream: hits the     -- appendExpectedVersionSQL CTE with a non-trivial conflict check.     r2 <-

@@ -26,7 +26,6 @@ import Hasql.Transaction qualified as Tx import Hasql.Transaction.Sessions qualified as TxSessions import Kiroku.Store-import Kiroku.Store.SQL qualified as SQL import Kiroku.Test.Postgres (migrateTestDatabase, withMigratedTestDatabase, withSharedMigratedPostgres) import Test.Tasty.Bench @@ -853,7 +852,9 @@                                 sn <- nextStream "bench-seq"                                 r0 <- runStoreIO store $ appendToStream sn NoStream [makeEvent "Init"]                                 forceAppend r0-                                let Right res0 = r0+                                let res0 = case r0 of+                                        Right ok -> ok+                                        Left e -> error ("Sequential append failed: " <> show e)                                 let go _ 0 = pure ()                                     go v n = do                                         r' <- runStoreIO store $ appendToStream sn (ExactVersion v) [makeEvent "Seq"]

@@ -13,9 +13,7 @@ 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

@@ -20,7 +20,7 @@ import Data.Aeson qualified as Aeson import Data.Generics.Labels () import Data.HashMap.Strict qualified as HashMap-import Data.IORef (atomicModifyIORef', newIORef, readIORef)+import Data.IORef (atomicModifyIORef', newIORef) import Data.List (sort) import Data.Text (Text) import Data.Text qualified as T@@ -28,9 +28,7 @@ import Effectful (Eff, IOE, liftIO, runEff, (:>)) import EphemeralPg qualified as Pg import Kiroku.Store-import Kiroku.Store.Subscription (subscribe) import Kiroku.Store.Subscription.Stream (subscriptionStream)-import Kiroku.Store.Subscription.Types (OverflowPolicy (..), SubscriptionConfigM (..), SubscriptionHandleM (..)) import Shibuya.Adapter (Adapter (..)) import Shibuya.App (ProcessorId (..), defaultAppConfig, mkProcessor, runApp, stopApp) import Shibuya.Core.Ack (AckDecision (..))

@@ -1,6 +1,6 @@ cabal-version:   3.0 name:            kiroku-store-version:         0.7.0.1+version:         0.8.0.0 synopsis:        High-performance PostgreSQL event store description:   Kiroku is a PostgreSQL-backed event store for Haskell applications. It

@@ -129,11 +129,30 @@       description for diagnostics. Retryable in most cases.       -}       ConnectionLost !Text+    | {- | PostgreSQL aborted the transaction with a class-40+      (transaction rollback) @SQLSTATE@: @40001@ serialization_failure+      or @40P01@ deadlock_detected. The first 'Text' is the @SQLSTATE@+      code, the second is the human-readable message.++      __Retryable.__ PostgreSQL rolled the transaction back completely+      and expects the client to run it again; nothing was committed.+      The store already retries an append once on these codes before+      surfacing them, so reaching a caller means the conflict repeated+      — retry with backoff rather than escalating.++      Appends are safe to retry when the caller supplies+      'Kiroku.Store.Types.EventData.eventId' values, which makes the+      retry idempotent; see the "Idempotent retries" note on+      'Kiroku.Store.Append.appendToStream'.+      -}+      TransientTransactionFailure !Text !Text     | {- | PostgreSQL raised a server error whose @SQLSTATE@ code is       outside the set this store recognises (currently @23505@-      unique violation and @23503@ foreign key violation). The first-      'Text' is the @SQLSTATE@ code, the second is the human-readable-      message. This is *not* generally retryable — investigate.+      unique violation, @23503@ foreign key violation, and the+      class-40 codes carried by 'TransientTransactionFailure'). The+      first 'Text' is the @SQLSTATE@ code, the second is the+      human-readable message. This is *not* generally retryable —+      investigate.       -}       UnexpectedServerError !Text !Text     | {- | Catch-all for everything not matched by a more specific@@ -214,8 +233,11 @@         PoolAcquisitionTimeout     SessionUsageError sessionErr ->         case extractServerError (SessionUsageError sessionErr) of-            Just (Errors.ServerError code message _ _ _) ->-                UnexpectedServerError code message+            Just (Errors.ServerError code message _ _ _)+                | isTransientTransactionCode code ->+                    TransientTransactionFailure code message+                | otherwise ->+                    UnexpectedServerError code message             Nothing ->                 ConnectionError ("Session error: " <> T.pack (show sessionErr)) @@ -255,6 +277,7 @@ mapServerError streamName expected (Errors.ServerError code message detail _hint _position)     | code == "23505" = mapUniqueViolation streamName expected message detail     | code == "23503" = StreamNotFound (StreamName streamName)+    | isTransientTransactionCode code = TransientTransactionFailure code message     | otherwise = UnexpectedServerError code message  {- | Map a unique_violation (23505) to an StoreError.@@ -476,11 +499,21 @@             _ -> Nothing     _ -> Nothing +{- | True for PostgreSQL class-40 (transaction rollback) @SQLSTATE@ codes:+@40001@ serialization_failure and @40P01@ deadlock_detected.++These are the codes hasql-transaction retries, and the ones the store maps to+'TransientTransactionFailure'. Kept as one predicate so the retry decision and+the error classification can never drift apart.+-}+isTransientTransactionCode :: Text -> Bool+isTransientTransactionCode code = code == "40001" || code == "40P01"+ -- | True for PostgreSQL transient transaction aborts retried by hasql-transaction. isTransientSerializationError :: UsageError -> Bool isTransientSerializationError usageErr =     case extractServerError usageErr of         Just (Errors.ServerError code _ _ _ _) ->-            code == "40001" || code == "40P01"+            isTransientTransactionCode code         Nothing ->             False

@@ -1122,14 +1122,29 @@ concurrent multi-stream transactions that touch overlapping streams in different orders. -Streams that don't yet exist (NoStream variant on a fresh stream) are not-matched by the WHERE clause, so they aren't pre-locked here; concurrent-INSERTs of a fresh stream serialize on the unique index on @stream_name@. \$all is intentionally NOT included in the pre-lock — its row lock is acquired by each per-stream CTE inside the transaction, after the source-stream's row lock, so deadlocks between multi-stream and single-stream-transactions are avoided as long as both lock kinds in the same order-(source-first, then $all). See EP-1 F4.+stream's row lock. See EP-1 F4.++Streams that don't yet exist (NoStream variant on a fresh stream) are not+matched by the WHERE clause, so they aren't pre-locked here. Concurrent+INSERTs of a fresh stream do serialize on the unique index on+@stream_name@, but that serialization does /not/ preserve the+source-before-$all acquisition order this pre-lock exists to establish,+and a multi-stream append can still deadlock against a concurrent+single-stream append:++  * multi locks stream A, then @$all@ (both inside A's CTE), then stream B;+  * a single-stream append to the still-fresh B locks B, then wants @$all@.++Each then waits on the other. PostgreSQL detects it and aborts one side+with @40P01@; the interpreter retries the append once, and a repeated+conflict reaches the caller as+'Kiroku.Store.Error.TransientTransactionFailure', which is documented+retryable. Eliminating the cycle would mean establishing every source+lock before any @$all@ lock, including for streams that do not exist yet+— tracked as IR-7, deliberately not done here because it is a throughput+question that needs benchmarking, not a correctness gap. -} lockStreamsForMultiStmt :: Statement (Vector Text) () lockStreamsForMultiStmt =

@@ -7,9 +7,8 @@ import Control.Exception (SomeException) import Control.Exception qualified import Control.Lens ((&), (.~), (^.))-import Control.Monad (unless)+import Control.Monad (forM_, unless) import Control.Monad.IO.Class (liftIO)-import Data.Aeson (Value (..)) import Data.Aeson qualified as Aeson import Data.Generics.Labels () import Data.IORef (modifyIORef', newIORef, readIORef, writeIORef)@@ -22,10 +21,8 @@ import Hasql.Errors qualified as Errors import Hasql.Pool (UsageError (..)) import Kiroku.Store-import Kiroku.Store.Error (extractStreamNameFromDetail) import Kiroku.Store.Subscription.Effect qualified as SubEff import Kiroku.Store.Subscription.EventPublisher (publisherPosition)-import Kiroku.Store.Subscription.Types (OverflowPolicy (..), SubscriptionConfigM (..), SubscriptionOverflowed (..)) import Kiroku.Test.Postgres (withMigratedTestDatabase) import Test.CatchupDbErrorNoPrematureSwitch qualified as CatchupDbErrorNoPrematureSwitch import Test.Category qualified as Category@@ -156,12 +153,12 @@              describe "empty event batches" $ do                 it "rejects NoStream without creating a phantom stream or advancing $all" $ \store -> do-                    Right beforeAll <- runStoreIO store $ readAllForward (GlobalPosition 0) 10+                    Right allBefore <- runStoreIO store $ readAllForward (GlobalPosition 0) 10                     result <- runStoreIO store $ appendToStream (StreamName "empty-nostream") NoStream []                     result `shouldBe` Left (EmptyAppendBatch (StreamName "empty-nostream"))                     runStoreIO store (getStream (StreamName "empty-nostream")) `shouldReturn` Right Nothing-                    Right afterAll <- runStoreIO store $ readAllForward (GlobalPosition 0) 10-                    fmap (^. #globalPosition) (V.toList afterAll) `shouldBe` fmap (^. #globalPosition) (V.toList beforeAll)+                    Right allAfter <- runStoreIO store $ readAllForward (GlobalPosition 0) 10+                    fmap (^. #globalPosition) (V.toList allAfter) `shouldBe` fmap (^. #globalPosition) (V.toList allBefore)                  it "rejects empty batches for existing-stream expectations without changing version" $ \store -> do                     Right _ <- runStoreIO store $ appendToStream (StreamName "empty-existing") NoStream [makeEvent "Created" (Aeson.object [])]@@ -999,10 +996,10 @@             it "removes orphan event payloads from the events table" $ \store -> do                 let evts = map (\i -> makeEvent ("F1Orphan" <> T.pack (show i)) (Aeson.object [])) [1 .. 3 :: Int]                 Right _ <- runStoreIO store $ appendToStream (StreamName "f1-orphan") NoStream evts-                before <- countEvents store+                countBefore <- countEvents store                 Right _ <- runStoreIO store $ hardDeleteStream (StreamName "f1-orphan")-                after <- countEvents store-                (before - after) `shouldBe` 3+                countAfter <- countEvents store+                (countBefore - countAfter) `shouldBe` 3              it "removes dead letters for orphaned events before deleting their payloads" $ \store -> do                 Right _ <- runStoreIO store $ appendToStream (StreamName "dl-hard-source") NoStream [makeEvent "DeadLettered" (Aeson.object [])]@@ -1293,7 +1290,7 @@                 let handler1 evt = do                         modifyIORef' firstRef (evt :)                         putMVar firstSeen ()-                        takeMVar block+                        _ <- takeMVar block                         pure Continue                     cfg1 =                         SubscriptionConfig@@ -1950,6 +1947,30 @@             isTransientSerializationError (serverUsage "40001") `shouldBe` True             isTransientSerializationError (serverUsage "23505") `shouldBe` False             isTransientSerializationError AcquisitionTimeoutUsageError `shouldBe` False++        -- The store retries an append once on these codes, so a caller only+        -- sees one when the conflict repeated. It must arrive as the retryable+        -- constructor: 'UnexpectedServerError' documents itself as "not+        -- generally retryable — investigate", which would send a caller to+        -- diagnose a condition whose correct handling is to retry.+        it "maps both transient SQLSTATEs to TransientTransactionFailure" $ do+            forM_ ["40001", "40P01"] $ \code -> do+                mapUsageError "orders-1" AnyVersion (serverUsage code)+                    `shouldBe` TransientTransactionFailure code "server error"+                mapGenericUsageError (serverUsage code)+                    `shouldBe` TransientTransactionFailure code "server error"+                mapTransactionUsageError (serverUsage code)+                    `shouldBe` TransientTransactionFailure code "server error"+                mapLinkUsageError (StreamName "orders-1") (serverUsage code)+                    `shouldBe` TransientTransactionFailure code "server error"+                attributeMultiStreamError+                    [(StreamName "orders-1", AnyVersion)]+                    (serverUsage code)+                    `shouldBe` TransientTransactionFailure code "server error"++        it "leaves non-transient server codes on UnexpectedServerError" $ do+            mapGenericUsageError (serverUsage "42883")+                `shouldBe` UnexpectedServerError "42883" "server error"      -- =================================================================     -- Notifier reconnection tests (EP-3 F1)

@@ -1,6 +1,6 @@ module Test.CatchupDbErrorNoPrematureSwitch (spec) where -import Control.Concurrent.MVar (newEmptyMVar, tryPutMVar)+import Control.Concurrent.MVar (newEmptyMVar) import Control.Concurrent.STM (atomically, check, newTVarIO, readTVar, writeTVar) import Control.Lens ((&), (.~), (^.)) import Data.Aeson qualified as Aeson

@@ -41,7 +41,6 @@ import Data.Generics.Labels () import Data.Text qualified as T import Kiroku.Store-import Kiroku.Store.Subscription.Types (ConsumerGroup (..), SubscriptionConfigM (..)) import Test.Helpers (caughtUpEventHandler, makeEvent, waitForPublisher, waitForSubscriptionLive, withTestStoreSettings) import Test.Hspec 

@@ -30,7 +30,7 @@ spec = around withTestStore $ do     describe "findCausationDescendants" $ do         it "returns the seed event and every descendant in global-position order" $ \store -> do-            uuids@[uA, uB, uC, uD, uE] <- replicateUuids 5+            uuids@[uA, _uB, _uC, _uD, _uE] <- replicateUuids 5             appendChain store uuids             Right found <- runStoreIO store $ findCausationDescendants (EventId uA)             eventIds found `shouldBe` map EventId uuids

@@ -95,10 +95,10 @@                     [ makeEvent (label <> "-" <> T.pack (show i)) (Aeson.object [])                     | i <- [1 .. n]                     ]-            Right r10 <- runStoreIO store $ appendToStream stream NoStream (mkBatch "Batch10" 10)+            Right r10 <- runStoreIO store $ appendToStream stream NoStream (mkBatch "Batch10" (10 :: Int))             (r10 ^. #streamVersion) `shouldBe` StreamVersion 10             (r10 ^. #globalPosition) `shouldBe` GlobalPosition 10-            Right r100 <- runStoreIO store $ appendToStream stream AnyVersion (mkBatch "Batch100" 100)+            Right r100 <- runStoreIO store $ appendToStream stream AnyVersion (mkBatch "Batch100" (100 :: Int))             (r100 ^. #streamVersion) `shouldBe` StreamVersion 110             (r100 ^. #globalPosition) `shouldBe` GlobalPosition 110             Right streamEvents <- runStoreIO store $ readStreamForward stream (StreamVersion 0) 200@@ -152,7 +152,7 @@             Right _ <- runStoreIO store $ appendToStream (StreamName "rollback-a") NoStream [makeEvent "init-a" (Aeson.object [])]             Right _ <- runStoreIO store $ appendToStream (StreamName "rollback-b") NoStream [makeEvent "init-b" (Aeson.object [])]             beforeCount <- countEvents store-            Right beforeAll <- runStoreIO store $ readAllForward (GlobalPosition 0) 100+            Right allBefore <- runStoreIO store $ readAllForward (GlobalPosition 0) 100             result <-                 runStoreIO store $                     appendMultiStream@@ -163,8 +163,8 @@                 Left (DuplicateEvent _) -> pure ()                 other -> expectationFailure ("duplicate event should abort the multi-stream transaction, got: " <> show other)             countEvents store `shouldReturn` beforeCount-            Right afterAll <- runStoreIO store $ readAllForward (GlobalPosition 0) 100-            globalPositions afterAll `shouldBe` globalPositions beforeAll+            Right allAfter <- runStoreIO store $ readAllForward (GlobalPosition 0) 100+            globalPositions allAfter `shouldBe` globalPositions allBefore             Right streamA <- runStoreIO store $ readStreamForward (StreamName "rollback-a") (StreamVersion 0) 100             Right streamB <- runStoreIO store $ readStreamForward (StreamName "rollback-b") (StreamVersion 0) 100             streamVersions streamA `shouldBe` [1]@@ -253,6 +253,16 @@     -- The race can pass vacuously on fast machines, but it must never surface     -- PostgreSQL's transient transaction SQLSTATEs (40001/40P01) to callers as     -- UnexpectedServerError.+    --+    -- Note what is and is not asserted. The store retries once, and no bounded+    -- retry can promise that a repeated conflict never reaches the caller --+    -- under contention this shape does deadlock, because a multi-stream append+    -- whose later stream is fresh takes $all before that stream's row while a+    -- single-stream append takes them in the opposite order (see IR-7). So a+    -- surfaced conflict is allowed; what is not allowed is surfacing it as+    -- UnexpectedServerError, which documents itself as "not generally+    -- retryable -- investigate" and would send a caller to diagnose a+    -- condition whose correct handling is to retry.     it "single-stream append retry does not leak transient SQLSTATEs" $         withTestStore $ \store -> do             forM_ [1 .. 25 :: Int] $ \i -> do@@ -284,9 +294,16 @@  assertNoTransientLeak :: String -> Either StoreError a -> IO () assertNoTransientLeak label = \case-    Left (UnexpectedServerError code _)+    Left (UnexpectedServerError code message)         | code == "40P01" || code == "40001" ->-            expectationFailure (label <> ": transient SQLSTATE leaked as UnexpectedServerError " <> show code)+            expectationFailure+                ( label+                    <> ": transient SQLSTATE "+                    <> show code+                    <> " surfaced as UnexpectedServerError instead of "+                    <> "TransientTransactionFailure: "+                    <> show message+                )     _ ->         pure () 

@@ -33,7 +33,6 @@ import Kiroku.Store import Kiroku.Store.SQL qualified as SQL import Kiroku.Store.Subscription.Stream (subscriptionStream)-import Kiroku.Store.Subscription.Types (ConsumerGroup (..), SubscriptionConfigM (..)) import Kiroku.Test.Postgres (withMigratedTestDatabase) import Streamly.Data.Stream qualified as Stream import Test.Helpers (makeEvent, waitForPublisher, waitWithTimeout, withTestStore, withTestStoreSettings)@@ -197,7 +196,7 @@     it "$all group partitions the whole store across members" $         withTestStore $ \store -> do             let cats = ["acct", "user", "order"]-                perCat = 10+                perCat = 10 :: Int                 perStream = 2                 streams = [c <> "-" <> T.pack (show i) | c <- cats, i <- [1 .. perCat]]                 total = length streams * perStream@@ -228,7 +227,7 @@      it "resumes member 2 from its own (name, member) checkpoint" $         withTestStore $ \store -> do-            let nStreams = 60+            let nStreams = 60 :: Int                 streams = ["rz-" <> T.pack (show i) | i <- [1 .. nStreams]]             seed store streams 1             waitForPublisher store (GlobalPosition (fromIntegral nStreams))

@@ -52,7 +52,6 @@ import Kiroku.Store import Kiroku.Store.SQL qualified as SQL import Kiroku.Store.Subscription.Effect qualified as SubEff-import Kiroku.Store.Subscription.Types (ConsumerGroup (..), SubscriptionConfigM (..)) import Test.Helpers (makeEvent, waitForPublisher, waitWithTimeout, withTestStore) import Test.Hspec 

@@ -158,7 +158,7 @@             $ \store -> do                 let stream = StreamName "history-retention-hard-delete"                 Right _ <- runStoreIO store $ appendToStream stream NoStream [makeEvent "Protected" (Aeson.object [])]-                before <- countEvents store+                countBefore <- countEvents store                 Right first <- runStoreIO store $ acquireHistoryRetentionLease (request "first" "protect" 60)                 Right second <- runStoreIO store $ acquireHistoryRetentionLease (request "second" "protect" 60)                 blocked <- runStoreIO store $ hardDeleteStream stream@@ -167,7 +167,7 @@                         actual `shouldBe` stream                         activeLeaseCount `shouldBe` 2                     other -> expectationFailure ("expected typed retention conflict, got " <> show other)-                countEvents store `shouldReturn` before+                countEvents store `shouldReturn` countBefore                 Right (Just _) <- runStoreIO store $ getStream stream                 Right HistoryRetentionReleased{} <- runStoreIO store $ releaseHistoryRetentionLease (leaseHandle first)                 stillBlocked <- runStoreIO store $ hardDeleteStream stream@@ -186,7 +186,7 @@         it "serializes lease-first acquisition ahead of raw deletion" $             withTestStore $ \store -> do                 Right _ <- runStoreIO store $ appendToStream (StreamName "raw-race-lease-first") NoStream [makeEvent "Raw" (Aeson.object [])]-                before <- countStreamEvents store+                countBefore <- countStreamEvents store                 acquisition <- Async.async $ runStoreIO store $ runTransaction $ do                     lease <- acquireHistoryRetentionLeaseTx (request "raw-race" "lease-first" 60)                     _ <- Tx.statement () holdCoordinatorStmt@@ -201,7 +201,7 @@                 acquired `shouldSatisfy` \case Right HistoryRetentionLease{} -> True; _ -> False                 rejected <- waitWithin "lease-first raw deletion" deletion                 rejected `shouldSatisfy` hasSqlState "KR001"-                countStreamEvents store `shouldReturn` before+                countStreamEvents store `shouldReturn` countBefore          it "serializes delete-first maintenance ahead of post-delete acquisition" $             withTestStore $ \store -> do@@ -221,11 +221,11 @@         it "rejects GUC-enabled DELETE with KR001 and permits it after release" $             withTestStore $ \store -> do                 Right _ <- runStoreIO store $ appendToStream (StreamName "raw-delete") NoStream (replicate 2 (makeEvent "Raw" (Aeson.object [])))-                before <- countStreamEvents store+                countBefore <- countStreamEvents store                 Right lease <- runStoreIO store $ acquireHistoryRetentionLease (request "raw" "delete" 60)                 rejected <- runRawDestruction store rawDeleteStreamEventsStmt                 rejected `shouldSatisfy` hasSqlState "KR001"-                countStreamEvents store `shouldReturn` before+                countStreamEvents store `shouldReturn` countBefore                 Right HistoryRetentionReleased{} <- runStoreIO store $ releaseHistoryRetentionLease (leaseHandle lease)                 runRawDestruction store rawDeleteStreamEventsStmt `shouldReturn` Right ()                 countStreamEvents store `shouldReturn` 0@@ -233,11 +233,11 @@         it "rejects GUC-enabled TRUNCATE with KR001 and permits it after release" $             withTestStore $ \store -> do                 Right _ <- runStoreIO store $ appendToStream (StreamName "raw-truncate") NoStream [makeEvent "Raw" (Aeson.object [])]-                before <- countEvents store+                countBefore <- countEvents store                 Right lease <- runStoreIO store $ acquireHistoryRetentionLease (request "raw" "truncate" 60)                 rejected <- runRawDestruction store rawTruncateDataStmt                 rejected `shouldSatisfy` hasSqlState "KR001"-                countEvents store `shouldReturn` before+                countEvents store `shouldReturn` countBefore                 Right HistoryRetentionReleased{} <- runStoreIO store $ releaseHistoryRetentionLease (leaseHandle lease)                 runRawDestruction store rawTruncateDataStmt `shouldReturn` Right ()                 countEvents store `shouldReturn` 0

@@ -16,7 +16,6 @@ import Hasql.Notifications qualified as Notifications import Kiroku.Store import Kiroku.Test.Postgres (withMigratedTestDatabase)-import System.IO.Error (userError) import Test.Helpers (makeEvent) import Test.Hspec 

@@ -8,7 +8,6 @@ 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.Int (Int64)@@ -36,30 +35,30 @@         it "rejects an empty appendToStream batch before pool checkout" $ do             checkouts <- newIORef (0 :: Int)             withObservedStore checkouts $ \store -> do-                before <- readIORef checkouts+                checkoutsBefore <- readIORef checkouts                 result <- runStoreIO store $ appendToStream (StreamName "performance-empty-append") AnyVersion []-                after <- readIORef checkouts+                checkoutsAfter <- readIORef checkouts                 result `shouldBe` Left (EmptyAppendBatch (StreamName "performance-empty-append"))-                after - before `shouldBe` 0+                checkoutsAfter - checkoutsBefore `shouldBe` 0          it "returns an empty appendMultiStream result before pool checkout" $ do             checkouts <- newIORef (0 :: Int)             withObservedStore checkouts $ \store -> do-                before <- readIORef checkouts+                checkoutsBefore <- readIORef checkouts                 result <- runStoreIO store $ appendMultiStream []-                after <- readIORef checkouts+                checkoutsAfter <- readIORef checkouts                 result `shouldBe` Right []-                after - before `shouldBe` 0+                checkoutsAfter - checkoutsBefore `shouldBe` 0          it "rejects invalid retention requests before pool checkout" $ do             checkouts <- newIORef (0 :: Int)             withObservedStore checkouts $ \_store -> do-                before <- readIORef checkouts+                checkoutsBefore <- readIORef checkouts                 mkHistoryRetentionLeaseOwner "" `shouldSatisfy` either (const True) (const False)                 mkHistoryRetentionLeaseReason "" `shouldSatisfy` either (const True) (const False)                 mkHistoryRetentionInventoryLimit 0 `shouldSatisfy` either (const True) (const False)-                after <- readIORef checkouts-                after - before `shouldBe` 0+                checkoutsAfter <- readIORef checkouts+                checkoutsAfter - checkoutsBefore `shouldBe` 0          it "keeps every ordinary statement free of retention coordination" $ do             let ordinarySql =

@@ -35,7 +35,7 @@ import Kiroku.Store import Test.Helpers (makeEvent, withTestStore) import Test.Hspec-import Test.Hspec.Hedgehog (PropertyT, hedgehog, modifyMaxSuccess)+import Test.Hspec.Hedgehog (hedgehog, modifyMaxSuccess)  -- | Operations the property generators can produce. data Op

@@ -1,5 +1,4 @@ {-# LANGUAGE DeriveAnyClass #-}-{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE OverloadedStrings #-}  module Test.PublisherCallbackResilience (spec) where@@ -7,11 +6,10 @@ import Control.Concurrent (threadDelay) import Control.Concurrent.Async qualified as Async import Control.Concurrent.MVar (newEmptyMVar, tryPutMVar)-import Control.Concurrent.STM (atomically, check, newTVarIO, readTVar, writeTVar)+import Control.Concurrent.STM (atomically, newTVarIO, readTVar, writeTVar) import Control.Exception (Exception, throwIO) import Control.Lens ((&), (.~), (^.)) import Data.Aeson qualified as Aeson-import Data.Data (Typeable) import Data.Generics.Labels () import Data.IORef (atomicModifyIORef', modifyIORef', newIORef, readIORef) import Kiroku.Store@@ -19,7 +17,7 @@ import Test.Hspec  data CallbackBoom = CallbackBoom-    deriving stock (Show, Typeable)+    deriving stock (Show)     deriving anyclass (Exception)  timeoutMicros :: Int

@@ -91,11 +91,11 @@                 cancel                 $ \_handle -> do                     within "category subscription live" (waitForSubscriptionLive caughtUp)-                    before <- readIORef counter+                    countBefore <- readIORef counter                     tailPos <- appendEvents store 30 "pubidleb"                     waitForPublisher store tailPos-                    after <- readIORef counter-                    (after - before) `shouldBe` 0+                    countAfter <- readIORef counter+                    (countAfter - countBefore) `shouldBe` 0                     publisherSubscriberCount store `shouldReturn` 0                      Right wakePos <-

@@ -11,7 +11,6 @@ import Data.IORef (modifyIORef', newIORef, readIORef) import Data.Text qualified as T import Kiroku.Store-import Kiroku.Store.Subscription.Types (OverflowPolicy (..), SubscriptionConfigM (..)) import Kiroku.Test.Postgres (withMigratedTestDatabase) import Test.Helpers (caughtUpEventHandler, makeEvent, waitForPublisher, waitForSubscriptionLive, waitWithTimeout) import Test.Hspec

@@ -107,14 +107,14 @@                     events = [makeEvent (T.pack ("E" <> show i)) (Aeson.object []) | i <- [1 .. 5 :: Int]]                 Right appendResult <- runStoreIO store $ appendToStream name NoStream events                 waitForPublisher store (appendResult ^. #globalPosition)-                before <- readIORef ref+                countBefore <- readIORef ref                 streamed <- runStoreIO store $ Stream.toList (readStreamForwardStream name (StreamVersion 0) 2)-                after <- readIORef ref+                countAfter <- readIORef ref                 case streamed of                     Right xs -> do                         map (^. #streamVersion) xs                             `shouldBe` [StreamVersion 1, StreamVersion 2, StreamVersion 3, StreamVersion 4, StreamVersion 5]-                        after - before `shouldBe` 3+                        countAfter - countBefore `shouldBe` 3                     Left err -> expectationFailure ("Unexpected error: " <> show err)      describe "eventExistsInStream" $

@@ -3,7 +3,7 @@ module Test.StartupFailureSurfacing (spec) where  import Control.Concurrent.Async qualified as Async-import Control.Concurrent.STM (atomically, newTVarIO, readTVar, readTVarIO, writeTVar)+import Control.Concurrent.STM (atomically, newTVarIO, readTVarIO, writeTVar) import Control.Exception (SomeException, fromException) import Control.Lens ((&), (.~), (^.)) import Control.Monad (replicateM_)@@ -87,9 +87,9 @@             -- 200 iterations gives async exceptions many chances to land in the             -- pre-fork window without making the test expensive on normal runs.             replicateM_ 200 $ do-                pending <- Async.async (subscribe store cfg)-                Async.cancel pending-                outcome <- Async.waitCatch pending+                pendingAsync <- Async.async (subscribe store cfg)+                Async.cancel pendingAsync+                outcome <- Async.waitCatch pendingAsync                 case outcome of                     Right handle -> cancel handle >> (() <$ wait handle)                     Left (_ :: SomeException) -> pure ()

@@ -1,5 +1,4 @@ {-# LANGUAGE DeriveAnyClass #-}-{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE OverloadedStrings #-}  module Test.StreamBridgeTermination (spec) where@@ -10,9 +9,7 @@ import Control.Exception (Exception, SomeException, finally, fromException, throwIO, try) import Control.Lens ((^.)) import Data.Aeson qualified as Aeson-import Data.Data (Typeable) import Data.Generics.Labels ()-import Hasql.Pool qualified as Pool import Kiroku.Store import Kiroku.Store.Subscription.Stream (AckItem (..), InvalidStreamBufferSize (..), subscriptionAckStream) import Kiroku.Store.Subscription.Worker (withFetchBatchHookForTest)@@ -21,7 +18,7 @@ import Test.Hspec  data TestBoom = TestBoom-    deriving stock (Show, Typeable)+    deriving stock (Show)     deriving anyclass (Exception)  timeoutMicros :: Int@@ -91,7 +88,7 @@         withTestStore $ \store -> do             let cfg = defaultSubscriptionConfig (SubscriptionName "bridge-full-cancel-sub") AllStreams (\_ -> pure Continue)             (stream0, cancelStream) <- subscriptionAckStream store cfg 1-            pos1 <- appendOne store (StreamName "bridge-full-cancel-1") (EventType "BridgeFullCancel1")+            _pos1 <- appendOne store (StreamName "bridge-full-cancel-1") (EventType "BridgeFullCancel1")             pos2 <- appendOne store (StreamName "bridge-full-cancel-2") (EventType "BridgeFullCancel2")             waitForPublisher store pos2 

@@ -4,7 +4,6 @@ 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)

@@ -63,7 +63,7 @@         withTestStore $ \store -> do             appendEvents store "reset-rollback-events" 10             initializeRows store [("rollback", 0)]-            reset store (SubscriptionName "rollback" :| []) (GlobalPosition 9)+            _ <- reset store (SubscriptionName "rollback" :| []) (GlobalPosition 9)             createSentinelTable store              result <- runStoreIO store $ runTransaction $ do@@ -90,8 +90,8 @@         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)+            _ <- reset store (SubscriptionName "rewind" :| []) (GlobalPosition 8)+            _ <- reset store (SubscriptionName "rewind" :| []) (GlobalPosition 4)             inventoryKeys <$> inventory store `shouldReturn` [("rewind", 0, 4)]              saveCheckpoint store "rewind" 0 2

@@ -37,7 +37,7 @@                 _ -> pure ()             tweak settings = settings & #eventHandler .~ Just observe         withTestStoreSettings tweak $ \store -> do-            appendBatch store "worker-from-beginning-events" 3+            _ <- appendBatch store "worker-from-beginning-events" 3             waitForPublisher store (GlobalPosition 3)             let handler event = do                     modifyIORef' eventsRef (event ^. #globalPosition :)@@ -70,7 +70,7 @@                 _ -> pure ()             tweak settings = settings & #eventHandler .~ Just observe         withTestStoreSettings tweak $ \store -> do-            appendBatch store "worker-current-head-race-events" 10+            _ <- appendBatch store "worker-current-head-race-events" 10             waitForPublisher store (GlobalPosition 10)             gate <- newEmptyMVar             let handler event = do@@ -114,7 +114,7 @@                 _ -> pure ()             tweak settings = settings & #eventHandler .~ Just observe         withTestStoreSettings tweak $ \store -> do-            appendBatch store "worker-current-head-members-events" 6+            _ <- appendBatch store "worker-current-head-members-events" 6             waitForPublisher store (GlobalPosition 6)             let config member =                     ( defaultSubscriptionConfig name AllStreams $ \_ -> do@@ -169,7 +169,7 @@                 _ -> pure ()             tweak settings = settings & #eventHandler .~ Just observe         withTestStoreSettings tweak $ \store -> do-            appendBatch store "worker-effect-current-head-events" 5+            _ <- appendBatch store "worker-effect-current-head-events" 5             waitForPublisher store (GlobalPosition 5)             runEff $ SubEff.runSubscription store $ do                 let config =

@@ -43,7 +43,6 @@ import Hasql.Session qualified as Session import Kiroku.Store import Kiroku.Store.SQL qualified as SQL-import Kiroku.Store.Subscription.Types (OverflowPolicy (..), SubscriptionConfigM (..), SubscriptionOverflowed (..)) import Test.Helpers (makeEvent, waitForPublisher, waitWithTimeout, withTestStoreSettings) import Test.Hspec 

@@ -34,7 +34,6 @@ import Data.Text (Text) import Kiroku.Store import Kiroku.Store.Subscription.EventPublisher qualified as Pub-import Kiroku.Store.Subscription.Types (SubscriptionConfigM (..)) import Test.Helpers (makeEvent, waitForPublisher, withTestStore) import Test.Hspec 

@@ -31,7 +31,6 @@ import Kiroku.Store import Kiroku.Store.SQL qualified as SQL import Kiroku.Store.Subscription.Fsm (SubscriptionState (..))-import Kiroku.Store.Subscription.Types (SubscriptionConfigM (..)) import Test.Helpers (caughtUpEventHandler, makeEvent, waitForPublisher, waitForSubscriptionLive, waitWithTimeout, withTestStoreSettings) import Test.Hspec 

@@ -51,7 +51,7 @@         it "leaves the $all global log and category reads intact" $ \store -> do             let name = StreamName "preference-abc"             seedStream store name 6-            before <- countEvents store+            countBefore <- countEvents store             Right (Just _) <- runStoreIO store $ setStreamTruncateBefore name (StreamVersion 6)              -- \$all global log still returns the full history.@@ -63,8 +63,8 @@             length (V.toList catEvents) `shouldBe` 6              -- Nothing was physically deleted.-            after <- countEvents store-            after `shouldBe` before+            countAfter <- countEvents store+            countAfter `shouldBe` countBefore          it "is reversible via clearStreamTruncateBefore" $ \store -> do             let name = StreamName "preference-abc"