shibuya-kiroku-adapter 0.7.0.1 → 0.8.0.0

raw patch · 5 files changed

+348/−43 lines, 5 filesdep +hs-opentelemetry-exporter-in-memorydep ~shibuya-corePVP ok

version bump matches the API change (PVP)

Dependencies added: hs-opentelemetry-exporter-in-memory

Dependency ranges changed: shibuya-core

API changes (from Hackage documentation)

Files

• CHANGELOG.md +39/−3
• shibuya-kiroku-adapter.cabal +27/−25
• src/Shibuya/Adapter/Kiroku.hs +59/−12
• test/CorrelationSpec.hs +216/−0
• test/Main.hs +7/−3

@@ -1,5 +1,40 @@ # Changelog +## 0.8.0.0 — 2026-11-14++### Breaking Changes++- **Default trace shape changed: each processed event is now its own trace,+  linked to the span that appended it.** `KirokuAdapterConfig` and+  `KirokuConsumerGroupConfig` gain a strict `telemetry :: AdapterTelemetry`+  field, defaulting to `kirokuDefaultTelemetry`+  (`defaultAdapterTelemetry {traceCorrelation = LinkOnly}`). A subscription+  replays and catches up on events written long ago, and several subscriptions+  read the same event, so making processing a child of the appending span+  stretched the writer's trace indefinitely. Restore the previous single-trace+  shape, now with an added link, with `telemetry = defaultAdapterTelemetry`+  (`ParentAndLink`). Every consumer-group member uses the group's value.+- Require `shibuya-core >=0.13 && <0.14` (also in `kiroku-store`). Core 0.13+  renames processing spans from `<processor> process` to `process <processor>`+  and adds `messaging.operation.name = "process"`; update trace queries that+  match the old name. Full `KirokuAdapterConfig` and `KirokuConsumerGroupConfig`+  literals must set `telemetry`.++### New Features++- Re-export `AdapterTelemetry (traceCorrelation)`, `defaultAdapterTelemetry`,+  `TraceCorrelation (..)` and the new `kirokuDefaultTelemetry` from+  `Shibuya.Adapter.Kiroku`.++### Other Changes++- Add PostgreSQL-backed correlation tests: both policies, two subscriptions to+  one event, consumer-group members, and events without `traceparent`.+- Correct the 0.5.0.0 note below: `kiroku-otel`'s native subscription spans are+  independent roots, not part of the writer's or Shibuya's trace. They and+  Shibuya's processing spans are separate traces correlated by matching+  `kiroku.*` attributes.+ ## 0.7.0.1 — 2026-10-28  ### Other Changes@@ -261,9 +296,10 @@   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.+  with `kiroku-otel`'s native subscription spans, a kiroku subscription can be+  followed into Shibuya's processing: the keys match on both sides. (Corrected+  in 0.8.0.0: these are separate traces correlated by those attributes, not a+  single trace.) * `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

