shibuya-kiroku-adapter 0.2.0.0 → 0.4.0.0

raw patch · 5 files changed

+508/−119 lines, 5 filesdep ~effectful-coredep ~kiroku-storedep ~shibuya-corePVP ok

version bump matches the API change (PVP)

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

API changes (from Hackage documentation)

Files

• CHANGELOG.md +42/−0
• shibuya-kiroku-adapter.cabal +11/−12
• src/Shibuya/Adapter/Kiroku.hs +159/−85
• src/Shibuya/Adapter/Kiroku/Convert.hs +1/−0
• test/Main.hs +295/−22

@@ -2,6 +2,48 @@  ## Unreleased +## 0.4.0.0 — 2026-07-11++### Breaking Changes++* `KirokuAdapterConfig` and `KirokuConsumerGroupConfig` gain a+  `queueCapacity` field, defaulting to 16 in the smart constructors.+* Requires `shibuya-core >=0.8 && <0.9` (was `>=0.7`) and `kiroku-store ^>=0.3`.++### Fixed++* Adapter subscriptions now rely on Kiroku's default lossless+  `PauseAndResume` overflow policy instead of pinning fail-fast+  `DropSubscription`.+* `guardKirokuHandlerWith` and `guardKirokuHandler` convert synchronous handler+  exceptions into finalized ack decisions so Kiroku workers are not abandoned+  mid-ack. `kirokuConsumerGroupProcessors` applies the default guard+  automatically.+* `Adapter.source` now has regression coverage for EP-1's termination contract:+  clean shutdown ends the stream normally, while a worker crash is rethrown to+  the stream consumer.+* `kirokuConsumerGroupProcessors` now throws `InvalidConsumerGroup` for+  non-positive group sizes before opening subscriptions, and cleans up already+  created member adapters if a later member fails during construction.+* The package description and Haddocks describe the ack-coupled bridge instead+  of the obsolete no-op ack behavior.++### Known Limitations++* `shibuya-core` still needs upstream fixes for finalize-on-exception in its+  supervised processor and for propagating ingester stream failures from the+  supervised runner. The adapter-side guard is defensive and remains correct+  after those fixes land.++## 0.3.0.0 — 2026-06-05++### Changed++- Require `shibuya-core >=0.7 && <0.8`. `Envelope` now carries a+  `headers :: Maybe Headers` field; `toEnvelope` sets it to `Nothing`,+  because Kiroku events carry no ordered, raw broker headers. W3C trace+  context continues to be surfaced via `traceContext` from event metadata.+ ## 0.2.0.0 — 2026-05-31  ### Breaking Changes

