shibuya-pgmq-adapter 0.18.0.0 → 0.19.0.0

raw patch · 10 files changed

+376/−17 lines, 10 filesdep ~shibuya-corePVP ok

version bump matches the API change (PVP)

Dependency ranges changed: shibuya-core

API changes (from Hackage documentation)

Files

• CHANGELOG.md +35/−0
• shibuya-pgmq-adapter.cabal +4/−3
• src/Shibuya/Adapter/Pgmq.hs +8/−2
• src/Shibuya/Adapter/Pgmq/Config.hs +15/−2
• test/Main.hs +2/−0
• test/Shibuya/Adapter/Pgmq/ChaosSpec.hs +3/−3
• test/Shibuya/Adapter/Pgmq/ConfigSpec.hs +9/−1
• test/Shibuya/Adapter/Pgmq/CorrelationSpec.hs +286/−0
• test/Shibuya/Adapter/Pgmq/IntegrationSpec.hs +5/−2
• test/Shibuya/Adapter/Pgmq/InternalSpec.hs +9/−4

@@ -1,5 +1,40 @@ # Changelog +## 0.19.0.0 — 2026-11-14++### Breaking Changes++- Require [`shibuya-core`](mori://shinzui/shibuya/packages/shibuya-core) and+  `shibuya-metrics` `^>=0.13.0.0`. Core 0.13 renames processing spans from+  `<processor> process` to `process <processor>` and adds+  `messaging.operation.name = "process"`; update trace queries, dashboards and+  alerts that match the old span name.+- Add the strict field `telemetry :: AdapterTelemetry` to `PgmqAdapterConfig`.+  Configurations built from `defaultConfig` with record update need no change;+  a full `PgmqAdapterConfig { .. }` literal must set it, normally to+  `defaultAdapterTelemetry`.++### New Features++- Choose how a processing span relates to the span that sent the message with+  `telemetry = defaultAdapterTelemetry {traceCorrelation = LinkOnly}`. The+  default, `ParentAndLink`, keeps the processing span in the sender's trace as+  its child and now also links to the sender. `LinkOnly` starts each delivery+  in its own trace, linked to the sender, which suits delayed sends and long+  retry cycles. Without a usable `traceparent` neither adds a link. Re-export+  `AdapterTelemetry (traceCorrelation)`, `defaultAdapterTelemetry` and+  `TraceCorrelation (..)` from `Shibuya.Adapter.Pgmq`.+- Under `LinkOnly`, a dead-lettered message's `traceparent` names the+  consumer's processing span in its own trace; the sender's original header is+  still preserved under `x-shibuya-upstream-traceparent`. Dead-letter header+  handling itself is unchanged.++### Other Changes++- Add PostgreSQL-backed correlation tests covering both policies, redelivery+  and the dead-letter headers, and let the example consumer select `LinkOnly`+  with `OTEL_TRACE_LINK_ONLY=true`.+ ## 0.18.0.0 — 2026-10-27  ### Breaking Changes

@@ -1,6 +1,6 @@ cabal-version: 3.12 name: shibuya-pgmq-adapter-version: 0.18.0.0+version: 0.19.0.0 synopsis: PGMQ adapter for the Shibuya queue processing framework description:   A Shibuya adapter that integrates with pgmq (PostgreSQL Message Queue)@@ -51,7 +51,7 @@     pgmq-core ^>=0.7.0.0,     pgmq-effectful ^>=0.7.0.0,     pgmq-hasql ^>=0.7.0.0,-    shibuya-core ^>=0.12.0.0,+    shibuya-core ^>=0.13.0.0,     stm ^>=2.5,     streamly ^>=0.11,     streamly-core ^>=0.3,@@ -101,6 +101,7 @@     Shibuya.Adapter.Pgmq.ConfigSpec     Shibuya.Adapter.Pgmq.Convert     Shibuya.Adapter.Pgmq.ConvertSpec+    Shibuya.Adapter.Pgmq.CorrelationSpec     Shibuya.Adapter.Pgmq.IntegrationSpec     Shibuya.Adapter.Pgmq.Internal     Shibuya.Adapter.Pgmq.InternalSpec@@ -129,7 +130,7 @@     pgmq-migration ^>=0.7.0.0,     quickcheck-instances ^>=0.3,     random,-    shibuya-core ^>=0.12.0.0,+    shibuya-core ^>=0.13.0.0,     shibuya-pgmq-adapter,     stm,     streamly ^>=0.11,

