diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 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.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 diff --git a/src/Kiroku/Store/Subscription/Stream.hs b/src/Kiroku/Store/Subscription/Stream.hs --- a/src/Kiroku/Store/Subscription/Stream.hs +++ b/src/Kiroku/Store/Subscription/Stream.hs @@ -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, ())) diff --git a/src/Kiroku/Store/Subscription/Worker.hs b/src/Kiroku/Store/Subscription/Worker.hs --- a/src/Kiroku/Store/Subscription/Worker.hs +++ b/src/Kiroku/Store/Subscription/Worker.hs @@ -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)) diff --git a/test/Test/StreamBridgeTermination.hs b/test/Test/StreamBridgeTermination.hs --- a/test/Test/StreamBridgeTermination.hs +++ b/test/Test/StreamBridgeTermination.hs @@ -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