kiroku-store 0.8.0.1 → 0.8.0.2

raw patch · 5 files changed

+110/−40 lines, 5 filesdep ~effectfuldep ~effectful-coredep ~shibuya-corePVP: minor bump suggested

API additions: PVP suggests at least a minor version bump

Dependency ranges changed: effectful, effectful-core, shibuya-core

API changes (from Hackage documentation)

Files

• CHANGELOG.md +20/−0
• kiroku-store.cabal +27/−27
• src/Kiroku/Store/Subscription/Stream.hs +14/−10
• src/Kiroku/Store/Subscription/Worker.hs +32/−3
• test/Test/StreamBridgeTermination.hs +17/−0

@@ -1,5 +1,25 @@ # Changelog +## 0.8.0.2 — 2026-09-21++### Bug Fixes++* `subscriptionAckStream` now masks the ownership transfer from the underlying+  subscription to its monitor and cancels the subscription if monitor startup+  fails. Its cancel action waits for the monitor to terminate, remains+  idempotent, and returns only after the subscription registry entry is gone.++### Other Changes++* The Shibuya integration benchmark now targets `shibuya-core >=0.10 && <0.11`+  so the store can participate in the coordinated lifecycle release candidate.+* Support `effectful-core` 2.6.1 and 2.7.1.1 or later, while excluding 2.7.0.0+  through 2.7.1.0 because upstream records a per-operation performance+  regression for dynamically dispatched effects.+* Subscription lifecycle tests gain a checkpoint-save boundary hook so+  cancellation after an acknowledgement but before persistence can be proved+  to replay the event without leaking a worker.+ ## 0.8.0.1 — 2026-09-18  ### Other Changes

@@ -1,6 +1,6 @@ cabal-version:   3.0 name:            kiroku-store-version:         0.8.0.1+version:         0.8.0.2 synopsis:        High-performance PostgreSQL event store description:   Kiroku is a PostgreSQL-backed event store for Haskell applications. It@@ -67,28 +67,28 @@     Kiroku.Store.Subscription.CheckpointInventory.SQL    build-depends:-    , aeson                 >=2.1  && <2.3-    , async                 >=2.2  && <2.3-    , base                  >=4.18 && <5-    , bytestring            >=0.11 && <0.13-    , containers            >=0.6  && <0.8-    , contravariant-extras  >=0.3  && <0.4-    , effectful-core        >=2.4  && <2.7-    , generic-lens          >=2.2  && <2.4-    , hasql                 >=1.10 && <1.11-    , hasql-notifications   >=0.2  && <0.3-    , hasql-pool            >=1.2  && <1.5-    , hasql-transaction     >=1.1  && <1.3-    , lens                  >=5.2  && <5.4-    , mmzk-typeid           >=0.6  && <0.8-    , mtl                   >=2.3  && <2.4-    , stm                   >=2.5  && <2.6-    , streamly-core         >=0.3  && <0.4-    , text                  >=2.0  && <2.2-    , time                  >=1.12 && <1.15-    , unliftio-core         >=0.2  && <0.3-    , uuid                  >=1.3  && <1.4-    , vector                >=0.13 && <0.14+    , aeson                 >=2.1   && <2.3+    , async                 >=2.2   && <2.3+    , base                  >=4.18  && <5+    , bytestring            >=0.11  && <0.13+    , containers            >=0.6   && <0.8+    , contravariant-extras  >=0.3   && <0.4+    , effectful-core        >=2.6.1 && <2.7  || >=2.7.1.1 && <2.8+    , generic-lens          >=2.2   && <2.4+    , hasql                 >=1.10  && <1.11+    , hasql-notifications   >=0.2   && <0.3+    , hasql-pool            >=1.2   && <1.5+    , hasql-transaction     >=1.1   && <1.3+    , lens                  >=5.2   && <5.4+    , mmzk-typeid           >=0.6   && <0.8+    , mtl                   >=2.3   && <2.4+    , stm                   >=2.5   && <2.6+    , streamly-core         >=0.3   && <0.4+    , text                  >=2.0   && <2.2+    , time                  >=1.12  && <1.15+    , unliftio-core         >=0.2   && <0.3+    , uuid                  >=1.3   && <1.4+    , vector                >=0.13  && <0.14    hs-source-dirs:  src @@ -148,7 +148,7 @@     , containers            >=0.6   && <0.8     , contravariant-extras  >=0.3     , directory-    , effectful-core        >=2.4   && <2.7+    , effectful-core        >=2.6.1 && <2.7  || >=2.7.1.1 && <2.8     , ephemeral-pg          >=0.3.1 && <0.4     , generic-lens          >=2.2   && <2.4     , hasql                 >=1.10  && <1.11@@ -230,14 +230,14 @@     , aeson                 >=2.1   && <2.3     , base                  >=4.18  && <5     , containers            >=0.6   && <0.8-    , effectful             >=2.4   && <2.7-    , effectful-core        >=2.4   && <2.7+    , effectful             >=2.6.1 && <2.8+    , effectful-core        >=2.6.1 && <2.7  || >=2.7.1.1 && <2.8     , ephemeral-pg          >=0.3.1 && <0.4     , generic-lens          >=2.2   && <2.4     , kiroku-store     , kiroku-test-support     , lens                  >=5.2   && <5.4-    , shibuya-core          >=0.9   && <0.10+    , shibuya-core          >=0.10  && <0.11     , stm                   >=2.5   && <2.6     , streamly              >=0.11     , streamly-core         >=0.3   && <0.4

