shibuya-kiroku-adapter (empty) → 0.2.0.0
raw patch · 5 files changed
Files
- CHANGELOG.md +106/−0
- shibuya-kiroku-adapter.cabal +92/−0
- src/Shibuya/Adapter/Kiroku.hs +454/−0
- src/Shibuya/Adapter/Kiroku/Convert.hs +223/−0
- test/Main.hs +804/−0
+ CHANGELOG.md view
@@ -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 "<name>-member-<m>"`), 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.
+ shibuya-kiroku-adapter.cabal view
@@ -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
+ src/Shibuya/Adapter/Kiroku.hs view
@@ -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+@\"\<subscriptionName\>-member-\<m\>\"@ so member identity is readable off the id+and two members never collide. The group-validity invariant (@groupSize >= 1@,+@0 <= member < groupSize@) is enforced downstream by+'Kiroku.Store.Subscription.subscribe' (throwing+'Kiroku.Store.Subscription.Types.InvalidConsumerGroup'); 'groupSize >= 1' is a+documented precondition of this helper.+-}+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)
+ src/Shibuya/Adapter/Kiroku/Convert.hs view
@@ -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
+ test/Main.hs view
@@ -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