@@ -1,6 +1,6 @@ cabal-version:   3.0 name:            shibuya-kiroku-adapter-version:         0.7.0.1+version:         0.8.0.0 synopsis:   Kiroku event store adapter for the Shibuya queue processing framework @@ -66,7 +66,7 @@     , hs-opentelemetry-api                   ^>=1.0     , hs-opentelemetry-semantic-conventions  ^>=1.40     , kiroku-store                           ^>=0.13.0.0-    , shibuya-core                           >=0.12      && <0.13+    , shibuya-core                           >=0.13      && <0.14     , shibuya-kiroku-adapter-internal     , stm                                    >=2.5       && <2.6     , streamly-core                          >=0.3       && <0.4@@ -82,32 +82,34 @@   type:           exitcode-stdio-1.0   main-is:        Main.hs   hs-source-dirs: test+  other-modules:  CorrelationSpec   ghc-options:    -threaded -rtsopts -with-rtsopts=-N   build-depends:-    , aeson                            >=2.1       && <2.3-    , base                             >=4.18      && <5-    , containers                       >=0.6       && <0.8+    , aeson                                >=2.1       && <2.3+    , base                                 >=4.18      && <5+    , containers                           >=0.6       && <0.8     , directory-    , effectful                        >=2.6.1     && <2.8-    , effectful-core                   >=2.6.1     && <2.7  || >=2.7.1.1 && <2.8-    , ephemeral-pg                     >=0.3.1     && <0.4-    , 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.13.0.0-    , kiroku-test-support              ^>=0.1.0.0-    , lens                             >=5.2       && <5.4-    , shibuya-core                     >=0.12      && <0.13+    , effectful                            >=2.6.1     && <2.8+    , effectful-core                       >=2.6.1     && <2.7  || >=2.7.1.1 && <2.8+    , ephemeral-pg                         >=0.3.1     && <0.4+    , generic-lens                         >=2.2       && <2.4+    , hasql                                >=1.10      && <1.11+    , hasql-pool                           >=1.2       && <1.5+    , hs-opentelemetry-api                 ^>=1.0+    , hs-opentelemetry-exporter-in-memory  ^>=1.0+    , hspec                                >=2.10      && <2.12+    , kiroku-store                         ^>=0.13.0.0+    , kiroku-test-support                  ^>=0.1.0.0+    , lens                                 >=5.2       && <5.4+    , shibuya-core                         >=0.13      && <0.14     , shibuya-kiroku-adapter     , shibuya-kiroku-adapter-internal-    , stm                              >=2.5       && <2.6-    , streamly-core                    >=0.3       && <0.4-    , text                             >=2.0       && <2.2-    , time                             >=1.12      && <1.15-    , unordered-containers             >=0.2       && <0.3-    , uuid                             >=1.3       && <1.4+    , stm                                  >=2.5       && <2.6+    , streamly-core                        >=0.3       && <0.4+    , text                                 >=2.0       && <2.2+    , time                                 >=1.12      && <1.15+    , unordered-containers                 >=0.2       && <0.3+    , uuid                                 >=1.3       && <1.4  -- EP-45 live-store performance, restart, backlog, and retained-memory fixture. -- The executable owns an ephemeral PostgreSQL instance and emits the common@@ -132,7 +134,7 @@     , ephemeral-pg            >=0.3.1     && <0.4     , kiroku-store            ^>=0.13.0.0     , kiroku-test-support     ^>=0.1.0.0-    , shibuya-core            >=0.12      && <0.13+    , shibuya-core            >=0.13      && <0.14     , shibuya-kiroku-adapter     , stm                     >=2.5       && <2.6     , text                    >=2.0       && <2.2@@ -159,7 +161,7 @@     , hasql-pool              >=1.2       && <1.5     , kiroku-store            ^>=0.13.0.0     , kiroku-test-support-    , shibuya-core            >=0.12      && <0.13+    , shibuya-core            >=0.13      && <0.14     , shibuya-kiroku-adapter     , text                    >=2.0       && <2.2     , vector                  >=0.13      && <0.14