@@ -54,7 +54,7 @@     writeTBQueue,     writeTVar,  )-import Control.Exception (Exception, SomeException, fromException, throwIO)+import Control.Exception (Exception, SomeException, fromException, mask, onException, throwIO) import Control.Monad (when) import Data.IORef (atomicModifyIORef', newIORef) import Kiroku.Store.Connection (KirokuStore)@@ -176,7 +176,7 @@     -- | TBQueue capacity for the bridge; must be at least 1.     Natural ->     IO (Stream IO AckItem, IO ())-subscriptionAckStream store config bufferSize = do+subscriptionAckStream store config bufferSize = mask $ \restore -> do     when (bufferSize < 1) $         throwIO (InvalidStreamBufferSize bufferSize)     queue <- newTBQueueIO bufferSize@@ -200,17 +200,21 @@      let bridgeConfig = config{handler = bridgeHandler} -    subHandle <- subscribe store bridgeConfig-    _monitor <- Async.async $ do-        outcome <- wait subHandle-        atomically . closeBridge closedVar $ case outcome of-            Right () -> BridgeClosedCleanly-            Left e-                | Just Async.AsyncCancelled <- fromException e -> BridgeClosedCleanly-                | otherwise -> BridgeCrashed e+    subHandle <- restore (subscribe store bridgeConfig)+    monitor <-+        ( Async.async $ do+            outcome <- wait subHandle+            atomically . closeBridge closedVar $ case outcome of+                Right () -> BridgeClosedCleanly+                Left e+                    | Just Async.AsyncCancelled <- fromException e -> BridgeClosedCleanly+                    | otherwise -> BridgeCrashed e+        )+            `onException` cancel subHandle      let cancelAction = do             cancel subHandle+            Async.wait monitor             atomically (closeBridge closedVar BridgeClosedCleanly)      let step :: () -> IO (Maybe (AckItem, ()))

@@ -18,8 +18,9 @@ the single delivery primitive shared by every live path, so behaviour is identical for @AllStreams@, @Category@, and consumer-group subscriptions. -'withFetchBatchHookForTest' and 'withLoadCheckpointHookForTest' are test-only-seams for injecting fetch and checkpoint-load failures.+'withFetchBatchHookForTest', 'withLoadCheckpointHookForTest', and+'withSaveCheckpointHookForTest' are test-only seams for controlling fetch,+checkpoint-load, and checkpoint-save boundaries. -} module Kiroku.Store.Subscription.Worker (     LiveSource (..),@@ -27,6 +28,7 @@     configMember,     withFetchBatchHookForTest,     withLoadCheckpointHookForTest,+    withSaveCheckpointHookForTest, ) where  import Contravariant.Extras (contrazip2)@@ -95,6 +97,11 @@             )         ) +type SaveCheckpointHook =+    SubscriptionConfig ->+    GlobalPosition ->+    IO ()+ {-# NOINLINE fetchBatchHookRef #-} fetchBatchHookRef :: IORef (Maybe FetchBatchHook) fetchBatchHookRef = unsafePerformIO (newIORef Nothing)@@ -103,6 +110,10 @@ loadCheckpointHookRef :: IORef (Maybe LoadCheckpointHook) loadCheckpointHookRef = unsafePerformIO (newIORef Nothing) +{-# NOINLINE saveCheckpointHookRef #-}+saveCheckpointHookRef :: IORef (Maybe SaveCheckpointHook)+saveCheckpointHookRef = unsafePerformIO (newIORef Nothing)+ {- | Install a process-local fetch hook for tests that need deterministic subscription-worker fault injection. Production code leaves the hook unset. -}@@ -132,6 +143,22 @@         (writeIORef loadCheckpointHookRef)         (const action) +{- | Install a process-local checkpoint-save boundary hook for lifecycle tests.++The hook runs immediately before the database statement. Production code leaves+it unset.+-}+withSaveCheckpointHookForTest :: SaveCheckpointHook -> IO a -> IO a+withSaveCheckpointHookForTest hook action =+    bracket+        ( do+            previous <- readIORef saveCheckpointHookRef+            writeIORef saveCheckpointHookRef (Just hook)+            pure previous+        )+        (writeIORef saveCheckpointHookRef)+        (const action)+ fetchRetryDelayMicros :: Int -> Int fetchRetryDelayMicros attempt =     min categorySafetyPollMicros (100_000 * (2 ^ min attempt 9 :: Int))@@ -794,9 +821,11 @@     GlobalPosition ->     (KirokuEvent -> IO ()) ->     IO ()-saveCheckpoint pool config (GlobalPosition pos) emit = do+saveCheckpoint pool config position@(GlobalPosition pos) emit = do     let subName@(SubscriptionName name') = name config         mem = configMember config+    mHook <- readIORef saveCheckpointHookRef+    mapM_ (\hook -> hook config position) mHook     result <- Pool.use pool (Session.statement (name', mem, pos) SQL.saveCheckpointMemberStmt)     case result of         Left err -> emit (KirokuEventSubscriptionDbError subName SaveCheckpoint err (groupCtxOf config))

@@ -10,6 +10,7 @@ import Control.Lens ((^.)) import Data.Aeson qualified as Aeson import Data.Generics.Labels ()+import Data.Map.Strict qualified as Map import Kiroku.Store import Kiroku.Store.Subscription.Stream (AckItem (..), InvalidStreamBufferSize (..), subscriptionAckStream) import Kiroku.Store.Subscription.Worker (withFetchBatchHookForTest)@@ -103,3 +104,19 @@                     case mNext of                         Nothing -> pure ()                         Just _ -> expectationFailure "expected stream to end after cancel"++    it "cancelAction is idempotent and leaves no subscription thread registered" $+        withTestStore $ \store -> do+            let subscriptionName = SubscriptionName "bridge-idempotent-cancel-sub"+                cfg = defaultSubscriptionConfig subscriptionName AllStreams (\_ -> pure Continue)+                key = (subscriptionName, 0)+            (_stream, cancelStream) <- subscriptionAckStream store cfg 1++            statesBefore <- subscriptionStates store+            Map.member key statesBefore `shouldBe` True++            within "first bridge cancellation" cancelStream+            within "repeated bridge cancellation" cancelStream++            statesAfter <- subscriptionStates store+            Map.member key statesAfter `shouldBe` False