packages feed

shibuya-core 0.4.0.0 → 0.5.0.0

raw patch · 11 files changed

+279/−25 lines, 11 files

Files

CHANGELOG.md view
@@ -1,5 +1,43 @@ # Changelog +## 0.5.0.0 — 2026-05-05++### Breaking Changes++- `Envelope` gained an `attributes :: !(HashMap Text Attribute)` field+  carrying adapter-supplied OpenTelemetry attributes for the per-message+  processing span. Direct constructions of `Envelope` must add the+  field; pass `Data.HashMap.Strict.empty` when the adapter has nothing+  to contribute (the common case). `Envelope`'s `NFData` instance is+  now hand-written rather than derived (because `Attribute` from+  `hs-opentelemetry-api` does not ship `NFData`); the strictness shape+  is unchanged for every other field.++### New Features++- `Shibuya.Runner.Supervised`'s `processOne` now applies+  `envelope.attributes` to its Consumer-kind span after setting the+  framework-default `messaging.*` attributes, so adapter-supplied+  keys override framework defaults of the same name. This lets+  broker-aware adapters (Kafka in particular) emit typed attributes+  (`messaging.kafka.destination.partition`,+  `messaging.kafka.message.offset`) and override the+  `messaging.system` default — without opening a second span. The+  per-record `Shibuya.Adapter.Kafka.Tracing.traced` wrapper that+  previously bolted these on can now be removed; see the audit+  document `docs/plans/9-otel-audit-findings.md` (Finding F1, P0)+  for the full motivation.+- `Shibuya.Telemetry.Propagation.currentTraceHeaders ::+  (Tracing :> es, IOE :> es) => Eff es (Maybe TraceHeaders)` looks+  up the currently-active OTel span (via thread-local context) and+  encodes its trace context as W3C headers, ready for an adapter+  to attach to an outgoing message. Returns `Nothing` when tracing+  is disabled or there is no active span. The intended call sites+  are adapter-side DLQ writes (so the failing-consumer's trace+  links to the resulting DLQ message) and ad-hoc producer paths.+  See the audit document `docs/plans/9-otel-audit-findings.md`+  (Findings F3 and F5).+ ## 0.4.0.0 — 2026-04-29  ### Breaking Changes
shibuya-core.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.14 name: shibuya-core-version: 0.4.0.0+version: 0.5.0.0 synopsis: Supervised queue processing framework for Haskell description:   A supervised queue processing framework inspired by Broadway (Elixir).
src/Shibuya/Core/Types.hs view
@@ -19,9 +19,11 @@   ) where -import Control.DeepSeq (NFData)+import Control.DeepSeq (NFData (..)) import Data.ByteString (ByteString)+import Data.HashMap.Strict (HashMap) import Data.String (IsString)+import OpenTelemetry.Attributes (Attribute (..), PrimitiveAttribute (..)) import Shibuya.Prelude  -- | Stable identity for idempotency & observability.@@ -68,8 +70,41 @@     -- 'Just (Attempt 0)' on first delivery; 'Nothing' if the adapter     -- does not track redeliveries (e.g., Kafka).     attempt :: !(Maybe Attempt),+    -- | Adapter-supplied OpenTelemetry attributes for the per-message+    -- processing span.+    --+    -- The framework's @processOne@ adds these to its Consumer-kind+    -- span after setting the spec-aligned @messaging.*@ defaults, so+    -- adapter-supplied keys override framework defaults of the same+    -- name. Use 'Data.HashMap.Strict.empty' when the adapter has+    -- nothing to contribute (the common case).+    --+    -- Adapters that emit broker-specific typed attributes+    -- (e.g. Kafka's @messaging.kafka.destination.partition@) should+    -- populate this field at envelope-construction time. The+    -- previous opt-in @Shibuya.Adapter.Kafka.Tracing.traced@+    -- transformer existed only to bolt these on; this field replaces+    -- that mechanism without the duplicate-span hazard.+    attributes :: !(HashMap Text Attribute),     -- | The actual message payload     payload :: !msg   }   deriving stock (Eq, Show, Functor, Generic)-  deriving anyclass (NFData)++-- | Manual 'NFData' so the @attributes@ field's 'Attribute' values do+-- not require an upstream NFData instance (which+-- @hs-opentelemetry-api@ does not currently ship). Forces every other+-- field deeply and reduces 'attributes' to WHNF — every 'Attribute'+-- leaf is a small primitive ('Text', 'Bool', 'Double', 'Int64'), so+-- WHNF is enough to evaluate the contained values when the HashMap+-- is itself in WHNF.+instance (NFData msg) => NFData (Envelope msg) where+  rnf e =+    rnf e.messageId `seq`+      rnf e.cursor `seq`+        rnf e.partition `seq`+          rnf e.enqueuedAt `seq`+            rnf e.traceContext `seq`+              rnf e.attempt `seq`+                e.attributes `seq`+                  rnf e.payload
src/Shibuya/Runner/Supervised.hs view
@@ -36,11 +36,13 @@     writeTVar,   ) import Control.Monad (unless)+import Data.HashMap.Strict qualified as HashMap import Data.IORef (IORef, atomicWriteIORef, newIORef, readIORef) import Data.Text qualified as Text import Effectful (Eff, IOE, liftIO, withEffToIO, (:>)) import Effectful.Dispatch.Static (unsafeEff_) import Effectful.Internal.Unlift (Limit (..), Persistence (..), UnliftStrategy (..))+import OpenTelemetry.Attributes (toAttribute) import OpenTelemetry.Trace.Core qualified as OTel import Shibuya.Adapter (Adapter (..)) import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), HaltReason (..))@@ -66,6 +68,7 @@ import Shibuya.Telemetry.Effect   ( Tracing,     addAttribute,+    addAttributes,     addEvent,     recordException,     setStatus,@@ -373,17 +376,26 @@    withExtractedContext parentCtx $     withSpan' (processSpanName pidText) consumerSpanArgs $ \traceSpan -> do-      -- Add messaging attributes+      -- 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 msgIdText = ingested.envelope.messageId-      addAttribute traceSpan attrMessagingSystem ("shibuya" :: Text)-      addAttribute traceSpan attrMessagingDestinationName pidText-      addAttribute traceSpan attrMessagingOperation ("process" :: Text)-      addAttribute traceSpan attrMessagingMessageId msgIdText--      -- Add partition if present-      case ingested.envelope.partition of-        Just p -> addAttribute traceSpan attrShibuyaPartition p-        Nothing -> pure ()+          frameworkAttrs =+            HashMap.fromList $+              [ (attrMessagingSystem, toAttribute ("shibuya" :: Text)),+                (attrMessagingDestinationName, toAttribute pidText),+                (attrMessagingOperation, toAttribute ("process" :: Text)),+                (attrMessagingMessageId, toAttribute msgIdText)+              ]+                <> case ingested.envelope.partition of+                  Just p -> [(attrShibuyaPartition, toAttribute p)]+                  Nothing -> []+          mergedAttrs = HashMap.union ingested.envelope.attributes frameworkAttrs+      addAttributes traceSpan mergedAttrs        -- Increment in-flight and add inflight attributes       now <- liftIO getCurrentTime
src/Shibuya/Telemetry/Propagation.hs view
@@ -6,15 +6,20 @@      -- * Injection     injectTraceContext,+    currentTraceHeaders,      -- * Re-export     TraceHeaders,   ) where +import Effectful (Eff, IOE, liftIO, (:>))+import OpenTelemetry.Context qualified as Ctx+import OpenTelemetry.Context.ThreadLocal qualified as Ctx import OpenTelemetry.Propagator.W3CTraceContext qualified as W3C import OpenTelemetry.Trace.Core (Span, SpanContext) import Shibuya.Core.Types (TraceHeaders)+import Shibuya.Telemetry.Effect (Tracing, isTracingEnabled)  -- | Extract SpanContext from W3C trace headers. -- Returns Nothing if headers are missing or malformed.@@ -49,3 +54,37 @@     [ ("traceparent", traceparent),       ("tracestate", tracestate)     ]++-- | Look up the currently-active OTel span and encode its context as+-- W3C trace headers, ready to attach to an outgoing message.+--+-- Returns 'Nothing' when tracing is disabled or when there is no+-- active span at the call site (e.g. a producer running outside any+-- 'Shibuya.Telemetry.Effect.withSpan'/@withSpan'@ scope).+--+-- This is the higher-level helper most call sites want — adapter+-- code that forwards a message (a DLQ write from @AckDeadLetter@,+-- a producer publishing a follow-on event) can do:+--+-- @+-- consumerHeaders <- currentTraceHeaders+-- let outgoingHeaders =+--       maybe originalHeaders (\\h -> h <> originalHeaders) consumerHeaders+-- @+--+-- and the failing-consumer's trace shows up linked to the resulting+-- message in the downstream trace store. The lower-level+-- 'injectTraceContext' is still exported for callers that already+-- hold a 'Span' handle from inside a 'withSpan''.+currentTraceHeaders ::+  (Tracing :> es, IOE :> es) =>+  Eff es (Maybe TraceHeaders)+currentTraceHeaders = do+  enabled <- isTracingEnabled+  if not enabled+    then pure Nothing+    else liftIO $ do+      ctx <- Ctx.getContext+      case Ctx.lookupSpan ctx of+        Nothing -> pure Nothing+        Just sp -> Just <$> injectTraceContext sp
test/Shibuya/Core/RetrySpec.hs view
@@ -2,6 +2,7 @@  module Shibuya.Core.RetrySpec (spec) where +import Data.HashMap.Strict qualified as HashMap import Data.Time (nominalDiffTimeToSeconds) import Effectful (runEff) import Shibuya.Core.Ack (AckDecision (..), RetryDelay (..))@@ -22,6 +23,7 @@       enqueuedAt = Nothing,       traceContext = Nothing,       attempt = a,+      attributes = HashMap.empty,       payload = ()     }
test/Shibuya/Core/TypesSpec.hs view
@@ -2,6 +2,7 @@  module Shibuya.Core.TypesSpec (spec) where +import Data.HashMap.Strict qualified as HashMap import Data.Time (UTCTime (..), fromGregorian) import Shibuya.Core.Types import Test.Hspec@@ -88,6 +89,7 @@       enqueuedAt = Just testTime,       traceContext = Nothing,       attempt = Nothing,+      attributes = HashMap.empty,       payload = msg     }
test/Shibuya/Runner/SupervisedSpec.hs view
@@ -5,6 +5,7 @@ import Control.Concurrent.NQE.Supervisor (Strategy (..)) import Control.Concurrent.STM (readTVarIO) import Control.Monad (forM, replicateM)+import Data.HashMap.Strict qualified as HashMap import Data.IORef (IORef, atomicModifyIORef', modifyIORef', newIORef, readIORef) import Data.Text qualified as Text import Data.Time (UTCTime (..), fromGregorian)@@ -854,6 +855,7 @@                 enqueuedAt = Just testTime,                 traceContext = Nothing,                 attempt = Nothing,+                attributes = HashMap.empty,                 payload = "message-" <> show i               }           ackHandle = AckHandle $ \_ -> pure ()@@ -877,6 +879,7 @@             enqueuedAt = Just testTime,             traceContext = Nothing,             attempt = Nothing,+            attributes = HashMap.empty,             payload = "message-" <> show i           }       ackHandle = AckHandle $ \_ -> pure ()
test/Shibuya/RunnerSpec.hs view
@@ -2,6 +2,7 @@  module Shibuya.RunnerSpec (spec) where +import Data.HashMap.Strict qualified as HashMap import Data.IORef (IORef, modifyIORef', newIORef, readIORef) import Data.Text qualified as Text import Data.Time (UTCTime (..), fromGregorian)@@ -206,6 +207,7 @@                 enqueuedAt = Just testTime,                 traceContext = Nothing,                 attempt = Nothing,+                attributes = HashMap.empty,                 payload = "message-" <> show i               }           ackHandle = AckHandle $ \_ -> pure () -- No-op ack@@ -230,6 +232,7 @@                 enqueuedAt = Just testTime,                 traceContext = Nothing,                 attempt = Nothing,+                attributes = HashMap.empty,                 payload = "message-" <> show i               }           ackHandle = trackingAckHandle tracking msgId
test/Shibuya/Telemetry/PropagationSpec.hs view
@@ -5,9 +5,21 @@  import Data.ByteArray.Encoding (Base (..)) import Data.ByteString (ByteString)+import Effectful (runEff)+import OpenTelemetry.Attributes (emptyAttributes)+import OpenTelemetry.Trace.Core+  ( InstrumentationLibrary (..),+    createTracerProvider,+    defaultSpanArguments,+    emptyTracerProviderOptions,+    makeTracer,+    shutdownTracerProvider,+    tracerOptions,+  ) import OpenTelemetry.Trace.Core qualified as OTel import OpenTelemetry.Trace.Id qualified as OTel.Id-import Shibuya.Telemetry.Propagation (extractTraceContext)+import Shibuya.Telemetry.Effect (runTracing, runTracingNoop, withSpan)+import Shibuya.Telemetry.Propagation (currentTraceHeaders, extractTraceContext) import Test.Hspec  spec :: Spec@@ -119,3 +131,46 @@         Nothing -> expectationFailure "Expected to extract trace context despite malformed tracestate"         Just ctx -> do           OTel.Id.traceIdBaseEncodedText Base16 (OTel.traceId ctx) `shouldBe` "0af7651916cd43dd8448eb211c80319c"++  describe "currentTraceHeaders" $ do+    it "returns Nothing when tracing is disabled" $ do+      result <- runEff $ runTracingNoop currentTraceHeaders+      result `shouldBe` Nothing++    it "returns Nothing outside any active span" $ do+      provider <- createTracerProvider [] emptyTracerProviderOptions+      let tracer =+            makeTracer+              provider+              ( InstrumentationLibrary+                  { libraryName = "shibuya-test",+                    libraryVersion = "",+                    librarySchemaUrl = "",+                    libraryAttributes = emptyAttributes+                  }+              )+              tracerOptions+      result <- runEff $ runTracing tracer currentTraceHeaders+      _ <- shutdownTracerProvider provider+      result `shouldBe` Nothing++    it "returns headers carrying the active span's traceparent" $ do+      provider <- createTracerProvider [] emptyTracerProviderOptions+      let tracer =+            makeTracer+              provider+              ( InstrumentationLibrary+                  { libraryName = "shibuya-test",+                    libraryVersion = "",+                    librarySchemaUrl = "",+                    libraryAttributes = emptyAttributes+                  }+              )+              tracerOptions+      result <- runEff $ runTracing tracer $ withSpan "outer" defaultSpanArguments currentTraceHeaders+      _ <- shutdownTracerProvider provider+      case result of+        Nothing -> expectationFailure "expected currentTraceHeaders to return Just"+        Just hs -> case lookup "traceparent" hs of+          Nothing -> expectationFailure "expected traceparent header to be present"+          Just _ -> pure ()
test/Shibuya/Telemetry/SemanticSpec.hs view
@@ -14,6 +14,7 @@ import Data.HashMap.Strict (HashMap) import Data.HashMap.Strict qualified as HashMap import Data.IORef (readIORef)+import Data.Int (Int64) import Data.Text (Text) import Effectful (runEff) import OpenTelemetry.Attributes@@ -21,6 +22,7 @@     PrimitiveAttribute (..),     emptyAttributes,     getAttributeMap,+    toAttribute,   ) import OpenTelemetry.Exporter.InMemory.Span (inMemoryListExporter) import OpenTelemetry.Trace.Core@@ -49,17 +51,7 @@   it "emits a process span with conventions-aligned attributes and events" $ do     (processor, spansRef) <- inMemoryListExporter     provider <- createTracerProvider [processor] emptyTracerProviderOptions-    let tracer =-          makeTracer-            provider-            ( InstrumentationLibrary-                { libraryName = "shibuya-test",-                  libraryVersion = "",-                  librarySchemaUrl = "",-                  libraryAttributes = emptyAttributes-                }-            )-            tracerOptions+    let tracer = mkTestTracer provider      runEff $ runTracing tracer $ do       let envelope =@@ -70,6 +62,7 @@                 enqueuedAt = Nothing,                 traceContext = Nothing,                 attempt = Nothing,+                attributes = HashMap.empty,                 payload = ("hello" :: Text)               }           ingested =@@ -102,7 +95,79 @@         evNames `shouldContain` ["shibuya.handler.completed"]       _ ->         expectationFailure $ "expected exactly one span, got " <> show (length spans)++  it "applies envelope.attributes onto the framework span (P0 fix, plan 9 F1/F2)" $ do+    -- The adapter contributes broker-specific typed attributes via+    -- 'Envelope.attributes'. The framework's processOne span must carry+    -- them, and adapter-supplied keys must override framework defaults+    -- of the same name (here: messaging.system flips from "shibuya" to+    -- "kafka" because the adapter set it).+    (processor, spansRef) <- inMemoryListExporter+    provider <- createTracerProvider [processor] emptyTracerProviderOptions+    let tracer = mkTestTracer provider++    runEff $ runTracing tracer $ do+      let envelope =+            Envelope+              { messageId = MessageId "orders-2-42",+                cursor = Nothing,+                partition = Nothing,+                enqueuedAt = Nothing,+                traceContext = Nothing,+                attempt = Nothing,+                attributes =+                  HashMap.fromList+                    [ ("messaging.system", toAttribute ("kafka" :: Text)),+                      ( "messaging.kafka.destination.partition",+                        toAttribute (2 :: Int64)+                      ),+                      ( "messaging.kafka.message.offset",+                        toAttribute (42 :: Int64)+                      )+                    ],+                payload = ("hello" :: Text)+              }+          ingested =+            Ingested+              { envelope = envelope,+                ack = AckHandle (\_ -> pure ()),+                lease = Nothing+              }+          adapter = listAdapter [ingested]+          handler _ = pure AckOk+          procId = ProcessorId "orders-consumer"+      _ <- runWithMetrics 4 procId adapter handler+      pure ()++    _ <- shutdownTracerProvider provider+    spans <- readIORef spansRef+    case spans of+      [s] -> do+        let attrs = getAttributeMap (spanAttributes s)+        -- Adapter override wins.+        attrs `shouldHaveTextAttribute` ("messaging.system", "kafka")+        -- Adapter-typed attributes appear on the framework span.+        attrs `shouldHaveIntAttribute` ("messaging.kafka.destination.partition", 2)+        attrs `shouldHaveIntAttribute` ("messaging.kafka.message.offset", 42)+        -- Framework defaults still set where the adapter did not override.+        attrs `shouldHaveTextAttribute` ("messaging.destination.name", "orders-consumer")+        attrs `shouldHaveTextAttribute` ("messaging.operation", "process")+        attrs `shouldHaveTextAttribute` ("messaging.message.id", "orders-2-42")+      _ ->+        expectationFailure $+          "expected exactly one span, got " <> show (length spans)   where+    mkTestTracer p =+      makeTracer+        p+        ( InstrumentationLibrary+            { libraryName = "shibuya-test",+              libraryVersion = "",+              librarySchemaUrl = "",+              libraryAttributes = emptyAttributes+            }+        )+        tracerOptions     shouldHaveTextAttribute :: HashMap Text Attribute -> (Text, Text) -> Expectation     shouldHaveTextAttribute attrs (k, expected) =       case HashMap.lookup k attrs of