@@ -117,6 +117,21 @@ -- @handlerStallWarnAfter@ to receive advisory store handler-stall events; warnings -- never finalize, retry or checkpoint the item. @retryPolicy@ controls total -- deliveries independently of the delay chosen by 'AckRetry'.+--+-- == Trace Correlation+--+-- When an event's metadata carries a W3C @traceparent@ (written, for example,+-- by @kiroku-otel@'s @injectTraceContext@ on append), Shibuya opens a+-- Consumer span named @process \<processor\>@ around the handler and always+-- adds a link to the appending span. By default ('kirokuDefaultTelemetry',+-- 'LinkOnly') that span is the root of a new trace: subscriptions replay and+-- catch up on events written long ago and several subscriptions read the+-- same event, so making each handler a child of the appending span would+-- stretch the writer's trace indefinitely. Set @telemetry@ to+-- 'defaultAdapterTelemetry' ('ParentAndLink') to make processing the+-- appending span's child instead; a consumer group applies its one value to+-- every member. @kiroku-otel@'s own subscription spans are separate roots,+-- correlated with these spans by the shared @kiroku.*@ attributes. module Shibuya.Adapter.Kiroku   ( -- * Adapter     kirokuAdapter,@@ -128,6 +143,12 @@     KirokuAdapterConfig (..),     defaultKirokuAdapterConfig, +    -- * Telemetry+    AdapterTelemetry (traceCorrelation),+    defaultAdapterTelemetry,+    kirokuDefaultTelemetry,+    TraceCorrelation (..),+     -- * Consumer-group helpers     KirokuConsumerGroupConfig (..),     defaultConsumerGroupConfig,@@ -178,7 +199,7 @@ import Kiroku.Store.Subscription.Types qualified as Sub import Kiroku.Store.Types (RecordedEvent) import Numeric.Natural (Natural)-import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter (Adapter (..), AdapterTelemetry (traceCorrelation), TraceCorrelation (..), defaultAdapterTelemetry) import Shibuya.Adapter.Kiroku.Convert (kirokuEnvelopeAttrs, toIngestedAck) import Shibuya.Adapter.Kiroku.Internal (acquireAllAndTransfer) import Shibuya.App (ProcessorId (..), QueueProcessor (..), mkProcessor)@@ -257,7 +278,18 @@     --     dead-lettered and the checkpoint still advances past it. See     --     'Kiroku.Store.Subscription.Types.selector' for when to prefer it over the     --     introspectable 'eventTypeFilter'.-    selector :: !(Maybe (RecordedEvent -> Bool))+    selector :: !(Maybe (RecordedEvent -> Bool)),+    -- | Telemetry options copied unchanged into the 'Shibuya.Adapter.Adapter'.+    --+    -- The default is @'defaultAdapterTelemetry' {traceCorrelation = 'LinkOnly'}@:+    -- each processed event's span starts its own trace and links to the span+    -- that appended the event. This differs from queue adapters, which default+    -- to 'ParentAndLink', because a subscription replays and catches up on+    -- events written long ago and several subscriptions read the same event,+    -- so parenting would stretch the writer's trace indefinitely and attach+    -- every subscriber's work to it. Set @telemetry = 'defaultAdapterTelemetry'@+    -- ('ParentAndLink') to make processing a child of the appending span again.+    telemetry :: !AdapterTelemetry   }   deriving stock (Generic) @@ -265,8 +297,8 @@ -- @bufferSize = defaultStreamBufferSize@, @queueCapacity = 16@, @consumerGroup = 'Nothing'@ -- (ordinary single-consumer subscription), @missingCheckpointPolicy = -- 'FromBeginning'@, @eventTypeFilter = 'AllEventTypes'@--- (deliver every type), and @selector = 'Nothing'@ (no extra predicate--- filtering). Supply the subscription name and target; override individual fields+-- (deliver every type), @selector = 'Nothing'@ (no extra predicate+-- filtering), and @telemetry = 'kirokuDefaultTelemetry'@ ('LinkOnly'). 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@@ -292,9 +324,17 @@       consumerGroup = Nothing,       missingCheckpointPolicy = FromBeginning,       eventTypeFilter = AllEventTypes,-      selector = Nothing+      selector = Nothing,+      telemetry = kirokuDefaultTelemetry     } +-- | The adapter's default telemetry: 'defaultAdapterTelemetry' with+-- @traceCorrelation = 'LinkOnly'@, so each processed event is its own trace+-- linked to the appending span. Shared by 'defaultKirokuAdapterConfig' and+-- 'defaultConsumerGroupConfig'.+kirokuDefaultTelemetry :: AdapterTelemetry+kirokuDefaultTelemetry = defaultAdapterTelemetry {traceCorrelation = LinkOnly}+ -- | Convert any synchronous exception thrown by a Shibuya handler into an -- 'AckDecision'. --@@ -341,7 +381,7 @@   KirokuStore ->   KirokuAdapterConfig ->   Eff es (Adapter es RecordedEvent)-kirokuAdapter store KirokuAdapterConfig {subscriptionName = subName, subscriptionTarget = subTarget, batchSize = bs, bufferSize = buf, queueCapacity = qCap, retryPolicy = attempts, handlerStallWarnAfter = stallWarn, consumerGroup = cg, missingCheckpointPolicy = checkpointPolicy, eventTypeFilter = etf, selector = sel} = do+kirokuAdapter store KirokuAdapterConfig {subscriptionName = subName, subscriptionTarget = subTarget, batchSize = bs, bufferSize = buf, queueCapacity = qCap, retryPolicy = attempts, handlerStallWarnAfter = stallWarn, consumerGroup = cg, missingCheckpointPolicy = checkpointPolicy, eventTypeFilter = etf, selector = sel, telemetry = tel} = 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@@ -378,7 +418,8 @@     Adapter       { adapterName = "kiroku",         source = ingestedStream,-        shutdown = liftIO cancelAction+        shutdown = liftIO cancelAction,+        telemetry = tel       }  -- | Configuration for a whole kiroku consumer group presented as a single@@ -439,7 +480,10 @@     --     selector-filtered single subscription (worker-side, per member, checkpoint     --     still advances past rejected events). See     --     'Kiroku.Store.Subscription.Types.selector'.-    selector :: !(Maybe (RecordedEvent -> Bool))+    selector :: !(Maybe (RecordedEvent -> Bool)),+    -- | Telemetry options given unchanged to /every/ member adapter. Defaults to+    --     'kirokuDefaultTelemetry' ('LinkOnly'); see 'KirokuAdapterConfig'.telemetry.+    telemetry :: !AdapterTelemetry   }   deriving stock (Generic) @@ -447,7 +491,8 @@ -- 'Serial'@ (the only legal per-member concurrency), @batchSize = Sub.defaultBatchSize@, -- @bufferSize = defaultStreamBufferSize@, @queueCapacity = 16@, @missingCheckpointPolicy = -- 'FromBeginning'@, @eventTypeFilter = 'AllEventTypes'@--- (deliver every type), @selector = 'Nothing'@ (no extra predicate filtering).+-- (deliver every type), @selector = 'Nothing'@ (no extra predicate filtering),+-- @telemetry = 'kirokuDefaultTelemetry'@ ('LinkOnly'). -- Supply the subscription name, target, and group size. defaultConsumerGroupConfig ::   SubscriptionName -> SubscriptionTarget -> ConsumerGroupSize -> KirokuConsumerGroupConfig@@ -464,7 +509,8 @@       memberConcurrency = Serial,       missingCheckpointPolicy = FromBeginning,       eventTypeFilter = AllEventTypes,-      selector = Nothing+      selector = Nothing,+      telemetry = kirokuDefaultTelemetry     }  -- | Map a requested per-member concurrency onto the group's validated Shibuya@@ -506,7 +552,7 @@   KirokuConsumerGroupConfig ->   Handler es RecordedEvent ->   Eff es (Either PolicyError [(ProcessorId, QueueProcessor es)])-kirokuConsumerGroupProcessors store cfg@KirokuConsumerGroupConfig {subscriptionName = subName, subscriptionTarget = subTarget, groupSize = n, batchSize = bs, bufferSize = buf, queueCapacity = qCap, retryPolicy = attempts, handlerStallWarnAfter = stallWarn, missingCheckpointPolicy = checkpointPolicy, eventTypeFilter = etf, selector = sel} handler =+kirokuConsumerGroupProcessors store cfg@KirokuConsumerGroupConfig {subscriptionName = subName, subscriptionTarget = subTarget, groupSize = n, batchSize = bs, bufferSize = buf, queueCapacity = qCap, retryPolicy = attempts, handlerStallWarnAfter = stallWarn, missingCheckpointPolicy = checkpointPolicy, eventTypeFilter = etf, selector = sel, telemetry = tel} handler =   kirokuConsumerGroupProcessorsWith mkMemberAdapter cfg handler   where     mkMemberAdapter m =@@ -523,7 +569,8 @@             consumerGroup = Just (either (error . show) Prelude.id (mkConsumerGroup m n)),             missingCheckpointPolicy = checkpointPolicy,             eventTypeFilter = etf,-            selector = sel+            selector = sel,+            telemetry = tel           }  -- | Factory-parameterized core for 'kirokuConsumerGroupProcessors'.

