kiroku-store 0.13.0.0 → 0.13.0.1

raw patch · 8 files changed

+232/−12 lines, 8 filesPVP: minor bump suggested

API additions: PVP suggests at least a minor version bump

API changes (from Hackage documentation)

Files

• CHANGELOG.md +15/−0
• kiroku-store.cabal +2/−1
• src/Kiroku/Store/Notification.hs +31/−5
• src/Kiroku/Store/Observability.hs +3/−3
• src/Kiroku/Store/Subscription/EventPublisher.hs +3/−3
• test/Main.hs +2/−0
• test/Test/AsyncCancellation.hs +172/−0
• test/Test/PublisherCallbackResilience.hs +4/−0

@@ -1,5 +1,20 @@ # Changelog +## 0.13.0.1 — 2026-10-30++### Bug Fixes++- Preserve asynchronous cancellation through observability callbacks, publisher+  recovery and initial notifier LISTEN setup, including concrete exception payloads.+- Release the acquired notifier connection when startup LISTEN is cancelled;+  retain synchronous callback suppression and publisher recovery behavior.++### Other Changes++- Export `withListenHookForTest` from `Kiroku.Store.Notification` for sequential,+  process-local startup LISTEN injection. The previous hook is restored on exit;+  production leaves it unset.+ ## 0.13.0.0 — 2026-10-28  ### Breaking Changes

@@ -1,6 +1,6 @@ cabal-version:   3.0 name:            kiroku-store-version:         0.13.0.0+version:         0.13.0.1 synopsis:        High-performance PostgreSQL event store description:   Kiroku is a PostgreSQL-backed event store for Haskell applications. It@@ -108,6 +108,7 @@   hs-source-dirs:  test   other-modules:     Paths_kiroku_store+    Test.AsyncCancellation     Test.BrowseQueryPlans     Test.BrowseReads     Test.BrowseReadsMock

@@ -3,6 +3,7 @@     NotifierStartError (..),     startNotifier,     stopNotifier,+    withListenHookForTest,     NamespaceRegistry,     registerNamespace,     notifyNamespace,@@ -13,11 +14,12 @@ import Control.Concurrent.Async (Async) import Control.Concurrent.Async qualified as Async import Control.Concurrent.STM (STM, TChan, TVar, atomically, modifyTVar', newBroadcastTChanIO, newTVar, newTVarIO, readTVar, readTVarIO, writeTChan, writeTVar)-import Control.Exception (Exception, SomeAsyncException, SomeException, asyncExceptionFromException, bracketOnError, catch, throwIO, toException)+import Control.Exception (Exception, SomeAsyncException, SomeException, asyncExceptionFromException, bracket, bracketOnError, catch, fromException, throwIO, toException) import Control.Monad (forM_, unless, void) import Control.Monad.IO.Class (MonadIO, liftIO) import Data.ByteString (ByteString) import Data.ByteString.Char8 qualified as BC+import Data.IORef (IORef, newIORef, readIORef, writeIORef) import Data.Map.Strict (Map) import Data.Map.Strict qualified as Map import Data.Text (Text)@@ -36,6 +38,7 @@ import Hasql.Statement (Statement, unpreparable) import Kiroku.Store.Observability (KirokuEvent (..), emitOrDrop) import Kiroku.Store.Types (CategoryName (..), StreamName (..), categoryName)+import System.IO.Unsafe (unsafePerformIO)  -- | A Notifier manages a dedicated PostgreSQL connection for LISTEN/NOTIFY. -- It writes a @()@ tick to a broadcast 'TChan' on every notification,@@ -128,6 +131,28 @@   | n <= 0 = 1_000_000   | otherwise = min maxReconnectDelayMicros (1_000_000 * 2 ^ (min (n - 1) 30)) +{-# NOINLINE listenHookRef #-}+listenHookRef :: IORef (Maybe (IO () -> IO ()))+listenHookRef = unsafePerformIO (newIORef Nothing)++-- | Test-only process-local startup LISTEN hook. Use sequentially; the previous+-- hook is restored even if the test throws. Production leaves it unset.+withListenHookForTest :: (IO () -> IO ()) -> IO a -> IO a+withListenHookForTest hook action =+  bracket+    ( do+        previous <- readIORef listenHookRef+        writeIORef listenHookRef (Just hook)+        pure previous+    )+    (writeIORef listenHookRef)+    (const action)++startupListen :: IO () -> IO ()+startupListen action = do+  hook <- readIORef listenHookRef+  maybe action ($ action) hook+ -- | Start a Notifier. Acquires a dedicated connection, sets -- @application_name@ to @kiroku-listener@ for operator visibility, issues -- LISTEN on the @\<schema\>.events@ channel, and spawns a thread that@@ -148,7 +173,8 @@ -- -- On 'Async.AsyncCancelled' (from cancellation), exits cleanly. ----- Initial acquire or LISTEN failure raises 'NotifierStartError'.+-- Synchronous initial acquire or LISTEN failure raises 'NotifierStartError'.+-- Asynchronous LISTEN exceptions propagate unchanged after releasing the connection. startNotifier ::   (MonadIO m) =>   -- | libpq connection string@@ -164,10 +190,10 @@   namespaces <- newTVarIO Map.empty   let channel = toPgIdentifier (schema <> ".events")   bracketOnError (acquireOrThrow connString) Connection.release $ \conn -> do-    Notifications.listen conn channel+    startupListen (Notifications.listen conn channel)       `catch` \(e :: SomeException) ->-        case asyncExceptionFromException e of-          Just (ae :: SomeAsyncException) -> throwIO ae+        case fromException e of+          Just (_ :: SomeAsyncException) -> throwIO e           Nothing -> throwIO (NotifierListenError e)     connRef <- newTVarIO conn     thread <- Async.async (listenerLoop chan catGenVar namespaces connRef channel connString mHandler)

@@ -51,7 +51,7 @@   ) where -import Control.Exception (SomeAsyncException, SomeException, asyncExceptionFromException, catch, throwIO)+import Control.Exception (SomeAsyncException, SomeException, catch, fromException, throwIO) import Data.Foldable (for_) import Data.Int (Int32, Int64) import Data.Time (NominalDiffTime)@@ -246,8 +246,8 @@ emitOrDrop mHandler evt =   for_ mHandler $ \handler ->     handler evt `catch` \(e :: SomeException) ->-      case asyncExceptionFromException e of-        Just (ae :: SomeAsyncException) -> throwIO ae+      case fromException e of+        Just (_ :: SomeAsyncException) -> throwIO e         Nothing -> pure ()  -- | Which database phase a 'KirokuEventSubscriptionDbError' fired in.