@@ -88,6 +88,11 @@     FifoReadStrategy (..),     validateConfig, +    -- * Telemetry+    AdapterTelemetry (traceCorrelation),+    defaultAdapterTelemetry,+    TraceCorrelation (..),+     -- * Smart Constructors     directDeadLetter,     topicDeadLetter,@@ -146,7 +151,7 @@     routingKeyToText,     topicPatternToText,   )-import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter (Adapter (..), AdapterTelemetry (traceCorrelation), TraceCorrelation (..), defaultAdapterTelemetry) import Shibuya.Adapter.Pgmq.Config   ( DeadLetterConfig (..),     DeadLetterTarget (..),@@ -216,7 +221,8 @@           Adapter             { adapterName = "pgmq:" <> queueNameToText validConfig.queueName,               source = messageSource,-              shutdown = liftIO $ atomically $ writeTVar shutdownVar True+              shutdown = liftIO $ atomically $ writeTVar shutdownVar True,+              telemetry = validConfig.telemetry             }  -- | Poll in chunks so a shutdown can release messages that were read from pgmq

@@ -42,6 +42,7 @@ import Pgmq.Effectful (PgmqRuntimeError) import Pgmq.Types (QueueName, RoutingKey) import Pgmq.Types qualified as Pgmq+import Shibuya.Adapter (AdapterTelemetry, defaultAdapterTelemetry)  -- | Runtime resources and callbacks used by the adapter. data PgmqAdapterEnv = PgmqAdapterEnv@@ -99,7 +100,18 @@     -- See 'PrefetchConfig' for the visibility-timeout trade-off and the bounded,     -- loss-free shutdown behaviour (buffered messages are redelivered after     -- their visibility timeout rather than released immediately).-    prefetchConfig :: !(Maybe PrefetchConfig)+    prefetchConfig :: !(Maybe PrefetchConfig),+    -- | Telemetry options copied unchanged into the 'Shibuya.Adapter.Adapter'.+    --+    -- The default is 'defaultAdapterTelemetry', whose @traceCorrelation@ is+    -- 'Shibuya.Adapter.ParentAndLink': the processing span is a child of the+    -- span that sent the message and also links to it, so a job appears in+    -- the sending request's trace. Choose 'Shibuya.Adapter.LinkOnly' with+    -- @defaultAdapterTelemetry {traceCorrelation = LinkOnly}@ when messages+    -- are sent with long delays or retried over long visibility timeouts:+    -- each delivery then starts its own trace and is related to the sender+    -- only by the link, instead of extending a trace that ended long ago.+    telemetry :: !AdapterTelemetry   }   deriving stock (Show, Eq, Generic) @@ -295,5 +307,6 @@       haltVisibilityTimeout = Nothing,       maxRetries = 3,       fifoConfig = Nothing,-      prefetchConfig = Nothing+      prefetchConfig = Nothing,+      telemetry = defaultAdapterTelemetry     }

@@ -3,6 +3,7 @@ import Shibuya.Adapter.Pgmq.ChaosSpec qualified as ChaosSpec import Shibuya.Adapter.Pgmq.ConfigSpec qualified as ConfigSpec import Shibuya.Adapter.Pgmq.ConvertSpec qualified as ConvertSpec+import Shibuya.Adapter.Pgmq.CorrelationSpec qualified as CorrelationSpec import Shibuya.Adapter.Pgmq.DeadLetterSpec qualified as DeadLetterSpec import Shibuya.Adapter.Pgmq.IntegrationSpec qualified as IntegrationSpec import Shibuya.Adapter.Pgmq.InternalSpec qualified as InternalSpec@@ -19,4 +20,5 @@     describe "Shibuya.Adapter.Pgmq Properties" PropertySpec.spec   describe "Integration Tests" IntegrationSpec.spec   describe "Chaos Tests" ChaosSpec.spec+  describe "Trace Correlation Tests" CorrelationSpec.spec   DeadLetterSpec.spec

@@ -40,7 +40,7 @@ import Pgmq.Hasql.Sessions qualified as Sessions import Pgmq.Hasql.Statements.Types (BindTopic (..), ReadMessage (..), SendMessage (..), SendMessageWithHeaders (..)) import Pgmq.Types (Message (..), MessageBody (..), MessageHeaders (..), QueueName, parseQueueName, parseRoutingKey, parseTopicPattern)-import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter (Adapter (..), defaultAdapterTelemetry) import Shibuya.Adapter.Pgmq   ( PgmqAcknowledgementException,     PgmqAdapterConfig (..),@@ -386,7 +386,7 @@         Vector.length messages `shouldBe` 1         spans <- readIORef spansRef         named <- mapM (\sp -> do hot <- readIORef (OTel.spanHot sp); pure (OTel.hotName hot, sp)) spans-        case (Vector.uncons messages, [sp | ("dlq-identity process", sp) <- named]) of+        case (Vector.uncons messages, [sp | ("process dlq-identity", sp) <- named]) of           (Just (msg, _), [consumer]) -> case msg.headers of             Just (Object hdrs) -> do               let stringHeader k = case KeyMap.lookup k hdrs of@@ -695,7 +695,7 @@           case ingestedResult of             Nothing -> liftIO (expectationFailure "message should not auto-DLQ") >> pure ""             Just ingested -> do-              let adapter = Adapter "pgmq:test-terminal-ack-failure" (Stream.fromList [ingested]) (pure ())+              let adapter = Adapter "pgmq:test-terminal-ack-failure" (Stream.fromList [ingested]) (pure ()) defaultAdapterTelemetry                   processor = mkProcessor adapter (\_ -> pure $ AckDeadLetter $ PoisonPill "fail loudly")               appResult <- runApp defaultAppConfig [(processorId, processor)]               case appResult of

@@ -2,7 +2,7 @@  import Data.Either (isLeft, isRight) import Pgmq.Types (parseRoutingKey)-import Shibuya.Adapter.Pgmq (parseQueueName)+import Shibuya.Adapter.Pgmq (AdapterTelemetry (traceCorrelation), TraceCorrelation (..), defaultAdapterTelemetry, parseQueueName) import Shibuya.Adapter.Pgmq.Config import Test.Hspec @@ -55,6 +55,14 @@    it "sets fifoConfig to Nothing" $ do     config.fifoConfig `shouldBe` Nothing++  it "uses the default adapter telemetry, whose correlation is ParentAndLink" $ do+    config.telemetry `shouldBe` defaultAdapterTelemetry+    config.telemetry.traceCorrelation `shouldBe` ParentAndLink++  it "validates a LinkOnly override and keeps it" $ do+    let linkOnly = config {telemetry = defaultAdapterTelemetry {traceCorrelation = LinkOnly}}+    fmap (\c -> c.telemetry.traceCorrelation) (validateConfig linkOnly) `shouldBe` Right LinkOnly  -- | Tests for defaultPollingConfig defaultPollingConfigSpec :: Spec

@@ -0,0 +1,286 @@+-- | Trace correlation of PGMQ processing spans against a real PostgreSQL.+--+-- Each case sends a message carrying a known producer @traceparent@, runs a+-- traced Shibuya app over the adapter with an in-memory span exporter, and+-- inspects the exported Consumer spans: their trace, their parent and their+-- links. The dead-letter cases also read the dead-letter queue message's+-- headers, which 'Shibuya.Adapter.Pgmq.Internal.mergeDlqHeaders' writes from+-- the active processing span.+--+-- Run with: cabal test shibuya-pgmq-adapter-test --test-option=--match --test-option=Correlation+module Shibuya.Adapter.Pgmq.CorrelationSpec (spec) where++import Control.Concurrent (threadDelay)+import Control.Exception qualified as Exception+import Control.Monad (void)+import Data.Aeson (Value (..), object, (.=))+import Data.Aeson.Key qualified as Key+import Data.Aeson.KeyMap qualified as KeyMap+import Data.Foldable (toList)+import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef)+import Data.Maybe (isNothing)+import Data.Text (Text)+import Data.Text.Encoding qualified as Text+import Data.Vector qualified as Vector+import Effectful (Eff, IOE, liftIO, runEff, (:>))+import Effectful.Error.Static (Error, runErrorNoCallStack)+import Hasql.Pool qualified as Pool+import OpenTelemetry.Attributes (emptyAttributes)+import OpenTelemetry.Exporter.InMemory (inMemoryListExporter)+import OpenTelemetry.Trace.Core qualified as OTel+import OpenTelemetry.Trace.Id (Base (..))+import OpenTelemetry.Trace.Id qualified as OTel.Id+import OpenTelemetry.Trace.Id.Generator (IdGenerator (..))+import OpenTelemetry.Util (appendOnlyBoundedCollectionValues)+import Pgmq.Effectful (Pgmq, PgmqRuntimeError, runPgmq)+import Pgmq.Hasql.Sessions qualified as Sessions+import Pgmq.Hasql.Statements.Types (ReadMessage (..), SendMessageWithHeaders (..))+import Pgmq.Types (Message (..), MessageBody (..), MessageHeaders (..))+import Shibuya.Adapter (Adapter (..))+import Shibuya.Adapter.Pgmq+  ( AdapterTelemetry (traceCorrelation),+    PgmqAdapterConfig (..),+    PollingConfig (..),+    TraceCorrelation (..),+    defaultAdapterTelemetry,+    defaultConfig,+    directDeadLetter,+    mkPgmqAdapterEnv,+    pgmqAdapter,+  )+import Shibuya.App (ProcessorId (..), ShutdownConfig (..), defaultAppConfig, defaultShutdownConfig, mkProcessor, runApp, stopAppGracefully)+import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), RetryDelay (..))+import Shibuya.Core.Ingested (Message (..))+import Shibuya.Core.Types (Attempt (..), Envelope (..))+import Shibuya.Handler (Handler)+import Shibuya.Telemetry.Effect (Tracing, runTracing)+import Shibuya.Telemetry.Propagation (extractTraceContext)+import Shibuya.Telemetry.Semantic (processSpanName)+import System.Environment (lookupEnv)+import Test.Hspec+import TmpPostgres (TestFixture (..), runPgmqSession, withPgmqDb, withTestFixture)++spec :: Spec+spec = around withTempDbFixture $ describe "Correlation" $ do+  it "ParentAndLink: process span is a child of the producer with one link" $ \fixture -> do+    run <- runCorrelated fixture defaultAdapterTelemetry False (countingHandler AckOk) 1+    run.adapterPolicy `shouldBe` ParentAndLink+    [sp] <- pure run.spans+    sp.name `shouldBe` processSpanName processorName+    hexTrace sp.context `shouldBe` producerTraceHex+    fmap hexSpan sp.parent `shouldBe` Just producerSpanHex+    linkIds sp `shouldBe` [(producerTraceHex, producerSpanHex)]++  it "LinkOnly: process span is a linked root" $ \fixture -> do+    run <- runCorrelated fixture linkOnly False (countingHandler AckOk) 1+    run.adapterPolicy `shouldBe` LinkOnly+    [sp] <- pure run.spans+    hexTrace sp.context `shouldNotBe` producerTraceHex+    isNothing sp.parent `shouldBe` True+    linkIds sp `shouldBe` [(producerTraceHex, producerSpanHex)]++  it "ParentAndLink: every redelivery stays in the producer's trace" $ \fixture -> do+    run <- runCorrelated fixture defaultAdapterTelemetry False retryOnceHandler 2+    length run.spans `shouldBe` 2+    map (hexTrace . (.context)) run.spans `shouldBe` [producerTraceHex, producerTraceHex]+    map (fmap hexSpan . (.parent)) run.spans `shouldBe` [Just producerSpanHex, Just producerSpanHex]+    map linkIds run.spans `shouldBe` replicate 2 [(producerTraceHex, producerSpanHex)]++  it "LinkOnly: each redelivery is its own linked trace" $ \fixture -> do+    run <- runCorrelated fixture linkOnly False retryOnceHandler 2+    case run.spans of+      [first, second] -> do+        hexTrace first.context `shouldNotBe` hexTrace second.context+        mapM_ (\sp -> hexTrace sp.context `shouldNotBe` producerTraceHex) [first, second]+        mapM_ (\sp -> isNothing sp.parent `shouldBe` True) [first, second]+        mapM_ (\sp -> linkIds sp `shouldBe` [(producerTraceHex, producerSpanHex)]) [first, second]+      other -> expectationFailure ("expected two processing spans, got " <> show (length other))++  it "LinkOnly: DLQ traceparent is the consumer's, upstream preserved" $ \fixture -> do+    run <- runCorrelated fixture linkOnly True (countingHandler (AckDeadLetter (PoisonPill "correlation"))) 1+    [sp] <- pure run.spans+    isNothing sp.parent `shouldBe` True+    linkIds sp `shouldBe` [(producerTraceHex, producerSpanHex)]+    hdrs <- readDlqHeaders fixture+    let stringHeader :: Text -> Maybe Text+        stringHeader k = case KeyMap.lookup (Key.fromText k) hdrs of+          Just (String v) -> Just v+          _ -> Nothing+        active = [(Text.encodeUtf8 k, Text.encodeUtf8 v) | k <- ["traceparent", "tracestate"], Just v <- [stringHeader k]]+    case extractTraceContext active of+      Nothing -> expectationFailure ("expected a decodable consumer traceparent in the DLQ, got " <> show hdrs)+      Just ctx -> do+        hexTrace ctx `shouldBe` hexTrace sp.context+        hexSpan ctx `shouldBe` hexSpan sp.context+        hexTrace ctx `shouldNotBe` producerTraceHex+        hexSpan ctx `shouldNotBe` producerSpanHex+    stringHeader "x-shibuya-upstream-traceparent" `shouldBe` Just producerParent+    stringHeader "x-shibuya-upstream-tracestate" `shouldBe` Just producerState++--------------------------------------------------------------------------------+-- Fixtures+--------------------------------------------------------------------------------++producerParent :: Text+producerParent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"++producerState :: Text+producerState = "vendor=opaque"++producerTraceHex :: Text+producerTraceHex = "0af7651916cd43dd8448eb211c80319c"++producerSpanHex :: Text+producerSpanHex = "b7ad6b7169203331"++processorName :: Text+processorName = "pgmq-correlation"++linkOnly :: AdapterTelemetry+linkOnly = defaultAdapterTelemetry {traceCorrelation = LinkOnly}++data CorrelatedRun = CorrelatedRun+  { adapterPolicy :: TraceCorrelation,+    spans :: [ConsumerSpan]+  }++-- | Send one message with the producer headers, process it with the given+-- telemetry until the handler has been called @deliveries@ times, and return+-- the exported Consumer spans in the order they ended.+runCorrelated ::+  TestFixture ->+  AdapterTelemetry ->+  Bool ->+  (IORef Int -> Handler '[Tracing, Pgmq, Error PgmqRuntimeError, IOE] Value) ->+  Int ->+  IO CorrelatedRun+runCorrelated TestFixture {pool, queueName, dlqName} telemetry withDlq handler deliveries = do+  runPgmqSession pool $+    void $+      Sessions.sendMessageWithHeaders $+        SendMessageWithHeaders+          { queueName = queueName,+            messageBody = MessageBody (String "correlated"),+            messageHeaders = MessageHeaders (object ["traceparent" .= producerParent, "tracestate" .= producerState]),+            delay = Just 0+          }+  let config =+        (defaultConfig queueName)+          { visibilityTimeout = 5,+            batchSize = 1,+            polling = StandardPolling {pollInterval = 0.05},+            deadLetterConfig = if withDlq then Just (directDeadLetter dlqName True) else Nothing,+            telemetry = telemetry+          }+  calls <- newIORef (0 :: Int)+  (spanProcessor, spansRef) <- inMemoryListExporter+  policy <-+    Exception.bracket+      (OTel.createTracerProvider [spanProcessor] OTel.emptyTracerProviderOptions {OTel.tracerProviderOptionsIdGenerator = DefaultIdGenerator})+      (\p -> OTel.shutdownTracerProvider p (Just 5_000_000))+      $ \provider -> do+        let tracer = OTel.makeTracer provider (OTel.InstrumentationLibrary "pgmq-correlation" "" "" emptyAttributes) OTel.tracerOptions+        runAdapterIO pool $ runTracing tracer $ do+          adapter <- requireAdapter pool config+          result <- runApp defaultAppConfig [(ProcessorId processorName, mkProcessor adapter (handler calls))]+          case result of+            Left err -> liftIO (expectationFailure (show err))+            Right appHandle -> do+              liftIO $ waitForCalls calls deliveries 5_000_000+              drained <- stopAppGracefully defaultShutdownConfig {drainTimeout = 5} appHandle+              liftIO $ drained `shouldBe` True+          pure adapter.telemetry.traceCorrelation+  readIORef calls `shouldReturn` deliveries+  exported <- readIORef spansRef+  consumers <- consumerSpans exported+  pure CorrelatedRun {adapterPolicy = policy, spans = consumers}++countingHandler :: (IOE :> es) => AckDecision -> IORef Int -> Handler es Value+countingHandler decision calls _ = do+  liftIO $ atomicModifyIORef' calls (\n -> (n + 1, ()))+  pure decision++-- | Ask for an immediate redelivery on the first attempt, succeed afterwards.+retryOnceHandler :: (IOE :> es) => IORef Int -> Handler es Value+retryOnceHandler calls message = do+  liftIO $ atomicModifyIORef' calls (\n -> (n + 1, ()))+  pure $ case message.envelope.attempt of+    Just (Attempt n) | n > 0 -> AckOk+    _ -> AckRetry (RetryDelay 0)++readDlqHeaders :: TestFixture -> IO (KeyMap.KeyMap Value)+readDlqHeaders TestFixture {pool, dlqName} = do+  messages <- runPgmqSession pool $ Sessions.readMessage $ ReadMessage dlqName 30 (Just 10) Nothing+  case Vector.toList messages of+    [msg] -> case msg.headers of+      Just (Object hdrs) -> pure hdrs+      other -> expectationFailure ("expected DLQ headers, got " <> show other) >> pure KeyMap.empty+    other -> expectationFailure ("expected one DLQ message, got " <> show (length other)) >> pure KeyMap.empty++waitForCalls :: IORef Int -> Int -> Int -> IO ()+waitForCalls ref target = go+  where+    go remaining+      | remaining <= 0 = pure ()+      | otherwise = do+          n <- readIORef ref+          if n >= target then pure () else threadDelay 20_000 >> go (remaining - 20_000)++withTempDbFixture :: (TestFixture -> IO ()) -> IO ()+withTempDbFixture action = do+  skipDb <- lookupEnv "PGMQ_TEST_SKIP_DB"+  case skipDb of+    Just "1" -> pendingWith "Database tests skipped (PGMQ_TEST_SKIP_DB=1)"+    _ -> do+      result <- withPgmqDb $ \pool -> withTestFixture pool action+      case result of+        Left err -> expectationFailure $ "Failed to start temp database: " <> show err+        Right () -> pure ()++runAdapterIO :: Pool.Pool -> Eff '[Pgmq, Error PgmqRuntimeError, IOE] a -> IO a+runAdapterIO pool action = do+  result <- runEff $ runErrorNoCallStack $ runPgmq pool action+  either (\err -> error ("Pgmq error: " <> show err)) pure result++requireAdapter ::+  (Pgmq :> es, Error PgmqRuntimeError :> es, IOE :> es, Tracing :> es) =>+  Pool.Pool ->+  PgmqAdapterConfig ->+  Eff es (Adapter es Value)+requireAdapter pool config =+  pgmqAdapter (mkPgmqAdapterEnv pool) config+    >>= either (\err -> liftIO (error ("Invalid PGMQ adapter config: " <> show err))) pure++--------------------------------------------------------------------------------+-- Span inspection+--------------------------------------------------------------------------------++data ConsumerSpan = ConsumerSpan+  { name :: Text,+    context :: OTel.SpanContext,+    parent :: Maybe OTel.SpanContext,+    links :: [OTel.SpanContext]+  }++consumerSpans :: [OTel.ImmutableSpan] -> IO [ConsumerSpan]+consumerSpans exported = traverse describeSpan [sp | sp <- exported, OTel.spanKind sp == OTel.Consumer]+  where+    describeSpan sp = do+      hot <- readIORef (OTel.spanHot sp)+      parent <- traverse OTel.getSpanContext (OTel.spanParent sp)+      pure+        ConsumerSpan+          { name = OTel.hotName hot,+            context = OTel.spanContext sp,+            parent,+            links = map OTel.frozenLinkContext (toList (appendOnlyBoundedCollectionValues (OTel.hotLinks hot)))+          }++linkIds :: ConsumerSpan -> [(Text, Text)]+linkIds sp = [(hexTrace l, hexSpan l) | l <- sp.links]++hexTrace :: OTel.SpanContext -> Text+hexTrace ctx = OTel.Id.traceIdBaseEncodedText Base16 ctx.traceId++hexSpan :: OTel.SpanContext -> Text+hexSpan ctx = OTel.Id.spanIdBaseEncodedText Base16 ctx.spanId

@@ -34,6 +34,7 @@     VisibilityTimeoutQuery (..),   ) import Pgmq.Types (Message (..), MessageBody (..), MessageHeaders (..), MessageId, QueueName)+import Shibuya.Adapter (defaultAdapterTelemetry) import Shibuya.Adapter.Pgmq.Config   ( FifoConfig (..),     FifoReadStrategy (..),@@ -448,7 +449,8 @@       haltVisibilityTimeout = Nothing,       maxRetries = 3,       fifoConfig = Nothing,-      prefetchConfig = Nothing+      prefetchConfig = Nothing,+      telemetry = defaultAdapterTelemetry     }  -- | Create config with custom batch size@@ -465,5 +467,6 @@       haltVisibilityTimeout = Nothing,       maxRetries = 3,       fifoConfig = Nothing,-      prefetchConfig = Nothing+      prefetchConfig = Nothing,+      telemetry = defaultAdapterTelemetry     }