@@ -1,6 +1,6 @@ cabal-version:   3.0 name:            shibuya-kiroku-adapter-version:         0.2.0.0+version:         0.4.0.0 synopsis:   Kiroku event store adapter for the Shibuya queue processing framework @@ -10,12 +10,10 @@   interface, enabling supervised multi-subscription processing with failure   isolation, per-subscription metrics, and coordinated graceful shutdown.   .-  Events flow through a bounded TBQueue bridge (provided by-  @kiroku-store@'s @subscriptionStream@) and are lifted into Shibuya's-  effect stack via @Stream.morphInner@. Checkpoint management is handled-  internally by Kiroku's subscription worker — the adapter's ack semantics-  are no-op for AckOk\/AckRetry\/AckDeadLetter and trigger subscription-  cancellation for AckHalt.+  Events flow through @kiroku-store@'s ack-coupled @subscriptionAckStream@+  bridge and are lifted into Shibuya's effect stack via @Stream.morphInner@.+  Each finalized Shibuya @AckDecision@ drives Kiroku checkpointing, retry,+  dead-lettering, or halt before the worker moves to the next event.  author:          Nadeem Bitar maintainer:      nadeem@gmail.com@@ -49,11 +47,11 @@   build-depends:     , aeson                                  >=2.1   && <2.3     , base                                   >=4.18  && <5-    , effectful-core                         >=2.4   && <2.7+    , effectful-core                         >=2.5   && <2.7     , hs-opentelemetry-api                   ^>=1.0     , hs-opentelemetry-semantic-conventions  ^>=1.40-    , kiroku-store                           ^>=0.2-    , shibuya-core                           >=0.6   && <0.7+    , kiroku-store                           ^>=0.3+    , shibuya-core                           >=0.8   && <0.9     , stm                                    >=2.5   && <2.6     , streamly-core                          >=0.3   && <0.4     , text                                   >=2.0   && <2.2@@ -80,12 +78,13 @@     , hasql-pool              >=1.2  && <1.5     , hs-opentelemetry-api    ^>=1.0     , hspec                   >=2.10 && <2.12-    , kiroku-store            ^>=0.2+    , kiroku-store            ^>=0.3     , kiroku-test-support     , lens                    >=5.2  && <5.4-    , shibuya-core            >=0.6  && <0.7+    , shibuya-core            >=0.8  && <0.9     , shibuya-kiroku-adapter     , stm                     >=2.5  && <2.6+    , streamly-core           >=0.3  && <0.4     , text                    >=2.0  && <2.2     , time                    >=1.12 && <1.15     , unordered-containers    >=0.2  && <0.3

@@ -18,21 +18,14 @@ main :: IO () main = withStore settings $ \\store ->     runEff $ runTracingNoop $ do-        adapter <- kirokuAdapter store-            KirokuAdapterConfig-                { subscriptionName = SubscriptionName \"my-projection\"-                , subscriptionTarget = AllStreams-                , batchSize = 100-                , bufferSize = 256-                , consumerGroup = Nothing-                , eventTypeFilter = AllEventTypes-                }+        let cfg = defaultKirokuAdapterConfig (SubscriptionName \"my-projection\") AllStreams+        adapter <- kirokuAdapter store cfg          let handler ingested = do                 -- process ingested.envelope.payload :: RecordedEvent                 pure AckOk -        Right appHandle <- runApp IgnoreFailures 100+        Right appHandle <- runApp defaultAppConfig             [(ProcessorId \"my-projection\", mkProcessor adapter handler)]          waitApp appHandle@@ -58,7 +51,7 @@                 4   -- group size          Right processors <- kirokuConsumerGroupProcessors store cfg handler-        Right appHandle <- runApp IgnoreFailures 100 processors+        Right appHandle <- runApp defaultAppConfig processors         waitApp appHandle   where     handler ingested = do@@ -95,16 +88,33 @@ The envelope's @attempt@ reports the zero-based redelivery count, so a handler can observe how many times Kiroku has redelivered an event. -== Backpressure+== Backpressure and Handler Exceptions -The @bufferSize@ field in 'KirokuAdapterConfig' controls the 'TBQueue'-capacity. Because delivery is ack-coupled, the Kiroku subscription worker blocks-on each event until the Shibuya handler finalizes its decision, providing natural-backpressure.+Because delivery is ack-coupled, the Kiroku subscription worker blocks on each+event until the Shibuya handler finalizes its decision, providing natural+backpressure. The @bufferSize@ field is only the capacity of the bridge queue+between the worker and the Shibuya stream consumer; for this adapter the+effective depth is at most one event because the worker waits for an ack before+delivering the next event, and @bufferSize@ must be at least 1.++The @queueCapacity@ field is the publisher-side burst knob: it is the number of+publisher batches, up to 1000 events each, that can be buffered before Kiroku+pauses this subscriber. The adapter uses Kiroku's lossless @PauseAndResume@+overflow policy, so a paused subscriber catches up from its checkpoint instead+of being killed.++A Shibuya handler used with this adapter must not let synchronous exceptions+escape. Shibuya's supervised runner records handler exceptions without+finalizing the ack; with Kiroku's ack-coupled bridge, an unfinalized ack blocks+the Kiroku worker forever. Direct 'mkProcessor' users should wrap handlers in+'guardKirokuHandler' or handle exceptions themselves.+'kirokuConsumerGroupProcessors' applies 'guardKirokuHandler' automatically. -} module Shibuya.Adapter.Kiroku (     -- * Adapter     kirokuAdapter,+    guardKirokuHandlerWith,+    guardKirokuHandler,      -- * Configuration     KirokuAdapterConfig (..),@@ -115,6 +125,7 @@     defaultConsumerGroupConfig,     consumerGroupPolicy,     kirokuConsumerGroupProcessors,+    kirokuConsumerGroupProcessorsWith,      -- * Re-exports from kiroku-store     SubscriptionName (..),@@ -123,41 +134,48 @@     EventTypeFilter (..), ) where +import Control.Exception (SomeException) import Data.Int (Int32) import Data.Text qualified as T import Effectful (Eff, IOE, liftIO, (:>))+import Effectful.Exception (catchSync, onException, throwIO) import GHC.Generics (Generic) import Kiroku.Store.Connection (KirokuStore) import Kiroku.Store.Subscription.Stream (subscriptionAckStream) import Kiroku.Store.Subscription.Types (     ConsumerGroup (..),     EventTypeFilter (..),-    OverflowPolicy (..),-    SubscriptionConfigM (..),+    InvalidConsumerGroup (..),+    SubscriptionConfig,     SubscriptionName (..),     SubscriptionResult (..),     SubscriptionTarget (..),     defaultSubscriptionConfig,  )+import Kiroku.Store.Subscription.Types qualified as Sub import Kiroku.Store.Types (RecordedEvent) import Numeric.Natural (Natural) import Shibuya.Adapter (Adapter (..)) import Shibuya.Adapter.Kiroku.Convert (kirokuEnvelopeAttrs, toIngestedAck) import Shibuya.App (ProcessorId (..), QueueProcessor (..))+import Shibuya.Core.Ack (AckDecision (..), RetryDelay (..)) import Shibuya.Core.Error (PolicyError (..)) import Shibuya.Handler (Handler)-import Shibuya.Policy (Concurrency (..), Ordering (..), validatePolicy)+import Shibuya.Policy (Concurrency (..), OrderingPolicy (..), validatePolicy) import Streamly.Data.Stream qualified as Stream-import Prelude hiding (Ordering)  {- | Configuration for creating a Kiroku adapter.  @subscriptionName@ must be unique across all active subscriptions — it identifies the checkpoint row in the @subscriptions@ table. -@bufferSize@ controls backpressure: the Kiroku worker blocks when the-internal queue is full, throttling database polling to match the handler's-consumption rate.+@bufferSize@ is the bridge queue capacity and must be at least 1. With this+ack-coupled adapter the effective depth is at most one event, because the+worker blocks until the handler's decision is finalized.++@queueCapacity@ is the publisher-side burst capacity in batches. When a burst+exceeds it, the adapter relies on Kiroku's default @PauseAndResume@ policy:+Kiroku pauses the subscriber and later resumes losslessly from its checkpoint. -} data KirokuAdapterConfig = KirokuAdapterConfig     { subscriptionName :: !SubscriptionName@@ -167,7 +185,12 @@     , batchSize :: !Int32     -- ^ Events per database fetch during catch-up     , bufferSize :: !Natural-    -- ^ TBQueue capacity (backpressure threshold)+    -- ^ Bridge 'TBQueue' capacity; must be at least 1.+    , queueCapacity :: !Natural+    {- ^ Publisher-side capacity in batches, where each batch contains up to+    Kiroku's publisher batch size (currently 1000 events). When this fills,+    Kiroku pauses and later resumes the subscriber losslessly.+    -}     , consumerGroup :: !(Maybe ConsumerGroup)     {- ^ Optional consumer-group membership for this adapter instance.     'Nothing' (the default) = ordinary single-consumer subscription.@@ -206,10 +229,11 @@     deriving stock (Generic)  {- | A 'KirokuAdapterConfig' with sensible defaults: @batchSize = 100@,-@bufferSize = 256@, @consumerGroup = 'Nothing'@ (ordinary single-consumer-subscription), @eventTypeFilter = 'AllEventTypes'@ (deliver every type), and-@selector = 'Nothing'@ (no extra predicate filtering). Supply the subscription-name and target; override individual fields with record-update syntax.+@bufferSize = 256@, @queueCapacity = 16@, @consumerGroup = 'Nothing'@+(ordinary single-consumer subscription), @eventTypeFilter = 'AllEventTypes'@+(deliver every type), and @selector = 'Nothing'@ (no extra predicate+filtering). Supply the subscription name and target; override individual fields+with record-update syntax.  Prefer this over a full record literal so that any field added to 'KirokuAdapterConfig' later is inherited at its default automatically:@@ -229,46 +253,69 @@         , subscriptionTarget = target         , batchSize = 100         , bufferSize = 256+        , queueCapacity = 16         , consumerGroup = Nothing         , eventTypeFilter = AllEventTypes         , selector = Nothing         } +{- | Convert any synchronous exception thrown by a Shibuya handler into an+'AckDecision'.++This ensures Shibuya still finalizes the ack. Asynchronous exceptions such as+thread cancellation are not caught.+-}+guardKirokuHandlerWith ::+    (SomeException -> AckDecision) ->+    Handler es msg ->+    Handler es msg+guardKirokuHandlerWith handleException h ingested =+    h ingested `catchSync` (pure . handleException)++{- | Recommended handler guard for this adapter.++A synchronous exception becomes @'AckRetry' ('RetryDelay' 1)@. Kiroku then+redelivers after one second and eventually dead-letters persistent failures+according to the subscription retry policy.+-}+guardKirokuHandler :: Handler es msg -> Handler es msg+guardKirokuHandler = guardKirokuHandlerWith (const (AckRetry (RetryDelay 1)))+ {- | Create a Shibuya 'Adapter' backed by a Kiroku subscription.  The adapter:  1. Calls 'subscriptionStream' to start a Kiroku subscription with a-   bounded 'TBQueue' bridge.+   ack-coupled bounded bridge. 2. Lifts the @Stream IO RecordedEvent@ to @Stream (Eff es)@ via    @Stream.morphInner liftIO@. 3. Wraps each 'RecordedEvent' into an 'Ingested' value with an    'Envelope' (mapping event ID → message ID, global position → cursor)-   and a no-op 'AckHandle' (except 'AckHalt' which cancels the-   subscription).+   and an 'AckHandle' whose finalized decision drives Kiroku checkpointing,+   retries, dead-lettering, or halt.  The returned adapter's @shutdown@ action cancels the underlying-subscription and flushes the sentinel through the queue so any-blocked stream reader terminates.+subscription and wakes any blocked stream reader. If the subscription worker+dies with an exception, @source@ terminates with that exception. -} kirokuAdapter ::     (IOE :> es) =>     KirokuStore ->     KirokuAdapterConfig ->     Eff es (Adapter es RecordedEvent)-kirokuAdapter store (KirokuAdapterConfig subName subTarget bs buf cg etf sel) = do+kirokuAdapter store KirokuAdapterConfig{subscriptionName = subName, subscriptionTarget = subTarget, batchSize = bs, bufferSize = buf, queueCapacity = qCap, consumerGroup = cg, eventTypeFilter = etf, selector = sel} = do     -- Build from 'defaultSubscriptionConfig' and override only the non-default     -- fields. Using the smart constructor (rather than a full record literal)     -- means any future field added to 'SubscriptionConfigM' is inherited at its     -- default automatically — e.g. EP-2's 'consumerGroupGuard', left 'False' here.-    let subConfig =+    let subConfig :: SubscriptionConfig+        subConfig =             (defaultSubscriptionConfig subName subTarget (\_ -> pure Continue))-                { batchSize = bs-                , queueCapacity = 16-                , overflowPolicy = DropSubscription-                , consumerGroup = cg-                , eventTypeFilter = etf-                , selector = sel+                { Sub.batchSize = bs+                , Sub.queueCapacity = qCap+                , Sub.consumerGroup = cg+                , Sub.eventTypeFilter = etf+                , Sub.selector = sel                 }      (ioStream, cancelAction) <- liftIO $ subscriptionAckStream store subConfig buf@@ -326,7 +373,11 @@     , batchSize :: !Int32     -- ^ Events per database fetch during catch-up (per member).     , bufferSize :: !Natural-    -- ^ Per-member 'TBQueue' capacity (backpressure threshold).+    -- ^ Per-member bridge 'TBQueue' capacity; must be at least 1.+    , queueCapacity :: !Natural+    {- ^ Per-member publisher-side capacity in batches. When this fills, Kiroku+    pauses and later resumes the member losslessly.+    -}     , memberConcurrency :: !Concurrency     -- ^ Per-member concurrency; must be 'Serial' (validated).     , eventTypeFilter :: !EventTypeFilter@@ -353,9 +404,9 @@  {- | A 'KirokuConsumerGroupConfig' with sensible defaults: @memberConcurrency = 'Serial'@ (the only legal per-member concurrency), @batchSize = 100@,-@bufferSize = 256@, @eventTypeFilter = 'AllEventTypes'@ (deliver every type),-@selector = 'Nothing'@ (no extra predicate filtering). Supply the subscription-name, target, and group size.+@bufferSize = 256@, @queueCapacity = 16@, @eventTypeFilter = 'AllEventTypes'@+(deliver every type), @selector = 'Nothing'@ (no extra predicate filtering).+Supply the subscription name, target, and group size. -} defaultConsumerGroupConfig ::     SubscriptionName -> SubscriptionTarget -> Int32 -> KirokuConsumerGroupConfig@@ -366,13 +417,14 @@         , groupSize = n         , batchSize = 100         , bufferSize = 256+        , queueCapacity = 16         , memberConcurrency = Serial         , eventTypeFilter = AllEventTypes         , selector = Nothing         }  {- | Map a requested per-member concurrency onto the group's validated Shibuya-@('Ordering', 'Concurrency')@.+@('OrderingPolicy', 'Concurrency')@.  The group's ordering contract is always 'PartitionedInOrder'; a member's own ordered stream must be processed serially. This reuses Shibuya's own@@ -383,7 +435,7 @@ @('PartitionedInOrder', 'Serial')@ also passes 'validatePolicy', so 'runApp' will not reject it later. -}-consumerGroupPolicy :: Concurrency -> Either PolicyError (Ordering, Concurrency)+consumerGroupPolicy :: Concurrency -> Either PolicyError (OrderingPolicy, Concurrency) consumerGroupPolicy conc = do     -- A member delivers one strictly-ordered stream; only Serial is honest.     validatePolicy StrictInOrder conc -- rejects Ahead/Async with PolicyError@@ -392,6 +444,9 @@ {- | Present a whole kiroku consumer group as a single 'PartitionedInOrder' unit: one call yields @groupSize@ named 'QueueProcessor's, each backed by its own member adapter and each pinned to @('PartitionedInOrder', 'Serial')@.+The supplied handler is wrapped in 'guardKirokuHandler' automatically so a+synchronous handler exception finalizes a retry disposition instead of+abandoning Kiroku's ack reply.  This replaces the manual @mapM mkMemberAdapter [0 .. N-1]@ boilerplate. The member→policy mapping is validated once up front via 'consumerGroupPolicy'; if@@ -401,11 +456,9 @@  Each processor's 'ProcessorId' is @\"\<subscriptionName\>-member-\<m\>\"@ so member identity is readable off the id-and two members never collide. The group-validity invariant (@groupSize >= 1@,-@0 <= member < groupSize@) is enforced downstream by-'Kiroku.Store.Subscription.subscribe' (throwing-'Kiroku.Store.Subscription.Types.InvalidConsumerGroup'); 'groupSize >= 1' is a-documented precondition of this helper.+and two members never collide. @groupSize >= 1@ is validated before any+subscription opens, throwing 'InvalidConsumerGroup' with member 0 when it is+violated. -} kirokuConsumerGroupProcessors ::     (IOE :> es) =>@@ -413,42 +466,63 @@     KirokuConsumerGroupConfig ->     Handler es RecordedEvent ->     Eff es (Either PolicyError [(ProcessorId, QueueProcessor es)])-kirokuConsumerGroupProcessors-    store+kirokuConsumerGroupProcessors store cfg@KirokuConsumerGroupConfig{subscriptionName = subName, subscriptionTarget = subTarget, groupSize = n, batchSize = bs, bufferSize = buf, queueCapacity = qCap, eventTypeFilter = etf, selector = sel} handler =+    kirokuConsumerGroupProcessorsWith mkMemberAdapter cfg handler+  where+    mkMemberAdapter m =+        kirokuAdapter+            store+            KirokuAdapterConfig+                { subscriptionName = subName+                , subscriptionTarget = subTarget+                , batchSize = bs+                , bufferSize = buf+                , queueCapacity = qCap+                , consumerGroup = Just (ConsumerGroup{member = m, size = n})+                , eventTypeFilter = etf+                , selector = sel+                }++{- | Factory-parameterized core for 'kirokuConsumerGroupProcessors'.++This is exported primarily for tests and advanced integration points. It+validates the group configuration, creates each member adapter with the+provided factory, and shuts down already-created adapters if a later factory+call throws before ownership can be returned to the caller.+-}+kirokuConsumerGroupProcessorsWith ::+    (IOE :> es) =>+    (Int32 -> Eff es (Adapter es RecordedEvent)) ->+    KirokuConsumerGroupConfig ->+    Handler es RecordedEvent ->+    Eff es (Either PolicyError [(ProcessorId, QueueProcessor es)])+kirokuConsumerGroupProcessorsWith+    mkMemberAdapter     KirokuConsumerGroupConfig         { subscriptionName = subName-        , subscriptionTarget = subTarget         , groupSize = n-        , batchSize = bs-        , bufferSize = buf         , memberConcurrency = mc-        , eventTypeFilter = etf-        , selector = sel         }     handler =-        case consumerGroupPolicy mc of-            Left e -> pure (Left e)-            Right (ordering, conc) -> do-                let SubscriptionName name = subName-                processors <--                    mapM-                        ( \m -> do-                            adapter <--                                kirokuAdapter-                                    store-                                    KirokuAdapterConfig-                                        { subscriptionName = subName-                                        , subscriptionTarget = subTarget-                                        , batchSize = bs-                                        , bufferSize = buf-                                        , consumerGroup = Just (ConsumerGroup{member = m, size = n})-                                        , eventTypeFilter = etf-                                        , selector = sel-                                        }-                            let pid = ProcessorId (name <> "-member-" <> T.pack (show m))-                            -- Built directly (not via 'mkProcessor', which hardcodes-                            -- Unordered/Serial) so the group policy is pinned.-                            pure (pid, QueueProcessor adapter handler ordering conc)-                        )-                        [0 .. n - 1]-                pure (Right processors)+        if n < 1+            then throwIO (InvalidConsumerGroup 0 n)+            else case consumerGroupPolicy mc of+                Left e -> pure (Left e)+                Right (ordering, conc) -> do+                    let SubscriptionName name = subName+                    adapters <- createAdapters [] 0+                    let processors =+                            [ let pid = ProcessorId (name <> "-member-" <> T.pack (show m))+                               in (pid, QueueProcessor adapter (guardKirokuHandler handler) ordering conc)+                            | (m, adapter) <- zip [0 .. n - 1] adapters+                            ]+                    pure (Right processors)+      where+        createAdapters created m+            | m >= n = pure (reverse created)+            | otherwise = do+                adapter <- mkMemberAdapter m `onException` shutdownCreated created+                createAdapters (adapter : created) (m + 1)++        shutdownCreated =+            mapM_ $ \Adapter{shutdown = shutdownAction} -> shutdownAction

@@ -176,6 +176,7 @@             , partition = Nothing             , enqueuedAt = Just ts             , traceContext = metadataTraceContext meta+            , headers = Nothing             , attempt = Nothing             , attributes = eventAttributes attrs etype pos             , payload = event

@@ -1,8 +1,10 @@ module Main where  import Control.Concurrent (threadDelay)+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar) import Control.Concurrent.STM (atomically, newTVarIO, readTVar, registerDelay, writeTVar) import Control.Concurrent.STM qualified as STM+import Control.Exception qualified as E import Control.Lens ((&), (.~), (^.)) import Control.Monad (unless) import Control.Monad.IO.Class (liftIO)@@ -25,20 +27,24 @@ import Hasql.Session qualified as Session import Kiroku.Store import Kiroku.Store.SQL qualified as SQL+import Kiroku.Store.Subscription.Worker (withFetchBatchHookForTest) import Kiroku.Test.Postgres (withMigratedTestDatabase, withSharedMigratedPostgres) import OpenTelemetry.Attributes (toAttribute)+import Shibuya.Adapter (Adapter (..)) import Shibuya.Adapter.Kiroku (     consumerGroupPolicy,     defaultConsumerGroupConfig,     defaultKirokuAdapterConfig,+    guardKirokuHandlerWith,     kirokuAdapter,     kirokuConsumerGroupProcessors,+    kirokuConsumerGroupProcessorsWith,  ) import Shibuya.Adapter.Kiroku.Convert (KirokuEnvelopeAttrs, kirokuEnvelopeAttrs, toEnvelope) import Shibuya.App (     ProcessorId (..),     QueueProcessor (..),-    SupervisionStrategy (..),+    defaultAppConfig,     getAppMetrics,     mkProcessor,     runApp,@@ -49,11 +55,14 @@ import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..)) import Shibuya.Core.Ack qualified as Ack import Shibuya.Core.Error (PolicyError (..))-import Shibuya.Core.Ingested (Ingested (..))-import Shibuya.Core.Types (Envelope (..))-import Shibuya.Policy (Concurrency (..), Ordering (..))-import Shibuya.Runner.Metrics (ProcessorState (..))+import Shibuya.Core.Ingested (Message (..))+import Shibuya.Core.Metrics (ProcessorState (..))+import Shibuya.Core.Types (Attempt (..), Envelope (..))+import Shibuya.Policy (Concurrency (..), OrderingPolicy (..)) import Shibuya.Telemetry.Effect (runTracingNoop)+import Streamly.Data.Fold qualified as Fold+import Streamly.Data.Stream qualified as Stream+import System.Timeout (timeout) import Test.Hspec  main :: IO ()@@ -92,6 +101,17 @@             missingTraceparent `shouldBe` Nothing             nonStringTraceparent `shouldBe` Nothing +        it "does not surface broker headers (headers is Nothing), even with trace metadata" $ do+            let Envelope{headers = noMeta} =+                    toEnvelope sampleEnvelopeAttrs (makeRecordedEvent Nothing)+                Envelope{headers = withTraceMeta} =+                    toEnvelope+                        sampleEnvelopeAttrs+                        (makeRecordedEvent (Just (Aeson.object ["traceparent" Aeson..= ("00-abc-def-01" :: Text)])))++            noMeta `shouldBe` Nothing+            withTraceMeta `shouldBe` Nothing+         it "stamps kiroku identity attributes for a non-grouped subscription (EP-5 M2)" $ do             let attrs = kirokuEnvelopeAttrs "orders-proj" Nothing                 Envelope{attributes} = toEnvelope attrs (makeRecordedEvent Nothing)@@ -156,7 +176,7 @@                                     c <- readTVar countVar                                     writeTVar countVar (c + 1)                             pure AckOk-                    res <- runApp IgnoreFailures 100 [(ProcessorId "etf", mkProcessor adapter handler)]+                    res <- runApp defaultAppConfig [(ProcessorId "etf", mkProcessor adapter handler)]                     case res of                         Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)                         Right appHandle -> do@@ -199,7 +219,7 @@                     case res of                         Left err -> liftIO $ expectationFailure ("kirokuConsumerGroupProcessors failed: " <> show err)                         Right processors -> do-                            appRes <- runApp IgnoreFailures 100 processors+                            appRes <- runApp defaultAppConfig processors                             case appRes of                                 Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)                                 Right appHandle -> do@@ -241,7 +261,7 @@                                     c <- readTVar countVar                                     writeTVar countVar (c + 1)                             pure AckOk-                    res <- runApp IgnoreFailures 100 [(ProcessorId "sel", mkProcessor adapter handler)]+                    res <- runApp defaultAppConfig [(ProcessorId "sel", mkProcessor adapter handler)]                     case res of                         Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)                         Right appHandle -> do@@ -274,7 +294,7 @@                                     writeTVar countVar (c + 1)                             pure AckOk -                    res <- runApp IgnoreFailures 100 [(ProcessorId "catchup", mkProcessor adapter handler)]+                    res <- runApp defaultAppConfig [(ProcessorId "catchup", mkProcessor adapter handler)]                     case res of                         Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)                         Right appHandle -> do@@ -301,7 +321,7 @@                                     writeTVar countVar (c + 1)                             pure AckOk -                    res <- runApp IgnoreFailures 100 [(ProcessorId "live", mkProcessor adapter handler)]+                    res <- runApp defaultAppConfig [(ProcessorId "live", mkProcessor adapter handler)]                     case res of                         Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)                         Right appHandle -> do@@ -316,6 +336,78 @@                 collected <- readIORef ref                 length collected `shouldBe` 5 +            it "survives an append burst while the handler is paused" $ \store -> do+                firstSeen <- newEmptyMVar+                release <- newEmptyMVar+                countVar <- newTVarIO (0 :: Int)+                seenRef <- newIORef ([] :: [RecordedEvent])++                runEff $ runTracingNoop $ do+                    adapter <-+                        kirokuAdapter store $+                            defaultKirokuAdapterConfig (SubscriptionName "burst-pause-resume") AllStreams+                                & #queueCapacity .~ 1++                    let handler ingested = do+                            n <- liftIO $ do+                                modifyIORef' seenRef (envelopePayload ingested :)+                                atomically $ do+                                    c <- readTVar countVar+                                    let c' = c + 1+                                    writeTVar countVar c'+                                    pure c'+                            if n == 1+                                then liftIO $ putMVar firstSeen () >> takeMVar release+                                else pure ()+                            pure AckOk++                    res <- runApp defaultAppConfig [(ProcessorId "burst", mkProcessor adapter handler)]+                    case res of+                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+                        Right appHandle -> do+                            liftIO $ do+                                let appendOne i =+                                        runStoreIO store $+                                            appendToStream+                                                (StreamName ("burst-" <> T.pack (show (i :: Int))))+                                                NoStream+                                                [makeEvent ("Burst" <> T.pack (show i)) (Aeson.object [])]+                                Right _ <- appendOne 1+                                takeMVar firstSeen+                                mapM_ (\i -> appendOne i >>= (`shouldSatisfy` either (const False) (const True))) [2 .. 40 :: Int]+                                putMVar release ()+                                waitForCount countVar 40 15_000_000+                            stopApp appHandle++                seen <- readIORef seenRef+                sort (map globalPos seen) `shouldBe` [1 .. 40]++            it "ends adapter.source cleanly after shutdown" $ \store -> do+                within "adapter source clean shutdown" $+                    runEff $+                        runTracingNoop $ do+                            adapter <- kirokuAdapter store $ defaultKirokuAdapterConfig (SubscriptionName "source-clean-end") AllStreams+                            let Adapter{source = sourceStream, shutdown = shutdownAction} = adapter+                            shutdownAction+                            Stream.fold Fold.drain sourceStream++            it "rethrows worker crashes through adapter.source" $ \store -> do+                Right _ <- runStoreIO store $ appendToStream (StreamName "source-crash-1") NoStream [makeEvent "SourceCrash" (Aeson.object [])]+                threadDelay 200_000+                let injectBoom _ _ = E.throwIO (userError "adapter source worker boom")+                result <-+                    withFetchBatchHookForTest injectBoom $+                        E.try $+                            within "adapter source worker crash" $+                                runEff $+                                    runTracingNoop $ do+                                        adapter <- kirokuAdapter store $ defaultKirokuAdapterConfig (SubscriptionName "source-crash") AllStreams+                                        let Adapter{source = sourceStream} = adapter+                                        Stream.fold Fold.drain sourceStream+                case result of+                    Left (e :: E.SomeException) -> show e `shouldContain` "adapter source worker boom"+                    Right _ -> expectationFailure "expected adapter.source to rethrow the worker exception"+             it "runs multiple category subscriptions concurrently" $ \store -> do                 Right _ <- runStoreIO store $ appendToStream (StreamName "orders-s1") NoStream [makeEvent "OrderCreated" (Aeson.object [])]                 Right _ <- runStoreIO store $ appendToStream (StreamName "orders-s2") NoStream [makeEvent "OrderShipped" (Aeson.object [])]@@ -349,8 +441,7 @@                      res <-                         runApp-                            IgnoreFailures-                            100+                            defaultAppConfig                             [ (ProcessorId "orders", mkProcessor ordersAdapter (mkHandler ordersRef ordersCount))                             , (ProcessorId "payments", mkProcessor paymentsAdapter (mkHandler paymentsRef paymentsCount))                             , (ProcessorId "inventory", mkProcessor inventoryAdapter (mkHandler inventoryRef inventoryCount))@@ -414,8 +505,7 @@                      res <-                         runApp-                            IgnoreFailures-                            100+                            defaultAppConfig                             [ (ProcessorId "good", mkProcessor goodAdapter goodHandler)                             , (ProcessorId "bad", mkProcessor badAdapter badHandler)                             , (ProcessorId "other", mkProcessor otherAdapter otherHandler)@@ -470,8 +560,7 @@                      res <-                         runApp-                            IgnoreFailures-                            100+                            defaultAppConfig                             [ (ProcessorId "a", mkProcessor adapterA handlerA)                             , (ProcessorId "b", mkProcessor adapterB handlerB)                             ]@@ -507,7 +596,7 @@                             -- Retry the first delivery, accept the second.                             pure (if n == 1 then AckRetry (Ack.RetryDelay 0) else AckOk) -                    res <- runApp IgnoreFailures 100 [(ProcessorId "ackretry", mkProcessor adapter handler)]+                    res <- runApp defaultAppConfig [(ProcessorId "ackretry", mkProcessor adapter handler)]                     case res of                         Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)                         Right appHandle -> do@@ -546,7 +635,7 @@                                     liftIO $ modifyIORef' okPayloads (envelopePayload ingested :)                                     pure AckOk -                    res <- runApp IgnoreFailures 100 [(ProcessorId "ackdl", mkProcessor adapter handler)]+                    res <- runApp defaultAppConfig [(ProcessorId "ackdl", mkProcessor adapter handler)]                     case res of                         Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)                         Right appHandle -> do@@ -560,6 +649,79 @@                 oks <- readIORef okPayloads                 map globalPos oks `shouldBe` [2] +            it "guardKirokuHandlerWith retries a transient handler exception and preserves attempts" $ \store -> do+                Right _ <- runStoreIO store $ appendToStream (StreamName "guardretry-1") NoStream [makeEvent "GR1" (Aeson.object [])]+                threadDelay 200_000++                deliveries <- newIORef (0 :: Int)+                attempts <- newIORef ([] :: [Maybe Attempt])+                countVar <- newTVarIO (0 :: Int)++                runEff $ runTracingNoop $ do+                    adapter <-+                        kirokuAdapter store $+                            defaultKirokuAdapterConfig (SubscriptionName "guardretry-proj") AllStreams++                    let throwingOnce ingested = do+                            let Message{envelope = Envelope{attempt = deliveryAttempt}} = ingested+                            n <- liftIO $ do+                                modifyIORef' attempts (deliveryAttempt :)+                                modifyIORef' deliveries (+ 1)+                                atomically $ do+                                    c <- readTVar countVar+                                    let c' = c + 1+                                    writeTVar countVar c'+                                    pure c'+                            if n == 1+                                then liftIO $ E.throwIO (userError "transient projection failure")+                                else pure AckOk+                        handler = guardKirokuHandlerWith (const (AckRetry (Ack.RetryDelay 0))) throwingOnce++                    res <- runApp defaultAppConfig [(ProcessorId "guardretry", mkProcessor adapter handler)]+                    case res of+                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+                        Right appHandle -> do+                            liftIO $ waitForCount countVar 2 10_000_000+                            stopApp appHandle++                n <- readIORef deliveries+                n `shouldBe` 2+                reverse <$> readIORef attempts `shouldReturn` [Just (Attempt 0), Just (Attempt 1)]+                dls <- readDeadLetters store "guardretry-proj"+                length dls `shouldBe` 0++            it "guardKirokuHandlerWith dead-letters a persistent handler exception and continues" $ \store -> do+                Right _ <- runStoreIO store $ appendToStream (StreamName "guarddl-1") NoStream [makeEvent "GDL1" (Aeson.object [])]+                Right _ <- runStoreIO store $ appendToStream (StreamName "guarddl-2") NoStream [makeEvent "GDL2" (Aeson.object [])]+                threadDelay 200_000++                nextSeen <- newTVarIO (0 :: Int)++                runEff $ runTracingNoop $ do+                    adapter <-+                        kirokuAdapter store $+                            defaultKirokuAdapterConfig (SubscriptionName "guarddl-proj") AllStreams++                    let alwaysThrowFirst ingested =+                            if globalPos (envelopePayload ingested) == 1+                                then liftIO $ E.throwIO (userError "persistent projection failure")+                                else do+                                    liftIO $ atomically $ do+                                        c <- readTVar nextSeen+                                        writeTVar nextSeen (c + 1)+                                    pure AckOk+                        handler = guardKirokuHandlerWith (const (AckRetry (Ack.RetryDelay 0))) alwaysThrowFirst++                    res <- runApp defaultAppConfig [(ProcessorId "guarddl", mkProcessor adapter handler)]+                    case res of+                        Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+                        Right appHandle -> do+                            liftIO $ waitForCount nextSeen 1 15_000_000+                            stopApp appHandle++                dls <- readDeadLetters store "guarddl-proj"+                map SQL.deadLetterGlobalPosition dls `shouldBe` [1]+         describe "consumer groups" $ do             it "four-member group delivers a disjoint, complete partition of the stream" $ \store -> do                 -- 20 streams × 2 events = 40 events, global positions 1..40, category "cg".@@ -604,7 +766,7 @@                             | m <- [0 .. 3 :: Int]                             ] -                    res <- runApp IgnoreFailures 100 processors+                    res <- runApp defaultAppConfig processors                     case res of                         Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)                         Right appHandle -> do@@ -639,6 +801,102 @@                     Left err -> err `shouldBe` InvalidPolicyCombo "StrictInOrder requires Serial concurrency"                     Right _ -> expectationFailure "expected Left PolicyError for Async member concurrency" +                withResult <-+                    runEff $+                        runTracingNoop $+                            kirokuConsumerGroupProcessorsWith+                                ( \_ -> do+                                    liftIO $ expectationFailure "factory should not be called for an invalid policy"+                                    pure stubAdapter+                                )+                                ( defaultConsumerGroupConfig (SubscriptionName "cgp-reject-with") (Category (CategoryName "cgp-reject-with")) 4+                                    & #memberConcurrency .~ Async 4+                                )+                                ( \ingested -> do+                                    let _ = envelopePayload ingested+                                    pure AckOk+                                )+                case withResult of+                    Left err -> err `shouldBe` InvalidPolicyCombo "StrictInOrder requires Serial concurrency"+                    Right _ -> expectationFailure "expected Left PolicyError for Async member concurrency"++            it "throws InvalidConsumerGroup for non-positive group sizes" $ \store -> do+                let handler ingested = do+                        let _ = envelopePayload ingested+                        pure AckOk+                    throwsSize n =+                        runEff+                            ( runTracingNoop $+                                kirokuConsumerGroupProcessors+                                    store+                                    (defaultConsumerGroupConfig (SubscriptionName ("cgp-invalid-" <> T.pack (show n))) AllStreams n)+                                    handler+                            )+                            `shouldThrow` ( \InvalidConsumerGroup{invalidMember = member, invalidSize = size} ->+                                                member == 0 && size == n+                                          )+                throwsSize 0+                throwsSize (-1)++            it "shuts down already-created member adapters after a later factory failure" $ \_store -> do+                shutdowns <- newIORef ([] :: [Int32])+                let sentinel = userError "member 2 failed"+                    factory m+                        | m == 2 = liftIO $ E.throwIO sentinel+                        | otherwise =+                            pure+                                Adapter+                                    { adapterName = "stub-kiroku"+                                    , source = Stream.nil+                                    , shutdown = liftIO $ modifyIORef' shutdowns (<> [m])+                                    }+                    handler ingested = do+                        let _ = envelopePayload ingested+                        pure AckOk+                result <-+                    E.try $+                        runEff $+                            runTracingNoop $+                                kirokuConsumerGroupProcessorsWith+                                    factory+                                    (defaultConsumerGroupConfig (SubscriptionName "cgp-partial-cleanup") AllStreams 3)+                                    handler+                case result of+                    Left (e :: E.SomeException) -> show e `shouldContain` "member 2 failed"+                    Right _ -> expectationFailure "expected the member factory failure to rethrow"+                readIORef shutdowns `shouldReturn` [1, 0]++            it "wraps consumer-group handlers so exceptions finalize a retry decision" $ \store -> do+                Right _ <- runStoreIO store $ appendToStream (StreamName "cgp-guard-1") NoStream [makeEvent "CGGuard" (Aeson.object [])]+                threadDelay 200_000++                countVar <- newTVarIO (0 :: Int)+                runEff $+                    runTracingNoop $ do+                        result <-+                            kirokuConsumerGroupProcessors+                                store+                                (defaultConsumerGroupConfig (SubscriptionName "cgp-guard") AllStreams 1)+                                ( \_ -> do+                                    n <- liftIO $ atomically $ do+                                        c <- readTVar countVar+                                        let c' = c + 1+                                        writeTVar countVar c'+                                        pure c'+                                    if n == 1+                                        then liftIO $ E.throwIO (userError "guard me")+                                        else pure AckOk+                                )+                        case result of+                            Left err -> liftIO $ expectationFailure ("kirokuConsumerGroupProcessors failed: " <> show err)+                            Right processors -> do+                                appRes <- runApp defaultAppConfig processors+                                case appRes of+                                    Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)+                                    Right appHandle -> do+                                        liftIO $ waitForCount countVar 2 15_000_000+                                        stopApp appHandle+             it "one call yields N PartitionedInOrder processors; members partition the stream disjointly" $ \store -> do                 -- 20 streams × 2 events = 40 events, global positions 1..40, category "cgp".                 let streams = ["cgp-" <> T.pack (show i) | i <- [1 .. 20 :: Int]]@@ -687,7 +945,7 @@                                 pids                                     `shouldBe` ["cgp-shibuya-group-member-" <> T.pack (show m) | m <- [0 .. 3 :: Int]] -                            appRes <- runApp IgnoreFailures 100 processors+                            appRes <- runApp defaultAppConfig processors                             case appRes of                                 Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err)                                 Right appHandle -> do@@ -707,9 +965,17 @@  -- Helpers -envelopePayload :: Ingested es RecordedEvent -> RecordedEvent-envelopePayload ing = let Ingested{envelope = env} = ing in env ^. #payload+envelopePayload :: Message es RecordedEvent -> RecordedEvent+envelopePayload ing = let Message{envelope = env} = ing in env ^. #payload +stubAdapter :: Adapter es RecordedEvent+stubAdapter =+    Adapter+        { adapterName = "stub-kiroku"+        , source = Stream.nil+        , shutdown = pure ()+        }+ -- | Read the dead letters recorded for a non-group subscription (member 0). readDeadLetters :: KirokuStore -> Text -> IO [SQL.DeadLetterRecord] readDeadLetters store subName = do@@ -764,6 +1030,13 @@     unless result $ do         actual <- atomically $ readTVar countVar         expectationFailure ("Timed out waiting for count " <> show target <> ", got " <> show actual)++within :: String -> IO a -> IO a+within label action = do+    result <- timeout 10_000_000 action+    case result of+        Nothing -> fail ("Timed out waiting for " <> label)+        Just a -> pure a  makeEvent :: Text -> Value -> EventData makeEvent typ p =