@@ -53,7 +53,7 @@     writeTVar,   ) import Control.Concurrent.STM qualified as STM-import Control.Exception (SomeAsyncException, SomeException, asyncExceptionFromException, catch, throwIO)+import Control.Exception (SomeAsyncException, SomeException, catch, fromException, throwIO) import Control.Monad (when) import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Foldable (for_)@@ -232,8 +232,8 @@       -- partial delivery, lastPublished is not advanced, so the next tick       -- re-fetches under Kiroku's at-least-once delivery contract.       fetchAndBroadcast `catch` \(e :: SomeException) ->-        case asyncExceptionFromException e of-          Just (ae :: SomeAsyncException) -> throwIO ae+        case fromException e of+          Just (_ :: SomeAsyncException) -> throwIO e           Nothing -> emitOrDrop mHandler (KirokuEventPublisherLoopError e)       loop 

@@ -25,6 +25,7 @@ import Kiroku.Store.Subscription.EventPublisher (publisherPosition) import Kiroku.Store.Subscription.Fsm qualified as Fsm import Kiroku.Test.Postgres (withMigratedTestDatabase)+import Test.AsyncCancellation qualified as AsyncCancellation import Test.BrowseQueryPlans qualified as BrowseQueryPlans import Test.BrowseReads qualified as BrowseReads import Test.BrowseReadsMock qualified as BrowseReadsMock@@ -138,6 +139,7 @@     StreamNameLookup.noOpSpec   CategoryIdleNoSpin.spec   HandlerStall.spec+  AsyncCancellation.spec   PublisherCallbackResilience.spec   PublisherDropCounter.spec   PublisherIdleAdvance.spec

