shibuya-core 0.12.0.0 → 0.13.0.0

raw patch · 26 files changed

+1615/−311 lines, 26 filesdep ~bytestringdep ~hs-opentelemetry-apidep ~streamly-corePVP ok

version bump matches the API change (PVP)

Dependency ranges changed: bytestring, hs-opentelemetry-api, streamly-core, text

API changes (from Hackage documentation)

Files

• CHANGELOG.md +72/−0
• shibuya-core.cabal +29/−1
• src/Shibuya.hs +4/−1
• src/Shibuya/Adapter.hs +12/−1
• src/Shibuya/Adapter/Mock.hs +3/−2
• src/Shibuya/Internal/Runner/BatchProcessor.hs +141/−124
• src/Shibuya/Internal/Runner/Supervised.hs +137/−134
• src/Shibuya/Telemetry.hs +8/−0
• src/Shibuya/Telemetry/Correlation.hs +171/−0
• src/Shibuya/Telemetry/Effect.hs +19/−0
• src/Shibuya/Telemetry/Process.hs +120/−0
• src/Shibuya/Telemetry/Semantic.hs +31/−10
• test-gc/Finished.hs +7/−4
• test-gc/Main.hs +3/−2
• test-retention/Main.hs +305/−0
• test/Main.hs +4/−0
• test/Shibuya/App/BatchSpec.hs +4/−3
• test/Shibuya/App/LifecycleSpec.hs +11/−6
• test/Shibuya/Batch/ReliabilitySpec.hs +5/−3
• test/Shibuya/Runner/BatchProcessorSpec.hs +2/−2
• test/Shibuya/Runner/PauseSpec.hs +4/−4
• test/Shibuya/Runner/SupervisedSpec.hs +15/−10
• test/Shibuya/RunnerSpec.hs +3/−2
• test/Shibuya/Telemetry/BatchCorrelationSpec.hs +220/−0
• test/Shibuya/Telemetry/CorrelationSpec.hs +282/−0
• test/Shibuya/Telemetry/SemanticSpec.hs +3/−2

@@ -1,5 +1,77 @@ # Changelog +## 0.13.0.0 — 2026-11-12++### Breaking Changes++- `Adapter` gains a strict `telemetry :: AdapterTelemetry` field. Record+  constructions must add `telemetry = defaultAdapterTelemetry`; positional+  construction requires the extra argument. Adapters should expose the value in+  their own configuration so users can override it.+- Per-message processing spans now carry a link to the producer's span when+  `Envelope.traceContext` decodes. Under the default `ParentAndLink` policy the span+  is still the producer span's child, so existing traces keep their shape and gain+  one link.+- Batch processing spans are always roots and no longer join the first member's+  trace. They link every member whose trace headers decode, in member order, each+  link carrying `messaging.message.id`, up to 128 links. Valid contexts beyond+  that are counted in `shibuya.batch.links.dropped`. Batch spans also carry+  `messaging.batch.message_count`.+- Processing and batch spans are renamed from `<destination> process` to+  `process <destination>`, following the messaging conventions'+  `{messaging.operation.name} {destination}`. Saved searches, alerts and sampling+  rules that match the span or operation name must change: a Jaeger or Tempo+  search for operation `orders process` becomes `process orders`. Lifecycle span+  names are unchanged.++### New Features++- Processing and batch spans carry `messaging.operation.name = "process"`, which+  the conventions mark Required, alongside `messaging.operation.type`.+  `Shibuya.Telemetry.Semantic` exports `attrMessagingOperationName`.+- New module `Shibuya.Telemetry.Correlation`, re-exported from `Shibuya.Adapter`,+  `Shibuya.Telemetry` and `Shibuya`. It provides `TraceCorrelation`+  (`ParentAndLink`, `LinkOnly`), `AdapterTelemetry` (constructor hidden, built from+  `defaultAdapterTelemetry` by record update), and the link helpers `creationLink`,+  `creationLinkWith` and `consumerSpanArgsLinkedTo`.+- `LinkOnly` makes each processing span with a usable creation context the root of+  a new trace, related to the producer only by the link. Use it for replayed,+  fanned-out or long-delayed sources and with tail-based sampling. Without usable+  headers both policies keep the ambient parent and add no link.+- New module `Shibuya.Telemetry.Process` with `withProcessSpan` (in `Tracing`) and+  `withProcessSpanIO` (for a plain `Tracer`). They open the same processing span+  as the runner for code that drains an adapter outside it. With valid headers they+  neither parent on nor link an ambient span; they are meant for independent+  message processing.+- `Shibuya.Telemetry.Correlation` also exports `maxBatchLinks`, `BatchLinks`,+  `collectBatchLinks` and `collectBatchLinksBy`, and `Shibuya.Telemetry.Semantic`+  exports `attrMessagingBatchMessageCount` and `attrShibuyaBatchLinksDropped`.+- `Shibuya.Telemetry.Effect.withRootContext` runs an action with an empty+  OpenTelemetry context attached.++### Other Changes++- With tracing disabled, per-message processing no longer decodes the W3C trace+  headers. Before, each message with a `traceparent` was decoded even under+  `runTracingNoop`.+- Per-message processing allocates about 208 bytes less on every single-message+  path. With tracing enabled and trace headers present it allocates about 210 bytes+  less, even with the added link, because the context attachment and the span now+  share one unlift.+- Accept a measured allocation increase on `Async` runners whose handlers wait:+  `concurrency-modes` `async-5` cells allocate 20–30% more than 0.12.0.0 (for+  example 1.17–1.21 MB to 1.47–1.57 MB for 100 messages with a 1 ms handler delay),+  `async-10` 4–5% and `async-20` 2–3% more, with unchanged time. `async-2`, every+  `Ahead` cell and `hot-path/async8` are flat or lower. The per-message code these+  cells run is the path that got cheaper everywhere else; the increase comes from+  streamly's worker dispatch, which is sensitive to the shape of per-message work+  (see `docs/plans/30-investigate-and-reduce-the-async-ahead-concurrency-allocation-regression.md`).+- The release comparison against 0.12.0.0 measured 10,000-message processing+  (`hot-path/serial-noop-10000`) at 9.247 to 9.098 ms, with allocation falling+  from 19,918,565 to 17,837,870 bytes. Every timing comparison passed the 10% gate+  in interleaved confirming runs at `--stdev 3`; measurements used package-default+  RTS settings on a shared workstation.+ ## 0.12.0.0 — 2026-10-26  ### Breaking Changes

@@ -1,6 +1,6 @@ cabal-version: 3.12 name: shibuya-core-version: 0.12.0.0+version: 0.13.0.0 synopsis: Supervised queue processing framework for Haskell description:   A supervised queue processing framework inspired by Broadway (Elixir).@@ -47,7 +47,9 @@     Shibuya.Policy     Shibuya.Stream     Shibuya.Telemetry+    Shibuya.Telemetry.Correlation     Shibuya.Telemetry.Effect+    Shibuya.Telemetry.Process     Shibuya.Telemetry.Propagation     Shibuya.Telemetry.Semantic @@ -132,6 +134,8 @@     Shibuya.Runner.PauseSpec     Shibuya.Runner.SupervisedSpec     Shibuya.RunnerSpec+    Shibuya.Telemetry.BatchCorrelationSpec+    Shibuya.Telemetry.CorrelationSpec     Shibuya.Telemetry.EffectSpec     Shibuya.Telemetry.LifecycleSpec     Shibuya.Telemetry.PropagationSpec@@ -195,3 +199,27 @@     shibuya-core,     streamly-core,     unliftio,++-- Trace retention probes for the trace correlation work. Process-isolated for+-- the same reason as the GC suites: weak-pointer and live-bytes assertions+-- need nothing else in the process holding references (ADR 0001).+test-suite shibuya-core-trace-retention-test+  import: warnings+  default-language: GHC2024+  type: exitcode-stdio-1.0+  hs-source-dirs: test-retention+  main-is: Main.hs+  ghc-options:+    -threaded+    -rtsopts+    "-with-rtsopts=-N2 -T"++  build-depends:+    base ^>=4.21.0.0,+    bytestring,+    effectful-core (>=2.6.1 && <2.7) || (>=2.7.1.1 && <2.8),+    hs-opentelemetry-api,+    hs-opentelemetry-exporter-in-memory ^>=1.0,+    shibuya-core,+    streamly-core,+    text,

@@ -76,6 +76,9 @@      -- * Adapter authoring     Adapter (..),+    AdapterTelemetry (traceCorrelation),+    defaultAdapterTelemetry,+    TraceCorrelation (..),     AckHandle (..),     Lease (..),     Ingested (..),@@ -114,7 +117,7 @@   ) where -import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter (Adapter (..), AdapterTelemetry (traceCorrelation), TraceCorrelation (..), defaultAdapterTelemetry) import Shibuya.App   ( AppConfig (..),     AppError (..),

@@ -3,12 +3,18 @@ -- Adapter owns queue semantics; runner never touches offsets directly. module Shibuya.Adapter   ( Adapter (..),++    -- * Telemetry options+    AdapterTelemetry (traceCorrelation),+    defaultAdapterTelemetry,+    TraceCorrelation (..),   ) where  import Data.Text (Text) import Effectful (Eff) import Shibuya.Core.Ingested (Ingested)+import Shibuya.Telemetry.Correlation (AdapterTelemetry (traceCorrelation), TraceCorrelation (..), defaultAdapterTelemetry) import Streamly.Data.Stream (Stream)  -- | Queue adapter interface.@@ -19,5 +25,10 @@     -- | Stream of leased messages     source :: Stream (Eff es) (Ingested es msg),     -- | Stop polling, release resources-    shutdown :: Eff es ()+    shutdown :: Eff es (),+    -- | Telemetry options for this adapter's processing spans, such as the+    -- 'TraceCorrelation' policy. Start from 'defaultAdapterTelemetry' and+    -- expose the value in the adapter's own configuration so users can+    -- override it. Read once per processor.+    telemetry :: !AdapterTelemetry   }

@@ -16,7 +16,7 @@  import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef) import Effectful (Eff, IOE, liftIO, (:>))-import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry) import Shibuya.Core.Ack (AckDecision) import Shibuya.Core.AckHandle (AckHandle (..)) import Shibuya.Core.Ingested (Ingested, mkIngested)@@ -30,7 +30,8 @@   Adapter     { adapterName = "mock:list",       source = Stream.fromList msgs,-      shutdown = pure ()+      shutdown = pure (),+      telemetry = defaultAdapterTelemetry     }  -- | Tracking state for ack decisions.

