shibuya-core 0.1.0.0 → 0.2.0.0

raw patch · 7 files changed

+258/−45 lines, 7 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• CHANGELOG.md +23/−0
• shibuya-core.cabal +7/−1
• src/Shibuya/Core/Types.hs +4/−0
• src/Shibuya/Runner/Supervised.hs +22/−15
• src/Shibuya/Telemetry/Semantic.hs +76/−29
• test/Main.hs +2/−0
• test/Shibuya/Telemetry/SemanticSpec.hs +124/−0

@@ -1,5 +1,28 @@ # Changelog +## 0.2.0.0 — 2026-04-22++### Breaking Changes++- `Shibuya.Telemetry.Semantic`: rename `processMessageSpanName :: Text` to+  `processSpanName :: Text -> Text`. The span name is now built from the+  destination (processor id), yielding e.g. `"shibuya-consumer process"`, in+  line with the OpenTelemetry messaging-spans recommendation.+- `Shibuya.Telemetry.Semantic`: remove `attrMessagingDestinationPartitionId`+  (replaced by the Shibuya-specific `attrShibuyaPartition`).+- `Shibuya.Telemetry.Semantic`: remove `eventHandlerException`.++### New Features++- `Shibuya.Telemetry.Semantic`: add `attrMessagingOperation` and+  `attrShibuyaPartition` attribute keys. Messaging attribute keys are now+  sourced from the typed `AttributeKey` values exported by+  `OpenTelemetry.SemanticConventions`, so upstream renames surface as+  compile errors rather than silent wire-format drift.+- `Shibuya.Core.Types`: add `NFData` instances for `MessageId`, `Cursor`,+  and `Envelope a` (when `a` itself has an `NFData` instance). Benchmark+  authors no longer need to declare these as orphans.+ ## 0.1.0.0 — 2026-02-24  Initial release.

@@ -1,6 +1,6 @@ cabal-version: 3.14 name: shibuya-core-version: 0.1.0.0+version: 0.2.0.0 synopsis: Supervised queue processing framework for Haskell description:   A supervised queue processing framework inspired by Broadway (Elixir).@@ -65,11 +65,13 @@     base ^>=4.21.0.0,     bytestring ^>=0.12.2.0,     containers ^>=0.7,+    deepseq ^>=1.5,     effectful ^>=2.6.1.0,     effectful-core ^>=2.6.1.0,     generic-lens ^>=2.3.0.0,     hs-opentelemetry-api ^>=0.3,     hs-opentelemetry-propagator-w3c ^>=0.1,+    hs-opentelemetry-semantic-conventions ^>=0.1,     lens ^>=5.3.5,     nqe ^>=0.6,     stm ^>=2.5,@@ -115,6 +117,7 @@     Shibuya.RunnerSpec     Shibuya.Telemetry.EffectSpec     Shibuya.Telemetry.PropagationSpec+    Shibuya.Telemetry.SemanticSpec    build-depends:     QuickCheck ^>=2.15,@@ -122,6 +125,7 @@     bytestring,     effectful,     hs-opentelemetry-api,+    hs-opentelemetry-exporter-in-memory ^>=0.0,     hspec ^>=2.11,     memory,     nqe,@@ -132,3 +136,5 @@     text,     time,     unliftio,+    unordered-containers,+    vector,

@@ -16,6 +16,7 @@   ) where +import Control.DeepSeq (NFData) import Data.ByteString (ByteString) import Data.String (IsString) import Shibuya.Prelude@@ -25,6 +26,7 @@ newtype MessageId = MessageId {unMessageId :: Text}   deriving stock (Eq, Ord, Show, Generic)   deriving newtype (IsString)+  deriving anyclass (NFData)  -- | Optional cursor / offset / global position. -- Used to track position in ordered streams.@@ -32,6 +34,7 @@   = CursorInt !Int   | CursorText !Text   deriving stock (Eq, Ord, Show, Generic)+  deriving anyclass (NFData)  -- | W3C Trace Context headers for distributed tracing. -- Contains traceparent and optionally tracestate headers.@@ -54,3 +57,4 @@     payload :: !msg   }   deriving stock (Eq, Show, Functor, Generic)+  deriving anyclass (NFData)

