diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,106 @@ +# Changelog + +## Unreleased + +## 0.2.0.0 — 2026-05-31 + +### Breaking Changes + +#### Shibuya 0.6 and OpenTelemetry 1.0 semantics + +* The adapter now targets `shibuya-core >=0.6 && <0.7` and + `hs-opentelemetry-api ^>=1.0`. +* `Envelope.attributes` now also carries current OpenTelemetry messaging + semantic-convention attributes for the Kiroku source: + `messaging.system = "kiroku"` and `messaging.destination.name` set to the + subscription name. These override Shibuya's framework defaults on the + per-message process span while preserving the existing `kiroku.*` attributes. + +#### Ack decisions now drive Kiroku checkpointing (plan 40) + +* The adapter bridges through `kiroku-store`'s ack-coupled + `subscriptionAckStream`: each event blocks the Kiroku worker until the Shibuya + handler's `AckDecision` is finalized, so the decision drives Kiroku + checkpointing per event. Previously `AckRetry` / `AckDeadLetter` were no-ops. + * `AckOk` — checkpoint past the event. + * `AckRetry delay` — redeliver the same event after `delay`, bounded by the + subscription's retry policy, then dead-letter on exhaustion. + * `AckDeadLetter reason` — record the event in `kiroku.dead_letters` (reason + translated to a Kiroku-native `DeadLetterReason`) and advance past it. + * `AckHalt` — cancels the subscription (unchanged). +* The envelope `attempt` now reports the zero-based redelivery count. +* `Shibuya.Adapter.Kiroku.Convert` exposes `toIngestedAck` (replacing + `toIngested`), `toKirokuResult`, and `toKirokuDeadLetterReason`. + +### New Features + +#### Kiroku identity on envelope attributes for end-to-end tracing (MasterPlan 6 EP-5) + +* `Envelope.attributes` (previously always empty) is now populated with the + kiroku identity so it appears on Shibuya's per-message span: + `kiroku.subscription.name`, `kiroku.event.type`, `kiroku.event.global_position`, + and — for a consumer-group member — `kiroku.consumer_group.member`. Combined + with `kiroku-otel`'s native subscription spans, a single trace is followable + from a kiroku subscription into Shibuya's processing; the keys match on both + sides. +* `Shibuya.Adapter.Kiroku.Convert` exposes `KirokuEnvelopeAttrs`; `toEnvelope` + and `toIngestedAck` take it. `kirokuAdapter` derives the subscription name and + member from its config, so the single-adapter and consumer-group paths are both + covered. No `shibuya-core` change; the ack-handle logic and trace-context + propagation are unchanged. + +#### `defaultKirokuAdapterConfig` smart constructor + +* `defaultKirokuAdapterConfig name target` builds a `KirokuAdapterConfig` with + recommended defaults (batch size 100, buffer 256, no consumer group, + `AllEventTypes`, no selector); override individual fields with record-update or + a generic-lens label (`& #eventTypeFilter .~ …`). Prefer it over a full record + literal so a future field is inherited at its default automatically — mirrors + `defaultConsumerGroupConfig` and `kiroku-store`'s `defaultSubscriptionConfig`. +* `KirokuAdapterConfig` and `KirokuConsumerGroupConfig` now derive `Generic`, + enabling generic-lens label access/update of their fields. + +#### Consumer groups as a single partitioned subscription (plan 42) + +* `kirokuConsumerGroupProcessors` — present a whole kiroku consumer group as one + `PartitionedInOrder` unit: a single call yields `N` named `QueueProcessor`s + (one member adapter each, `ProcessorId "-member-"`), each pinned to + `(PartitionedInOrder, Serial)`. Replaces the manual `mapM mkMemberAdapter + [0..N-1]` wiring. +* `KirokuConsumerGroupConfig` + `defaultConsumerGroupConfig` — describe a whole + group (subscription name, target, group size, batch size, buffer size, and a + per-member `Concurrency` that must be `Serial`). +* `consumerGroupPolicy` — map a requested per-member `Concurrency` onto the + validated group policy `(PartitionedInOrder, Serial)`, reusing Shibuya's own + `validatePolicy` so `Ahead`/`Async` are rejected early with + `InvalidPolicyCombo` before any subscription opens. +* No `shibuya-core` changes: the helper only consumes existing exports. + +#### Event-type filter and selector forwarding (plan 43) + +* `KirokuAdapterConfig` and `KirokuConsumerGroupConfig` gain an `eventTypeFilter` + field (`AllEventTypes` / `OnlyEventTypes (Set EventType)`, default + `AllEventTypes`), forwarded into the underlying subscription so the adapter + delivers only the chosen event types. Filtering is worker-side, ahead of the + ack-coupled bridge, so a filtered-out event never reaches the Shibuya handler, + is never retried or dead-lettered, and the checkpoint still advances past it. + `EventTypeFilter (..)` is re-exported. +* Both config records also gain an optional `selector :: Maybe (RecordedEvent -> + Bool)` field (default `Nothing`) — the opaque escape hatch for filtering on a + property the type set cannot express (payload, metadata, correlation ids). It + composes with `eventTypeFilter` as a logical AND and is applied worker-side + with the same no-stall / checkpoint-advances guarantee. For a consumer group + the same predicate is applied to every member. + +## 0.1.0.0 — 2026-05-23 + +### New Features + +* Initial release. +* `kirokuAdapter` — create a Shibuya `Adapter es RecordedEvent` from a + Kiroku store handle and subscription configuration. +* `KirokuAdapterConfig` — subscription name, target, batch size, and + TBQueue buffer size. +* Ack semantics: AckOk/AckRetry/AckDeadLetter are no-ops (checkpoint + managed by Kiroku); AckHalt cancels the subscription. +* `Shibuya.Adapter.Kiroku.Convert` — RecordedEvent to Envelope mapping. diff --git a/shibuya-kiroku-adapter.cabal b/shibuya-kiroku-adapter.cabal new file mode 100644 --- /dev/null +++ b/shibuya-kiroku-adapter.cabal @@ -0,0 +1,92 @@ +cabal-version: 3.0 +name: shibuya-kiroku-adapter +version: 0.2.0.0 +synopsis: + Kiroku event store adapter for the Shibuya queue processing framework + +description: + A Shibuya adapter that integrates with Kiroku, a PostgreSQL event store. + Wraps Kiroku's push-based subscriptions into Shibuya's pull-based Adapter + 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. + +author: Nadeem Bitar +maintainer: nadeem@gmail.com +license: BSD-3-Clause +build-type: Simple +category: Concurrency, Database, Eventing +extra-doc-files: CHANGELOG.md +homepage: https://github.com/shinzui/kiroku +bug-reports: https://github.com/shinzui/kiroku/issues + +source-repository head + type: git + location: https://github.com/shinzui/kiroku.git + +common common + default-language: GHC2024 + default-extensions: + DeriveAnyClass + DuplicateRecordFields + OverloadedLabels + OverloadedStrings + + ghc-options: -Wall + +library + import: common + exposed-modules: + Shibuya.Adapter.Kiroku + Shibuya.Adapter.Kiroku.Convert + + build-depends: + , aeson >=2.1 && <2.3 + , base >=4.18 && <5 + , effectful-core >=2.4 && <2.7 + , hs-opentelemetry-api ^>=1.0 + , hs-opentelemetry-semantic-conventions ^>=1.40 + , kiroku-store ^>=0.2 + , shibuya-core >=0.6 && <0.7 + , stm >=2.5 && <2.6 + , streamly-core >=0.3 && <0.4 + , text >=2.0 && <2.2 + , unordered-containers >=0.2 && <0.3 + , uuid >=1.3 && <1.4 + + hs-source-dirs: src + +test-suite shibuya-kiroku-adapter-test + import: common + type: exitcode-stdio-1.0 + main-is: Main.hs + hs-source-dirs: test + ghc-options: -threaded -rtsopts -with-rtsopts=-N + build-depends: + , aeson >=2.1 && <2.3 + , base >=4.18 && <5 + , containers >=0.6 && <0.8 + , directory + , effectful >=2.4 && <2.7 + , ephemeral-pg >=0.2 && <0.3 + , generic-lens >=2.2 && <2.4 + , hasql >=1.10 && <1.11 + , hasql-pool >=1.2 && <1.5 + , hs-opentelemetry-api ^>=1.0 + , hspec >=2.10 && <2.12 + , kiroku-store ^>=0.2 + , kiroku-test-support + , lens >=5.2 && <5.4 + , shibuya-core >=0.6 && <0.7 + , shibuya-kiroku-adapter + , stm >=2.5 && <2.6 + , text >=2.0 && <2.2 + , time >=1.12 && <1.15 + , unordered-containers >=0.2 && <0.3 + , uuid >=1.3 && <1.4 diff --git a/src/Shibuya/Adapter/Kiroku.hs b/src/Shibuya/Adapter/Kiroku.hs new file mode 100644 --- /dev/null +++ b/src/Shibuya/Adapter/Kiroku.hs @@ -0,0 +1,454 @@ +{- | Kiroku event store adapter for the Shibuya queue processing framework. + +This adapter wraps Kiroku's push-based subscriptions into Shibuya's +pull-based 'Adapter' interface. Events are bridged through a bounded +'TBQueue' (via @kiroku-store@'s 'subscriptionStream') and lifted into the +effectful stack with @Stream.morphInner@. + +== Example + +@ +import Effectful (runEff) +import Kiroku.Store (withStore, defaultConnectionSettings) +import Shibuya.Adapter.Kiroku +import Shibuya.App +import Shibuya.Core.Ack (AckDecision (..)) +import Shibuya.Telemetry.Effect (runTracingNoop) + +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 handler ingested = do + -- process ingested.envelope.payload :: RecordedEvent + pure AckOk + + Right appHandle <- runApp IgnoreFailures 100 + [(ProcessorId \"my-projection\", mkProcessor adapter handler)] + + waitApp appHandle +@ + +== Consumer-Group Example (size 4) + +A consumer group splits one logical subscription across @N@ members. Each +originating stream is deterministically assigned to exactly one member (by a +hash computed in PostgreSQL), so same-stream events stay ordered while distinct +streams are processed in parallel. To run a whole group in one process, use +'kirokuConsumerGroupProcessors': one call yields @N@ named processors, each a +member adapter pinned to the group-level @('PartitionedInOrder', 'Serial')@ +policy — no manual @[0..N-1]@ wiring. + +@ +main :: IO () +main = withStore settings $ \\store -> + runEff $ runTracingNoop $ do + let cfg = defaultConsumerGroupConfig + (SubscriptionName \"orders-projection\") + (Category (CategoryName \"orders\")) + 4 -- group size + + Right processors <- kirokuConsumerGroupProcessors store cfg handler + Right appHandle <- runApp IgnoreFailures 100 processors + waitApp appHandle + where + handler ingested = do + -- process ingested.envelope.payload :: RecordedEvent + pure AckOk +@ + +To run members across separate processes instead, give each process one +'kirokuAdapter' with its own 'member' index and the same 'subscriptionName'. +Kiroku's per-member checkpoint (keyed by @(subscriptionName, member)@) lets each +process resume from its own position after a restart. Exactly one live process +must own each member index at a time. + +== Ack Semantics + +The adapter bridges through @kiroku-store@'s __ack-coupled__ stream +('subscriptionAckStream'): for each event the Kiroku worker blocks until the +Shibuya handler's 'AckDecision' is finalized, then acts on it. The handler's +decision therefore drives Kiroku checkpointing per event: + +* 'AckOk' — the worker checkpoints past the event (the normal case). +* 'AckRetry' @delay@ — the worker redelivers the /same/ event after @delay@, + bounded by the subscription's retry policy + ('Kiroku.Store.Subscription.Types.RetryPolicy', default five attempts); on + exhaustion the event is dead-lettered with + 'Kiroku.Store.Subscription.Types.DeadLetterMaxAttempts'. +* 'AckDeadLetter' @reason@ — the worker records the event in + @kiroku.dead_letters@ (with the reason translated to a Kiroku-native + 'Kiroku.Store.Subscription.Types.DeadLetterReason') and atomically advances + the checkpoint past it. +* 'AckHalt' — cancels the underlying Kiroku subscription (no checkpoint advance, + so the halting event replays on restart). + +The envelope's @attempt@ reports the zero-based redelivery count, so a handler +can observe how many times Kiroku has redelivered an event. + +== Backpressure + +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. +-} +module Shibuya.Adapter.Kiroku ( + -- * Adapter + kirokuAdapter, + + -- * Configuration + KirokuAdapterConfig (..), + defaultKirokuAdapterConfig, + + -- * Consumer-group helpers + KirokuConsumerGroupConfig (..), + defaultConsumerGroupConfig, + consumerGroupPolicy, + kirokuConsumerGroupProcessors, + + -- * Re-exports from kiroku-store + SubscriptionName (..), + SubscriptionTarget (..), + ConsumerGroup (..), + EventTypeFilter (..), +) where + +import Data.Int (Int32) +import Data.Text qualified as T +import Effectful (Eff, IOE, liftIO, (:>)) +import GHC.Generics (Generic) +import Kiroku.Store.Connection (KirokuStore) +import Kiroku.Store.Subscription.Stream (subscriptionAckStream) +import Kiroku.Store.Subscription.Types ( + ConsumerGroup (..), + EventTypeFilter (..), + OverflowPolicy (..), + SubscriptionConfigM (..), + SubscriptionName (..), + SubscriptionResult (..), + SubscriptionTarget (..), + defaultSubscriptionConfig, + ) +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.Error (PolicyError (..)) +import Shibuya.Handler (Handler) +import Shibuya.Policy (Concurrency (..), Ordering (..), 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. +-} +data KirokuAdapterConfig = KirokuAdapterConfig + { subscriptionName :: !SubscriptionName + -- ^ Unique subscription identifier (checkpoint key) + , subscriptionTarget :: !SubscriptionTarget + -- ^ 'AllStreams' or @'Category' categoryName@ + , batchSize :: !Int32 + -- ^ Events per database fetch during catch-up + , bufferSize :: !Natural + -- ^ TBQueue capacity (backpressure threshold) + , consumerGroup :: !(Maybe ConsumerGroup) + {- ^ Optional consumer-group membership for this adapter instance. + 'Nothing' (the default) = ordinary single-consumer subscription. + @'Just' ('ConsumerGroup' { member = m, size = n })@ = this adapter is + member @m@ of a group of size @n@, receiving only the events whose + originating stream hashes to slot @m@ (in global-position order). To run a + full size-@n@ group, create @n@ adapters with the same 'subscriptionName' + and distinct 'member' indices, each backed by its own Shibuya processor. + + The validity invariant (@size >= 1@, @0 <= member < size@) is enforced by + the underlying 'Kiroku.Store.Subscription.subscribe' call, which throws + 'Kiroku.Store.Subscription.Types.InvalidConsumerGroup' on violation. + -} + , eventTypeFilter :: !EventTypeFilter + {- ^ Which event types this adapter delivers. Pass 'AllEventTypes' (deliver + everything) or @'OnlyEventTypes' s@ to receive only events whose type is in + @s@. Forwarded into the underlying subscription; filtering is worker-side + (before the ack-coupled bridge), so a filtered-out event never reaches the + Shibuya handler, is never retried or dead-lettered, and the checkpoint still + advances past it. 'Shibuya.Adapter.Kiroku.Convert' and the 'AckHandle' are + unaffected. + -} + , selector :: !(Maybe (RecordedEvent -> Bool)) + {- ^ Optional opaque per-event predicate, the escape hatch for filtering this + adapter's stream on a property 'eventTypeFilter' cannot express (e.g. + payload, metadata, or correlation\/causation ids). Default 'Nothing' (no + extra filtering). Forwarded into the underlying subscription and composed + with 'eventTypeFilter' as a logical AND: an event reaches the Shibuya handler + only when it passes both. Like 'eventTypeFilter' it is applied worker-side + before the ack-coupled bridge, so a rejected event is never retried or + dead-lettered and the checkpoint still advances past it. See + 'Kiroku.Store.Subscription.Types.selector' for when to prefer it over the + introspectable 'eventTypeFilter'. + -} + } + 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. + +Prefer this over a full record literal so that any field added to +'KirokuAdapterConfig' later is inherited at its default automatically: + +@ +let cfg = + (defaultKirokuAdapterConfig "my-projection" 'AllStreams') + { eventTypeFilter = 'OnlyEventTypes' (Set.fromList [EventType "OrderPlaced"]) } +adapter <- kirokuAdapter store cfg +@ +-} +defaultKirokuAdapterConfig :: + SubscriptionName -> SubscriptionTarget -> KirokuAdapterConfig +defaultKirokuAdapterConfig name target = + KirokuAdapterConfig + { subscriptionName = name + , subscriptionTarget = target + , batchSize = 100 + , bufferSize = 256 + , consumerGroup = Nothing + , eventTypeFilter = AllEventTypes + , selector = Nothing + } + +{- | Create a Shibuya 'Adapter' backed by a Kiroku subscription. + +The adapter: + +1. Calls 'subscriptionStream' to start a Kiroku subscription with a + bounded 'TBQueue' 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). + +The returned adapter's @shutdown@ action cancels the underlying +subscription and flushes the sentinel through the queue so any +blocked stream reader terminates. +-} +kirokuAdapter :: + (IOE :> es) => + KirokuStore -> + KirokuAdapterConfig -> + Eff es (Adapter es RecordedEvent) +kirokuAdapter store (KirokuAdapterConfig subName subTarget bs buf cg etf 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 = + (defaultSubscriptionConfig subName subTarget (\_ -> pure Continue)) + { batchSize = bs + , queueCapacity = 16 + , overflowPolicy = DropSubscription + , consumerGroup = cg + , eventTypeFilter = etf + , selector = sel + } + + (ioStream, cancelAction) <- liftIO $ subscriptionAckStream store subConfig buf + + -- The subscription name and consumer-group member are known only here (not on + -- the RecordedEvent), so thread them into the conversion as OTel attributes + -- that ride onto Shibuya's per-message span (EP-5 M2). + let SubscriptionName subNameText = subName + -- Precompute the constant kiroku.* attributes once per adapter (not per + -- event): kirokuEnvelopeAttrs builds the base map, and the per-event + -- conversion only inserts the event type and global position. + envAttrs = + kirokuEnvelopeAttrs + subNameText + (fmap (\ConsumerGroup{member = m} -> fromIntegral m) cg) + ingestedStream = fmap (toIngestedAck envAttrs cancelAction) (Stream.morphInner liftIO ioStream) + + pure + Adapter + { adapterName = "kiroku" + , source = ingestedStream + , shutdown = liftIO cancelAction + } + +{- | Configuration for a whole kiroku consumer group presented as a single +Shibuya partitioned-ordering unit. + +Unlike 'KirokuAdapterConfig' (which describes one member), this describes the +__entire__ group: 'groupSize' members of one subscription, each receiving the +streams whose originating-stream hash maps to its slot. Hand to +'kirokuConsumerGroupProcessors' to obtain @groupSize@ ready-to-run Shibuya +processors with no manual @[0..N-1]@ wiring. + +@memberConcurrency@ is the per-member concurrency. Because kiroku delivers each +member a single strictly global-position-ordered stream, only 'Serial' honestly +preserves per-stream ordering; any 'Ahead'/'Async' is rejected by +'consumerGroupPolicy' before any subscription opens. The /group/ as a whole is +'PartitionedInOrder' (ordered within each member's partition, parallel across +members). +-} +data KirokuConsumerGroupConfig = KirokuConsumerGroupConfig + { subscriptionName :: !SubscriptionName + {- ^ Shared subscription identifier; each member checkpoints under + @(subscriptionName, member)@. + -} + , subscriptionTarget :: !SubscriptionTarget + {- ^ 'AllStreams' or @'Category' categoryName@ — the same source for every + member; kiroku partitions it across members in SQL. + -} + , groupSize :: !Int32 + {- ^ @N@ members; must be @>= 1@ (enforced by the underlying + 'Kiroku.Store.Subscription.subscribe', which throws + 'Kiroku.Store.Subscription.Types.InvalidConsumerGroup' otherwise). + -} + , batchSize :: !Int32 + -- ^ Events per database fetch during catch-up (per member). + , bufferSize :: !Natural + -- ^ Per-member 'TBQueue' capacity (backpressure threshold). + , memberConcurrency :: !Concurrency + -- ^ Per-member concurrency; must be 'Serial' (validated). + , eventTypeFilter :: !EventTypeFilter + {- ^ Event-type filter applied to /every/ member (the same filter on each). + 'AllEventTypes' delivers everything; @'OnlyEventTypes' s@ delivers only the + named types. Forwarded into each per-member 'KirokuAdapterConfig', so a + filtered partitioned group behaves like a filtered single subscription: + filtering is worker-side and per member, the checkpoint still advances past + filtered events, and the partition's completeness is preserved over the + delivered types. + -} + , selector :: !(Maybe (RecordedEvent -> Bool)) + {- ^ Optional opaque per-event predicate applied to /every/ member (the same + predicate on each), the escape hatch for filtering a property + 'eventTypeFilter' cannot express. Default 'Nothing'. Forwarded into each + per-member 'KirokuAdapterConfig' and composed with 'eventTypeFilter' as a + logical AND, so a selector-filtered partitioned group behaves like a + selector-filtered single subscription (worker-side, per member, checkpoint + still advances past rejected events). See + 'Kiroku.Store.Subscription.Types.selector'. + -} + } + deriving stock (Generic) + +{- | 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. +-} +defaultConsumerGroupConfig :: + SubscriptionName -> SubscriptionTarget -> Int32 -> KirokuConsumerGroupConfig +defaultConsumerGroupConfig name target n = + KirokuConsumerGroupConfig + { subscriptionName = name + , subscriptionTarget = target + , groupSize = n + , batchSize = 100 + , bufferSize = 256 + , memberConcurrency = Serial + , eventTypeFilter = AllEventTypes + , selector = Nothing + } + +{- | Map a requested per-member concurrency onto the group's validated Shibuya +@('Ordering', 'Concurrency')@. + +The group's ordering contract is always 'PartitionedInOrder'; a member's own +ordered stream must be processed serially. This reuses Shibuya's own +'validatePolicy' rule (@'StrictInOrder' => 'Serial'@) so the adapter never +invents its own legality check: 'Ahead'/'Async' yield +@'Left' ('InvalidPolicyCombo' ...)@ and 'Serial' yields +@'Right' ('PartitionedInOrder', 'Serial')@. The returned +@('PartitionedInOrder', 'Serial')@ also passes 'validatePolicy', so 'runApp' +will not reject it later. +-} +consumerGroupPolicy :: Concurrency -> Either PolicyError (Ordering, Concurrency) +consumerGroupPolicy conc = do + -- A member delivers one strictly-ordered stream; only Serial is honest. + validatePolicy StrictInOrder conc -- rejects Ahead/Async with PolicyError + pure (PartitionedInOrder, conc) + +{- | 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')@. + +This replaces the manual @mapM mkMemberAdapter [0 .. N-1]@ boilerplate. The +member→policy mapping is validated once up front via 'consumerGroupPolicy'; if +the caller requests a 'memberConcurrency' kiroku cannot honor per member +('Ahead'/'Async'), the result is @'Left' ('InvalidPolicyCombo' ...)@ and __no +kiroku subscription is opened__. + +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. +-} +kirokuConsumerGroupProcessors :: + (IOE :> es) => + KirokuStore -> + KirokuConsumerGroupConfig -> + Handler es RecordedEvent -> + Eff es (Either PolicyError [(ProcessorId, QueueProcessor es)]) +kirokuConsumerGroupProcessors + store + 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) diff --git a/src/Shibuya/Adapter/Kiroku/Convert.hs b/src/Shibuya/Adapter/Kiroku/Convert.hs new file mode 100644 --- /dev/null +++ b/src/Shibuya/Adapter/Kiroku/Convert.hs @@ -0,0 +1,223 @@ +{- | Conversion from Kiroku's 'RecordedEvent' to Shibuya's 'Ingested' and +'Envelope' types. + +== Envelope Mapping + +@ +RecordedEvent field → Envelope field +───────────────────────────────────────── +eventId (UUID) → messageId (Text) +globalPosition → cursor (CursorInt) +createdAt → enqueuedAt +metadata.traceparent → traceContext +(the event itself) → payload +(none) → partition = Nothing +@ + +The adapter preserves W3C trace-context metadata when @metadata@ is a JSON +object containing a string @traceparent@ key. A string @tracestate@ key is +included when present. +-} +module Shibuya.Adapter.Kiroku.Convert ( + -- * Conversion + toIngestedAck, + toEnvelope, + + -- * Envelope attribute source + KirokuEnvelopeAttrs, + kirokuEnvelopeAttrs, + + -- * Ack-decision translation + toKirokuResult, + toKirokuDeadLetterReason, +) where + +import Control.Concurrent.STM (atomically, tryPutTMVar) +import Control.Monad (void) +import Data.Aeson (Value (..)) +import Data.Aeson.Key qualified as Key +import Data.Aeson.KeyMap qualified as KM +import Data.HashMap.Strict (HashMap) +import Data.HashMap.Strict qualified as HashMap +import Data.Int (Int64) +import Data.Text (Text) +import Data.Text qualified as T +import Data.Text.Encoding qualified as TE +import Data.UUID qualified as UUID +import Effectful (IOE, liftIO, (:>)) +import Kiroku.Store.Subscription.Stream (AckItem (..)) +import Kiroku.Store.Subscription.Types ( + DeadLetterReason (..), + RetryDelay (..), + SubscriptionResult (..), + ) +import Kiroku.Store.Types ( + EventId (..), + EventType (..), + GlobalPosition (..), + RecordedEvent (..), + ) +import OpenTelemetry.Attributes (Attribute, AttributeKey (..), toAttribute) +import OpenTelemetry.SemanticConventions qualified as Sem +import Shibuya.Core.Ack (AckDecision (..)) +import Shibuya.Core.Ack qualified as Ack +import Shibuya.Core.AckHandle (AckHandle (..)) +import Shibuya.Core.Ingested (Ingested (..)) +import Shibuya.Core.Types (Attempt (..), Cursor (..), Envelope (..), MessageId (..), TraceHeaders) + +{- | The kiroku identity stamped onto each Shibuya 'Envelope' as OpenTelemetry +attributes, so the kiroku subscription is visible on Shibuya's per-message span. + +Construct it once per adapter with 'kirokuEnvelopeAttrs'. It holds the +/constant/ per-subscription attributes — the subscription name and, for a +grouped subscription, the consumer-group member — already built into an +attribute map, so the per-event conversion path only inserts this event's type +and global position rather than rebuilding the whole map for every event. The +subscription name and member are not carried on a 'RecordedEvent' (they are +known only at the adapter-config level), so the adapter threads them in through +this value; the event type and global position come from the 'RecordedEvent' +itself. +-} +newtype KirokuEnvelopeAttrs = KirokuEnvelopeAttrs + { baseAttributes :: HashMap Text Attribute + -- ^ The constant per-subscription attributes, precomputed once. + } + +{- | Build a 'KirokuEnvelopeAttrs' from a subscription name and an optional +consumer-group member index (@'Nothing'@ for a non-grouped subscription), +precomputing the constant @kiroku.*@ attribute map once. The member attribute is +included only when a member index is given. +-} +kirokuEnvelopeAttrs :: Text -> Maybe Int -> KirokuEnvelopeAttrs +kirokuEnvelopeAttrs subscriptionName member = + KirokuEnvelopeAttrs $ + HashMap.fromList $ + (attrKirokuSubscriptionName, toAttribute subscriptionName) + : (attrMessagingSystem, toAttribute ("kiroku" :: Text)) + : (attrMessagingDestinationName, toAttribute subscriptionName) + : maybe + [] + (\m -> [(attrKirokuConsumerGroupMember, toAttribute (fromIntegral m :: Int64))]) + member + +{- | Wrap an ack-coupled 'AckItem' (from @kiroku-store@'s 'subscriptionAckStream') +into an 'Ingested' value suitable for Shibuya handlers. + +The underlying Kiroku worker is blocked waiting for this item's reply, so the +'AckHandle.finalize' translates the Shibuya 'AckDecision' into a Kiroku +'SubscriptionResult' and writes it back — driving the worker's checkpointing: + +* 'AckOk' — reply 'Continue'; the worker checkpoints past the event. +* 'AckRetry' @delay@ — reply 'Retry'; the worker redelivers the same event after + @delay@, bounded by the subscription's retry policy, then dead-letters it. +* 'AckDeadLetter' @reason@ — reply 'DeadLetter'; the worker records the event in + @kiroku.dead_letters@ and advances past it. +* 'AckHalt' — cancels the underlying Kiroku subscription (no checkpoint advance, + so the halting event replays on restart), preserving the prior adapter + behavior; the blocked worker is interrupted by the cancellation. + +'finalize' is idempotent: the reply is written with 'tryPutTMVar' so a second +call is a no-op. + +The envelope's @attempt@ is set from the item's redelivery counter so a Shibuya +handler can observe how many times Kiroku has redelivered the event. +-} +toIngestedAck :: (IOE :> es) => KirokuEnvelopeAttrs -> IO () -> AckItem -> Ingested es RecordedEvent +toIngestedAck attrs cancelAction (AckItem event attempt reply) = + Ingested + { envelope = (toEnvelope attrs event){attempt = Just (Attempt attempt)} + , ack = + AckHandle + { finalize = \case + AckHalt _ -> liftIO cancelAction + decision -> + liftIO $ + atomically $ + void $ + tryPutTMVar reply (toKirokuResult attempt decision) + } + , lease = Nothing + } + +{- | Translate a non-halt Shibuya 'AckDecision' into a Kiroku +'SubscriptionResult'. 'AckHalt' is handled separately (it cancels the +subscription) and maps to 'Continue' here only defensively. The 'Word' is the +event's current redelivery attempt, used to annotate a 'MaxRetriesExceeded' +reason. +-} +toKirokuResult :: Word -> AckDecision -> SubscriptionResult +toKirokuResult attempt = \case + AckOk -> Continue + AckRetry (Ack.RetryDelay d) -> Retry (RetryDelay d) + AckDeadLetter reason -> DeadLetter (toKirokuDeadLetterReason attempt reason) + AckHalt _ -> Continue + +-- | Translate a Shibuya 'Ack.DeadLetterReason' into a Kiroku 'DeadLetterReason'. +toKirokuDeadLetterReason :: Word -> Ack.DeadLetterReason -> DeadLetterReason +toKirokuDeadLetterReason attempt = \case + Ack.PoisonPill detail -> DeadLetterPoison detail + Ack.InvalidPayload detail -> DeadLetterInvalid detail + Ack.MaxRetriesExceeded -> DeadLetterMaxAttempts (fromIntegral attempt) + +{- | Convert a 'RecordedEvent' to a Shibuya 'Envelope'. + +The event's UUID is formatted as text for the 'MessageId', and the +global position is used as an integer 'Cursor' for ordering. The 'Envelope's +@attributes@ are populated with the kiroku identity (subscription name, +consumer-group member, event type, global position) from 'KirokuEnvelopeAttrs' +and the event, so Shibuya's per-message span carries the kiroku context. +-} +toEnvelope :: KirokuEnvelopeAttrs -> RecordedEvent -> Envelope RecordedEvent +toEnvelope attrs event = + let RecordedEvent{eventId = EventId uuid, eventType = EventType etype, globalPosition = GlobalPosition pos, createdAt = ts, metadata = meta} = event + in Envelope + { messageId = MessageId (T.pack (UUID.toString uuid)) + , cursor = Just (CursorInt (fromIntegral pos)) + , partition = Nothing + , enqueuedAt = Just ts + , traceContext = metadataTraceContext meta + , attempt = Nothing + , attributes = eventAttributes attrs etype pos + , payload = event + } + +{- | The per-event @kiroku.*@ attribute map: the precomputed constant attributes +('baseAttributes') plus this event's type and global position. Only the two +per-event keys are inserted, so the constant attributes are not rebuilt for +every event. Keys mirror the native-span attribute keys in +@Kiroku.Otel.Subscription@ so a trace reads consistently across the kiroku and +Shibuya sides. +-} +eventAttributes :: KirokuEnvelopeAttrs -> Text -> Int64 -> HashMap Text Attribute +eventAttributes attrs etype pos = + HashMap.insert attrKirokuEventType (toAttribute etype) $ + HashMap.insert attrKirokuEventGlobalPosition (toAttribute pos) (baseAttributes attrs) + +attrMessagingSystem :: Text +attrMessagingSystem = unkey Sem.messaging_system + +attrMessagingDestinationName :: Text +attrMessagingDestinationName = unkey Sem.messaging_destination_name + +attrKirokuSubscriptionName :: Text +attrKirokuSubscriptionName = "kiroku.subscription.name" + +attrKirokuConsumerGroupMember :: Text +attrKirokuConsumerGroupMember = "kiroku.consumer_group.member" + +attrKirokuEventType :: Text +attrKirokuEventType = "kiroku.event.type" + +attrKirokuEventGlobalPosition :: Text +attrKirokuEventGlobalPosition = "kiroku.event.global_position" + +metadataTraceContext :: Maybe Value -> Maybe TraceHeaders +metadataTraceContext (Just (Object metadata)) = do + String traceparent <- KM.lookup (Key.fromString "traceparent") metadata + let traceparentHeader = ("traceparent", TE.encodeUtf8 traceparent) + traceHeaders = + case KM.lookup (Key.fromString "tracestate") metadata of + Just (String tracestate) -> [traceparentHeader, ("tracestate", TE.encodeUtf8 tracestate)] + _ -> [traceparentHeader] + pure traceHeaders +metadataTraceContext _ = Nothing diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -0,0 +1,804 @@ +module Main where + +import Control.Concurrent (threadDelay) +import Control.Concurrent.STM (atomically, newTVarIO, readTVar, registerDelay, writeTVar) +import Control.Concurrent.STM qualified as STM +import Control.Lens ((&), (.~), (^.)) +import Control.Monad (unless) +import Control.Monad.IO.Class (liftIO) +import Data.Aeson (Value) +import Data.Aeson qualified as Aeson +import Data.Foldable (toList) +import Data.Generics.Labels () +import Data.HashMap.Strict qualified as HashMap +import Data.IORef (modifyIORef', newIORef, readIORef) +import Data.Int (Int32, Int64) +import Data.List (nub, sort) +import Data.Map.Strict qualified as Map +import Data.Set qualified as Set +import Data.Text (Text) +import Data.Text qualified as T +import Data.Time (UTCTime (..), fromGregorian) +import Data.UUID qualified as UUID +import Effectful (runEff) +import Hasql.Pool qualified as Pool +import Hasql.Session qualified as Session +import Kiroku.Store +import Kiroku.Store.SQL qualified as SQL +import Kiroku.Test.Postgres (withMigratedTestDatabase, withSharedMigratedPostgres) +import OpenTelemetry.Attributes (toAttribute) +import Shibuya.Adapter.Kiroku ( + consumerGroupPolicy, + defaultConsumerGroupConfig, + defaultKirokuAdapterConfig, + kirokuAdapter, + kirokuConsumerGroupProcessors, + ) +import Shibuya.Adapter.Kiroku.Convert (KirokuEnvelopeAttrs, kirokuEnvelopeAttrs, toEnvelope) +import Shibuya.App ( + ProcessorId (..), + QueueProcessor (..), + SupervisionStrategy (..), + getAppMetrics, + mkProcessor, + runApp, + stopApp, + stopAppGracefully, + ) +import Shibuya.App qualified as Shibuya +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.Telemetry.Effect (runTracingNoop) +import Test.Hspec + +main :: IO () +main = withSharedMigratedPostgres $ hspec $ do + describe "toEnvelope" $ do + it "copies W3C trace metadata into Shibuya trace headers" $ do + let traceparent :: Text + traceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01" + tracestate :: Text + tracestate = "rojo=00f067aa0ba902b7" + Envelope{traceContext} = + toEnvelope + sampleEnvelopeAttrs + ( makeRecordedEvent + ( Just $ + Aeson.object + [ "traceparent" Aeson..= traceparent + , "tracestate" Aeson..= tracestate + , "other" Aeson..= ("preserved" :: Text) + ] + ) + ) + + traceContext + `shouldBe` Just + [ ("traceparent", "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01") + , ("tracestate", "rojo=00f067aa0ba902b7") + ] + + it "omits trace headers when traceparent is absent or not a string" $ do + let Envelope{traceContext = missingTraceparent} = + toEnvelope sampleEnvelopeAttrs (makeRecordedEvent (Just (Aeson.object ["tracestate" Aeson..= ("state" :: Text)]))) + Envelope{traceContext = nonStringTraceparent} = + toEnvelope sampleEnvelopeAttrs (makeRecordedEvent (Just (Aeson.object ["traceparent" Aeson..= Aeson.Number 1]))) + + missingTraceparent `shouldBe` Nothing + nonStringTraceparent `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) + -- makeRecordedEvent: eventType "TraceEvent", globalPosition 1. + HashMap.lookup "kiroku.subscription.name" attributes + `shouldBe` Just (toAttribute ("orders-proj" :: Text)) + HashMap.lookup "messaging.system" attributes + `shouldBe` Just (toAttribute ("kiroku" :: Text)) + HashMap.lookup "messaging.destination.name" attributes + `shouldBe` Just (toAttribute ("orders-proj" :: Text)) + HashMap.lookup "kiroku.event.type" attributes + `shouldBe` Just (toAttribute ("TraceEvent" :: Text)) + HashMap.lookup "kiroku.event.global_position" attributes + `shouldBe` Just (toAttribute (1 :: Int64)) + -- No member key for a non-grouped subscription. + HashMap.lookup "kiroku.consumer_group.member" attributes `shouldBe` Nothing + + it "stamps the consumer-group member attribute for a grouped subscription (EP-5 M2)" $ do + let attrs = kirokuEnvelopeAttrs "orders-proj" (Just 2) + Envelope{attributes} = toEnvelope attrs (makeRecordedEvent Nothing) + HashMap.lookup "kiroku.subscription.name" attributes + `shouldBe` Just (toAttribute ("orders-proj" :: Text)) + HashMap.lookup "kiroku.consumer_group.member" attributes + `shouldBe` Just (toAttribute (2 :: Int64)) + + describe "consumer group policy" $ do + it "accepts Serial member concurrency as (PartitionedInOrder, Serial)" $ + consumerGroupPolicy Serial `shouldBe` Right (PartitionedInOrder, Serial) + + it "rejects Ahead member concurrency with a PolicyError" $ + consumerGroupPolicy (Ahead 4) + `shouldBe` Left (InvalidPolicyCombo "StrictInOrder requires Serial concurrency") + + it "rejects Async member concurrency with a PolicyError" $ + consumerGroupPolicy (Async 4) + `shouldBe` Left (InvalidPolicyCombo "StrictInOrder requires Serial concurrency") + + around withTestStore $ do + describe "kirokuAdapter" $ do + it "delivers only matching event types when an eventTypeFilter is set (EP-43)" $ \store -> do + -- A, B, A, B, A on one stream (global positions 1..5). The adapter + -- is filtered to type A; the Shibuya handler must see only the As. + let events = + [ makeEvent "A" (Aeson.object []) + , makeEvent "B" (Aeson.object []) + , makeEvent "A" (Aeson.object []) + , makeEvent "B" (Aeson.object []) + , makeEvent "A" (Aeson.object []) + ] + Right _ <- runStoreIO store $ appendToStream (StreamName "etf-adapter-1") NoStream events + ref <- newIORef ([] :: [RecordedEvent]) + countVar <- newTVarIO (0 :: Int) + runEff $ runTracingNoop $ do + adapter <- + kirokuAdapter store $ + defaultKirokuAdapterConfig (SubscriptionName "etf-adapter") AllStreams + & #eventTypeFilter .~ OnlyEventTypes (Set.fromList [EventType "A"]) + let handler ingested = do + liftIO $ do + modifyIORef' ref (envelopePayload ingested :) + atomically $ do + c <- readTVar countVar + writeTVar countVar (c + 1) + pure AckOk + res <- runApp IgnoreFailures 100 [(ProcessorId "etf", mkProcessor adapter handler)] + case res of + Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err) + Right appHandle -> do + liftIO $ waitForCount countVar 3 10_000_000 + stopApp appHandle + + collected <- readIORef ref + -- Only the three A events reached the handler; no Bs. + map (^. #eventType) collected `shouldBe` replicate 3 (EventType "A") + map globalPos (reverse collected) `shouldBe` [1, 3, 5] + + it "delivers only matching types across a filtered consumer group (EP-43)" $ \store -> do + -- 20 streams in category "etfg", each [A, B]; A at odd global + -- positions 1,3,..,39. A size-4 group filtered to type A must + -- deliver exactly the 20 As, disjoint and complete. + let streams = [0 .. 19 :: Int] + mapM_ + ( \i -> do + let sn = StreamName ("etfg-" <> T.pack (show i)) + Right _ <- + runStoreIO store $ + appendToStream sn NoStream [makeEvent "A" (Aeson.object []), makeEvent "B" (Aeson.object [])] + pure () + ) + streams + ref <- newIORef ([] :: [RecordedEvent]) + countVar <- newTVarIO (0 :: Int) + runEff $ runTracingNoop $ do + let cfg = + defaultConsumerGroupConfig (SubscriptionName "etfg") (Category (CategoryName "etfg")) 4 + & #eventTypeFilter .~ OnlyEventTypes (Set.fromList [EventType "A"]) + handler ingested = do + liftIO $ do + modifyIORef' ref (envelopePayload ingested :) + atomically $ do + c <- readTVar countVar + writeTVar countVar (c + 1) + pure AckOk + res <- kirokuConsumerGroupProcessors store cfg handler + case res of + Left err -> liftIO $ expectationFailure ("kirokuConsumerGroupProcessors failed: " <> show err) + Right processors -> do + appRes <- runApp IgnoreFailures 100 processors + case appRes of + Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err) + Right appHandle -> do + liftIO $ waitForCount countVar 20 15_000_000 + stopApp appHandle + + collected <- readIORef ref + -- Every delivered event is an A (filter honored per member)... + map (^. #eventType) collected `shouldBe` replicate 20 (EventType "A") + -- ...and the union over members is the complete, disjoint set of A + -- positions [1,3,..,39] — no A dropped, none delivered twice. + sort (map globalPos collected) `shouldBe` [1, 3 .. 39] + + it "delivers only events matching an opaque selector (EP-43 follow-up)" $ \store -> do + -- All one type "A" (so eventTypeFilter cannot distinguish them); + -- only the payload tag {keep} differs. keep, skip, keep, skip, keep + -- at positions 1..5. The selector admits only keep=True. + let keepObj = Aeson.object [("keep", Aeson.Bool True)] + skipObj = Aeson.object [("keep", Aeson.Bool False)] + events = + [ makeEvent "A" keepObj + , makeEvent "A" skipObj + , makeEvent "A" keepObj + , makeEvent "A" skipObj + , makeEvent "A" keepObj + ] + Right _ <- runStoreIO store $ appendToStream (StreamName "sel-adapter-1") NoStream events + ref <- newIORef ([] :: [RecordedEvent]) + countVar <- newTVarIO (0 :: Int) + runEff $ runTracingNoop $ do + adapter <- + kirokuAdapter store $ + defaultKirokuAdapterConfig (SubscriptionName "sel-adapter") AllStreams + & #selector .~ Just (\e -> (e ^. #payload) == keepObj) + let handler ingested = do + liftIO $ do + modifyIORef' ref (envelopePayload ingested :) + atomically $ do + c <- readTVar countVar + writeTVar countVar (c + 1) + pure AckOk + res <- runApp IgnoreFailures 100 [(ProcessorId "sel", mkProcessor adapter handler)] + case res of + Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err) + Right appHandle -> do + liftIO $ waitForCount countVar 3 10_000_000 + stopApp appHandle + + collected <- readIORef ref + -- Only the three keep events reached the handler (positions 1,3,5); + -- the two skip events were filtered worker-side and never delivered. + map globalPos (reverse collected) `shouldBe` [1, 3, 5] + + it "delivers catch-up events through Shibuya pipeline" $ \store -> do + let events = map (\i -> makeEvent ("CU" <> T.pack (show i)) (Aeson.object [])) [1 .. 10 :: Int] + Right _ <- runStoreIO store $ appendToStream (StreamName "shibuya-catchup-1") NoStream events + threadDelay 200_000 + + ref <- newIORef ([] :: [RecordedEvent]) + countVar <- newTVarIO (0 :: Int) + + runEff $ runTracingNoop $ do + adapter <- + kirokuAdapter store $ + defaultKirokuAdapterConfig (SubscriptionName "shibuya-catchup-test") AllStreams + + let handler ingested = do + liftIO $ do + modifyIORef' ref (envelopePayload ingested :) + atomically $ do + c <- readTVar countVar + writeTVar countVar (c + 1) + pure AckOk + + res <- runApp IgnoreFailures 100 [(ProcessorId "catchup", mkProcessor adapter handler)] + case res of + Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err) + Right appHandle -> do + liftIO $ waitForCount countVar 10 10_000_000 + stopApp appHandle + + collected <- readIORef ref + length collected `shouldBe` 10 + + it "delivers live events through Shibuya pipeline" $ \store -> do + ref <- newIORef ([] :: [RecordedEvent]) + countVar <- newTVarIO (0 :: Int) + + runEff $ runTracingNoop $ do + adapter <- + kirokuAdapter store $ + defaultKirokuAdapterConfig (SubscriptionName "shibuya-live-test") AllStreams + + let handler ingested = do + liftIO $ do + modifyIORef' ref (envelopePayload ingested :) + atomically $ do + c <- readTVar countVar + writeTVar countVar (c + 1) + pure AckOk + + res <- runApp IgnoreFailures 100 [(ProcessorId "live", mkProcessor adapter handler)] + case res of + Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err) + Right appHandle -> do + liftIO $ threadDelay 200_000 + liftIO $ do + let events = map (\i -> makeEvent ("Live" <> T.pack (show i)) (Aeson.object [])) [1 .. 5 :: Int] + Right _ <- runStoreIO store $ appendToStream (StreamName "shibuya-live-1") NoStream events + pure () + liftIO $ waitForCount countVar 5 10_000_000 + stopApp appHandle + + collected <- readIORef ref + length collected `shouldBe` 5 + + 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 [])] + Right _ <- runStoreIO store $ appendToStream (StreamName "payments-s1") NoStream [makeEvent "PaymentReceived" (Aeson.object [])] + Right _ <- runStoreIO store $ appendToStream (StreamName "inventory-s1") NoStream [makeEvent "StockUpdated" (Aeson.object [])] + Right _ <- runStoreIO store $ appendToStream (StreamName "inventory-s2") NoStream [makeEvent "StockDepleted" (Aeson.object [])] + threadDelay 200_000 + + ordersRef <- newIORef ([] :: [RecordedEvent]) + paymentsRef <- newIORef ([] :: [RecordedEvent]) + inventoryRef <- newIORef ([] :: [RecordedEvent]) + ordersCount <- newTVarIO (0 :: Int) + paymentsCount <- newTVarIO (0 :: Int) + inventoryCount <- newTVarIO (0 :: Int) + + runEff $ runTracingNoop $ do + let mkCatAdapter nm cat = + kirokuAdapter store $ + defaultKirokuAdapterConfig (SubscriptionName nm) (Category (CategoryName cat)) + let mkHandler ref' cVar ingested = do + liftIO $ do + modifyIORef' ref' (envelopePayload ingested :) + atomically $ do + c <- readTVar cVar + writeTVar cVar (c + 1) + pure AckOk + + ordersAdapter <- mkCatAdapter "orders-proj" "orders" + paymentsAdapter <- mkCatAdapter "payments-proj" "payments" + inventoryAdapter <- mkCatAdapter "inventory-proj" "inventory" + + res <- + runApp + IgnoreFailures + 100 + [ (ProcessorId "orders", mkProcessor ordersAdapter (mkHandler ordersRef ordersCount)) + , (ProcessorId "payments", mkProcessor paymentsAdapter (mkHandler paymentsRef paymentsCount)) + , (ProcessorId "inventory", mkProcessor inventoryAdapter (mkHandler inventoryRef inventoryCount)) + ] + case res of + Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err) + Right appHandle -> do + liftIO $ do + waitForCount ordersCount 2 10_000_000 + waitForCount paymentsCount 1 10_000_000 + waitForCount inventoryCount 2 10_000_000 + stopApp appHandle + + ordersCollected <- readIORef ordersRef + paymentsCollected <- readIORef paymentsRef + inventoryCollected <- readIORef inventoryRef + length ordersCollected `shouldBe` 2 + length paymentsCollected `shouldBe` 1 + length inventoryCollected `shouldBe` 2 + + it "isolates a failing subscription from healthy ones" $ \store -> do + Right _ <- runStoreIO store $ appendToStream (StreamName "good-s1") NoStream (map (\i -> makeEvent ("Good" <> T.pack (show i)) (Aeson.object [])) [1 .. 3 :: Int]) + Right _ <- runStoreIO store $ appendToStream (StreamName "bad-s1") NoStream [makeEvent "Bad1" (Aeson.object []), makeEvent "Bad2" (Aeson.object [])] + Right _ <- runStoreIO store $ appendToStream (StreamName "other-s1") NoStream [makeEvent "Other1" (Aeson.object []), makeEvent "Other2" (Aeson.object [])] + threadDelay 200_000 + + goodRef <- newIORef ([] :: [RecordedEvent]) + otherRef <- newIORef ([] :: [RecordedEvent]) + goodCount <- newTVarIO (0 :: Int) + otherCount <- newTVarIO (0 :: Int) + + runEff $ runTracingNoop $ do + goodAdapter <- + kirokuAdapter store $ + defaultKirokuAdapterConfig (SubscriptionName "good-proj") (Category (CategoryName "good")) + badAdapter <- + kirokuAdapter store $ + defaultKirokuAdapterConfig (SubscriptionName "bad-proj") (Category (CategoryName "bad")) + otherAdapter <- + kirokuAdapter store $ + defaultKirokuAdapterConfig (SubscriptionName "other-proj") (Category (CategoryName "other")) + + let goodHandler ingested = do + liftIO $ do + modifyIORef' goodRef (envelopePayload ingested :) + atomically $ do + c <- readTVar goodCount + writeTVar goodCount (c + 1) + pure AckOk + + let badHandler _ingested = do + liftIO $ error "handler crash!" + + let otherHandler ingested = do + liftIO $ do + modifyIORef' otherRef (envelopePayload ingested :) + atomically $ do + c <- readTVar otherCount + writeTVar otherCount (c + 1) + pure AckOk + + res <- + runApp + IgnoreFailures + 100 + [ (ProcessorId "good", mkProcessor goodAdapter goodHandler) + , (ProcessorId "bad", mkProcessor badAdapter badHandler) + , (ProcessorId "other", mkProcessor otherAdapter otherHandler) + ] + case res of + Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err) + Right appHandle -> do + liftIO $ do + waitForCount goodCount 3 10_000_000 + waitForCount otherCount 2 10_000_000 + threadDelay 500_000 + metrics <- getAppMetrics appHandle + case Map.lookup (ProcessorId "bad") metrics of + Just m -> case m ^. #state of + Failed _ _ -> pure () + other -> liftIO $ expectationFailure ("Expected Failed state, got: " <> show other) + Nothing -> liftIO $ expectationFailure "No metrics for bad processor" + stopApp appHandle + + goodCollected <- readIORef goodRef + otherCollected <- readIORef otherRef + length goodCollected `shouldBe` 3 + length otherCollected `shouldBe` 2 + + it "shuts down all subscriptions coordinately" $ \store -> do + Right _ <- runStoreIO store $ appendToStream (StreamName "shut-a") NoStream [makeEvent "A" (Aeson.object [])] + Right _ <- runStoreIO store $ appendToStream (StreamName "shut-b") NoStream [makeEvent "B" (Aeson.object [])] + threadDelay 200_000 + + countA <- newTVarIO (0 :: Int) + countB <- newTVarIO (0 :: Int) + + runEff $ runTracingNoop $ do + adapterA <- + kirokuAdapter store $ + defaultKirokuAdapterConfig (SubscriptionName "shut-a-proj") (Category (CategoryName "shut")) + adapterB <- + kirokuAdapter store $ + defaultKirokuAdapterConfig (SubscriptionName "shut-b-proj") (Category (CategoryName "shut")) + + let handlerA _ingested = do + liftIO $ atomically $ do + c <- readTVar countA + writeTVar countA (c + 1) + pure AckOk + + let handlerB _ingested = do + liftIO $ atomically $ do + c <- readTVar countB + writeTVar countB (c + 1) + pure AckOk + + res <- + runApp + IgnoreFailures + 100 + [ (ProcessorId "a", mkProcessor adapterA handlerA) + , (ProcessorId "b", mkProcessor adapterB handlerB) + ] + case res of + Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err) + Right appHandle -> do + liftIO $ do + waitForCount countA 2 10_000_000 + waitForCount countB 2 10_000_000 + drained <- stopAppGracefully Shibuya.defaultShutdownConfig appHandle + liftIO $ drained `shouldBe` True + + describe "ack dispositions" $ do + it "AckRetry redelivers the same event, then AckOk advances (EP-40 M3)" $ \store -> do + Right _ <- runStoreIO store $ appendToStream (StreamName "ackretry-1") NoStream [makeEvent "R1" (Aeson.object [])] + threadDelay 200_000 + + deliveries <- newIORef (0 :: Int) + countVar <- newTVarIO (0 :: Int) + + runEff $ runTracingNoop $ do + adapter <- + kirokuAdapter store $ + defaultKirokuAdapterConfig (SubscriptionName "ackretry-proj") AllStreams + + let handler _ingested = do + n <- liftIO $ do + modifyIORef' deliveries (+ 1) + atomically $ do + c <- readTVar countVar + writeTVar countVar (c + 1) + pure (c + 1) + -- 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)] + case res of + Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err) + Right appHandle -> do + liftIO $ waitForCount countVar 2 10_000_000 + stopApp appHandle + + -- The one event was delivered twice (initial + one retry). + n <- readIORef deliveries + n `shouldBe` 2 + -- It was not dead-lettered (the retry succeeded). + dls <- readDeadLetters store "ackretry-proj" + length dls `shouldBe` 0 + + it "AckDeadLetter records the event and the next event continues (EP-40 M3)" $ \store -> do + Right _ <- runStoreIO store $ appendToStream (StreamName "ackdl-1") NoStream [makeEvent "D1" (Aeson.object [])] + Right _ <- runStoreIO store $ appendToStream (StreamName "ackdl-2") NoStream [makeEvent "D2" (Aeson.object [])] + threadDelay 200_000 + + okPayloads <- newIORef ([] :: [RecordedEvent]) + countVar <- newTVarIO (0 :: Int) + + runEff $ runTracingNoop $ do + adapter <- + kirokuAdapter store $ + defaultKirokuAdapterConfig (SubscriptionName "ackdl-proj") AllStreams + + let handler ingested = do + n <- liftIO $ atomically $ do + c <- readTVar countVar + writeTVar countVar (c + 1) + pure (c + 1) + -- Dead-letter the first event; accept the second. + if n == 1 + then pure (AckDeadLetter (PoisonPill "poison")) + else do + liftIO $ modifyIORef' okPayloads (envelopePayload ingested :) + pure AckOk + + res <- runApp IgnoreFailures 100 [(ProcessorId "ackdl", 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 + + -- The first event (global position 1) is recorded as dead-letter. + dls <- readDeadLetters store "ackdl-proj" + map SQL.deadLetterGlobalPosition dls `shouldBe` [1] + -- The second event was delivered and accepted. + oks <- readIORef okPayloads + map globalPos oks `shouldBe` [2] + + 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". + let streams = ["cg-" <> T.pack (show i) | i <- [1 .. 20 :: Int]] + mapM_ + ( \sn -> do + let evs = [makeEvent ("EV" <> T.pack (show k)) (Aeson.object []) | k <- [1 .. 2 :: Int]] + Right _ <- runStoreIO store $ appendToStream (StreamName sn) NoStream evs + pure () + ) + streams + threadDelay 200_000 + + -- One IORef + one TVar counter per member. + refs <- mapM (const (newIORef ([] :: [RecordedEvent]))) [0 .. 3 :: Int] + cvars <- mapM (const (newTVarIO (0 :: Int))) [0 .. 3 :: Int] + + runEff $ runTracingNoop $ do + -- Build one adapter per member index; each carries the same + -- subscriptionName and a distinct member of a size-4 group. + adapters <- + mapM + ( \m -> + kirokuAdapter store $ + defaultKirokuAdapterConfig (SubscriptionName "cg-shibuya-group") (Category (CategoryName "cg")) + & #consumerGroup .~ Just (ConsumerGroup{member = m, size = 4}) + ) + [0, 1, 2, 3] + + let mkHandler ref' cvar ingested = do + liftIO $ do + modifyIORef' ref' (envelopePayload ingested :) + atomically $ do + c <- readTVar cvar + writeTVar cvar (c + 1) + pure AckOk + + processors = + [ ( ProcessorId ("cg-member-" <> T.pack (show m)) + , mkProcessor (adapters !! m) (mkHandler (refs !! m) (cvars !! m)) + ) + | m <- [0 .. 3 :: Int] + ] + + res <- runApp IgnoreFailures 100 processors + case res of + Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err) + Right appHandle -> do + liftIO $ waitForTotal cvars 40 15_000_000 + stopApp appHandle + + collected <- mapM readIORef refs + -- Disjoint + complete: the union of every member's delivered global + -- positions is exactly [1..40] — no event dropped, none delivered twice. + let allPositions = sort (concatMap (map globalPos) collected) + allPositions `shouldBe` [1 .. 40] + -- No member starved: 20 streams over 4 members gives each ≥ 1 stream. + mapM_ (\c -> length c `shouldSatisfy` (>= 1)) collected + + describe "consumer group policy" $ do + it "rejects an invalid member concurrency before opening any subscription" $ \store -> do + -- A request the helper cannot honor per member returns Left without + -- starting any worker. The (unused) store proves no subscription opens. + result <- + runEff $ + runTracingNoop $ + kirokuConsumerGroupProcessors + store + ( defaultConsumerGroupConfig (SubscriptionName "cgp-reject") (Category (CategoryName "cgp-reject")) 4 + & #memberConcurrency .~ Async 4 + ) + ( \ingested -> do + let _ = envelopePayload ingested + pure AckOk + ) + case result of + Left err -> err `shouldBe` InvalidPolicyCombo "StrictInOrder requires Serial concurrency" + Right _ -> expectationFailure "expected Left PolicyError for Async member concurrency" + + 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]] + mapM_ + ( \sn -> do + let evs = [makeEvent ("EV" <> T.pack (show k)) (Aeson.object []) | k <- [1 .. 2 :: Int]] + Right _ <- runStoreIO store $ appendToStream (StreamName sn) NoStream evs + pure () + ) + streams + threadDelay 200_000 + + -- A single shared handler records every delivery (consumer-group + -- members all run the same handler). One counter for the whole group. + ref <- newIORef ([] :: [RecordedEvent]) + countVar <- newTVarIO (0 :: Int) + + let cfg = + defaultConsumerGroupConfig + (SubscriptionName "cgp-shibuya-group") + (Category (CategoryName "cgp")) + 4 + + runEff $ runTracingNoop $ do + let handler ingested = do + liftIO $ do + modifyIORef' ref (envelopePayload ingested :) + atomically $ do + c <- readTVar countVar + writeTVar countVar (c + 1) + pure AckOk + + -- One call replaces the manual `mapM mkMemberAdapter [0..3]` wiring. + res <- kirokuConsumerGroupProcessors store cfg handler + case res of + Left err -> liftIO $ expectationFailure ("kirokuConsumerGroupProcessors failed: " <> show err) + Right processors -> do + -- The group is presented as N processors, each pinned to the + -- group-level PartitionedInOrder contract + per-member Serial. + liftIO $ length processors `shouldBe` 4 + let policies = map (\(_, QueueProcessor _ _ ord conc) -> (ord, conc)) processors + liftIO $ policies `shouldBe` replicate 4 (PartitionedInOrder, Serial) + -- The member index is readable off the ProcessorId. + let pids = map (\(ProcessorId p, _) -> p) processors + liftIO $ + pids + `shouldBe` ["cgp-shibuya-group-member-" <> T.pack (show m) | m <- [0 .. 3 :: Int]] + + appRes <- runApp IgnoreFailures 100 processors + case appRes of + Left err -> liftIO $ expectationFailure ("runApp failed: " <> show err) + Right appHandle -> do + liftIO $ waitForCount countVar 40 15_000_000 + stopApp appHandle + + collected <- readIORef ref + -- Disjoint + complete through the helper: the union of every member's + -- delivered global positions is exactly [1..40] — no event dropped, none + -- delivered twice. No duplicates ⇒ the member partitions are disjoint; + -- ==[1..40] ⇒ the union is the complete source. This is the same + -- disjoint-complete property MasterPlan 4 used, now asserted through the + -- single-call helper and the Shibuya pipeline. + sort (map globalPos collected) `shouldBe` [1 .. 40] + -- Every one of the 20 originating streams was covered. + length (nub (map origStreamId collected)) `shouldBe` 20 + +-- Helpers + +envelopePayload :: Ingested es RecordedEvent -> RecordedEvent +envelopePayload ing = let Ingested{envelope = env} = ing in env ^. #payload + +-- | Read the dead letters recorded for a non-group subscription (member 0). +readDeadLetters :: KirokuStore -> Text -> IO [SQL.DeadLetterRecord] +readDeadLetters store subName = do + result <- Pool.use (store ^. #pool) (Session.statement (subName, 0 :: Int32) SQL.readDeadLettersStmt) + case result of + Left err -> error ("readDeadLetters failed: " <> show err) + Right v -> pure (toList v) + +-- | The raw global position of a recorded event (for disjoint/complete checks). +globalPos :: RecordedEvent -> Int64 +globalPos e = case e ^. #globalPosition of GlobalPosition p -> p + +-- | The originating stream's surrogate id (for stream-coverage checks). +origStreamId :: RecordedEvent -> Int64 +origStreamId e = case e ^. #originalStreamId of StreamId p -> p + +-- | Wait until the sum of all TVar counts reaches the target or the timeout fires. +waitForTotal :: [STM.TVar Int] -> Int -> Int -> IO () +waitForTotal vars target timeoutMicros = do + timeoutVar <- registerDelay timeoutMicros + result <- + atomically $ + ( do + total <- sum <$> mapM readTVar vars + STM.check (total >= target) + pure True + ) + `STM.orElse` ( do + t <- readTVar timeoutVar + STM.check t + pure False + ) + unless result $ do + actual <- atomically $ sum <$> mapM readTVar vars + expectationFailure ("Timed out waiting for total " <> show target <> ", got " <> show actual) + +waitForCount :: STM.TVar Int -> Int -> Int -> IO () +waitForCount countVar target timeoutMicros = do + timeoutVar <- registerDelay timeoutMicros + result <- + atomically $ + ( do + c <- readTVar countVar + STM.check (c >= target) + pure True + ) + `STM.orElse` ( do + t <- readTVar timeoutVar + STM.check t + pure False + ) + unless result $ do + actual <- atomically $ readTVar countVar + expectationFailure ("Timed out waiting for count " <> show target <> ", got " <> show actual) + +makeEvent :: Text -> Value -> EventData +makeEvent typ p = + EventData + { eventId = Nothing + , eventType = EventType typ + , payload = p + , metadata = Nothing + , causationId = Nothing + , correlationId = Nothing + } + +{- | A throwaway attribute source for the trace-header tests, whose assertions +do not depend on the kiroku identity attributes. +-} +sampleEnvelopeAttrs :: KirokuEnvelopeAttrs +sampleEnvelopeAttrs = kirokuEnvelopeAttrs "test-sub" Nothing + +makeRecordedEvent :: Maybe Value -> RecordedEvent +makeRecordedEvent meta = + RecordedEvent + { eventId = EventId UUID.nil + , eventType = EventType "TraceEvent" + , streamVersion = StreamVersion 1 + , globalPosition = GlobalPosition 1 + , originalStreamId = StreamId 1 + , originalVersion = StreamVersion 1 + , payload = Aeson.object [] + , metadata = meta + , causationId = Nothing + , correlationId = Nothing + , createdAt = UTCTime (fromGregorian 2026 5 16) 0 + } + +withTestStore :: (KirokuStore -> IO ()) -> IO () +withTestStore action = + withMigratedTestDatabase $ \connStr -> + withStore (defaultConnectionSettings connStr) action