@@ -87,22 +87,26 @@ import Shibuya.Internal.Runner.KeyedScheduler (runKeyedScheduler) import Shibuya.Policy (Concurrency (..)) import Shibuya.Prelude+import Shibuya.Telemetry.Correlation (BatchLinks (..), collectBatchLinksBy, consumerSpanArgsLinkedTo) import Shibuya.Telemetry.Effect   ( Tracing,     addAttribute,     addAttributes,     addEvent,+    isTracingEnabled,     recordException,     setStatus,-    withExtractedContext,+    withRootContext,     withSpan',   )-import Shibuya.Telemetry.Propagation (extractTraceContext) import Shibuya.Telemetry.Semantic-  ( attrMessagingDestinationName,+  ( attrMessagingBatchMessageCount,+    attrMessagingDestinationName,     attrMessagingOperation,+    attrMessagingOperationName,     attrMessagingSystem,     attrShibuyaBatchKey,+    attrShibuyaBatchLinksDropped,     attrShibuyaBatchSize,     attrShibuyaBatchTrigger,     attrShibuyaInflightCount,@@ -184,138 +188,151 @@   (BatchInfo, NonEmpty (Ingested es msg)) ->   Eff es () processCurrentBatchWithDetail detailed metricsHandle procId maxConc stopSignal exitPublisher handler (info, batch) = do-  -- Use the first message's trace context as the batch span's parent. A batch-  -- may span several traces; picking the first is a pragmatic single parent-  -- (full fan-in links are a later refinement).-  let firstMsg = NE.head batch-      parentCtx = firstMsg.envelope.traceContext >>= extractTraceContext-      ProcessorId pidText = procId+  enabled <- isTracingEnabled+  let ProcessorId pidText = procId+      spanName = processSpanName pidText+      -- A batch has many producers and a span has one parent, so the batch+      -- span is always a root and links each member's creation context+      -- instead (ADR 0014). Disabled tracing reads no member headers.+      openBatchSpan body+        | not enabled = withSpan' spanName consumerSpanArgs body+        | otherwise =+            let !batchLinks = collectBatchLinksBy (.envelope.messageId) (.envelope.traceContext) batch+                !args = case batchLinks.links of+                  [] -> consumerSpanArgs+                  links -> consumerSpanArgsLinkedTo links+             in withRootContext $+                  withSpan' spanName args $ \traceSpan -> do+                    when (batchLinks.dropped > 0) $+                      addAttribute traceSpan attrShibuyaBatchLinksDropped batchLinks.dropped+                    body traceSpan -  withExtractedContext parentCtx $-    withSpan' (processSpanName pidText) consumerSpanArgs $ \traceSpan -> do-      -- Framework messaging.* attributes plus batch-scoped attributes.-      let BatchKey keyText = info.batchKey-          frameworkAttrs =-            HashMap.fromList-              [ (attrMessagingSystem, toAttribute ("shibuya" :: Text)),-                (attrMessagingDestinationName, toAttribute pidText),-                (attrMessagingOperation, toAttribute ("process" :: Text)),-                (attrShibuyaBatchKey, toAttribute keyText),-                (attrShibuyaBatchSize, toAttribute info.size),-                (attrShibuyaBatchTrigger, toAttribute (triggerText info.trigger))-              ]-      addAttributes traceSpan frameworkAttrs+  openBatchSpan $ \traceSpan -> do+    -- Framework messaging.* attributes plus batch-scoped attributes.+    let BatchKey keyText = info.batchKey+        frameworkAttrs =+          HashMap.fromList+            [ (attrMessagingSystem, toAttribute ("shibuya" :: Text)),+              (attrMessagingDestinationName, toAttribute pidText),+              (attrMessagingOperation, toAttribute ("process" :: Text)),+              (attrMessagingOperationName, toAttribute ("process" :: Text)),+              (attrShibuyaBatchKey, toAttribute keyText),+              (attrShibuyaBatchSize, toAttribute info.size),+              (attrMessagingBatchMessageCount, toAttribute info.size),+              (attrShibuyaBatchTrigger, toAttribute (triggerText info.trigger))+            ]+    addAttributes traceSpan frameworkAttrs -      let process currentInflight = do-            addAttribute traceSpan attrShibuyaInflightCount currentInflight-            addAttribute traceSpan attrShibuyaInflightMax maxConc+    let process currentInflight = do+          addAttribute traceSpan attrShibuyaInflightCount currentInflight+          addAttribute traceSpan attrShibuyaInflightMax maxConc -            addEvent traceSpan (mkEvent eventBatchStarted [])+          addEvent traceSpan (mkEvent eventBatchStarted []) -            terminal <- liftIO $ readProcessorExit stopSignal+          terminal <- liftIO $ readProcessorExit stopSignal -            -- Run the handler under exception isolation. On any exception, record it-            -- on the span and substitute the whole-batch retry default.-            (handlerResult, skippedAfterHalt) <--              case terminal of-                Just _ -> pure (Left (), True)-                Nothing -> do-                  result <--                    catchAny-                      (Right <$> handler info (toMessage <$> batch))-                      ( \ex -> do-                          recordException traceSpan ex-                          pure (Left ())-                      )-                  pure (result, False)-            let (resolvedAck, handlerThrew) = case handlerResult of-                  Right a -> (a, False)-                  Left () -> (ackAll (AckRetry (RetryDelay 0)), True)+          -- Run the handler under exception isolation. On any exception, record it+          -- on the span and substitute the whole-batch retry default.+          (handlerResult, skippedAfterHalt) <-+            case terminal of+              Just _ -> pure (Left (), True)+              Nothing -> do+                result <-+                  catchAny+                    (Right <$> handler info (toMessage <$> batch))+                    ( \ex -> do+                        recordException traceSpan ex+                        pure (Left ())+                    )+                pure (result, False)+          let (resolvedAck, handlerThrew) = case handlerResult of+                Right a -> (a, False)+                Left () -> (ackAll (AckRetry (RetryDelay 0)), True) -            -- RELIABLE FINALIZATION: iterate OUR OWN retained list, never the-            -- handler's output. For each retained message, choose its decision once-            -- via findWithDefault, then call the idempotent adapter finalizer with-            -- bounded retry. Do not let one adapter failure prevent attempts for the-            -- rest of the batch.-            results <--              mapM-                ( \ingested -> do-                    let explicitDecision = Map.lookup ingested.envelope.messageId resolvedAck.decisions-                        d = fromMaybe resolvedAck.fallback explicitDecision-                    finalResult <- finalizeWithRetry traceSpan ingested d-                    let recordCursor = liftIO $ Metrics.recordAcknowledgedCursor metricsHandle ingested.envelope.partition ingested.envelope.cursor-                    when detailed $ case (finalResult, d) of-                      (Right (), AckOk) -> recordCursor-                      (Right (), AckDeadLetter _) -> recordCursor-                      _ -> pure ()-                    pure (ingested.envelope.messageId, isJust explicitDecision, d, finalResult)-                )-                (NE.toList batch)-            let decisions = [d | (_, _, d, _) <- results]-                finalizeFailures = [(mid, ex) | (mid, _, _, Left ex) <- results]+          -- RELIABLE FINALIZATION: iterate OUR OWN retained list, never the+          -- handler's output. For each retained message, choose its decision once+          -- via findWithDefault, then call the idempotent adapter finalizer with+          -- bounded retry. Do not let one adapter failure prevent attempts for the+          -- rest of the batch.+          results <-+            mapM+              ( \ingested -> do+                  let explicitDecision = Map.lookup ingested.envelope.messageId resolvedAck.decisions+                      d = fromMaybe resolvedAck.fallback explicitDecision+                  finalResult <- finalizeWithRetry traceSpan ingested d+                  let recordCursor = liftIO $ Metrics.recordAcknowledgedCursor metricsHandle ingested.envelope.partition ingested.envelope.cursor+                  when detailed $ case (finalResult, d) of+                    (Right (), AckOk) -> recordCursor+                    (Right (), AckDeadLetter _) -> recordCursor+                    _ -> pure ()+                  pure (ingested.envelope.messageId, isJust explicitDecision, d, finalResult)+              )+              (NE.toList batch)+          let decisions = [d | (_, _, d, _) <- results]+              finalizeFailures = [(mid, ex) | (mid, _, _, Left ex) <- results] -            -- Compute halt and partial-failure signals from the resolved decisions.-            let finalizationFailure =-                  case finalizeFailures of-                    [] -> Nothing-                    failed@((firstMessageId, _) : _) ->-                      Just $-                        ProcessorFailed-                          ( "batch finalization failed for message ids: "-                              <> Text.intercalate ", " [tshow mid | (mid, _) <- failed]-                          )-                          (Just firstMessageId)-                firstHalt = listToMaybe [r | AckHalt r <- decisions]-                requestedExit = finalizationFailure <|> (ProcessorHalted <$> firstHalt)-                overrideFailures =-                  [ ()-                  | (_, explicitlyNamed, d, _) <- results,-                    explicitlyNamed,-                    isFailing d-                  ]-                partialInc = not handlerThrew && not (null overrideFailures)+          -- Compute halt and partial-failure signals from the resolved decisions.+          let finalizationFailure =+                case finalizeFailures of+                  [] -> Nothing+                  failed@((firstMessageId, _) : _) ->+                    Just $+                      ProcessorFailed+                        ( "batch finalization failed for message ids: "+                            <> Text.intercalate ", " [tshow mid | (mid, _) <- failed]+                        )+                        (Just firstMessageId)+              firstHalt = listToMaybe [r | AckHalt r <- decisions]+              requestedExit = finalizationFailure <|> (ProcessorHalted <$> firstHalt)+              overrideFailures =+                [ ()+                | (_, explicitlyNamed, d, _) <- results,+                  explicitlyNamed,+                  isFailing d+                ]+              partialInc = not handlerThrew && not (null overrideFailures) -            -- Span status: error on halt or exception, otherwise Ok.-            addEvent traceSpan (mkEvent eventBatchCompleted [])-            case requestedExit of-              Just processorExit -> setStatus traceSpan (OTel.Error (processorExitText processorExit))-              Nothing ->-                if skippedAfterHalt-                  then setStatus traceSpan (OTel.Error "batch skipped after halt")-                  else-                    if handlerThrew-                      then setStatus traceSpan (OTel.Error "batch handler exception")-                      else setStatus traceSpan OTel.Ok+          -- Span status: error on halt or exception, otherwise Ok.+          addEvent traceSpan (mkEvent eventBatchCompleted [])+          case requestedExit of+            Just processorExit -> setStatus traceSpan (OTel.Error (processorExitText processorExit))+            Nothing ->+              if skippedAfterHalt+                then setStatus traceSpan (OTel.Error "batch skipped after halt")+                else+                  if handlerThrew+                    then setStatus traceSpan (OTel.Error "batch handler exception")+                    else setStatus traceSpan OTel.Ok -            traverse_ (recordException traceSpan . snd) finalizeFailures+          traverse_ (recordException traceSpan . snd) finalizeFailures -            -- Record metrics: decrement in-flight, add per-message counter deltas,-            -- advance batch counters, set Failed state on halt or exhausted-            -- finalization retry.-            liftIO $-              recordBatchOutcomeMetrics-                metricsHandle-                (triggerMetric info.trigger)-                info.size-                handlerThrew-                partialInc-                (decisionMetric <$> decisions)-                (processorExitText <$> requestedExit)+          -- Record metrics: decrement in-flight, add per-message counter deltas,+          -- advance batch counters, set Failed state on halt or exhausted+          -- finalization retry.+          liftIO $+            recordBatchOutcomeMetrics+              metricsHandle+              (triggerMetric info.trigger)+              info.size+              handlerThrew+              partialInc+              (decisionMetric <$> decisions)+              (processorExitText <$> requestedExit) -            -- Halt: set the shared flag; do NOT throw (let the stream drain).-            for_ requestedExit $ \processorExit ->-              liftIO $ requestProcessorExit exitPublisher processorExit-      if not detailed-        then do-          currentInflight <- liftIO $ beginProcessing metricsHandle maxConc-          process currentInflight-        else Exception.mask $ \restore -> do-          currentInflight <- liftIO $ beginProcessing metricsHandle maxConc-          slot <- liftIO $ Metrics.claimInFlightSlot metricsHandle firstMsg.envelope.messageId-          restore (process currentInflight)-            `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)-          liftIO (Metrics.releaseInFlightSlot metricsHandle slot)-            `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)+          -- Halt: set the shared flag; do NOT throw (let the stream drain).+          for_ requestedExit $ \processorExit ->+            liftIO $ requestProcessorExit exitPublisher processorExit+    if not detailed+      then do+        currentInflight <- liftIO $ beginProcessing metricsHandle maxConc+        process currentInflight+      else Exception.mask $ \restore -> do+        currentInflight <- liftIO $ beginProcessing metricsHandle maxConc+        slot <- liftIO $ Metrics.claimInFlightSlot metricsHandle (NE.head batch).envelope.messageId+        restore (process currentInflight)+          `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)+        liftIO (Metrics.releaseInFlightSlot metricsHandle slot)+          `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)   where     isFailing :: AckDecision -> Bool     isFailing (AckDeadLetter _) = True

@@ -109,6 +109,7 @@ import Shibuya.Internal.Runner.Pause (PauseHandle, newPauseHandle) import Shibuya.Policy (Concurrency (..), OrderingPolicy (..)) import Shibuya.Prelude+import Shibuya.Telemetry.Correlation (AdapterTelemetry (traceCorrelation), TraceCorrelation) import Shibuya.Telemetry.Effect   ( Tracing,     addAttribute,@@ -116,21 +117,19 @@     addEvent,     recordException,     setStatus,-    withExtractedContext,-    withSpan',   )-import Shibuya.Telemetry.Propagation (extractTraceContext)+import Shibuya.Telemetry.Process (withProcessSpan) import Shibuya.Telemetry.Semantic   ( attrMessagingDestinationName,     attrMessagingMessageId,     attrMessagingOperation,+    attrMessagingOperationName,     attrMessagingSystem,     attrShibuyaAckDecision,     attrShibuyaDeadLetterReasonCode,     attrShibuyaInflightCount,     attrShibuyaInflightMax,     attrShibuyaPartition,-    consumerSpanArgs,     eventAckDecision,     eventHandlerCompleted,     eventHandlerStarted,@@ -308,6 +307,9 @@   -- Create bounded inbox (this is where inboxSize is used for backpressure)   inbox <- liftIO $ newBoundedInbox inboxSize +  -- The adapter's trace correlation policy, read once per processor.+  let !correlation = adapter.telemetry.traceCorrelation+   -- Signal when ingester completes (stream exhausted)   streamDoneVar <- liftIO $ newTVarIO False @@ -348,8 +350,8 @@       -- the detail ownership action from the basic specialization. Selecting a       -- partially applied worker retains a per-message closure.       case Metrics.metricsDetailLevel metricsHandle of-        Metrics.MetricsBasic -> runInIO $ processUntilDrained False metricsHandle procId ordering concurrency handler inbox streamDoneVar-        Metrics.MetricsDetailed _ -> runInIO $ processUntilDrained True metricsHandle procId ordering concurrency handler inbox streamDoneVar+        Metrics.MetricsBasic -> runInIO $ processUntilDrained False metricsHandle procId correlation ordering concurrency handler inbox streamDoneVar+        Metrics.MetricsDetailed _ -> runInIO $ processUntilDrained True metricsHandle procId correlation ordering concurrency handler inbox streamDoneVar       UIO.waitCatch ingesterAsync >>= \case         Left ingesterErr -> do           now <- getCurrentTime@@ -613,13 +615,14 @@   Bool ->   MetricsHandle ->   ProcessorId ->+  TraceCorrelation ->   OrderingPolicy ->   Concurrency ->   Handler es msg ->   Inbox (Ingested es msg) ->   TVar Bool ->   Eff es ()-processUntilDrained detailed metricsHandle procId ordering concurrency handler inbox streamDoneVar = do+processUntilDrained detailed metricsHandle procId correlation ordering concurrency handler inbox streamDoneVar = do   stopSignal <- liftIO newProcessorSignal   let exitPublisher = newProcessorExitPublisherWithWake stopSignal streamDoneVar @@ -633,13 +636,14 @@         HashMap.fromList           [ (attrMessagingSystem, toAttribute ("shibuya" :: Text)),             (attrMessagingDestinationName, toAttribute pidText),-            (attrMessagingOperation, toAttribute ("process" :: Text))+            (attrMessagingOperation, toAttribute ("process" :: Text)),+            (attrMessagingOperationName, toAttribute ("process" :: Text))           ]    withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do     let inboxStream = inboxToStream inbox streamDoneVar stopSignal         runProcessAction ingested =-          runInIO $ processOne detailed metricsHandle spanName constantFrameworkAttrs maxConc exitPublisher handler ingested+          runInIO $ processOne detailed metricsHandle correlation spanName constantFrameworkAttrs maxConc exitPublisher handler ingested          -- Restore normal interruptibility for handlers and finalizers while the         -- concurrent Streamly scheduler retains its inherited mask.@@ -694,6 +698,7 @@   (IOE :> es, Tracing :> es) =>   Bool ->   MetricsHandle ->+  TraceCorrelation ->   Text ->   HashMap.HashMap Text Attribute ->   Int ->@@ -701,146 +706,144 @@   Handler es msg ->   Ingested es msg ->   Eff es ()-processOne detailed metricsHandle spanName constantFrameworkAttrs maxConc exitPublisher handler ingested = do-  -- Extract parent context from message headers for distributed tracing-  let parentCtx = ingested.envelope.traceContext >>= extractTraceContext+processOne detailed metricsHandle correlation spanName constantFrameworkAttrs maxConc exitPublisher handler ingested =+  -- Link to (and, by policy, parent on) the producer's span from the+  -- message's trace headers. With tracing disabled the headers are not read.+  withProcessSpan correlation spanName ingested.envelope.traceContext $ \traceSpan -> do+    -- Build the messaging.* attribute set: framework defaults first,+    -- then the adapter's HashMap layered on top so adapter keys with+    -- the same name override the framework's default. The explicit+    -- 'HashMap.union' here is left-biased (left wins), which keeps+    -- the precedence rule local and obvious instead of relying on+    -- the order of repeated 'addAttribute' / 'addAttributes' calls+    -- against the underlying mutable Span.+    let messageId@(MessageId msgIdText) = ingested.envelope.messageId+        frameworkAttrs =+          HashMap.insert attrMessagingMessageId (toAttribute msgIdText) $+            case ingested.envelope.partition of+              Just p -> HashMap.insert attrShibuyaPartition (toAttribute p) constantFrameworkAttrs+              Nothing -> constantFrameworkAttrs+        mergedAttrs = HashMap.union ingested.envelope.attributes frameworkAttrs+    addAttributes traceSpan mergedAttrs -  withExtractedContext parentCtx $-    withSpan' spanName consumerSpanArgs $ \traceSpan -> do-      -- Build the messaging.* attribute set: framework defaults first,-      -- then the adapter's HashMap layered on top so adapter keys with-      -- the same name override the framework's default. The explicit-      -- 'HashMap.union' here is left-biased (left wins), which keeps-      -- the precedence rule local and obvious instead of relying on-      -- the order of repeated 'addAttribute' / 'addAttributes' calls-      -- against the underlying mutable Span.-      let messageId@(MessageId msgIdText) = ingested.envelope.messageId-          frameworkAttrs =-            HashMap.insert attrMessagingMessageId (toAttribute msgIdText) $-              case ingested.envelope.partition of-                Just p -> HashMap.insert attrShibuyaPartition (toAttribute p) constantFrameworkAttrs-                Nothing -> constantFrameworkAttrs-          mergedAttrs = HashMap.union ingested.envelope.attributes frameworkAttrs-      addAttributes traceSpan mergedAttrs+    status <- ingested.deliveryStatus+    case status of+      DeliverySuperseded -> do+        addEvent traceSpan $+          mkEvent+            eventAckDecision+            [(attrShibuyaAckDecision, OTel.toAttribute ("superseded" :: Text))]+        setStatus traceSpan OTel.Ok+      DeliveryCurrent -> do+        let process currentInflight = do+              addAttribute traceSpan attrShibuyaInflightCount currentInflight+              addAttribute traceSpan attrShibuyaInflightMax maxConc -      status <- ingested.deliveryStatus-      case status of-        DeliverySuperseded -> do-          addEvent traceSpan $-            mkEvent-              eventAckDecision-              [(attrShibuyaAckDecision, OTel.toAttribute ("superseded" :: Text))]-          setStatus traceSpan OTel.Ok-        DeliveryCurrent -> do-          let process currentInflight = do-                addAttribute traceSpan attrShibuyaInflightCount currentInflight-                addAttribute traceSpan attrShibuyaInflightMax maxConc+              -- Record handler start event+              addEvent traceSpan handlerStartedEvent -                -- Record handler start event-                addEvent traceSpan handlerStartedEvent+              -- Call handler and finalizer separately. A handler exception is+              -- substituted with immediate retry so an admitted delivery always receives a+              -- finalization decision. Superseded deliveries are excluded.+              --+              -- This is a *separate* per-message 'catchAny' from the one inside+              -- 'finalizeWithRetry' below. 0.7.1.0 used a single combined catch around+              -- handler+finalize, which skipped finalization when the handler threw;+              -- splitting them is what makes "always finalize admitted deliveries" hold. The extra+              -- per-message exception frame costs ~126 bytes/message (measured, EP-31+              -- S3: docs/plans/31-…-shared-per-message-…-allocation-regression.md) and is+              -- accepted as the price of the always-finalize guarantee — do not merge the+              -- two catches back together to reclaim it.+              handlerResult <-+                catchAny+                  (Right <$> handler (toMessage ingested))+                  ( \ex -> do+                      recordException traceSpan ex+                      pure (Left ex)+                  )+              let (decision, result) = case handlerResult of+                    Right d -> (d, Right d)+                    Left ex ->+                      ( AckRetry (RetryDelay 0),+                        Left $ HandlerException $ Text.pack $ show ex+                      ) -                -- Call handler and finalizer separately. A handler exception is-                -- substituted with immediate retry so an admitted delivery always receives a-                -- finalization decision. Superseded deliveries are excluded.-                ---                -- This is a *separate* per-message 'catchAny' from the one inside-                -- 'finalizeWithRetry' below. 0.7.1.0 used a single combined catch around-                -- handler+finalize, which skipped finalization when the handler threw;-                -- splitting them is what makes "always finalize admitted deliveries" hold. The extra-                -- per-message exception frame costs ~126 bytes/message (measured, EP-31-                -- S3: docs/plans/31-…-shared-per-message-…-allocation-regression.md) and is-                -- accepted as the price of the always-finalize guarantee — do not merge the-                -- two catches back together to reclaim it.-                handlerResult <--                  catchAny-                    (Right <$> handler (toMessage ingested))-                    ( \ex -> do-                        recordException traceSpan ex-                        pure (Left ex)-                    )-                let (decision, result) = case handlerResult of-                      Right d -> (d, Right d)-                      Left ex ->-                        ( AckRetry (RetryDelay 0),-                          Left $ HandlerException $ Text.pack $ show ex-                        )+              when (isLeft handlerResult) $+                addEvent traceSpan $+                  mkEvent+                    eventAckDecision+                    [(attrShibuyaAckDecision, OTel.toAttribute ("ack_retry" :: Text))] -                when (isLeft handlerResult) $+              finalizeResult <- finalizeWithRetry traceSpan ingested decision++              -- Record completion event and set status+              case finalizeResult of+                Left _ -> do                   addEvent traceSpan $                     mkEvent                       eventAckDecision-                      [(attrShibuyaAckDecision, OTel.toAttribute ("ack_retry" :: Text))]--                finalizeResult <- finalizeWithRetry traceSpan ingested decision+                      [(attrShibuyaAckDecision, OTel.toAttribute ("finalization_failed" :: Text))]+                  setStatus traceSpan $ OTel.Error $ finalizationFailureText msgIdText+                Right () -> case result of+                  Right decision' -> do+                    let decisionText = showAckDecision decision'+                    addEvent traceSpan $+                      mkEvent+                        eventHandlerCompleted+                        [(attrShibuyaAckDecision, OTel.toAttribute decisionText)]+                    addAttribute traceSpan attrShibuyaAckDecision decisionText -                -- Record completion event and set status-                case finalizeResult of-                  Left _ -> do+                    -- Set span status based on decision+                    case decision' of+                      AckOk -> setStatus traceSpan OTel.Ok+                      AckRetry _ -> setStatus traceSpan OTel.Ok+                      AckDeadLetter reason -> do+                        addAttribute+                          traceSpan+                          attrShibuyaDeadLetterReasonCode+                          (deadLetterCodeText (deadLetterReasonCode reason))+                        setStatus traceSpan $ OTel.Error $ renderDeadLetterReason reason+                      AckHalt reason ->+                        setStatus traceSpan $ OTel.Error $ showHaltReason reason+                  Left err -> do                     addEvent traceSpan $                       mkEvent                         eventAckDecision-                        [(attrShibuyaAckDecision, OTel.toAttribute ("finalization_failed" :: Text))]-                    setStatus traceSpan $ OTel.Error $ finalizationFailureText msgIdText-                  Right () -> case result of-                    Right decision' -> do-                      let decisionText = showAckDecision decision'-                      addEvent traceSpan $-                        mkEvent-                          eventHandlerCompleted-                          [(attrShibuyaAckDecision, OTel.toAttribute decisionText)]-                      addAttribute traceSpan attrShibuyaAckDecision decisionText--                      -- Set span status based on decision-                      case decision' of-                        AckOk -> setStatus traceSpan OTel.Ok-                        AckRetry _ -> setStatus traceSpan OTel.Ok-                        AckDeadLetter reason -> do-                          addAttribute-                            traceSpan-                            attrShibuyaDeadLetterReasonCode-                            (deadLetterCodeText (deadLetterReasonCode reason))-                          setStatus traceSpan $ OTel.Error $ renderDeadLetterReason reason-                        AckHalt reason ->-                          setStatus traceSpan $ OTel.Error $ showHaltReason reason-                    Left err -> do-                      addEvent traceSpan $-                        mkEvent-                          eventAckDecision-                          [(attrShibuyaAckDecision, OTel.toAttribute ("error" :: Text))]-                      setStatus traceSpan $ OTel.Error $ handlerErrorToText err--                let recordCursor = liftIO $ Metrics.recordAcknowledgedCursor metricsHandle ingested.envelope.partition ingested.envelope.cursor-                when detailed $ case (finalizeResult, result) of-                  (Right (), Right AckOk) -> recordCursor-                  (Right (), Right (AckDeadLetter _)) -> recordCursor-                  _ -> pure ()+                        [(attrShibuyaAckDecision, OTel.toAttribute ("error" :: Text))]+                    setStatus traceSpan $ OTel.Error $ handlerErrorToText err -                -- Decrement in-flight and update stats.-                liftIO $-                  case finalizeResult of-                    Left _ -> finishFinalizationFailure metricsHandle (finalizationFailureText msgIdText)-                    Right () -> finishProcessing metricsHandle (metricForResult result)+              let recordCursor = liftIO $ Metrics.recordAcknowledgedCursor metricsHandle ingested.envelope.partition ingested.envelope.cursor+              when detailed $ case (finalizeResult, result) of+                (Right (), Right AckOk) -> recordCursor+                (Right (), Right (AckDeadLetter _)) -> recordCursor+                _ -> pure () -                -- Handle halt (set flag, don't throw - let stream drain)+              -- Decrement in-flight and update stats.+              liftIO $                 case finalizeResult of-                  Left _ ->-                    liftIO $-                      requestProcessorExit exitPublisher (ProcessorFailed (finalizationFailureText msgIdText) (Just messageId))-                  Right () -> case result of-                    Right (AckHalt reason) ->-                      liftIO $ requestProcessorExit exitPublisher (ProcessorHalted reason)-                    _ -> pure ()-          if not detailed-            then do-              currentInflight <- liftIO $ beginProcessing metricsHandle maxConc-              process currentInflight-            else Exception.mask $ \restore -> do-              currentInflight <- liftIO $ beginProcessing metricsHandle maxConc-              slot <- liftIO $ Metrics.claimInFlightSlot metricsHandle messageId-              restore (process currentInflight)-                `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)-              liftIO (Metrics.releaseInFlightSlot metricsHandle slot)-                `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)+                  Left _ -> finishFinalizationFailure metricsHandle (finalizationFailureText msgIdText)+                  Right () -> finishProcessing metricsHandle (metricForResult result)++              -- Handle halt (set flag, don't throw - let stream drain)+              case finalizeResult of+                Left _ ->+                  liftIO $+                    requestProcessorExit exitPublisher (ProcessorFailed (finalizationFailureText msgIdText) (Just messageId))+                Right () -> case result of+                  Right (AckHalt reason) ->+                    liftIO $ requestProcessorExit exitPublisher (ProcessorHalted reason)+                  _ -> pure ()+        if not detailed+          then do+            currentInflight <- liftIO $ beginProcessing metricsHandle maxConc+            process currentInflight+          else Exception.mask $ \restore -> do+            currentInflight <- liftIO $ beginProcessing metricsHandle maxConc+            slot <- liftIO $ Metrics.claimInFlightSlot metricsHandle messageId+            restore (process currentInflight)+              `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)+            liftIO (Metrics.releaseInFlightSlot metricsHandle slot)+              `Exception.onException` liftIO (Metrics.abandonInFlightSlot metricsHandle slot)   where     isLeft :: Either a b -> Bool     isLeft (Left _) = True

@@ -31,9 +31,17 @@      -- * Semantic Conventions     module Shibuya.Telemetry.Semantic,++    -- * Trace Correlation+    module Shibuya.Telemetry.Correlation,++    -- * Processing Span+    module Shibuya.Telemetry.Process,   ) where +import Shibuya.Telemetry.Correlation import Shibuya.Telemetry.Effect+import Shibuya.Telemetry.Process import Shibuya.Telemetry.Propagation import Shibuya.Telemetry.Semantic

