diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/bench/Explain.hs b/bench/Explain.hs --- a/bench/Explain.hs +++ b/bench/Explain.hs @@ -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 <- diff --git a/bench/Main.hs b/bench/Main.hs --- a/bench/Main.hs +++ b/bench/Main.hs @@ -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"] diff --git a/bench/RegressionGate.hs b/bench/RegressionGate.hs --- a/bench/RegressionGate.hs +++ b/bench/RegressionGate.hs @@ -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 diff --git a/bench/ShibuyaOverhead.hs b/bench/ShibuyaOverhead.hs --- a/bench/ShibuyaOverhead.hs +++ b/bench/ShibuyaOverhead.hs @@ -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 (..)) diff --git a/kiroku-store.cabal b/kiroku-store.cabal --- a/kiroku-store.cabal +++ b/kiroku-store.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: kiroku-store -version: 0.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 diff --git a/src/Kiroku/Store/Error.hs b/src/Kiroku/Store/Error.hs --- a/src/Kiroku/Store/Error.hs +++ b/src/Kiroku/Store/Error.hs @@ -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 diff --git a/src/Kiroku/Store/SQL.hs b/src/Kiroku/Store/SQL.hs --- a/src/Kiroku/Store/SQL.hs +++ b/src/Kiroku/Store/SQL.hs @@ -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 = diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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) diff --git a/test/Test/CatchupDbErrorNoPrematureSwitch.hs b/test/Test/CatchupDbErrorNoPrematureSwitch.hs --- a/test/Test/CatchupDbErrorNoPrematureSwitch.hs +++ b/test/Test/CatchupDbErrorNoPrematureSwitch.hs @@ -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 diff --git a/test/Test/CategoryIdleNoSpin.hs b/test/Test/CategoryIdleNoSpin.hs --- a/test/Test/CategoryIdleNoSpin.hs +++ b/test/Test/CategoryIdleNoSpin.hs @@ -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 diff --git a/test/Test/Causation.hs b/test/Test/Causation.hs --- a/test/Test/Causation.hs +++ b/test/Test/Causation.hs @@ -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 diff --git a/test/Test/Concurrency.hs b/test/Test/Concurrency.hs --- a/test/Test/Concurrency.hs +++ b/test/Test/Concurrency.hs @@ -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 () diff --git a/test/Test/ConsumerGroup.hs b/test/Test/ConsumerGroup.hs --- a/test/Test/ConsumerGroup.hs +++ b/test/Test/ConsumerGroup.hs @@ -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)) diff --git a/test/Test/ConsumerGroupEffect.hs b/test/Test/ConsumerGroupEffect.hs --- a/test/Test/ConsumerGroupEffect.hs +++ b/test/Test/ConsumerGroupEffect.hs @@ -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 diff --git a/test/Test/HistoryRetention.hs b/test/Test/HistoryRetention.hs --- a/test/Test/HistoryRetention.hs +++ b/test/Test/HistoryRetention.hs @@ -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 diff --git a/test/Test/NotifyGuard.hs b/test/Test/NotifyGuard.hs --- a/test/Test/NotifyGuard.hs +++ b/test/Test/NotifyGuard.hs @@ -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 diff --git a/test/Test/PerformanceStructure.hs b/test/Test/PerformanceStructure.hs --- a/test/Test/PerformanceStructure.hs +++ b/test/Test/PerformanceStructure.hs @@ -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 = diff --git a/test/Test/Properties.hs b/test/Test/Properties.hs --- a/test/Test/Properties.hs +++ b/test/Test/Properties.hs @@ -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 diff --git a/test/Test/PublisherCallbackResilience.hs b/test/Test/PublisherCallbackResilience.hs --- a/test/Test/PublisherCallbackResilience.hs +++ b/test/Test/PublisherCallbackResilience.hs @@ -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 diff --git a/test/Test/PublisherIdleAdvance.hs b/test/Test/PublisherIdleAdvance.hs --- a/test/Test/PublisherIdleAdvance.hs +++ b/test/Test/PublisherIdleAdvance.hs @@ -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 <- diff --git a/test/Test/PublisherRestartNoRebroadcast.hs b/test/Test/PublisherRestartNoRebroadcast.hs --- a/test/Test/PublisherRestartNoRebroadcast.hs +++ b/test/Test/PublisherRestartNoRebroadcast.hs @@ -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 diff --git a/test/Test/ReadStream.hs b/test/Test/ReadStream.hs --- a/test/Test/ReadStream.hs +++ b/test/Test/ReadStream.hs @@ -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" $ diff --git a/test/Test/StartupFailureSurfacing.hs b/test/Test/StartupFailureSurfacing.hs --- a/test/Test/StartupFailureSurfacing.hs +++ b/test/Test/StartupFailureSurfacing.hs @@ -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 () diff --git a/test/Test/StreamBridgeTermination.hs b/test/Test/StreamBridgeTermination.hs --- a/test/Test/StreamBridgeTermination.hs +++ b/test/Test/StreamBridgeTermination.hs @@ -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 diff --git a/test/Test/SubscriptionCheckpointInitialization.hs b/test/Test/SubscriptionCheckpointInitialization.hs --- a/test/Test/SubscriptionCheckpointInitialization.hs +++ b/test/Test/SubscriptionCheckpointInitialization.hs @@ -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) diff --git a/test/Test/SubscriptionCheckpointReset.hs b/test/Test/SubscriptionCheckpointReset.hs --- a/test/Test/SubscriptionCheckpointReset.hs +++ b/test/Test/SubscriptionCheckpointReset.hs @@ -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 diff --git a/test/Test/SubscriptionCheckpointWorker.hs b/test/Test/SubscriptionCheckpointWorker.hs --- a/test/Test/SubscriptionCheckpointWorker.hs +++ b/test/Test/SubscriptionCheckpointWorker.hs @@ -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 = diff --git a/test/Test/SubscriptionPauseResume.hs b/test/Test/SubscriptionPauseResume.hs --- a/test/Test/SubscriptionPauseResume.hs +++ b/test/Test/SubscriptionPauseResume.hs @@ -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 diff --git a/test/Test/SubscriptionRegistry.hs b/test/Test/SubscriptionRegistry.hs --- a/test/Test/SubscriptionRegistry.hs +++ b/test/Test/SubscriptionRegistry.hs @@ -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 diff --git a/test/Test/SubscriptionState.hs b/test/Test/SubscriptionState.hs --- a/test/Test/SubscriptionState.hs +++ b/test/Test/SubscriptionState.hs @@ -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 diff --git a/test/Test/TruncateBefore.hs b/test/Test/TruncateBefore.hs --- a/test/Test/TruncateBefore.hs +++ b/test/Test/TruncateBefore.hs @@ -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"