@@ -74,19 +74,20 @@   ) import Shibuya.Telemetry.Propagation (extractTraceContext) import Shibuya.Telemetry.Semantic-  ( attrMessagingDestinationPartitionId,+  ( attrMessagingDestinationName,     attrMessagingMessageId,+    attrMessagingOperation,     attrMessagingSystem,     attrShibuyaAckDecision,     attrShibuyaInflightCount,     attrShibuyaInflightMax,+    attrShibuyaPartition,     consumerSpanArgs,     eventAckDecision,     eventHandlerCompleted,-    eventHandlerException,     eventHandlerStarted,     mkEvent,-    processMessageSpanName,+    processSpanName,   ) import Streamly.Data.Fold qualified as Fold import Streamly.Data.Stream qualified as Stream@@ -160,7 +161,7 @@       runInIO $         -- Catch ProcessorHalt to prevent propagation via link         -- (Halt is intentional, not a failure - other processors should continue)-        ( runIngesterAndProcessor metricsVar doneVar inboxSize concurrency adapter handler+        ( runIngesterAndProcessor metricsVar procId doneVar inboxSize concurrency adapter handler             `catch` \(ProcessorHalt _) -> pure () -- Convert halt to graceful exit         )           `finally` unregisterProcessor master procId@@ -205,7 +206,7 @@   runIngesterWithMetrics metricsVar adapter.source inbox    -- Drain remaining messages from inbox-  drainInboxWithMetrics metricsVar handler inbox+  drainInboxWithMetrics metricsVar procId handler inbox    -- Mark done   liftIO $ atomically $ writeTVar doneVar True@@ -224,13 +225,14 @@ runIngesterAndProcessor ::   (IOE :> es, Tracing :> es) =>   TVar ProcessorMetrics ->+  ProcessorId ->   TVar Bool ->   Natural ->   Concurrency ->   Adapter es msg ->   Handler es msg ->   Eff es ()-runIngesterAndProcessor metricsVar doneVar inboxSize concurrency adapter handler = do+runIngesterAndProcessor metricsVar procId doneVar inboxSize concurrency adapter handler = do   -- Create bounded inbox (this is where inboxSize is used for backpressure)   inbox <- liftIO $ newBoundedInbox inboxSize @@ -247,7 +249,7 @@      UIO.withAsync ingesterWithSignal $ \_ ->       -- Processor: process messages, exit when stream done and inbox empty-      runInIO $ processUntilDrained metricsVar concurrency handler inbox streamDoneVar+      runInIO $ processUntilDrained metricsVar procId concurrency handler inbox streamDoneVar    -- Mark done when processor exits   liftIO $ atomically $ writeTVar doneVar True@@ -294,12 +296,13 @@ processUntilDrained ::   (IOE :> es, Tracing :> es) =>   TVar ProcessorMetrics ->+  ProcessorId ->   Concurrency ->   Handler es msg ->   Inbox (Ingested es msg) ->   TVar Bool ->   Eff es ()-processUntilDrained metricsVar concurrency handler inbox streamDoneVar = do+processUntilDrained metricsVar procId concurrency handler inbox streamDoneVar = do   haltRef <- liftIO $ newIORef Nothing    let maxConc = case concurrency of@@ -309,7 +312,7 @@    withEffToIO (ConcUnlift Persistent Unlimited) $ \runInIO -> do     let inboxStream = inboxToStream inbox streamDoneVar haltRef-        processAction = runInIO . processOne metricsVar maxConc haltRef handler+        processAction = runInIO . processOne metricsVar procId maxConc haltRef handler      case concurrency of       Serial ->@@ -333,10 +336,11 @@ drainInboxWithMetrics ::   (IOE :> es, Tracing :> es) =>   TVar ProcessorMetrics ->+  ProcessorId ->   Handler es msg ->   Inbox (Ingested es msg) ->   Eff es ()-drainInboxWithMetrics metricsVar handler inbox = do+drainInboxWithMetrics metricsVar procId handler inbox = do   haltRef <- liftIO $ newIORef Nothing   go haltRef   where@@ -344,7 +348,7 @@       empty <- liftIO $ mailboxEmpty inbox       unless empty $ do         ingested <- liftIO $ receive inbox-        processOne metricsVar 1 haltRef handler ingested+        processOne metricsVar procId 1 haltRef handler ingested         -- Check if halted         halted <- liftIO $ readIORef haltRef         case halted of@@ -356,25 +360,29 @@ processOne ::   (IOE :> es, Tracing :> es) =>   TVar ProcessorMetrics ->+  ProcessorId ->   Int ->   IORef (Maybe HaltReason) ->   Handler es msg ->   Ingested es msg ->   Eff es ()-processOne metricsVar maxConc haltRef handler ingested = do+processOne metricsVar procId maxConc haltRef handler ingested = do   -- Extract parent context from message headers for distributed tracing   let parentCtx = ingested.envelope.traceContext >>= extractTraceContext+      ProcessorId pidText = procId    withExtractedContext parentCtx $-    withSpan' processMessageSpanName consumerSpanArgs $ \traceSpan -> do+    withSpan' (processSpanName pidText) consumerSpanArgs $ \traceSpan -> do       -- Add messaging attributes       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 attrMessagingDestinationPartitionId p+        Just p -> addAttribute traceSpan attrShibuyaPartition p         Nothing -> pure ()        -- Increment in-flight and add inflight attributes@@ -406,7 +414,6 @@           )           ( \ex -> do               recordException traceSpan ex-              addEvent traceSpan (mkEvent eventHandlerException [])               pure $ Left $ HandlerException $ Text.pack $ show ex           ) 

@@ -1,26 +1,31 @@ -- | Semantic conventions for Shibuya OpenTelemetry instrumentation. -- Provides span names, attribute keys, and helpers following OTel conventions.+--+-- Attribute keys for the OTel @messaging.*@ namespace are derived from the+-- typed @AttributeKey@ values exported by @OpenTelemetry.SemanticConventions@,+-- so an upstream rename surfaces as a Haskell compile error rather than a+-- silent wire-format drift. module Shibuya.Telemetry.Semantic   ( -- * Span Names-    processMessageSpanName,+    processSpanName,     ingestSpanName,      -- * Attribute Keys (OTel Messaging Conventions)     attrMessagingSystem,     attrMessagingMessageId,     attrMessagingDestinationName,-    attrMessagingDestinationPartitionId,+    attrMessagingOperation,      -- * Attribute Keys (Shibuya-specific)     attrShibuyaProcessorId,     attrShibuyaInflightCount,     attrShibuyaInflightMax,     attrShibuyaAckDecision,+    attrShibuyaPartition,      -- * Event Names     eventHandlerStarted,     eventHandlerCompleted,-    eventHandlerException,     eventAckDecision,      -- * SpanArguments Helpers@@ -34,7 +39,8 @@  import Data.HashMap.Strict qualified as HashMap import Data.Text (Text)-import OpenTelemetry.Attributes (Attribute)+import OpenTelemetry.Attributes (Attribute, AttributeKey (..))+import OpenTelemetry.SemanticConventions qualified as Sem import OpenTelemetry.Trace.Core   ( NewEvent (..),     SpanArguments (..),@@ -46,8 +52,18 @@ --------------------------------------------------------------------------------  -- | Span name for processing a single message.-processMessageSpanName :: Text-processMessageSpanName = "shibuya.process.message"+--+-- 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\"@.+--+-- 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@.+processSpanName :: Text -> Text+processSpanName destination = destination <> " process"  -- | Span name for the ingester lifecycle. ingestSpanName :: Text@@ -56,63 +72,94 @@ -------------------------------------------------------------------------------- -- Attribute Keys (OTel Messaging Semantic Conventions) -- See: https://opentelemetry.io/docs/specs/semconv/messaging/messaging-spans/+--+-- Each constant is the wire string of a typed 'AttributeKey' exported by+-- 'OpenTelemetry.SemanticConventions'. Using @unkey@ (the AttributeKey+-- newtype selector) keeps these strings in lock-step with the upstream+-- code-generated keys. -------------------------------------------------------------------------------- --- | The messaging system identifier.+-- | The messaging system identifier (@messaging.system@). attrMessagingSystem :: Text-attrMessagingSystem = "messaging.system"+attrMessagingSystem = unkey Sem.messaging_system --- | The message identifier.+-- | The message identifier (@messaging.message.id@). attrMessagingMessageId :: Text-attrMessagingMessageId = "messaging.message.id"+attrMessagingMessageId = unkey Sem.messaging_message_id --- | The destination (queue/topic) name.+-- | The destination (queue/topic) name (@messaging.destination.name@). attrMessagingDestinationName :: Text-attrMessagingDestinationName = "messaging.destination.name"+attrMessagingDestinationName = unkey Sem.messaging_destination_name --- | The partition identifier.-attrMessagingDestinationPartitionId :: Text-attrMessagingDestinationPartitionId = "messaging.destination.partition.id"+-- | The messaging operation (@messaging.operation@).+--+-- One of the spec-defined enum values: @publish@, @receive@, @process@,+-- @settle@, @create@. Shibuya's per-message span uses @"process"@.+attrMessagingOperation :: Text+attrMessagingOperation = unkey Sem.messaging_operation  -------------------------------------------------------------------------------- -- Attribute Keys (Shibuya-specific)+--+-- These keys are intentionally outside the upstream OTel spec. The+-- semantic-conventions guide is explicit that non-spec attributes should+-- carry a vendor prefix; @shibuya.*@ satisfies that and cannot collide+-- with any current or future spec key. -------------------------------------------------------------------------------- --- | The processor identifier.+-- | The processor identifier (@shibuya.processor.id@).+-- Shibuya-specific: identifies a deployment-level processor instance. attrShibuyaProcessorId :: Text attrShibuyaProcessorId = "shibuya.processor.id" --- | Current number of in-flight messages.+-- | Current number of in-flight messages (@shibuya.inflight.count@).+-- Shibuya-specific: a property of Shibuya's supervision/concurrency model. attrShibuyaInflightCount :: Text attrShibuyaInflightCount = "shibuya.inflight.count" --- | Maximum concurrency limit.+-- | Maximum concurrency limit (@shibuya.inflight.max@).+-- Shibuya-specific: a property of Shibuya's supervision/concurrency model. attrShibuyaInflightMax :: Text attrShibuyaInflightMax = "shibuya.inflight.max" --- | The ack decision made by the handler.+-- | The ack decision made by the handler (@shibuya.ack.decision@).+-- Shibuya-specific: there is no upstream key for an explicit ack decision. attrShibuyaAckDecision :: Text attrShibuyaAckDecision = "shibuya.ack.decision" +-- | A generic partition identifier (@shibuya.partition@).+--+-- Shibuya-specific: in semantic-conventions v1.24 there is no portable+-- @messaging.destination.partition.id@ key — only system-specific ones+-- (@messaging.kafka.destination.partition@,+-- @messaging.eventhubs.destination.partition.id@). System-specific+-- adapters that know which broker is in use may emit those in addition,+-- but a generic Shibuya envelope only carries an opaque partition string,+-- which we surface under this vendor-prefixed key.+attrShibuyaPartition :: Text+attrShibuyaPartition = "shibuya.partition"+ -------------------------------------------------------------------------------- -- Event Names+--+-- Shibuya-namespaced; the upstream spec does not define handler.* events.+-- The standard @exception@ event (with @exception.type@/@exception.message@/+-- @exception.stacktrace@) is emitted by 'OpenTelemetry.Trace.Core.recordException'+-- and so is not duplicated here. -------------------------------------------------------------------------------- --- | Event recorded when handler execution starts.+-- | Event recorded when handler execution starts (@shibuya.handler.started@). eventHandlerStarted :: Text-eventHandlerStarted = "handler.started"+eventHandlerStarted = "shibuya.handler.started" --- | Event recorded when handler execution completes successfully.+-- | Event recorded when handler execution completes successfully+-- (@shibuya.handler.completed@). eventHandlerCompleted :: Text-eventHandlerCompleted = "handler.completed"---- | Event recorded when handler throws an exception.-eventHandlerException :: Text-eventHandlerException = "handler.exception"+eventHandlerCompleted = "shibuya.handler.completed" --- | Event recorded when ack decision is made.+-- | Event recorded when ack decision is made (@shibuya.ack.decision@). eventAckDecision :: Text-eventAckDecision = "ack.decision"+eventAckDecision = "shibuya.ack.decision"  -------------------------------------------------------------------------------- -- SpanArguments Helpers

@@ -9,6 +9,7 @@ import Shibuya.RunnerSpec qualified import Shibuya.Telemetry.EffectSpec qualified import Shibuya.Telemetry.PropagationSpec qualified+import Shibuya.Telemetry.SemanticSpec qualified import Test.Hspec  main :: IO ()@@ -20,3 +21,4 @@   Shibuya.Runner.SupervisedSpec.spec   Shibuya.Telemetry.EffectSpec.spec   Shibuya.Telemetry.PropagationSpec.spec+  Shibuya.Telemetry.SemanticSpec.spec

@@ -0,0 +1,124 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Asserts that spans emitted by 'processOne' carry the wire names+-- defined by the OpenTelemetry messaging semantic conventions and the+-- shibuya-namespaced fallbacks. This is the guard that catches drift+-- between Shibuya's emitted attributes and the upstream spec — a future+-- rename in @hs-opentelemetry-semantic-conventions@ will break both+-- compilation (because @Shibuya.Telemetry.Semantic@ derives the strings+-- from typed @AttributeKey@s) and these assertions (because the wire+-- string changed).+module Shibuya.Telemetry.SemanticSpec (spec) where++import Data.Foldable (toList)+import Data.HashMap.Strict (HashMap)+import Data.HashMap.Strict qualified as HashMap+import Data.IORef (readIORef)+import Data.Text (Text)+import Effectful (runEff)+import OpenTelemetry.Attributes+  ( Attribute (..),+    PrimitiveAttribute (..),+    emptyAttributes,+    getAttributeMap,+  )+import OpenTelemetry.Exporter.InMemory.Span (inMemoryListExporter)+import OpenTelemetry.Trace.Core+  ( Event (..),+    ImmutableSpan (..),+    InstrumentationLibrary (..),+    createTracerProvider,+    emptyTracerProviderOptions,+    makeTracer,+    shutdownTracerProvider,+    tracerOptions,+  )+import OpenTelemetry.Util (appendOnlyBoundedCollectionValues)+import Shibuya.Adapter.Mock (listAdapter)+import Shibuya.Core.Ack (AckDecision (..))+import Shibuya.Core.AckHandle (AckHandle (..))+import Shibuya.Core.Ingested (Ingested (..))+import Shibuya.Core.Types (Envelope (..), MessageId (..))+import Shibuya.Runner.Metrics (ProcessorId (..))+import Shibuya.Runner.Supervised (runWithMetrics)+import Shibuya.Telemetry.Effect (runTracing)+import Test.Hspec++spec :: Spec+spec = describe "Shibuya.Telemetry.Semantic (wire-format)" $ do+  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++    runEff $ runTracing tracer $ do+      let envelope =+            Envelope+              { messageId = MessageId "m-1",+                cursor = Nothing,+                partition = Nothing,+                enqueuedAt = Nothing,+                traceContext = Nothing,+                payload = ("hello" :: Text)+              }+          ingested =+            Ingested+              { envelope = envelope,+                ack = AckHandle (\_ -> pure ()),+                lease = Nothing+              }+          adapter = listAdapter [ingested]+          handler _ = pure AckOk+          procId = ProcessorId "test-proc"+      _ <- runWithMetrics 4 procId adapter handler+      pure ()++    _ <- shutdownTracerProvider provider+    spans <- readIORef spansRef+    case spans of+      [s] -> do+        spanName s `shouldBe` "test-proc process"+        let attrs = getAttributeMap (spanAttributes s)+        attrs `shouldHaveTextAttribute` ("messaging.system", "shibuya")+        attrs `shouldHaveTextAttribute` ("messaging.message.id", "m-1")+        attrs `shouldHaveTextAttribute` ("messaging.destination.name", "test-proc")+        attrs `shouldHaveTextAttribute` ("messaging.operation", "process")+        attrs `shouldHaveTextAttribute` ("shibuya.ack.decision", "ack_ok")+        attrs `shouldHaveIntAttribute` ("shibuya.inflight.count", 1)+        attrs `shouldHaveIntAttribute` ("shibuya.inflight.max", 1)+        let evNames = map eventName (toList (appendOnlyBoundedCollectionValues (spanEvents s)))+        evNames `shouldContain` ["shibuya.handler.started"]+        evNames `shouldContain` ["shibuya.handler.completed"]+      _ ->+        expectationFailure $ "expected exactly one span, got " <> show (length spans)+  where+    shouldHaveTextAttribute :: HashMap Text Attribute -> (Text, Text) -> Expectation+    shouldHaveTextAttribute attrs (k, expected) =+      case HashMap.lookup k attrs of+        Just (AttributeValue (TextAttribute v)) -> v `shouldBe` expected+        Just other ->+          expectationFailure $+            "attribute " <> show k <> " was not a Text: " <> show other+        Nothing ->+          expectationFailure $+            "attribute " <> show k <> " missing; have keys " <> show (HashMap.keys attrs)+    shouldHaveIntAttribute :: HashMap Text Attribute -> (Text, Int) -> Expectation+    shouldHaveIntAttribute attrs (k, expected) =+      case HashMap.lookup k attrs of+        Just (AttributeValue (IntAttribute v)) -> v `shouldBe` fromIntegral expected+        Just other ->+          expectationFailure $+            "attribute " <> show k <> " was not an Int: " <> show other+        Nothing ->+          expectationFailure $+            "attribute " <> show k <> " missing; have keys " <> show (HashMap.keys attrs)