@@ -0,0 +1,171 @@+-- | How a processing span relates to the span that created its message.+--+-- A message's /creation context/ is the span identity its producer encoded+-- in the W3C trace headers the adapter extracted into+-- 'Shibuya.Core.Types.Envelope'@.traceContext@. Every processing span links to+-- the creation context when one decodes, as the OpenTelemetry messaging+-- conventions recommend. Whether the span is also a /child/ of that context+-- is the adapter's declared 'TraceCorrelation' policy.+module Shibuya.Telemetry.Correlation+  ( -- * Policy+    TraceCorrelation (..),+    AdapterTelemetry (traceCorrelation),+    defaultAdapterTelemetry,++    -- * Links+    creationLink,+    creationLinkWith,+    consumerSpanArgsLinkedTo,++    -- * Batch links+    maxBatchLinks,+    BatchLinks (..),+    collectBatchLinks,+    collectBatchLinksBy,+  )+where++import Control.DeepSeq (NFData)+import Data.HashMap.Strict qualified as HashMap+import GHC.Generics (Generic)+import OpenTelemetry.Attributes (AttributeMap, toAttribute)+import OpenTelemetry.Trace.Core (NewLink (..), SpanArguments (..), SpanContext, SpanKind (..))+import Shibuya.Core.Types (MessageId (..), TraceHeaders)+import Shibuya.Telemetry.Propagation (extractTraceContext)+import Shibuya.Telemetry.Semantic (attrMessagingMessageId)++-- | How a processing span relates to the span that created the message.+--+-- Both policies link the processing span to the creation context when the+-- message's trace headers decode. Without usable headers neither policy adds+-- a link, and the span is opened in the ambient context exactly as before.+data TraceCorrelation+  = -- | Child of the creation context, and linked to it: one connected trace+    -- from producer to consumer. Suits queues consumed soon after production.+    ParentAndLink+  | -- | A new root trace, related to the creation context only by a link.+    -- Suits replayed event streams, long-delayed or retried jobs, fan-out to+    -- many consumers, and tail-sampling deployments: the consumer trace is+    -- sampled independently of the producer's.+    LinkOnly+  deriving stock (Eq, Show, Generic, Enum, Bounded)+  deriving anyclass (NFData)++-- | Telemetry options an adapter declares in its 'Shibuya.Adapter.Adapter'.+--+-- Build a value from 'defaultAdapterTelemetry' with record update:+--+-- @+-- defaultAdapterTelemetry {traceCorrelation = LinkOnly}+-- @+--+-- The constructor is not exported, so options can be added later as fields+-- with defaults without breaking any adapter. The framework reads it once per+-- processor, never per message.+data AdapterTelemetry = AdapterTelemetry+  { -- | Parent-versus-link policy for this adapter's processing spans.+    traceCorrelation :: !TraceCorrelation+  }+  deriving stock (Eq, Show, Generic)+  deriving anyclass (NFData)++-- | @traceCorrelation = 'ParentAndLink'@.+defaultAdapterTelemetry :: AdapterTelemetry+defaultAdapterTelemetry = AdapterTelemetry {traceCorrelation = ParentAndLink}++-- | Decode trace headers into a link to the message's creation context.+--+-- 'Nothing' when the headers are absent or malformed. The link carries no+-- attributes (the shared empty map), holds an evaluated+-- 'OpenTelemetry.Trace.Core.SpanContext', and keeps no reference to the+-- header list.+creationLink :: TraceHeaders -> Maybe NewLink+creationLink = creationLinkWith HashMap.empty+{-# INLINE creationLink #-}++-- | As 'creationLink', with attributes describing the linked message (for+-- example its @messaging.message.id@ on a batch span). The caller supplies an+-- already evaluated map.+creationLinkWith :: AttributeMap -> TraceHeaders -> Maybe NewLink+creationLinkWith attrs headers = case extractTraceContext headers of+  Nothing -> Nothing+  Just ctx ->+    let !link = NewLink {linkContext = ctx, linkAttributes = attrs}+     in Just link+{-# INLINE creationLinkWith #-}++-- | Consumer-kind span arguments carrying the given links. The list must be+-- fully evaluated by the caller, so the span references nothing else.+--+-- 'Shibuya.Telemetry.Semantic.consumerSpanArgs' remains the constant for+-- spans without links.+consumerSpanArgsLinkedTo :: [NewLink] -> SpanArguments+consumerSpanArgsLinkedTo links =+  SpanArguments+    { kind = Consumer,+      attributes = HashMap.empty,+      links = links,+      startTime = Nothing+    }+{-# INLINE consumerSpanArgsLinkedTo #-}++-- | Maximum links Shibuya places on one batch span: the OpenTelemetry SDK's+-- default per-span link limit. A lower configured SDK limit+-- (@OTEL_SPAN_LINK_COUNT_LIMIT@) can still drop supplied links on its own.+maxBatchLinks :: Int+maxBatchLinks = 128++-- | The creation-context links collected for one batch.+data BatchLinks = BatchLinks+  { -- | One link per member with a decodable creation context, in member+    -- order, each carrying that member's @messaging.message.id@. Fully+    -- evaluated; at most 'maxBatchLinks' long.+    links :: ![NewLink],+    -- | Members with a decodable creation context beyond 'maxBatchLinks'.+    -- Absent and malformed contexts are never counted.+    dropped :: !Int+  }++-- | 'collectBatchLinksBy' over @(message id, trace headers)@ pairs.+collectBatchLinks :: [(MessageId, Maybe TraceHeaders)] -> BatchLinks+collectBatchLinks = collectBatchLinksBy fst snd+{-# INLINE collectBatchLinks #-}++-- | Collect one link per member that has a decodable creation context, in+-- order, each carrying @messaging.message.id@. At most 'maxBatchLinks' links+-- are stored. Decoding continues past the bound so that 'dropped' counts the+-- valid omitted contexts exactly, but no link or attribute map is built for+-- them.+--+-- One strict pass with strict accumulators. A member whose raw headers equal+-- the previous member's reuses that member's decoded context. Members that+-- share a context still get separate links, because the links carry+-- different message identifiers. The result references only evaluated span+-- contexts and identifier text, never the members.+collectBatchLinksBy :: (Foldable t) => (a -> MessageId) -> (a -> Maybe TraceHeaders) -> t a -> BatchLinks+collectBatchLinksBy messageIdOf headersOf members =+  case foldl' step (Acc 0 [] 0 Nothing Nothing) members of+    Acc _ collected omitted _ _ -> BatchLinks {links = reverse collected, dropped = omitted}+  where+    step acc@(Acc count collected omitted previousHeaders previousContext) member =+      case headersOf member of+        Nothing -> acc+        Just headers ->+          let !context =+                if Just headers == previousHeaders+                  then previousContext+                  else extractTraceContext headers+           in case context of+                Nothing -> Acc count collected omitted (Just headers) context+                Just spanContext+                  | count < maxBatchLinks ->+                      let MessageId idText = messageIdOf member+                          !attrs = HashMap.singleton attrMessagingMessageId $! toAttribute idText+                          !link = NewLink {linkContext = spanContext, linkAttributes = attrs}+                       in Acc (count + 1) (link : collected) omitted (Just headers) context+                  | otherwise -> Acc count collected (omitted + 1) (Just headers) context+{-# INLINE collectBatchLinksBy #-}++-- Strict fold state: links so far, reversed, with their count; the overflow+-- count; and the previous member's raw headers and decoded context.+data Acc = Acc !Int ![NewLink] !Int !(Maybe TraceHeaders) !(Maybe SpanContext)

@@ -25,6 +25,7 @@     getTracer,     isTracingEnabled,     withExtractedContext,+    withRootContext,      -- * Re-exports     OTel.Span,@@ -294,6 +295,24 @@           (Ctx.attachContext newContext)           Ctx.detachContext           (\_token -> runInIO action)++-- | Run an action with an empty OpenTelemetry context attached, so a span+-- opened inside it is a root in a new trace. The previous context is restored+-- afterwards, on normal exit, exception and asynchronous cancellation.+-- A no-op when tracing is disabled.+withRootContext ::+  (Tracing :> es, IOE :> es) =>+  Eff es a ->+  Eff es a+withRootContext action = do+  TracingRep {tracingEnabled} <- getStaticRep+  if not tracingEnabled+    then action+    else withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->+      bracket+        (Ctx.attachContext Ctx.empty)+        Ctx.detachContext+        (\_token -> runInIO action)  -------------------------------------------------------------------------------- -- Internal Helpers

@@ -0,0 +1,120 @@+-- | The Consumer-kind processing span for one message.+--+-- Shibuya's own runner opens every per-message processing span through+-- 'withProcessSpan'. It is exported so code that drains an adapter outside+-- the runner opens the same span, with the same link and parent rule, instead+-- of a hand-written copy that can drift.+--+-- The rule, decided once per call:+--+-- 1. Tracing disabled: the body receives the shared non-recording sentinel+--    span. The trace headers are not inspected.+-- 2. Tracing enabled, no usable creation context (headers absent or+--    malformed): the span is opened in the ambient context, with no link,+--    exactly as Shibuya did before trace correlation policies existed.+-- 3. Tracing enabled, the headers decode: the span links to the creation+--    context. Under 'ParentAndLink' it is also a child of it; under+--    'LinkOnly' it is a root in a new trace. The previous thread context is+--    restored afterwards, also on exceptions and cancellation.+--+-- In case 3 neither policy preserves an enclosing ambient operation: it is+-- neither the parent nor linked. These helpers are for /independent/ message+-- processing. A caller that processes a message inside a meaningful+-- enclosing operation, and wants that relationship recorded, must open its+-- own span rather than use these helpers.+--+-- The functions set no attributes and record no outcome; both remain the+-- caller's.+module Shibuya.Telemetry.Process+  ( withProcessSpan,+    withProcessSpanIO,+  )+where++import Control.Exception (bracket)+import Data.Text (Text)+import Effectful (Eff, IOE, Limit (..), Persistence (..), UnliftStrategy (..), withEffToIO, (:>))+import OpenTelemetry.Context qualified as Ctx+import OpenTelemetry.Context.ThreadLocal qualified as Ctx+import OpenTelemetry.Trace.Core (NewLink (..), Span, SpanArguments, Tracer)+import OpenTelemetry.Trace.Core qualified as OTel+import Shibuya.Core.Types (TraceHeaders)+import Shibuya.Telemetry.Correlation (TraceCorrelation (..), consumerSpanArgsLinkedTo, creationLink)+import Shibuya.Telemetry.Effect (Tracing, getTracer, isTracingEnabled, withSpan')+import Shibuya.Telemetry.Semantic (consumerSpanArgs)++-- | Open the Consumer-kind processing span for one message.+--+-- Linked to the creation context in the headers when one decodes; a child of+-- it under 'ParentAndLink'; a root under 'LinkOnly'. Without usable headers+-- the span is opened in the ambient context, as before. With tracing+-- disabled the body receives the sentinel span and no header is decoded.+-- With valid headers a meaningful ambient operation is neither parented nor+-- linked: callers opt into independent message processing by using this+-- helper (see the module documentation).+withProcessSpan ::+  (Tracing :> es, IOE :> es) =>+  TraceCorrelation ->+  -- | Full span name, e.g. @processSpanName destination@.+  Text ->+  -- | The message's trace headers ('Shibuya.Core.Types.Envelope'@.traceContext@).+  Maybe TraceHeaders ->+  (Span -> Eff es a) ->+  Eff es a+withProcessSpan policy name headers body = do+  enabled <- isTracingEnabled+  if not enabled+    then withSpan' name consumerSpanArgs body+    else case processSpanPlan policy headers of+      Nothing -> withSpan' name consumerSpanArgs body+      Just (context, args) -> do+        tracer <- getTracer+        -- One unlift covers both the context attachment and the span.+        withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO ->+          inContext context $ OTel.inSpan' tracer name args (runInIO . body)+{-# INLINE withProcessSpan #-}++-- | The same span for callers that hold a 'Tracer' outside the 'Tracing'+-- effect. It always behaves as tracing enabled; pass a tracer from a no-op+-- provider to disable it.+withProcessSpanIO ::+  Tracer ->+  TraceCorrelation ->+  -- | Full span name, e.g. @processSpanName destination@.+  Text ->+  Maybe TraceHeaders ->+  (Span -> IO a) ->+  IO a+withProcessSpanIO tracer policy name headers body =+  case processSpanPlan policy headers of+    Nothing -> OTel.inSpan' tracer name consumerSpanArgs body+    Just (context, args) -> inContext context $ OTel.inSpan' tracer name args body+{-# INLINE withProcessSpanIO #-}++-- | The decision shared by both entry points. 'Nothing': open the span in+-- the ambient context without links. @Just (context, args)@: attach+-- @context@ (the creation context for 'ParentAndLink', the empty context for+-- 'LinkOnly') and open the span with @args@, which carry the one link.+--+-- The headers are decoded once; the link, its one-element list and the+-- arguments are evaluated before the span exists, so the span references+-- neither the headers nor anything else of the message.+processSpanPlan :: TraceCorrelation -> Maybe TraceHeaders -> Maybe (Ctx.Context, SpanArguments)+processSpanPlan policy headers = case headers of+  Nothing -> Nothing+  Just hs -> case creationLink hs of+    Nothing -> Nothing+    Just link ->+      let !args = consumerSpanArgsLinkedTo [link]+          !context = case policy of+            ParentAndLink -> Ctx.insertSpan (OTel.wrapSpanContext link.linkContext) Ctx.empty+            LinkOnly -> Ctx.empty+       in Just (context, args)+{-# INLINE processSpanPlan #-}++-- | Attach a context for the duration of an action, restoring the previous+-- one on normal exit, exception and asynchronous cancellation.+inContext :: Ctx.Context -> IO a -> IO a+inContext context action =+  bracket (Ctx.attachContext context) Ctx.detachContext (const action)+{-# INLINE inContext #-}

@@ -15,6 +15,8 @@     attrMessagingMessageId,     attrMessagingDestinationName,     attrMessagingOperation,+    attrMessagingOperationName,+    attrMessagingBatchMessageCount,      -- * Attribute Keys (Shibuya-specific)     attrShibuyaProcessorId,@@ -26,6 +28,7 @@     attrShibuyaBatchKey,     attrShibuyaBatchSize,     attrShibuyaBatchTrigger,+    attrShibuyaBatchLinksDropped,      -- * Event Names     eventHandlerStarted,@@ -57,19 +60,21 @@ -- Span Names -------------------------------------------------------------------------------- --- | Span name for processing a single message.+-- | Span name for processing a message or a batch: @\"process <destination>\"@,+-- for example @\"process orders\"@. ----- Follows the OpenTelemetry messaging-spans recommendation that consumer--- spans be named @\"<destination> <operation>\"@ — for Shibuya the--- destination is the processor id and the operation is @\"process\"@,--- yielding e.g. @\"shibuya-consumer process\"@.+-- The OpenTelemetry messaging conventions say the span name "SHOULD be+-- @{messaging.operation.name} {destination}@". Shibuya's operation name is+-- @process@ and the destination is the processor id. The destination stays+-- the processor id even when an adapter overrides+-- @messaging.destination.name@ per message, so the name remains a+-- per-processor constant. ----- This intentionally departs from--- @hs-opentelemetry-instrumentation-hw-kafka-client@'s--- @\"process <topic>\"@ order in favor of the spec's order; see--- @docs/plans/2-align-opentelemetry-semantic-conventions.md@.+-- Before 0.13.0.0 the name was @\"<destination> process\"@, following an+-- earlier reading of the conventions+-- (@docs/plans/2-align-opentelemetry-semantic-conventions.md@). processSpanName :: Text -> Text-processSpanName destination = destination <> " process"+processSpanName destination = "process " <> destination  -- | Span name for the ingester lifecycle. ingestSpanName :: Text@@ -104,6 +109,17 @@ attrMessagingOperation :: Text attrMessagingOperation = unkey Sem.messaging_operation_type +-- | @messaging.operation.name@: the system-specific operation name, Required+-- by the messaging conventions. Shibuya's processing spans set @\"process\"@.+-- ('attrMessagingOperation' is the operation /type/ key.)+attrMessagingOperationName :: Text+attrMessagingOperationName = unkey Sem.messaging_operation_name++-- | @messaging.batch.message_count@: the number of messages in a batch, set+-- on batch processing spans.+attrMessagingBatchMessageCount :: Text+attrMessagingBatchMessageCount = unkey Sem.messaging_batch_messageCount+ -------------------------------------------------------------------------------- -- Attribute Keys (Shibuya-specific) --@@ -164,6 +180,11 @@ -- | Why the batch was emitted: size, timeout, or flush (@shibuya.batch.trigger@). attrShibuyaBatchTrigger :: Text attrShibuyaBatchTrigger = "shibuya.batch.trigger"++-- | Members of a batch with a valid creation context that were not linked+-- because the batch span reached its link bound. Set only when non-zero.+attrShibuyaBatchLinksDropped :: Text+attrShibuyaBatchLinksDropped = "shibuya.batch.links.dropped"  -------------------------------------------------------------------------------- -- Event Names

@@ -12,7 +12,7 @@ import Control.Monad (forM, replicateM_, unless, void) import Data.List (isInfixOf) import Effectful (IOE, liftIO, runEff)-import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry) import Shibuya.App   ( AppConfig (..),     ProcessorId (..),@@ -113,7 +113,8 @@   Adapter     { adapterName = "gc-regression:finite",       source = Stream.fromList (map message [1 .. count]),-      shutdown = pure ()+      shutdown = pure (),+      telemetry = defaultAdapterTelemetry     }  failedSource :: Adapter '[Tracing, IOE] String@@ -121,7 +122,8 @@   Adapter     { adapterName = "gc-regression:failed",       source = Stream.fromEffect (liftIO (throwIO (userError "the source failed on purpose"))),-      shutdown = pure ()+      shutdown = pure (),+      telemetry = defaultAdapterTelemetry     }  -- | One message, then a quiet queue: the source stays alive but produces nothing.@@ -130,7 +132,8 @@   Adapter     { adapterName = "gc-regression:one-then-idle",       source = Stream.unfoldrM step (0 :: Int),-      shutdown = pure ()+      shutdown = pure (),+      telemetry = defaultAdapterTelemetry     }   where     step 0 = pure (Just (message 0, 1))

@@ -9,7 +9,7 @@ import Control.Exception (SomeException, displayException, try) import Control.Monad (replicateM_) import Effectful (liftIO, runEff)-import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry) import Shibuya.App   ( AppConfig (..),     ProcessorId (..),@@ -56,7 +56,8 @@             source = Stream.fromEffect $ liftIO $ do               threadDelay 60_000_000               fail "the idle adapter unexpectedly woke during the GC probe",-            shutdown = pure ()+            shutdown = pure (),+            telemetry = defaultAdapterTelemetry           }   result <-     runApp

@@ -0,0 +1,305 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Trace retention probes.+--+-- This program owns its process (see ADR 0001): weak-pointer and live-bytes+-- assertions depend on major collections and on nothing else in the process+-- holding references, so they cannot share a process with the hspec suite.+--+-- Check A: a finished span, held by an exporter, does not keep the message+-- that produced it alive.+--+-- Check B: live heap after processing ten times as many traced messages is+-- not materially larger.+--+-- Each check prints one PASS/FAIL line; the program exits non-zero if any+-- check fails.+module Main (main) where++import Control.Exception (evaluate)+import Control.Monad (forM, unless, void)+import Data.ByteString (ByteString)+import Data.ByteString qualified as BS+import Data.ByteString.Char8 qualified as BS8+import Data.IORef (IORef, atomicModifyIORef', mkWeakIORef, newIORef, readIORef)+import Data.List.NonEmpty (NonEmpty)+import Data.Maybe (isJust)+import Data.Text qualified as Text+import Data.Word (Word8)+import Effectful (Eff, IOE, liftIO, runEff, (:>))+import Effectful.Exception qualified as Exception+import GHC.Stats (GCDetails (..), RTSStats (..), getRTSStats)+import Numeric (showHex)+import OpenTelemetry.Attributes (emptyAttributes)+import OpenTelemetry.Exporter.InMemory (inMemoryListExporter)+import OpenTelemetry.Processor.Span (FlushResult (..), ShutdownResult (..), SpanProcessor (..))+import OpenTelemetry.Trace.Core qualified as OTel+import Shibuya.Adapter (Adapter (..), AdapterTelemetry (traceCorrelation), TraceCorrelation (..), defaultAdapterTelemetry)+import Shibuya.App+  ( QueueProcessor (..),+    ackAllOk,+    defaultAppConfig,+    defaultBatchConfig,+    runApp,+    stopApp,+    waitApp,+  )+import Shibuya.Batch (BatchAck, BatchConfig (..), BatchInfo)+import Shibuya.Core.Ack (AckDecision (..))+import Shibuya.Core.AckHandle (AckHandle (..))+import Shibuya.Core.Ingested (Ingested, mkIngested)+import Shibuya.Core.Metrics (ProcessorId (..))+import Shibuya.Core.Types (Envelope (..), MessageId (..), TraceHeaders, mkEnvelope)+import Shibuya.Policy (Concurrency (..), OrderingPolicy (..))+import Shibuya.Telemetry.Effect (Tracing, runTracing)+import Streamly.Data.Stream qualified as Stream+import System.Exit (exitFailure)+import System.Mem (performMajorGC)+import System.Mem.Weak (Weak, deRefWeak)+import System.Timeout (timeout)++-- | 'BatchDistinct' is a batch whose members each carry a different producer+-- context, so the batch span holds one link per member up to the bound.+data Shape = SingleSerial | SingleAsync8 | Batch | BatchDistinct+  deriving stock (Eq, Show)++-- | Batch cases are labelled by shape alone: batch spans take no policy.+caseName :: Shape -> TraceCorrelation -> String+caseName Batch _ = shapeName Batch+caseName BatchDistinct _ = shapeName BatchDistinct+caseName shape policy = shapeName shape <> "/" <> show policy++shapeName :: Shape -> String+shapeName = \case+  SingleSerial -> "single-serial"+  SingleAsync8 -> "single-async8"+  Batch -> "batch"+  BatchDistinct -> "batch/distinct-contexts"++-- | A message payload. The 'token' is the identity the weak pointer watches:+-- a weak pointer keyed on an 'IORef' follows its underlying mutable cell,+-- which the optimiser cannot copy or unbox, so a dead weak pointer means the+-- payload record really is unreachable. The 'blob' gives a retention bug a+-- visible size.+data Payload = Payload+  { blob :: !ByteString,+    token :: !(IORef ())+  }++main :: IO ()+main = do+  results <-+    sequence+      [ checkPayloadsCollected SingleSerial ParentAndLink,+        checkPayloadsCollected SingleSerial LinkOnly,+        checkPayloadsCollected Batch ParentAndLink,+        checkPayloadsCollected BatchDistinct ParentAndLink,+        checkLiveBytesFlat SingleSerial ParentAndLink,+        checkLiveBytesFlat SingleSerial LinkOnly,+        checkLiveBytesFlat SingleAsync8 ParentAndLink,+        checkLiveBytesFlat Batch ParentAndLink,+        checkLiveBytesFlat BatchDistinct ParentAndLink+      ]+  unless (and results) exitFailure++--------------------------------------------------------------------------------+-- Check A+--------------------------------------------------------------------------------++payloadCount :: Int+payloadCount = 8++checkPayloadsCollected :: Shape -> TraceCorrelation -> IO Bool+checkPayloadsCollected shape policy = bounded label $ do+  (processor, exported) <- inMemoryListExporter+  provider <- OTel.createTracerProvider [processor] OTel.emptyTracerProviderOptions+  let tracer = OTel.makeTracer provider (OTel.InstrumentationLibrary "trace-retention-test" "" "" emptyAttributes) OTel.tracerOptions+  batches <- newIORef (0 :: Int)+  weaks <- runPayloadCase tracer shape policy batches+  performMajorGC+  performMajorGC+  alive <- length . filter isJust <$> traverse deRefWeak weaks+  -- Read the exporter only after collection, so it is demonstrably still+  -- reachable (and still holds every finished span) while the payloads die.+  spans <- readIORef exported+  let consumerSpans = length [() | sp <- spans, OTel.spanKind sp == OTel.Consumer]+  batchCount <- readIORef batches+  let expectedConsumer = if isBatch shape then batchCount else payloadCount+  void $ OTel.shutdownTracerProvider provider (Just 1_000_000)+  report+    label+    (alive == 0 && consumerSpans == expectedConsumer && expectedConsumer > 0)+    ( show consumerSpans+        <> " Consumer spans (expected "+        <> show expectedConsumer+        <> "), "+        <> show (length spans)+        <> " exported in total, "+        <> show (payloadCount - alive)+        <> " of "+        <> show payloadCount+        <> " payloads collected"+    )+  where+    label = caseName shape policy <> ": finished spans do not retain payloads"++-- | Build the messages, process them to completion, and return only the weak+-- pointers. Everything else this case allocated, including the application+-- handle (whose adapter still references the message list), is unreachable+-- once it returns.+runPayloadCase :: OTel.Tracer -> Shape -> TraceCorrelation -> IORef Int -> IO [Weak (IORef ())]+runPayloadCase tracer shape policy batches = do+  built <- forM [1 .. payloadCount] $ \i -> do+    token <- newIORef ()+    weak <- mkWeakToken token+    let payload = Payload {blob = BS.replicate (1024 * 1024) (fromIntegral i :: Word8), token}+    pure (weak, message shape i payload)+  let weaks = map fst built+      adapter =+        Adapter+          { adapterName = "trace-retention:payload",+            source = Stream.fromList (map snd built),+            shutdown = pure (),+            telemetry = defaultAdapterTelemetry+          }+  -- Force the weak pointers out of their tuples now; returning the lazy+  -- @map fst built@ would keep the whole message list reachable.+  mapM_ evaluate weaks+  runEff $ runTracing tracer $ runToCompletion shape policy batches adapter+  pure weaks+{-# NOINLINE runPayloadCase #-}++mkWeakToken :: IORef () -> IO (Weak (IORef ()))+mkWeakToken ref = mkWeakIORef ref (pure ())++--------------------------------------------------------------------------------+-- Check B+--------------------------------------------------------------------------------++smallRun :: Int+smallRun = 20_000++checkLiveBytesFlat :: Shape -> TraceCorrelation -> IO Bool+checkLiveBytesFlat shape policy = bounded label $ do+  -- No exporter retains finished spans; a single discarding processor is+  -- needed only because hs-opentelemetry makes spans non-recording when a+  -- provider has no processors at all.+  provider <- OTel.createTracerProvider [discardProcessor] OTel.emptyTracerProviderOptions+  let tracer = OTel.makeTracer provider (OTel.InstrumentationLibrary "trace-retention-test" "" "" emptyAttributes) OTel.tracerOptions+  runSmallMessages tracer shape policy smallRun+  before <- liveBytesAfterGC+  runSmallMessages tracer shape policy (10 * smallRun)+  after <- liveBytesAfterGC+  void $ OTel.shutdownTracerProvider provider (Just 1_000_000)+  let allowed = before + 1024 * 1024 + before `div` 10+  report+    label+    (after <= allowed)+    ( "N="+        <> show smallRun+        <> " -> "+        <> show (10 * smallRun)+        <> ", live bytes "+        <> show before+        <> " -> "+        <> show after+        <> " (limit "+        <> show allowed+        <> ")"+    )+  where+    label = caseName shape policy <> ": live bytes flat"++runSmallMessages :: OTel.Tracer -> Shape -> TraceCorrelation -> Int -> IO ()+runSmallMessages tracer shape policy n = do+  batches <- newIORef (0 :: Int)+  let adapter =+        Adapter+          { adapterName = "trace-retention:flat",+            -- A generator, not a list, so the source itself retains nothing.+            source = fmap (\i -> message shape i (Text.pack (show i))) (Stream.enumerateFromTo 1 n),+            shutdown = pure (),+            telemetry = defaultAdapterTelemetry+          }+  runEff $ runTracing tracer $ runToCompletion shape policy batches adapter+{-# NOINLINE runSmallMessages #-}++liveBytesAfterGC :: IO Integer+liveBytesAfterGC = do+  performMajorGC+  performMajorGC+  stats <- getRTSStats+  pure $ fromIntegral (gcdetails_live_bytes (gc stats))++discardProcessor :: SpanProcessor+discardProcessor =+  SpanProcessor+    { spanProcessorOnStart = \_ _ -> pure (),+      spanProcessorOnEnd = \_ -> pure (),+      spanProcessorShutdown = pure ShutdownSuccess,+      spanProcessorForceFlush = pure FlushSuccess+    }++--------------------------------------------------------------------------------+-- Shared+--------------------------------------------------------------------------------++-- | Run one processor over the adapter until its source is exhausted, then+-- stop the application so no runner thread outlives the call.+runToCompletion ::+  (IOE :> es, Tracing :> es) =>+  Shape ->+  TraceCorrelation ->+  IORef Int ->+  Adapter es msg ->+  Eff es ()+runToCompletion shape policy batches adapter0 = do+  -- Batch spans take no policy; single-message spans follow the adapter's.+  let adapter = adapter0 {telemetry = defaultAdapterTelemetry {traceCorrelation = policy}}+      processor = case shape of+        SingleSerial -> QueueProcessor adapter (\_ -> pure AckOk) Unordered Serial+        SingleAsync8 -> QueueProcessor adapter (\_ -> pure AckOk) Unordered (Async 8)+        Batch -> batching+        BatchDistinct -> batching+      batching = BatchingProcessor adapter (countBatch batches) defaultBatchConfig {batchSize = 100} Unordered Serial+  runApp defaultAppConfig [(ProcessorId "trace-retention", processor)] >>= \case+    Left err -> liftIO $ fail ("runApp failed: " <> show err)+    Right app -> Exception.finally (waitApp app) (stopApp app)++countBatch :: (IOE :> es) => IORef Int -> BatchInfo -> NonEmpty a -> Eff es BatchAck+countBatch batches _ _ = do+  liftIO $ atomicModifyIORef' batches (\n -> (n + 1, ()))+  pure ackAllOk++isBatch :: Shape -> Bool+isBatch = \case+  Batch -> True+  BatchDistinct -> True+  _ -> False++message :: (IOE :> es) => Shape -> Int -> msg -> Ingested es msg+message shape i payload =+  let headers = if shape == BatchDistinct then distinctTraceHeaders i else traceHeaders+      envelope = (mkEnvelope (MessageId (Text.pack (show i))) payload) {traceContext = Just headers}+   in mkIngested envelope (AckHandle (\_ -> pure ()))++-- | A valid W3C context unique to message @i@.+distinctTraceHeaders :: Int -> TraceHeaders+distinctTraceHeaders i =+  [("traceparent", BS8.pack ("00-" <> hex 32 (i + 0x1000) <> "-" <> hex 16 (i + 0x1000) <> "-01"))]+  where+    hex width n = let digits = showHex n "" in replicate (width - length digits) '0' <> digits++traceHeaders :: TraceHeaders+traceHeaders = [("traceparent", "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01")]++bounded :: String -> IO Bool -> IO Bool+bounded label check =+  timeout 60_000_000 check >>= \case+    Just ok -> pure ok+    Nothing -> report label False "timed out after 60 seconds"++report :: String -> Bool -> String -> IO Bool+report label ok detail = do+  putStrLn ((if ok then "PASS: " else "FAIL: ") <> label <> " (" <> detail <> ")")+  pure ok

@@ -18,6 +18,8 @@ import Shibuya.Runner.PauseSpec qualified as Pause import Shibuya.Runner.SupervisedSpec qualified import Shibuya.RunnerSpec qualified+import Shibuya.Telemetry.BatchCorrelationSpec qualified+import Shibuya.Telemetry.CorrelationSpec qualified import Shibuya.Telemetry.EffectSpec qualified import Shibuya.Telemetry.LifecycleSpec qualified import Shibuya.Telemetry.PropagationSpec qualified@@ -44,5 +46,7 @@   Shibuya.Runner.SupervisedSpec.spec   Shibuya.Telemetry.LifecycleSpec.spec   Shibuya.Telemetry.EffectSpec.spec+  Shibuya.Telemetry.CorrelationSpec.spec+  Shibuya.Telemetry.BatchCorrelationSpec.spec   Shibuya.Telemetry.PropagationSpec.spec   Shibuya.Telemetry.SemanticSpec.spec

@@ -8,7 +8,7 @@ import Data.Text qualified as Text import Data.Time (UTCTime (..), fromGregorian) import Effectful (Eff, IOE, liftIO, runEff, (:>))-import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry) import Shibuya.Adapter.Mock (TrackingAck (..), newTrackingAck, trackingAckHandle) import Shibuya.App   ( AppConfig (..),@@ -179,7 +179,7 @@ -- Finite adapter: ends as soon as its list is exhausted. listAdapter' :: [Ingested es String] -> Adapter es String listAdapter' messages =-  Adapter {adapterName = "test:batch", source = Stream.fromList messages, shutdown = pure ()}+  Adapter {adapterName = "test:batch", source = Stream.fromList messages, shutdown = pure (), telemetry = defaultAdapterTelemetry}  -- Gated adapter: emits the messages, then blocks until 'shutdown' opens the gate, -- so a partial batch stays pending until stopApp triggers the flush. Implemented@@ -190,7 +190,8 @@   Adapter     { adapterName = "test:gated",       source = Stream.unfoldrM step messages,-      shutdown = liftIO $ atomically $ writeTVar gate True+      shutdown = liftIO $ atomically $ writeTVar gate True,+      telemetry = defaultAdapterTelemetry     }   where     step [] = do

@@ -15,7 +15,7 @@ import Effectful.Exception qualified as Exception import OpenTelemetry.Attributes qualified as OTelAttributes import OpenTelemetry.Trace.Core qualified as OTel-import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry) import Shibuya.App   ( AppConfig (..),     QueueProcessor (..),@@ -581,7 +581,8 @@   Adapter     { adapterName = "test:list",       source = Stream.fromList messages,-      shutdown = pure ()+      shutdown = pure (),+      telemetry = defaultAdapterTelemetry     }  failingAfterAdapter :: (IOE :> es) => Int -> Text.Text -> Adapter es String@@ -589,7 +590,8 @@   Adapter     { adapterName = "test:failing",       source = Stream.unfoldrM step (0 :: Int),-      shutdown = pure ()+      shutdown = pure (),+      telemetry = defaultAdapterTelemetry     }   where     step n@@ -603,7 +605,8 @@   Adapter     { adapterName = "test:infinite",       source = Stream.unfoldrM step (1 :: Int),-      shutdown = pure ()+      shutdown = pure (),+      telemetry = defaultAdapterTelemetry     }   where     step n = do@@ -616,7 +619,8 @@   Adapter     { adapterName = "test:one-then-idle",       source = Stream.unfoldrM step False,-      shutdown = pure ()+      shutdown = pure (),+      telemetry = defaultAdapterTelemetry     }   where     step False = do@@ -629,7 +633,8 @@   Adapter     { adapterName = "test:permanent-finalizer-failure",       source = Stream.fromEffect (pure ingested),-      shutdown = pure ()+      shutdown = pure (),+      telemetry = defaultAdapterTelemetry     }   where     envelope = mkEnvelope (MessageId "finalizer-failure") "message"

@@ -21,7 +21,7 @@ import Data.Text (Text) import Data.Text qualified as Text import Effectful (Eff, IOE, liftIO, runEff, (:>))-import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry) import Shibuya.Adapter.Mock   ( TrackingAck,     getTrackedDecisions,@@ -155,7 +155,8 @@   Adapter     { adapterName = "test:gated-tracked",       source = Stream.unfoldrM step (map (mkTrackedIngested tracking) envs),-      shutdown = liftIO $ atomically $ writeTVar gate True+      shutdown = liftIO $ atomically $ writeTVar gate True,+      telemetry = defaultAdapterTelemetry     }   where     step [] = do@@ -556,7 +557,8 @@               Adapter                 { adapterName = "test:counting-source",                   source = source,-                  shutdown = pure ()+                  shutdown = pure (),+                  telemetry = defaultAdapterTelemetry                 }             cfg = (defaultBatchConfig @_ @Int) {batchSize = 1, batchTimeout = 30, batchKey = scenarioBatchKey}             handler _ _ = do

@@ -13,7 +13,7 @@ import Data.Time (UTCTime (..), fromGregorian) import Effectful (Eff, IOE, liftIO, runEff, (:>)) import Shibuya (ProcessorFailure (..), ProcessorHalt (..))-import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry) import Shibuya.Adapter.Mock   ( TrackingAck (..),     getTrackedDecisions,@@ -453,7 +453,7 @@ runStatusBatches detail handler batches = do   master <- startMaster IgnoreAll   let values = concatMap (NE.toList . snd) batches-      adapter = Adapter "status-batch" (Stream.fromList values) (pure ())+      adapter = Adapter "status-batch" (Stream.fromList values) (pure ()) defaultAdapterTelemetry       config = defaultBatchConfig {batchSize = length values}   proc <- runSupervisedBatchWith detail master 10 (ProcessorId "status-batch") Serial config adapter handler   liftIO $ atomically $ readTVar proc.done >>= check

@@ -10,7 +10,7 @@ import Effectful (liftIO, runEff) import Effectful.State.Static.Local qualified as State import Shibuya qualified as Public-import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry) import Shibuya.Batch (BatchConfig (..), ackAllOk, defaultBatchConfig) import Shibuya.Core.Ack (AckDecision (..), HaltReason (..)) import Shibuya.Core.AckHandle (AckHandle (..))@@ -95,13 +95,13 @@                     pure $ fmap (\() -> (mkIngested (mkEnvelope (MessageId "live") ()) (AckHandle (\_ -> liftIO $ atomically $ modifyTVar' finishes (+ 1))), n + 1)) next                 )                 (1 :: Int)-            adapter = Adapter "live" source (liftIO $ atomically $ writeTVar stop True)+            adapter = Adapter "live" source (liftIO $ atomically $ writeTVar stop True) defaultAdapterTelemetry             work = liftIO ((atomically (writeTQueue starts ()) >> atomically (takeTMVar handlerGate))) >> pure AckOk             processor =               if batching                 then Public.mkBatchProcessor adapter (\_ _ -> work >> pure ackAllOk) defaultBatchConfig {batchSize = 1}                 else Public.mkProcessor adapter (\_ -> work)-            siblingAdapter = Adapter "sibling" (Stream.fromList [mkIngested (mkEnvelope (MessageId "sibling") ()) (AckHandle (\_ -> pure ()))]) (pure ())+            siblingAdapter = Adapter "sibling" (Stream.fromList [mkIngested (mkEnvelope (MessageId "sibling") ()) (AckHandle (\_ -> pure ()))]) (pure ()) defaultAdapterTelemetry         result <- Public.runApp Public.defaultAppConfig [(ProcessorId "live", processor), (ProcessorId "sibling", Public.mkProcessor siblingAdapter (\_ -> liftIO (atomically $ putTMVar sibling ()) >> pure AckOk))]         app <- either (liftIO . ioError . userError . show) pure result         liftIO $ atomically (readTQueue polls) `shouldReturn` 1@@ -168,7 +168,7 @@       UIO.finally         ( runEff $ runTracingNoop $ do             let message = mkIngested (mkEnvelope (MessageId "terminal") ()) (AckHandle (\_ -> if failure then liftIO (ioError (userError "failure")) else pure ()))-                adapter = Adapter "terminal" (Stream.fromList [message]) (pure ())+                adapter = Adapter "terminal" (Stream.fromList [message]) (pure ()) defaultAdapterTelemetry             proc <-               runSupervised                 master

@@ -16,7 +16,7 @@ import Effectful (Eff, IOE, liftIO, runEff, (:>)) import Effectful.State.Static.Local qualified as State import Shibuya (ProcessorHalt (..))-import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry) import Shibuya.Adapter.Mock   ( getTrackedDecisions,     mkTrackedIngested,@@ -287,7 +287,7 @@                           pure $ Just (mkIngested (mkEnvelope (MessageId (Text.pack (show n))) value) (AckHandle (const $ pure ())), n + 1)                   )                   0-              adapter = Adapter {adapterName = "test:state", source = source, shutdown = pure ()}+              adapter = Adapter {adapterName = "test:state", source = source, shutdown = pure (), telemetry = defaultAdapterTelemetry}               observe message = do                 value <- State.get @Int                 State.modify @Int (+ 10)@@ -405,7 +405,7 @@               mkIngested                 (mkEnvelope (MessageId "failed-detail") ())                 (AckHandle (\_ -> liftIO $ ioError (userError "failed finalizer")))-            adapter = Adapter "failed-detail" (Stream.fromList [message]) (pure ())+            adapter = Adapter "failed-detail" (Stream.fromList [message]) (pure ()) defaultAdapterTelemetry         sp <-           runSupervisedWith             (Metrics.MetricsDetailed Metrics.defaultDetailOptions)@@ -432,7 +432,7 @@           master <- startMaster IgnoreAll           let waitHere = liftIO (UIO.putMVar entered () >> UIO.takeMVar blocker)               message = mkIngested (mkEnvelope (MessageId "cancel") ()) (AckHandle (\_ -> when inFinalizer waitHere))-              adapter = Adapter "cancel-detail" (Stream.fromList [message]) (pure ())+              adapter = Adapter "cancel-detail" (Stream.fromList [message]) (pure ()) defaultAdapterTelemetry           sp <-             runSupervisedWith               (Metrics.MetricsDetailed Metrics.defaultDetailOptions)@@ -1068,7 +1068,8 @@                         Adapter                           { adapterName = "test:failing-source",                             source = failingSource,-                            shutdown = pure ()+                            shutdown = pure (),+                            telemetry = defaultAdapterTelemetry                           }                    let handler _ = do@@ -1126,7 +1127,8 @@                         Adapter                           { adapterName = "test:failing-source",                             source = failingSource,-                            shutdown = pure ()+                            shutdown = pure (),+                            telemetry = defaultAdapterTelemetry                           }                    let handler _ = do@@ -1170,7 +1172,8 @@                     Adapter                       { adapterName = "test:failing-immediate",                         source = failingSource,-                        shutdown = pure ()+                        shutdown = pure (),+                        telemetry = defaultAdapterTelemetry                       }                   handler _ = pure AckOk @@ -1263,7 +1266,8 @@                         Adapter                           { adapterName = "test:failing-A",                             source = failingSourceA,-                            shutdown = pure ()+                            shutdown = pure (),+                            telemetry = defaultAdapterTelemetry                           }                    -- Handler A counts messages@@ -1522,7 +1526,8 @@   Adapter     { adapterName = "test:supervised",       source = Stream.fromList messages,-      shutdown = pure ()+      shutdown = pure (),+      telemetry = defaultAdapterTelemetry     }  validDeadLetterCode :: Text.Text -> DeadLetterCode@@ -1549,7 +1554,7 @@         mkIngested           ((mkEnvelope (MessageId "cursor") ()) {cursor = Just (CursorInt 5), partition = Just "p"})           (AckHandle (\_ -> when failFinalizer (liftIO $ UIO.throwIO (userError "finalization failed"))))-      adapter = Adapter "cursor" (Stream.fromList [message]) (pure ())+      adapter = Adapter "cursor" (Stream.fromList [message]) (pure ()) defaultAdapterTelemetry   proc <- runSupervisedWith (Metrics.MetricsDetailed Metrics.defaultDetailOptions) master 10 (ProcessorId "cursor") Unordered Serial adapter (\_ -> pure decision)   liftIO $ atomically $ readTVar proc.done >>= check   m <- getMetrics proc

@@ -6,7 +6,7 @@ import Data.Text qualified as Text import Data.Time (UTCTime (..), fromGregorian) import Effectful (Eff, IOE, liftIO, runEff, (:>))-import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry) import Shibuya.Adapter.Mock (TrackingAck (..), newTrackingAck, trackingAckHandle) import Shibuya.App (AppConfig (..), AppError (..), QueueProcessor (..), defaultAppConfig, mkBatchProcessor, mkProcessor, runApp, stopApp, waitApp) import Shibuya.Batch (ackAll, defaultBatchConfig)@@ -296,7 +296,8 @@   Adapter     { adapterName = "test:mock",       source = Stream.fromList messages,-      shutdown = pure ()+      shutdown = pure (),+      telemetry = defaultAdapterTelemetry     }  -- | Handler that records processed messages

@@ -0,0 +1,220 @@+-- | Fan-in links on batch processing spans.+--+-- A batch has many producers and a span has one parent, so the batch span is+-- a root that links every member's creation context, bounded at+-- 'maxBatchLinks', each link naming its member.+module Shibuya.Telemetry.BatchCorrelationSpec (spec) where++import Control.Exception (bracket)+import Control.Monad (forM_, void)+import Data.ByteString.Char8 qualified as BS8+import Data.Foldable (toList)+import Data.HashMap.Strict qualified as HashMap+import Data.IORef (modifyIORef', newIORef, readIORef, writeIORef)+import Data.Maybe (isNothing)+import Data.Text (Text)+import Data.Text qualified as Text+import Effectful (Eff, IOE, liftIO, runEff, (:>))+import Numeric (showHex)+import OpenTelemetry.Attributes (Attribute, emptyAttributes, getAttributeMap, toAttribute)+import OpenTelemetry.Context qualified as Ctx+import OpenTelemetry.Context.ThreadLocal qualified as Ctx+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 (..), AdapterTelemetry (traceCorrelation), TraceCorrelation (..), defaultAdapterTelemetry)+import Shibuya.Adapter.Mock (listAdapter)+import Shibuya.Batch (BatchConfig (..), ackAllOk, defaultBatchConfig)+import Shibuya.Core.AckHandle (AckHandle (..))+import Shibuya.Core.Ingested (DeliveryStatus (..), Ingested (..), mkIngested)+import Shibuya.Core.Metrics (ProcessorId (..))+import Shibuya.Core.Types (Envelope (..), MessageId (..), TraceHeaders, mkEnvelope)+import Shibuya.Internal.Runner.Supervised (runWithMetricsBatch)+import Shibuya.Policy (Concurrency (..))+import Shibuya.Telemetry.Correlation (BatchLinks (..), collectBatchLinks, maxBatchLinks)+import Shibuya.Telemetry.Effect (Tracing, runTracing, runTracingNoop, withSpan')+import Test.Hspec++spec :: Spec+spec = describe "Shibuya.Telemetry.Correlation (batches)" $ do+  describe "collectBatchLinks" $ do+    it "links members in order, each with its message id" $ do+      let result = collectBatchLinks [(MessageId "a", Just (producer 1)), (MessageId "b", Nothing), (MessageId "c", Just (producer 3))]+      map (hexSpan . (.linkContext)) result.links `shouldBe` [spanHex 1, spanHex 3]+      map (HashMap.lookup "messaging.message.id" . (.linkAttributes)) result.links+        `shouldBe` [Just (toAttribute ("a" :: Text)), Just (toAttribute ("c" :: Text))]+      result.dropped `shouldBe` 0++    it "keeps one link per member when members share a context" $ do+      let result = collectBatchLinks [(MessageId "a", Just (producer 1)), (MessageId "b", Just (producer 1))]+      map (hexSpan . (.linkContext)) result.links `shouldBe` [spanHex 1, spanHex 1]+      map (HashMap.lookup "messaging.message.id" . (.linkAttributes)) result.links+        `shouldBe` [Just (toAttribute ("a" :: Text)), Just (toAttribute ("b" :: Text))]++    it "stores at most maxBatchLinks and counts only valid omitted contexts" $ do+      let full = [(MessageId (tshow i), Just (producer i)) | i <- [1 .. maxBatchLinks]]+          tailMembers =+            [ (MessageId "absent", Nothing),+              (MessageId "malformed", Just malformed),+              (MessageId "valid-1", Just (producer 1000)),+              (MessageId "valid-2", Just (producer 1001))+            ]+          result = collectBatchLinks (full <> tailMembers)+      length result.links `shouldBe` maxBatchLinks+      result.dropped `shouldBe` 2++    it "keeps sparse valid contexts before the cap" $ do+      let members = concat [[(MessageId (tshow i), Just (producer i)), (MessageId "gap", Nothing), (MessageId "bad", Just malformed)] | i <- [1 .. 10]]+          result = collectBatchLinks members+      length result.links `shouldBe` 10+      result.dropped `shouldBe` 0++  describe "batch spans through the runner" $ do+    it "is a root with one link per producer, naming each member" $ do+      [sp] <- batchSpans =<< runBatch [member "m1" (Just (producer 1)), member "m2" (Just (producer 2)), member "m3" (Just (producer 3))]+      isNothing sp.parent `shouldBe` True+      map fst sp.links `shouldBe` [spanHex 1, spanHex 2, spanHex 3]+      map snd sp.links `shouldBe` map (Just . toAttribute) ["m1", "m2", "m3" :: Text]+      sp.traceHex `notElem` map traceHex [1, 2, 3] `shouldBe` True+      HashMap.lookup "messaging.batch.message_count" sp.attributes `shouldBe` Just (toAttribute (3 :: Int))+      sp.name `shouldBe` "process batch-correlation"+      HashMap.lookup "messaging.operation.name" sp.attributes `shouldBe` Just (toAttribute ("process" :: Text))+      HashMap.lookup "messaging.operation.type" sp.attributes `shouldBe` Just (toAttribute ("process" :: Text))+      HashMap.member "shibuya.batch.links.dropped" sp.attributes `shouldBe` False++    it "links members that share a context separately" $ do+      [sp] <- batchSpans =<< runBatch [member "m1" (Just (producer 1)), member "m2" (Just (producer 1)), member "m3" (Just (producer 3))]+      map fst sp.links `shouldBe` [spanHex 1, spanHex 1, spanHex 3]+      map snd sp.links `shouldBe` map (Just . toAttribute) ["m1", "m2", "m3" :: Text]++    it "has no links and no parent when no member has headers" $ do+      [sp] <- batchSpans =<< runBatch [member "m1" Nothing, member "m2" Nothing]+      sp.links `shouldBe` []+      isNothing sp.parent `shouldBe` True++    it "links only the members that have headers" $ do+      [sp] <- batchSpans =<< runBatch [member "m1" Nothing, member "m2" (Just (producer 2)), member "m3" (Just malformed)]+      map snd sp.links `shouldBe` [Just (toAttribute ("m2" :: Text))]++    it "bounds 200 distinct producers at 128 links and reports 72 dropped" $ do+      [sp] <- batchSpans =<< runBatch [member (tshow i) (Just (producer i)) | i <- [1 .. 200]]+      length sp.links `shouldBe` 128+      HashMap.lookup "shibuya.batch.links.dropped" sp.attributes `shouldBe` Just (toAttribute (72 :: Int))+      HashMap.lookup "messaging.batch.message_count" sp.attributes `shouldBe` Just (toAttribute (200 :: Int))++    it "does not link a superseded member" $ do+      let superseded = (member "m2" (Just (producer 2))) {deliveryStatus = pure DeliverySuperseded}+      [sp] <- batchSpans =<< runBatch [member "m1" (Just (producer 1)), superseded, member "m3" (Just (producer 3))]+      map snd sp.links `shouldBe` map (Just . toAttribute) ["m1", "m3" :: Text]+      HashMap.lookup "messaging.batch.message_count" sp.attributes `shouldBe` Just (toAttribute (2 :: Int))++    forM_ [minBound .. maxBound] $ \policy ->+      it ("ignores the adapter's " <> show policy <> " policy") $ do+        [sp] <- batchSpans =<< runBatchWith policy [member "m1" (Just (producer 1))]+        isNothing sp.parent `shouldBe` True+        map fst sp.links `shouldBe` [spanHex 1]++    it "is still a root inside an ambient span, which is restored afterwards" $ do+      restored <- newIORef False+      spans <- withExporter $ \tracer -> runEff $+        runTracing tracer $+          withSpan' "ambient" OTel.defaultSpanArguments $ \ambient -> do+            batchOnce ParentAndLink [member "m1" (Just (producer 1))]+            liftIO $ do+              ambientCtx <- OTel.getSpanContext ambient+              current <- traverse OTel.getSpanContext . Ctx.lookupSpan =<< Ctx.getContext+              writeIORef restored (current == Just ambientCtx)+      [sp] <- batchSpans spans+      isNothing sp.parent `shouldBe` True+      readIORef restored `shouldReturn` True++    it "runs the handler and exports nothing with tracing disabled" $ do+      handled <- newIORef (0 :: Int)+      spans <- withExporter $ \_ -> runEff $ runTracingNoop $ do+        let adapter = listAdapter [member "m1" (Just (producer 1))]+        void $ runWithMetricsBatch 16 (ProcessorId "batch-correlation") Serial batchConfig adapter $ \_ _ ->+          liftIO (modifyIORef' handled (+ 1)) >> pure ackAllOk+      readIORef handled `shouldReturn` 1+      length spans `shouldBe` 0++--------------------------------------------------------------------------------+-- Fixtures+--------------------------------------------------------------------------------++batchConfig :: BatchConfig es msg+batchConfig = defaultBatchConfig {batchSize = 1000, batchTimeout = 60}++member :: (IOE :> es) => Text -> Maybe TraceHeaders -> Ingested es ()+member msgId headers =+  mkIngested ((mkEnvelope (MessageId msgId) ()) {traceContext = headers}) (AckHandle (\_ -> pure ()))++-- | A distinct, valid producer context for each index.+producer :: Int -> TraceHeaders+producer i = [("traceparent", BS8.pack ("00-" <> Text.unpack (traceHex i) <> "-" <> Text.unpack (spanHex i) <> "-01"))]++malformed :: TraceHeaders+malformed = [("traceparent", "00-not-a-valid-traceparent-01")]++traceHex :: Int -> Text+traceHex i = padHex 32 (i + 0xabc)++spanHex :: Int -> Text+spanHex i = padHex 16 (i + 0x100)++padHex :: Int -> Int -> Text+padHex width n = let digits = showHex n "" in Text.pack (replicate (width - length digits) '0' <> digits)++tshow :: (Show a) => a -> Text+tshow = Text.pack . show++batchOnce :: (IOE :> es, Tracing :> es) => TraceCorrelation -> [Ingested es ()] -> Eff es ()+batchOnce policy members =+  let adapter = (listAdapter members) {telemetry = defaultAdapterTelemetry {traceCorrelation = policy}}+   in void $ runWithMetricsBatch 16 (ProcessorId "batch-correlation") Serial batchConfig adapter (\_ _ -> pure ackAllOk)++runBatch :: [Ingested '[Tracing, IOE] ()] -> IO [OTel.ImmutableSpan]+runBatch = runBatchWith ParentAndLink++runBatchWith :: TraceCorrelation -> [Ingested '[Tracing, IOE] ()] -> IO [OTel.ImmutableSpan]+runBatchWith policy members = withExporter $ \tracer -> runEff $ runTracing tracer $ batchOnce policy members++withExporter :: (OTel.Tracer -> IO ()) -> IO [OTel.ImmutableSpan]+withExporter action = do+  (processor, spansRef) <- inMemoryListExporter+  bracket+    (OTel.createTracerProvider [processor] OTel.emptyTracerProviderOptions {OTel.tracerProviderOptionsIdGenerator = DefaultIdGenerator})+    (\provider -> void $ OTel.shutdownTracerProvider provider (Just 5_000_000))+    (\provider -> action (OTel.makeTracer provider (OTel.InstrumentationLibrary "batch-correlation-test" "" "" emptyAttributes) OTel.tracerOptions))+  readIORef spansRef++data BatchSpan = BatchSpan+  { name :: Text,+    traceHex :: Text,+    parent :: Maybe OTel.SpanContext,+    links :: [(Text, Maybe Attribute)],+    attributes :: HashMap.HashMap Text Attribute+  }++batchSpans :: [OTel.ImmutableSpan] -> IO [BatchSpan]+batchSpans spans = traverse describeSpan [sp | sp <- spans, OTel.spanKind sp == OTel.Consumer]+  where+    describeSpan sp = do+      hot <- readIORef (OTel.spanHot sp)+      parent <- traverse OTel.getSpanContext (OTel.spanParent sp)+      pure+        BatchSpan+          { name = OTel.hotName hot,+            traceHex = OTel.Id.traceIdBaseEncodedText Base16 (OTel.spanContext sp).traceId,+            parent,+            links =+              [ (hexSpan (OTel.frozenLinkContext link), HashMap.lookup "messaging.message.id" (getAttributeMap (OTel.frozenLinkAttributes link)))+              | link <- toList (appendOnlyBoundedCollectionValues (OTel.hotLinks hot))+              ],+            attributes = getAttributeMap (OTel.hotAttributes hot)+          }++hexSpan :: OTel.SpanContext -> Text+hexSpan ctx = OTel.Id.spanIdBaseEncodedText Base16 ctx.spanId

@@ -0,0 +1,282 @@+-- | Trace correlation policies for per-message processing spans.+--+-- Every case drives a real tracer whose provider exports into memory, and+-- inspects the exported Consumer spans: their trace, their parent, and+-- their links.+module Shibuya.Telemetry.CorrelationSpec (spec) where++import Control.Exception (bracket)+import Control.Monad (forM_, void)+import Data.ByteString (ByteString)+import Data.Foldable (toList)+import Data.HashMap.Strict qualified as HashMap+import Data.IORef (IORef, modifyIORef', newIORef, readIORef, writeIORef)+import Data.Maybe (isNothing)+import Data.Text (Text)+import Effectful (Eff, IOE, liftIO, runEff, (:>))+import OpenTelemetry.Attributes (emptyAttributes, getAttributeMap, toAttribute)+import OpenTelemetry.Context qualified as Ctx+import OpenTelemetry.Context.ThreadLocal qualified as Ctx+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.Trace.TraceState qualified as TraceState+import OpenTelemetry.Util (appendOnlyBoundedCollectionValues)+import Shibuya.Adapter (Adapter (..), AdapterTelemetry (traceCorrelation), TraceCorrelation (..), defaultAdapterTelemetry)+import Shibuya.Adapter.Mock (listAdapter)+import Shibuya.Core.Ack (AckDecision (..))+import Shibuya.Core.AckHandle (AckHandle (..))+import Shibuya.Core.Ingested (DeliveryStatus (..), Ingested (..), mkIngested)+import Shibuya.Core.Metrics (ProcessorId (..))+import Shibuya.Core.Types (Envelope (..), MessageId (..), TraceHeaders, mkEnvelope)+import Shibuya.Internal.Runner.Supervised (runWithMetrics)+import Shibuya.Telemetry.Correlation (consumerSpanArgsLinkedTo, creationLink, creationLinkWith)+import Shibuya.Telemetry.Effect (Tracing, runTracing, runTracingNoop, withSpan')+import Shibuya.Telemetry.Process (withProcessSpan, withProcessSpanIO)+import Shibuya.Telemetry.Propagation (currentTraceHeaders, extractTraceContext)+import Shibuya.Telemetry.Semantic (processSpanName)+import Test.Hspec++spec :: Spec+spec = describe "Shibuya.Telemetry.Correlation" $ do+  describe "AdapterTelemetry" $ do+    it "defaults to ParentAndLink and is changed by record update" $ do+      defaultAdapterTelemetry.traceCorrelation `shouldBe` ParentAndLink+      (defaultAdapterTelemetry {traceCorrelation = LinkOnly}).traceCorrelation `shouldBe` LinkOnly++  describe "creationLink" $ do+    it "decodes a valid traceparent into an attribute-free link" $+      case creationLink producerHeaders of+        Nothing -> expectationFailure "expected a link"+        Just link -> do+          hexTrace link.linkContext `shouldBe` producerTraceHex+          hexSpan link.linkContext `shouldBe` producerSpanHex+          HashMap.null link.linkAttributes `shouldBe` True++    it "keeps tracestate on the linked context" $+      case creationLink (producerHeaders <> [("tracestate", "congo=t61rcWkgMzE")]) of+        Nothing -> expectationFailure "expected a link"+        Just link -> TraceState.toList link.linkContext.traceState `shouldBe` [(TraceState.Key "congo", TraceState.Value "t61rcWkgMzE")]++    it "returns Nothing without a traceparent" $+      isNothing (creationLink [("tracestate", "congo=t61rcWkgMzE")]) `shouldBe` True++    it "returns Nothing for a malformed traceparent" $+      isNothing (creationLink malformedHeaders) `shouldBe` True++    it "carries caller-supplied attributes with creationLinkWith" $ do+      let attrs = HashMap.singleton "messaging.message.id" (toAttribute ("m-1" :: Text))+      case creationLinkWith attrs producerHeaders of+        Nothing -> expectationFailure "expected a link"+        Just link -> link.linkAttributes `shouldBe` attrs++    it "builds Consumer-kind arguments carrying the links" $ do+      let links = maybe [] pure (creationLink producerHeaders)+          args = consumerSpanArgsLinkedTo links+      args.kind `shouldBe` OTel.Consumer+      length args.links `shouldBe` 1++  describe "processing spans through the runner" $ do+    it "ParentAndLink: child of the producer, with one link to it" $ do+      [sp] <- consumerSpans =<< runMessage ParentAndLink (Just producerHeaders) Nothing+      hexTrace sp.context `shouldBe` producerTraceHex+      fmap hexSpan sp.parent `shouldBe` Just producerSpanHex+      map (\l -> (hexTrace l, hexSpan l)) sp.links `shouldBe` [(producerTraceHex, producerSpanHex)]++    it "LinkOnly: a root in a new trace, with one link to the producer" $ do+      [sp] <- consumerSpans =<< runMessage LinkOnly (Just producerHeaders) Nothing+      hexTrace sp.context `shouldNotBe` producerTraceHex+      isNothing sp.parent `shouldBe` True+      map (\l -> (hexTrace l, hexSpan l)) sp.links `shouldBe` [(producerTraceHex, producerSpanHex)]++    it "LinkOnly: still a root when an ambient span is active, which is restored afterwards" $ do+      restored <- newIORef Nothing+      spans <- withExporter $ \tracer -> runEff $+        runTracing tracer $+          withSpan' "ambient" OTel.defaultSpanArguments $ \ambient -> do+            runOnce LinkOnly (Just producerHeaders) Nothing+            liftIO $ do+              ambientCtx <- OTel.getSpanContext ambient+              current <- Ctx.lookupSpan <$> Ctx.getContext+              currentCtx <- traverse OTel.getSpanContext current+              writeIORef restored (Just (Just ambientCtx == currentCtx))+      [sp] <- consumerSpans spans+      isNothing sp.parent `shouldBe` True+      length sp.links `shouldBe` 1+      readIORef restored `shouldReturn` Just True++    forM_ [minBound .. maxBound] $ \policy -> do+      forM_ [("absent", Nothing), ("malformed", Just malformedHeaders)] $ \(label, headers) -> do+        it (show policy <> ": " <> label <> " headers keep the ambient parent and add no link") $ do+          (ambientSpanHex, spans) <- withAmbient $ \_tracer ->+            runOnce policy headers Nothing+          [sp] <- consumerSpans spans+          sp.links `shouldBe` []+          fmap hexSpan sp.parent `shouldBe` Just ambientSpanHex++        it (show policy <> ": " <> label <> " headers without an ambient span give a root") $ do+          [sp] <- consumerSpans =<< runMessage policy headers Nothing+          sp.links `shouldBe` []+          isNothing sp.parent `shouldBe` True++    it "LinkOnly: currentTraceHeaders in the handler names the new trace and the Consumer span" $ do+      seen <- newIORef Nothing+      spans <- withExporter $ \tracer ->+        runEff $+          runTracing tracer $+            runOnce LinkOnly (Just producerHeaders) (Just seen)+      [sp] <- consumerSpans spans+      Just (Just headers) <- readIORef seen+      case extractTraceContext headers of+        Nothing -> expectationFailure "handler saw no decodable traceparent"+        Just ctx -> do+          hexTrace ctx `shouldBe` hexTrace sp.context+          hexSpan ctx `shouldBe` hexSpan sp.context+          hexTrace ctx `shouldNotBe` producerTraceHex++    it "LinkOnly: a superseded delivery still opens a linked span that ends successfully" $ do+      spans <- withExporter $ \tracer -> runEff $ runTracing tracer $ do+        let message = (producedMessage (Just producerHeaders)) {deliveryStatus = pure DeliverySuperseded}+        void $ runWithMetrics 4 (ProcessorId "correlation") (policyAdapter LinkOnly [message]) (\_ -> pure AckOk)+      [sp] <- consumerSpans spans+      isNothing sp.parent `shouldBe` True+      length sp.links `shouldBe` 1+      sp.status `shouldBe` OTel.Ok+      sp.ackDecisions `shouldBe` [toAttribute ("superseded" :: Text)]++    forM_ [minBound .. maxBound] $ \policy ->+      it (show policy <> ": disabled tracing runs the handler and exports nothing") $ do+        handled <- newIORef (0 :: Int)+        spans <- withExporter $ \_tracer -> runEff $ runTracingNoop $ do+          let adapter = policyAdapter policy [producedMessage (Just producerHeaders)]+          void $ runWithMetrics 4 (ProcessorId "correlation") adapter (\_ -> liftIO (modifyIORef' handled (+ 1)) >> pure AckOk)+        readIORef handled `shouldReturn` 1+        length spans `shouldBe` 0++  describe "withProcessSpan and withProcessSpanIO called directly" $ do+    forM_ [minBound .. maxBound] $ \policy -> do+      it (show policy <> ": both give the runner's span shape") $ do+        [runner] <- consumerSpans =<< runMessage policy (Just producerHeaders) Nothing+        [effect] <- consumerSpans =<< withExporter (\tracer -> runEff $ runTracing tracer $ withProcessSpan policy processName (Just producerHeaders) (\_ -> pure ()))+        [io] <- consumerSpans =<< withExporter (\tracer -> withProcessSpanIO tracer policy processName (Just producerHeaders) (\_ -> pure ()))+        forM_ [effect, io] $ \sp -> do+          sp.name `shouldBe` runner.name+          fmap hexSpan sp.parent `shouldBe` fmap hexSpan runner.parent+          map hexSpan sp.links `shouldBe` map hexSpan runner.links+          (hexTrace sp.context == producerTraceHex) `shouldBe` (hexTrace runner.context == producerTraceHex)++      it (show policy <> ": the IO helper keeps the ambient parent without headers") $ do+        (ambientSpanHex, spans) <- withAmbient $ \tracer ->+          liftIO $ withProcessSpanIO tracer policy processName Nothing (\_ -> pure ())+        [sp] <- consumerSpans spans+        sp.links `shouldBe` []+        fmap hexSpan sp.parent `shouldBe` Just ambientSpanHex++      it (show policy <> ": withProcessSpan under runTracingNoop passes the sentinel span") $ do+        recording <- runEff $ runTracingNoop $ withProcessSpan policy processName (Just producerHeaders) (liftIO . OTel.isRecording)+        recording `shouldBe` False++--------------------------------------------------------------------------------+-- Fixtures+--------------------------------------------------------------------------------++producerTraceHex :: Text+producerTraceHex = "0af7651916cd43dd8448eb211c80319c"++producerSpanHex :: Text+producerSpanHex = "b7ad6b7169203331"++producerHeaders :: TraceHeaders+producerHeaders = [("traceparent", "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01")]++malformedHeaders :: [(ByteString, ByteString)]+malformedHeaders = [("traceparent", "00-not-a-valid-traceparent-01")]++processName :: Text+processName = processSpanName "correlation"++producedMessage :: (IOE :> es) => Maybe TraceHeaders -> Ingested es ()+producedMessage headers =+  mkIngested ((mkEnvelope (MessageId "m-1") ()) {traceContext = headers}) (AckHandle (\_ -> pure ()))++policyAdapter :: (IOE :> es) => TraceCorrelation -> [Ingested es msg] -> Adapter es msg+policyAdapter policy messages =+  (listAdapter messages) {telemetry = defaultAdapterTelemetry {traceCorrelation = policy}}++-- | Process one message on the calling thread (the unsupervised runner path),+-- so an ambient span on this thread is visible to the processing span.+runOnce ::+  (IOE :> es, Tracing :> es) =>+  TraceCorrelation ->+  Maybe TraceHeaders ->+  Maybe (IORef (Maybe (Maybe TraceHeaders))) ->+  Eff es ()+runOnce policy headers seen =+  void $+    runWithMetrics 4 (ProcessorId "correlation") (policyAdapter policy [producedMessage headers]) $ \_ -> do+      forM_ seen $ \ref -> currentTraceHeaders >>= liftIO . writeIORef ref . Just+      pure AckOk++runMessage :: TraceCorrelation -> Maybe TraceHeaders -> Maybe (IORef (Maybe (Maybe TraceHeaders))) -> IO [OTel.ImmutableSpan]+runMessage policy headers seen =+  withExporter $ \tracer -> runEff $ runTracing tracer $ runOnce policy headers seen++-- | Run inside an ambient span and return that span's ID with the exports.+withAmbient :: (OTel.Tracer -> Eff '[Tracing, IOE] ()) -> IO (Text, [OTel.ImmutableSpan])+withAmbient action = do+  ambientRef <- newIORef ""+  spans <- withExporter $ \tracer -> runEff $+    runTracing tracer $+      withSpan' "ambient" OTel.defaultSpanArguments $ \ambient -> do+        ctx <- liftIO $ OTel.getSpanContext ambient+        liftIO $ writeIORef ambientRef (hexSpan ctx)+        action tracer+  ambientSpanHex <- readIORef ambientRef+  pure (ambientSpanHex, spans)++withExporter :: (OTel.Tracer -> IO ()) -> IO [OTel.ImmutableSpan]+withExporter action = do+  (processor, spansRef) <- inMemoryListExporter+  bracket+    (OTel.createTracerProvider [processor] OTel.emptyTracerProviderOptions {OTel.tracerProviderOptionsIdGenerator = DefaultIdGenerator})+    (\provider -> void $ OTel.shutdownTracerProvider provider (Just 5_000_000))+    (\provider -> action (OTel.makeTracer provider (OTel.InstrumentationLibrary "correlation-test" "" "" emptyAttributes) OTel.tracerOptions))+  readIORef spansRef++data ConsumerSpan = ConsumerSpan+  { name :: Text,+    context :: OTel.SpanContext,+    parent :: Maybe OTel.SpanContext,+    links :: [OTel.SpanContext],+    status :: OTel.SpanStatus,+    ackDecisions :: [OTel.Attribute]+  }++consumerSpans :: [OTel.ImmutableSpan] -> IO [ConsumerSpan]+consumerSpans spans = traverse describeSpan [sp | sp <- spans, OTel.spanKind sp == OTel.Consumer]+  where+    describeSpan sp = do+      hot <- readIORef (OTel.spanHot sp)+      parent <- traverse OTel.getSpanContext (OTel.spanParent sp)+      pure+        ConsumerSpan+          { name = OTel.hotName hot,+            context = OTel.spanContext sp,+            parent,+            links = map OTel.frozenLinkContext (toList (appendOnlyBoundedCollectionValues (OTel.hotLinks hot))),+            status = OTel.hotStatus hot,+            ackDecisions =+              [ value+              | event <- toList (appendOnlyBoundedCollectionValues (OTel.hotEvents hot)),+                OTel.eventName event == "shibuya.ack.decision",+                Just value <- [HashMap.lookup "shibuya.ack.decision" (getAttributeMap (OTel.eventAttributes event))]+              ]+          }++hexTrace :: OTel.SpanContext -> Text+hexTrace ctx = OTel.Id.traceIdBaseEncodedText Base16 ctx.traceId++hexSpan :: OTel.SpanContext -> Text+hexSpan ctx = OTel.Id.spanIdBaseEncodedText Base16 ctx.spanId

@@ -80,7 +80,7 @@     case spans of       [processingSpan] -> do         hot <- readIORef (spanHot processingSpan)-        hotName hot `shouldBe` "skip-proc process"+        hotName hot `shouldBe` "process skip-proc"         let events = toList (appendOnlyBoundedCollectionValues (hotEvents hot))         map eventName events `shouldBe` ["shibuya.ack.decision"]         hotStatus hot `shouldBe` Ok@@ -109,8 +109,9 @@     case spans of       [s] -> do         hot <- readIORef (spanHot s)-        hotName hot `shouldBe` "test-proc process"+        hotName hot `shouldBe` "process test-proc"         let attrs = getAttributeMap (hotAttributes hot)+        attrs `shouldHaveTextAttribute` ("messaging.operation.name", "process")         attrs `shouldHaveTextAttribute` ("messaging.system", "shibuya")         attrs `shouldHaveTextAttribute` ("messaging.message.id", "m-1")         attrs `shouldHaveTextAttribute` ("messaging.destination.name", "test-proc")