@@ -18,6 +18,7 @@ import Pgmq.Effectful.Effect qualified as PgmqEffect import Pgmq.Hasql.Statements.Types (ReadGrouped (..), ReadGroupedWithPoll (..), ReadMessage (..), ReadWithPollMessage (..)) import Pgmq.Types (Message (..), MessageBody (..), MessageId (..), parseQueueName, queueNameToText)+import Shibuya.Adapter (defaultAdapterTelemetry) import Shibuya.Adapter.Pgmq.Config   ( FifoConfig (..),     FifoReadStrategy (..),@@ -104,7 +105,8 @@             haltVisibilityTimeout = Nothing,             maxRetries = 3,             fifoConfig = Nothing,-            prefetchConfig = Nothing+            prefetchConfig = Nothing,+            telemetry = defaultAdapterTelemetry           }       ReadMessage         { queueName = queryQueueName,@@ -143,7 +145,8 @@             haltVisibilityTimeout = Nothing,             maxRetries = 3,             fifoConfig = Nothing,-            prefetchConfig = Nothing+            prefetchConfig = Nothing,+            telemetry = defaultAdapterTelemetry           }       ReadWithPollMessage         { queueName = queryQueueName,@@ -190,7 +193,8 @@             haltVisibilityTimeout = Nothing,             maxRetries = 3,             fifoConfig = Nothing,-            prefetchConfig = Nothing+            prefetchConfig = Nothing,+            telemetry = defaultAdapterTelemetry           }       ReadGrouped         { queueName = queryQueueName,@@ -503,7 +507,8 @@           haltVisibilityTimeout = Nothing,           maxRetries = 3,           fifoConfig = Nothing,-          prefetchConfig = Nothing+          prefetchConfig = Nothing,+          telemetry = defaultAdapterTelemetry         }  runStubPgmq ::