diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/shibuya-kiroku-adapter.cabal b/shibuya-kiroku-adapter.cabal --- a/shibuya-kiroku-adapter.cabal +++ b/shibuya-kiroku-adapter.cabal @@ -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 diff --git a/src/Shibuya/Adapter/Kiroku.hs b/src/Shibuya/Adapter/Kiroku.hs --- a/src/Shibuya/Adapter/Kiroku.hs +++ b/src/Shibuya/Adapter/Kiroku.hs @@ -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 @\"\-member-\\"@ 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 diff --git a/src/Shibuya/Adapter/Kiroku/Convert.hs b/src/Shibuya/Adapter/Kiroku/Convert.hs --- a/src/Shibuya/Adapter/Kiroku/Convert.hs +++ b/src/Shibuya/Adapter/Kiroku/Convert.hs @@ -176,6 +176,7 @@ , partition = Nothing , enqueuedAt = Just ts , traceContext = metadataTraceContext meta + , headers = Nothing , attempt = Nothing , attributes = eventAttributes attrs etype pos , payload = event diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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 =