@@ -0,0 +1,216 @@+-- | Trace correlation of Shibuya processing spans over a Kiroku subscription.+--+-- Each case appends an event whose metadata carries a known writer+-- @traceparent@, subscribes with a traced Shibuya app whose tracer exports into+-- memory, and inspects the exported Consumer spans: their trace, their parent+-- and their links.+module CorrelationSpec (spec) where++import Control.Concurrent.STM (atomically, modifyTVar', newTVarIO, readTVar, registerDelay)+import Control.Concurrent.STM qualified as STM+import Control.Exception qualified as E+import Control.Lens ((&), (.~), (^.))+import Control.Monad (forM_)+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.IORef (readIORef)+import Data.Maybe (isNothing)+import Data.Text (Text)+import Effectful (runEff)+import Kiroku.Store+import OpenTelemetry.Attributes (emptyAttributes)+import OpenTelemetry.Exporter.InMemory (inMemoryListExporter)+import OpenTelemetry.Trace.Core qualified as OTel+import OpenTelemetry.Trace.Id (Base (..))+import OpenTelemetry.Trace.Id qualified as OTel.Id+import OpenTelemetry.Trace.Id.Generator (IdGenerator (..))+import OpenTelemetry.Util (appendOnlyBoundedCollectionValues)+import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter.Kiroku+  ( AdapterTelemetry,+    ConsumerGroupSize,+    KirokuAdapterConfig,+    TraceCorrelation (..),+    defaultAdapterTelemetry,+    defaultConsumerGroupConfig,+    defaultKirokuAdapterConfig,+    kirokuAdapter,+    kirokuConsumerGroupProcessors,+    mkConsumerGroupSize,+  )+import Shibuya.App (ProcessorId (..), QueueProcessor (..), defaultAppConfig, mkProcessor, runApp, stopApp)+import Shibuya.Core.Ack (AckDecision (..))+import Shibuya.Telemetry.Effect (runTracing, runTracingNoop)+import Test.Hspec++spec :: SpecWith KirokuStore+spec = describe "trace correlation" $ do+  it "defaults to LinkOnly for single adapters and consumer groups" $ \_ -> do+    defaultKirokuAdapterConfig (SubscriptionName "c") AllStreams ^. #telemetry . #traceCorrelation `shouldBe` LinkOnly+    defaultConsumerGroupConfig (SubscriptionName "c") AllStreams (sizeOf 2) ^. #telemetry . #traceCorrelation `shouldBe` LinkOnly++  it "an adapter built with a ParentAndLink override exposes it" $ \store -> do+    policy <- runEff $ runTracingNoop $ do+      Adapter {shutdown, telemetry} <- kirokuAdapter store (defaultKirokuAdapterConfig (SubscriptionName "corr-override") AllStreams & #telemetry .~ defaultAdapterTelemetry)+      shutdown+      pure (telemetry ^. #traceCorrelation)+    policy `shouldBe` ParentAndLink++  it "every consumer-group member reports the configured policy" $ \store -> do+    forM_ [LinkOnly, ParentAndLink] $ \policy -> do+      let cfg = defaultConsumerGroupConfig (SubscriptionName "corr-group") AllStreams (sizeOf 3) & #telemetry .~ withPolicy policy+      policies <- runEff $ runTracingNoop $ do+        result <- kirokuConsumerGroupProcessors store cfg (\_ -> pure AckOk)+        case result of+          Left err -> liftIO (expectationFailure (show err)) >> pure []+          Right processors ->+            traverse+              ( \(_, processor) -> case processor of+                  QueueProcessor {adapter = Adapter {shutdown, telemetry}} -> shutdown >> pure (telemetry ^. #traceCorrelation)+                  BatchingProcessor {adapter = Adapter {shutdown, telemetry}} -> shutdown >> pure (telemetry ^. #traceCorrelation)+              )+              processors+      policies `shouldBe` replicate 3 policy++  it "LinkOnly (default): the processing span is a linked root" $ \store -> do+    appendTraced store "corrlo-1" (Just writerParent)+    [sp] <- processAll store [("corr-linkonly", defaultKirokuAdapterConfig (SubscriptionName "corr-linkonly") (category "corrlo"))]+    hexTrace (spanContext sp) `shouldNotBe` writerTraceHex+    isNothing (spanParent sp) `shouldBe` True+    linkIds sp `shouldBe` [(writerTraceHex, writerSpanHex)]++  it "ParentAndLink: the processing span is a child of the writer with one link" $ \store -> do+    appendTraced store "corrpal-1" (Just writerParent)+    [sp] <- processAll store [("corr-pal", defaultKirokuAdapterConfig (SubscriptionName "corr-pal") (category "corrpal") & #telemetry .~ defaultAdapterTelemetry)]+    hexTrace (spanContext sp) `shouldBe` writerTraceHex+    fmap hexSpan (spanParent sp) `shouldBe` Just writerSpanHex+    linkIds sp `shouldBe` [(writerTraceHex, writerSpanHex)]++  it "LinkOnly: two subscriptions to one event are two traces linking the same writer span" $ \store -> do+    appendTraced store "corrfan-1" (Just writerParent)+    spans <-+      processAll+        store+        [ ("corr-fan-a", defaultKirokuAdapterConfig (SubscriptionName "corr-fan-a") (category "corrfan")),+          ("corr-fan-b", defaultKirokuAdapterConfig (SubscriptionName "corr-fan-b") (category "corrfan"))+        ]+    case spans of+      [a, b] -> do+        hexTrace (spanContext a) `shouldNotBe` hexTrace (spanContext b)+        mapM_ (\sp -> isNothing (spanParent sp) `shouldBe` True) [a, b]+        mapM_ (\sp -> linkIds sp `shouldBe` [(writerTraceHex, writerSpanHex)]) [a, b]+      other -> expectationFailure ("expected two processing spans, got " <> show (length other))++  forM_ [LinkOnly, ParentAndLink] $ \policy ->+    it (show policy <> ": an event without traceparent yields no link") $ \store -> do+      let cat = if policy == LinkOnly then "corrnolo" else "corrnopal"+      appendTraced store (cat <> "-1") Nothing+      [sp] <- processAll store [("corr-none-" <> cat, defaultKirokuAdapterConfig (SubscriptionName ("corr-none-" <> cat)) (category cat) & #telemetry .~ withPolicy policy)]+      spanLinks sp `shouldBe` []+      isNothing (spanParent sp) `shouldBe` True++--------------------------------------------------------------------------------+-- Fixtures+--------------------------------------------------------------------------------++writerParent :: Text+writerParent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"++writerTraceHex :: Text+writerTraceHex = "0af7651916cd43dd8448eb211c80319c"++writerSpanHex :: Text+writerSpanHex = "b7ad6b7169203331"++sizeOf :: Int -> ConsumerGroupSize+sizeOf = either (error . show) Prelude.id . mkConsumerGroupSize . fromIntegral++category :: Text -> SubscriptionTarget+category = Category . CategoryName++withPolicy :: TraceCorrelation -> AdapterTelemetry+withPolicy policy = defaultAdapterTelemetry & #traceCorrelation .~ policy++-- | Append one event to the given stream, with the writer's traceparent in its+-- metadata when supplied.+appendTraced :: KirokuStore -> Text -> Maybe Text -> IO ()+appendTraced store stream traceparent = do+  let metadata = fmap (\tp -> Aeson.object ["traceparent" Aeson..= tp]) traceparent+      event =+        EventData+          { eventId = Nothing,+            eventType = EventType "Correlated",+            payload = Aeson.object [] :: Value,+            metadata = metadata,+            causationId = Nothing,+            correlationId = Nothing+          }+  result <- runStoreIO store $ appendToStream (StreamName stream) NoStream [event]+  either (\err -> expectationFailure ("append failed: " <> show err)) (const (pure ())) result++-- | Run one processor per configuration until each has handled one event, and+-- return the exported Consumer spans.+processAll :: KirokuStore -> [(Text, KirokuAdapterConfig)] -> IO [ConsumerSpan]+processAll store configs = do+  (spanProcessor, spansRef) <- inMemoryListExporter+  handled <- newTVarIO (0 :: Int)+  E.bracket+    (OTel.createTracerProvider [spanProcessor] OTel.emptyTracerProviderOptions {OTel.tracerProviderOptionsIdGenerator = DefaultIdGenerator})+    (\provider -> OTel.shutdownTracerProvider provider (Just 5_000_000))+    $ \provider -> do+      let tracer = OTel.makeTracer provider (OTel.InstrumentationLibrary "kiroku-correlation" "" "" emptyAttributes) OTel.tracerOptions+      runEff $ runTracing tracer $ do+        processors <-+          traverse+            ( \(name, cfg) -> do+                adapter <- kirokuAdapter store cfg+                pure (ProcessorId name, mkProcessor adapter (\_ -> liftIO (atomically (modifyTVar' handled (+ 1))) >> pure AckOk))+            )+            configs+        result <- runApp defaultAppConfig processors+        case result of+          Left err -> liftIO (expectationFailure (show err))+          Right appHandle -> do+            liftIO $ waitFor handled (length configs) 10_000_000+            stopApp appHandle+  consumerSpans =<< readIORef spansRef++waitFor :: STM.TVar Int -> Int -> Int -> IO ()+waitFor var target timeoutMicros = do+  deadline <- registerDelay timeoutMicros+  reached <-+    atomically $+      ((readTVar var >>= STM.check . (>= target)) >> pure True)+        `STM.orElse` (readTVar deadline >>= STM.check >> pure False)+  if reached then pure () else expectationFailure ("timed out waiting for " <> show target <> " handled events")++data ConsumerSpan = ConsumerSpan+  { spanContext :: OTel.SpanContext,+    spanParent :: Maybe OTel.SpanContext,+    spanLinks :: [OTel.SpanContext]+  }++consumerSpans :: [OTel.ImmutableSpan] -> IO [ConsumerSpan]+consumerSpans exported = traverse describeSpan [sp | sp <- exported, OTel.spanKind sp == OTel.Consumer]+  where+    describeSpan sp = do+      hot <- readIORef (OTel.spanHot sp)+      parent <- traverse OTel.getSpanContext (OTel.spanParent sp)+      pure+        ConsumerSpan+          { spanContext = OTel.spanContext sp,+            spanParent = parent,+            spanLinks = map OTel.frozenLinkContext (toList (appendOnlyBoundedCollectionValues (OTel.hotLinks hot)))+          }++linkIds :: ConsumerSpan -> [(Text, Text)]+linkIds sp = [(hexTrace l, hexSpan l) | l <- spanLinks sp]++hexTrace :: OTel.SpanContext -> Text+hexTrace ctx = OTel.Id.traceIdBaseEncodedText Base16 (OTel.traceId ctx)++hexSpan :: OTel.SpanContext -> Text+hexSpan ctx = OTel.Id.spanIdBaseEncodedText Base16 (OTel.spanId ctx)

@@ -8,6 +8,7 @@ import Control.Lens ((&), (.~), (^.)) import Control.Monad (forM_, unless) import Control.Monad.IO.Class (liftIO)+import CorrelationSpec qualified import Data.Aeson (Value) import Data.Aeson qualified as Aeson import Data.Foldable (toList)@@ -36,7 +37,7 @@ import Kiroku.Store.Subscription.Worker (withFetchBatchHookForTest, withSaveCheckpointHookForTest) import Kiroku.Test.Postgres (withMigratedTestDatabase, withSharedMigratedPostgres) import OpenTelemetry.Attributes (toAttribute)-import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry) import Shibuya.Adapter.Kiroku   ( consumerGroupPolicy,     defaultConsumerGroupConfig,@@ -195,6 +196,7 @@         `shouldBe` Left (InvalidPolicyCombo "StrictInOrder requires Serial concurrency")    acknowledgementSpec+  around withTestStore CorrelationSpec.spec   around withTestStore $ do     describe "kirokuAdapter" $ do       it "delivers only matching event types when an eventTypeFilter is set (EP-43)" $ \store -> do@@ -1119,7 +1121,8 @@                           modifyIORef' shutdowns (<> [m])                           if m == 1                             then E.throwIO (userError "member 1 cleanup failed")-                            else pure ()+                            else pure (),+                        telemetry = defaultAdapterTelemetry                       }             handler ingested = do               let _ = envelopePayload ingested@@ -1287,7 +1290,8 @@   Adapter     { adapterName = "stub-kiroku",       source = Stream.nil,-      shutdown = pure ()+      shutdown = pure (),+      telemetry = defaultAdapterTelemetry     }  -- | Read the dead letters recorded for a non-group subscription (member 0).