@@ -0,0 +1,172 @@+module Test.AsyncCancellation (spec, publisherSpec) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async qualified as Async+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar, tryPutMVar, tryReadMVar)+import Control.Concurrent.STM (atomically)+import Control.Exception (AsyncException (ThreadKilled, UserInterrupt), Exception (..), SomeException, asyncExceptionFromException, asyncExceptionToException, bracket, finally, throwIO, throwTo, try)+import Control.Lens ((&), (.~), (^.))+import Control.Monad (forM_, void)+import Data.Aeson qualified as Aeson+import Data.Generics.Labels ()+import Data.IORef (newIORef, readIORef, writeIORef)+import Data.Int (Int64)+import Data.Text (Text)+import Kiroku.Store+import Kiroku.Store.Notification qualified as N+import Kiroku.Store.Observability (emitOrDrop)+import Kiroku.Store.Subscription.EventPublisher qualified as P+import Kiroku.Test.Postgres (withMigratedTestDatabase)+import System.Timeout (timeout)+import Test.Helpers (listenerCount, makeEvent, withTestStoreSettings)+import Test.Hspec++-- The payload ensures rethrowing preserves a consumer-defined exception too.+data PayloadCancellation = PayloadCancellation Int deriving stock (Eq, Show)++instance Exception PayloadCancellation where+  toException = asyncExceptionToException+  fromException = asyncExceptionFromException++data SyncFailure = SyncFailure deriving stock (Eq, Show)++instance Exception SyncFailure++cases :: [(String, SomeException, SomeException -> Bool)]+cases =+  [ ("AsyncCancelled", toException Async.AsyncCancelled, matches Async.AsyncCancelled),+    ("ThreadKilled", toException ThreadKilled, matches ThreadKilled),+    ("UserInterrupt", toException UserInterrupt, matches UserInterrupt),+    ("custom payload", toException (PayloadCancellation 73), matches (PayloadCancellation 73))+  ]+  where+    matches :: (Exception e, Eq e) => e -> SomeException -> Bool+    matches expected e = fromException e == Just expected++bounded :: IO a -> IO a+bounded action = timeout 10_000_000 action >>= maybe (fail "asynchronous boundary timed out") pure++expectException :: (SomeException -> Bool) -> Either SomeException a -> Expectation+expectException matches = \case+  Left e -> e `shouldSatisfy` matches+  Right _ -> expectationFailure "expected asynchronous exception; action returned normally"++emit :: IO () -> IO ()+emit action = emitOrDrop (Just (const action)) KirokuEventNotifierReconnected++-- Release the blocked action before the scoped async finalizer on every exit.+-- This also makes the original, cancellation-swallowing implementation safe to test.+crossThread :: (IO () -> IO ()) -> (Async.Async () -> IO ()) -> (SomeException -> Bool) -> IO ()+crossThread boundary send matches = do+  entered <- newEmptyMVar+  release <- newEmptyMVar+  Async.withAsync (boundary (putMVar entered () >> takeMVar release)) $ \target ->+    ( do+        bounded (takeMVar entered)+        bounded (send target)+        bounded (Async.waitCatch target) >>= expectException matches+    )+      `finally` void (tryPutMVar release ())++awaitNoListeners :: Text -> Int64 -> IO ()+awaitNoListeners conn baseline = bounded loop+  where+    loop = do+      n <- listenerCount conn+      if n == baseline then pure () else threadDelay 1_000 >> loop++spec :: Spec+spec = describe "asynchronous cancellation" $ do+  it "allows absent and successful observation callbacks" $ do+    emitOrDrop Nothing KirokuEventNotifierReconnected+    called <- newIORef False+    emit (writeIORef called True)+    readIORef called `shouldReturn` True+  it "suppresses synchronous observation failures" $ emit (throwIO SyncFailure)+  forM_ cases $ \(label, exception, matches) -> do+    it ("preserves observation " <> label) $+      try (emit (throwIO exception)) >>= expectException matches+    it ("preserves notifier startup " <> label <> " and releases its connection") $+      withMigratedTestDatabase $ \conn -> do+        baseline <- listenerCount conn+        baseline `shouldBe` 0+        N.withListenHookForTest+          ( \_ -> do+              listenerCount conn `shouldReturn` (baseline + 1)+              throwIO exception+          )+          $ do+            result <- try (N.startNotifier conn "kiroku" Nothing)+            -- Even an incorrect successful startup must be cleaned up.+            either (const (pure ())) N.stopNotifier result+            expectException matches result+        awaitNoListeners conn baseline+  it "wraps synchronous notifier startup failure with its original cause and releases the connection" $+    withMigratedTestDatabase $ \conn -> do+      baseline <- listenerCount conn+      N.withListenHookForTest (\_ -> throwIO SyncFailure) $ do+        result <- try (N.startNotifier conn "kiroku" Nothing)+        either (const (pure ())) N.stopNotifier result+        case result of+          Left e+            | Just (N.NotifierListenError cause) <- fromException e ->+                fromException cause `shouldBe` Just SyncFailure+          _ -> expectationFailure "expected NotifierListenError with synchronous cause"+      awaitNoListeners conn baseline+  forM_ [False, True] $ \custom -> do+    let send target = if custom then throwTo (Async.asyncThreadId target) (PayloadCancellation 73) else Async.cancel target+        matches e = if custom then fromException e == Just (PayloadCancellation 73) else fromException e == Just Async.AsyncCancelled+    it ("terminates a blocked observation callback, custom=" <> show custom) $+      crossThread emit send matches+    it ("terminates blocked notifier startup and releases its connection, custom=" <> show custom) $+      withMigratedTestDatabase $ \conn -> do+        baseline <- listenerCount conn+        let boundary action =+              N.withListenHookForTest (\_ -> (listenerCount conn `shouldReturn` (baseline + 1)) >> action) $+                bracket (N.startNotifier conn "kiroku" Nothing) N.stopNotifier (const (pure ()))+        crossThread boundary send matches+        awaitNoListeners conn baseline++publisherSpec :: Spec+publisherSpec = describe "asynchronous publisher recovery" $ do+  forM_ [False, True] $ \callback -> do+    forM_ cases $ \(label, exception, matches) ->+      it ("preserves " <> label <> ", callback=" <> show callback) $+        publisherCase callback (throwIO exception) Nothing (pure ()) matches+    forM_ [False, True] $ \custom ->+      it ("terminates blocked action, callback=" <> show callback <> ", custom=" <> show custom) $ do+        entered <- newEmptyMVar+        release <- newEmptyMVar+        let action = putMVar entered () >> takeMVar release+            send target = do+              bounded (takeMVar entered)+              bounded (if custom then throwTo (Async.asyncThreadId target) (PayloadCancellation 73) else Async.cancel target)+            matches e = if custom then fromException e == Just (PayloadCancellation 73) else fromException e == Just Async.AsyncCancelled+        publisherCase callback action (Just send) (void (tryPutMVar release ())) matches++publisherCase :: Bool -> IO () -> Maybe (Async.Async () -> IO ()) -> IO () -> (SomeException -> Bool) -> IO ()+publisherCase callback action sender cleanup matches = do+  loopError <- newEmptyMVar+  let hook event =+        if callback+          then pure (Left (DecodeFailure (event ^. #eventId) "injected decode failure"))+          else action >> pure (Right event)+      observe event = case event of+        KirokuEventPublisherDecodeFailed {} | callback -> action+        KirokuEventPublisherLoopError {} -> void (tryPutMVar loopError ())+        _ -> pure ()+      tweak settings = settings & #storeSettings .~ defaultStoreSettings {decodeHook = Just hook} & #eventHandler .~ Just observe+  withTestStoreSettings tweak $ \store -> do+    let pub = store ^. #publisher+        target = P.publisherThread pub+    ( bracket (atomically (P.subscribePublisher pub 1 PauseAndResume)) (\(_, _, unsubscribe) -> unsubscribe) $ \_ -> do+        Right _ <- runStoreIO store $ appendToStream (StreamName "async-boundary") NoStream [makeEvent "Cancel" (Aeson.object [])]+        forM_ sender ($ target)+        outcome <- bounded $ Async.race (takeMVar loopError) (Async.waitCatch target)+        case outcome of+          Left () -> expectationFailure "asynchronous exception became a publisher loop error"+          Right result -> expectException matches result+        tryReadMVar loopError `shouldReturn` Nothing+        atomically (P.publisherPosition pub) `shouldReturn` GlobalPosition 0+      )+      `finally` cleanup

@@ -29,6 +29,7 @@ import Kiroku.Store.SQL qualified as SQL import Kiroku.Store.Subscription.Effect qualified as SubEff import Kiroku.Store.Subscription.Fsm (SubscriptionState (..))+import Test.AsyncCancellation qualified as AsyncCancellation import Test.Helpers (caughtUpEventHandler, makeEvent, validConsumerGroup, waitForPublisher, waitForSubscriptionLive, waitWithTimeout, withTestStoreSettings) import Test.Hspec @@ -90,6 +91,7 @@  spec :: Spec spec = describe "publisher callback resilience" $ do+  AsyncCancellation.publisherSpec   it "stops only the default subscriber while a sibling dead-letters and continues live" $ do     defaultLive <- newEmptyMVar     siblingLive <- newEmptyMVar@@ -553,6 +555,8 @@         Right (Right ()) -> pure ()         Left timeout -> expectationFailure timeout         Right (Left e) -> expectationFailure ("expected clean stop, got: " <> show e)+      awaitCheckpoint store "publisher-decode-hook-resilience" 2+      waitForPublisher store (GlobalPosition 2)      seen <- reverse <$> readIORef delivered     seen `shouldBe` [EventType "Boom", EventType "Ok"]