pgmq-effectful 0.1.3.0 → 0.2.0.0

raw patch · 12 files changed

+1235/−377 lines, 12 filesdep +case-insensitivedep +ephemeral-pgdep +hs-opentelemetry-exporter-in-memorydep ~pgmq-coredep ~pgmq-hasqlPVP ok

version bump matches the API change (PVP)

Dependencies added: case-insensitive, ephemeral-pg, hs-opentelemetry-exporter-in-memory, hs-opentelemetry-sdk, hs-opentelemetry-semantic-conventions, http-types, pgmq-effectful, pgmq-migration, random, tasty, tasty-hunit

Dependency ranges changed: pgmq-core, pgmq-hasql

API changes (from Hackage documentation)

Files

• CHANGELOG.md +137/−0
• pgmq-effectful.cabal +60/−15
• src/Pgmq/Effectful.hs +15/−2
• src/Pgmq/Effectful/Interpreter.hs +67/−5
• src/Pgmq/Effectful/Interpreter/Traced.hs +264/−192
• src/Pgmq/Effectful/Telemetry.hs +124/−90
• src/Pgmq/Effectful/Traced.hs +55/−73
• test/ClassificationSpec.hs +66/−0
• test/EphemeralDb.hs +44/−0
• test/Main.hs +24/−0
• test/PlainInterpreterSpec.hs +68/−0
• test/TracedInterpreterSpec.hs +311/−0

@@ -1,5 +1,142 @@ # Revision history for pgmq-effectful +## 0.2.0.0 -- 2026-04-23++### Breaking Changes++* **OpenTelemetry semantic conventions updated to spec v1.24.** The+  traced interpreter now emits attribute names compliant with+  OpenTelemetry Semantic Conventions+  [v1.24](https://github.com/open-telemetry/semantic-conventions/tree/v1.24.0),+  sourced as typed `AttributeKey` values from the+  [`hs-opentelemetry-semantic-conventions`](https://hackage.haskell.org/package/hs-opentelemetry-semantic-conventions)+  library. Breaking consequences:++  * `messaging.operation.type` → `messaging.operation`. Values change+    from the post-v1.24 vocabulary (`"send"`, `"receive"`) to the+    v1.24 vocabulary (`"publish"`, `"receive"`). Note the verb change:+    `send` → `publish`.+  * `db.operation.name` → `db.operation`. Values are the pgmq SQL+    function name (`"pgmq.send"`, `"pgmq.read"`, `"pgmq.archive"`, …),+    not the previous free-form label.+  * `messaging.destination.routing_key` (ad-hoc, not in v1.24) is+    gone. Topic-send operations now emit the routing key as+    `messaging.destination.name`, since for pgmq topics the routing+    key is the logical destination. `ValidateRoutingKey`,+    `ValidateTopicPattern`, and `TestRouting` no longer emit a+    routing-key attribute at all.+  * Span names changed from `"pgmq <op>"` to the v1.24+    `"<operation> <destination>"` form. Examples:+    `"pgmq send"` → `"publish my-queue"`,+    `"pgmq read"` → `"receive my-queue"`,+    `"pgmq archive"` → `"pgmq.archive my-queue"`,+    `"pgmq list_queues"` → `"pgmq.list_queues"`.+  * Queue management operations (`createQueue`, `dropQueue`,+    `createPartitionedQueue`, `createUnloggedQueue`) moved from+    `Producer` span kind to `Internal`. v1.24 reserves `Producer` for+    message publishes; queue administration is not a publish.+  * Span status on failure now carries a short non-PII label+    (`"pool.acquisition_timeout"`, `"pool.connection.networking"`,+    `"pool.session.statement"`, …) instead of the raw `show` of+    `UsageError` (which bakes in SQL text and parameter values).+    Full detail remains on the standard `exception` event via+    `OpenTelemetry.Trace.Core.recordException`.++  Dashboards, saved queries, and alerts keyed on the old attribute+  names or span-name format need to be updated.++* **Trace context propagation now uses the tracer provider's+  configured propagator** (W3C by default, but B3 / Datadog / … work+  out of the box when the provider is configured with them). The old+  implementation hard-wired `hs-opentelemetry-propagator-w3c`.++  * `Pgmq.Effectful.Telemetry.injectTraceContext` / `extractTraceContext`+    take a `TracerProvider` (previously an `OTel.Span` / raw carrier).+  * `Pgmq.Effectful.Telemetry.TraceHeaders` is now+    `Network.HTTP.Types.RequestHeaders` (case-insensitive header names),+    matching the carrier type every propagator uses. The previous+    `[(ByteString, ByteString)]` alias is gone.+  * `Pgmq.Effectful.Traced.sendMessageTraced` takes a `TracerProvider`.+  * `Pgmq.Effectful.Traced.readMessageWithContext` takes a+    `TracerProvider` and now returns+    `Vector (Message, OpenTelemetry.Context.Context)`.+    Callers that specifically need the raw `SpanContext` can recover+    it via `Context.lookupSpan >>= getSpanContext`.+  * Two new helpers, `traceHeadersToJson` and `jsonToTraceHeaders`,+    handle the pgmq-over-jsonb serialization boundary.++  The `hs-opentelemetry-propagator-w3c` dependency was dropped from+  the library; the SDK still installs the W3C propagator as the+  default.++* Renamed the interpreter error type from `PgmqError` to+  `PgmqRuntimeError` and replaced its opaque `PgmqPoolError UsageError`+  constructor with three structured constructors:++      data PgmqRuntimeError+        = PgmqAcquisitionTimeout+        | PgmqConnectionError Hasql.Errors.ConnectionError+        | PgmqSessionError Hasql.Errors.SessionError++  The old `PgmqError`/`PgmqPoolError` names are retained with a+  DEPRECATED pragma and will be removed in 0.3.0.0.++* `runPgmq`'s error constraint changed from `Error PgmqError :> es` to+  `Error PgmqRuntimeError :> es`. Update any+  `runError @PgmqError` annotation to `runError @PgmqRuntimeError`.++* `runPgmqTraced` and `runPgmqTracedWith` now require+  `Error PgmqRuntimeError :> es`. Previously they had *no* error+  constraint and threw a `fail`-derived `IOError` outside the Error+  effect channel, which meant any `runError` wrapper around a traced+  program was a no-op. Code that relied on that silent swallowing now+  receives typed errors; update call sites to wrap with+  `runError @PgmqRuntimeError`.++### New Features++* `fromUsageError :: Hasql.Pool.UsageError -> PgmqRuntimeError` —+  convert raw hasql-pool errors into the pgmq-effectful error type.+  Useful when layering `pgmq-effectful` over code that already calls+  `Pool.use` directly.++* `isTransient :: PgmqRuntimeError -> Bool` — classification helper for+  retry logic. Returns True for acquisition timeouts, networking+  connection errors, unrecognized libpq connection errors, and+  session-level connection drops; False for authentication,+  compatibility, missing-types, statement, script, and driver errors.++* New test suite `pgmq-effectful-test` asserts that both interpreters+  surface typed `PgmqRuntimeError` values through the Error channel.++### Migration Guide++Before:++    import Pgmq.Effectful (PgmqError (..), runPgmq)++    handler =+      runEff . runError @PgmqError . runPgmq pool $ action++After:++    import Pgmq.Effectful (PgmqRuntimeError (..), runPgmq)++    handler =+      runEff . runError @PgmqRuntimeError . runPgmq pool $ action++For retry logic:++    import Pgmq.Effectful (isTransient)++    retryIfTransient action = do+      result <- runError @PgmqRuntimeError action+      case result of+        Right a -> pure (Right a)+        Left (_, err)+          | isTransient err -> retryIfTransient action+          | otherwise -> pure (Left err)+ ## 0.1.3.0 -- 2026-03-12  ### Other Changes

@@ -1,6 +1,6 @@ cabal-version:   3.4 name:            pgmq-effectful-version:         0.1.3.0+version:         0.2.0.0 synopsis:        Effectful effects for PGMQ (PostgreSQL Message Queue) description:   Effectful effects and interpreters for pgmq-hs, a Haskell client@@ -47,20 +47,65 @@     TypeOperators    build-depends:-    , aeson                            >=2.0   && <2.3-    , base                             >=4.18  && <5-    , bytestring                       >=0.11  && <0.13-    , effectful-core                   ^>=2.5  || ^>=2.6-    , hasql                            ^>=1.10-    , hasql-pool                       ^>=1.4-    , hs-opentelemetry-api             >=0.2   && <0.4-    , hs-opentelemetry-propagator-w3c  >=0.0   && <0.2-    , pgmq-core                        >=0.1.3 && <0.2-    , pgmq-hasql                       >=0.1.3 && <0.2-    , text                             >=2.0   && <2.2-    , unliftio                         >=0.2   && <0.3-    , unordered-containers             >=0.2   && <0.3-    , vector                           ^>=0.13+    , aeson                                  >=2.0   && <2.3+    , base                                   >=4.18  && <5+    , bytestring                             >=0.11  && <0.13+    , case-insensitive                       >=1.2   && <1.3+    , effectful-core                         ^>=2.5  || ^>=2.6+    , hasql                                  ^>=1.10+    , hasql-pool                             ^>=1.4+    , hs-opentelemetry-api                   >=0.2   && <0.4+    , hs-opentelemetry-semantic-conventions  >=0.1   && <0.2+    , http-types                             >=0.12  && <0.13+    , pgmq-core                              >=0.2   && <0.3+    , pgmq-hasql                             >=0.2   && <0.3+    , text                                   >=2.0   && <2.2+    , unliftio                               >=0.2   && <0.3+    , unordered-containers                   >=0.2   && <0.3+    , vector                                 ^>=0.13    hs-source-dirs:     src   default-language:   GHC2024++test-suite pgmq-effectful-test+  import:             warnings+  default-language:   GHC2024+  type:               exitcode-stdio-1.0+  hs-source-dirs:     test+  main-is:            Main.hs+  ghc-options:        -threaded -rtsopts -with-rtsopts=-N+  other-modules:+    ClassificationSpec+    EphemeralDb+    PlainInterpreterSpec+    TracedInterpreterSpec++  default-extensions:+    DataKinds+    ImportQualifiedPost+    LambdaCase+    OverloadedRecordDot+    OverloadedStrings+    TypeOperators++  build-depends:+    , aeson                                  >=2.0   && <2.3+    , base                                   >=4.18  && <5+    , effectful-core                         ^>=2.5  || ^>=2.6+    , ephemeral-pg                           >=0.2.1+    , hasql                                  ^>=1.10+    , hasql-pool                             ^>=1.4+    , hs-opentelemetry-api                   >=0.2   && <0.4+    , hs-opentelemetry-exporter-in-memory    >=0.0   && <0.1+    , hs-opentelemetry-propagator-w3c        >=0.0   && <0.2+    , hs-opentelemetry-sdk                   >=0.0   && <0.2+    , hs-opentelemetry-semantic-conventions  >=0.1   && <0.2+    , pgmq-core                              >=0.2   && <0.3+    , pgmq-effectful+    , pgmq-migration                         >=0.2   && <0.3+    , random                                 ^>=1.2+    , tasty                                  ^>=1.5+    , tasty-hunit                            ^>=0.10+    , text                                   >=2.0   && <2.2+    , unordered-containers                   >=0.2   && <0.3+    , vector                                 ^>=0.13

@@ -4,7 +4,6 @@      -- * Interpreters     runPgmq,-    PgmqError (..),      -- ** Traced Interpreters     runPgmqTraced,@@ -12,6 +11,14 @@     TracingConfig (..),     defaultTracingConfig, +    -- * Errors+    PgmqRuntimeError (..),+    fromUsageError,+    isTransient,++    -- ** Deprecated Error Types+    PgmqError (..),+     -- ** Traced Operations     sendMessageTraced,     readMessageWithContext,@@ -185,7 +192,13 @@     validateRoutingKey,     validateTopicPattern,   )-import Pgmq.Effectful.Interpreter (PgmqError (..), runPgmq)+import Pgmq.Effectful.Interpreter+  ( PgmqError (..),+    PgmqRuntimeError (..),+    fromUsageError,+    isTransient,+    runPgmq,+  ) import Pgmq.Effectful.Interpreter.Traced   ( TracingConfig (..),     defaultTracingConfig,

@@ -3,28 +3,90 @@     runPgmq,      -- * Error Types+    PgmqRuntimeError (..),+    fromUsageError,+    isTransient,++    -- * Legacy Error Types (deprecated; will be removed in 0.3.0)     PgmqError (..),   ) where +import Control.Exception (Exception) import Effectful (Eff, IOE, (:>)) import Effectful qualified import Effectful.Dispatch.Dynamic (interpret) import Effectful.Error.Static (Error, throwError)+import GHC.Generics (Generic)+import Hasql.Errors qualified as HasqlErrors import Hasql.Pool (Pool, UsageError) import Hasql.Pool qualified as Pool import Hasql.Session qualified import Pgmq.Effectful.Effect (Pgmq (..)) import Pgmq.Hasql.Sessions qualified as Sessions --- | Error type for pgmq operations.+-- | Structured runtime error for pgmq-effectful operations.+--+-- Constructors mirror 'Hasql.Pool.UsageError' so the mapping is direct.+-- The inner 'HasqlErrors.ConnectionError' and 'HasqlErrors.SessionError'+-- types are sum types with detailed constructors; see the Hasql+-- documentation for their full shapes.+data PgmqRuntimeError+  = -- | Timed out waiting for a connection from the pool.+    PgmqAcquisitionTimeout+  | -- | Failed to establish a connection to PostgreSQL.+    PgmqConnectionError HasqlErrors.ConnectionError+  | -- | Error during session execution (SQL statement, decoding, etc.).+    PgmqSessionError HasqlErrors.SessionError+  deriving stock (Show, Eq, Generic)++instance Exception PgmqRuntimeError++-- | Convert hasql-pool's 'UsageError' into the structured+-- 'PgmqRuntimeError'.+fromUsageError :: UsageError -> PgmqRuntimeError+fromUsageError = \case+  Pool.AcquisitionTimeoutUsageError -> PgmqAcquisitionTimeout+  Pool.ConnectionUsageError e -> PgmqConnectionError e+  Pool.SessionUsageError e -> PgmqSessionError e++-- | Is this error plausibly transient — i.e., worth retrying?+--+-- Returns 'True' for acquisition timeouts, networking connection errors,+-- uncategorized libpq connection errors, and session-level connection+-- drops. All other errors (authentication failure, compatibility+-- mismatches, missing types, statement errors, driver bugs) are treated+-- as permanent.+--+-- Note: 'HasqlErrors.OtherConnectionError' is classed as transient here+-- despite hasql's documentation calling it \"not transient by default\",+-- because it is the catch-all for unrecognized libpq errors and classing+-- the unknown as transient errs toward letting retries happen.+isTransient :: PgmqRuntimeError -> Bool+isTransient = \case+  PgmqAcquisitionTimeout -> True+  PgmqConnectionError e -> case e of+    HasqlErrors.NetworkingConnectionError _ -> True+    HasqlErrors.AuthenticationConnectionError _ -> False+    HasqlErrors.CompatibilityConnectionError _ -> False+    HasqlErrors.OtherConnectionError _ -> True+  PgmqSessionError e -> case e of+    HasqlErrors.ConnectionSessionError _ -> True+    HasqlErrors.StatementSessionError {} -> False+    HasqlErrors.ScriptSessionError {} -> False+    HasqlErrors.MissingTypesSessionError _ -> False+    HasqlErrors.DriverSessionError _ -> False++-- | Legacy error type. Retained for one release cycle; migrate to+-- 'PgmqRuntimeError'. Will be removed in pgmq-effectful 0.3.0. newtype PgmqError = PgmqPoolError UsageError   deriving stock (Show)+{-# DEPRECATED PgmqError, PgmqPoolError "Use PgmqRuntimeError instead (available from Pgmq.Effectful). The legacy types will be removed in pgmq-effectful 0.3.0." #-}  -- | Run the Pgmq effect using a connection pool.--- Errors are thrown via the 'Error' effect.+-- Errors are thrown via the 'Error' effect as 'PgmqRuntimeError'. runPgmq ::-  (IOE :> es, Error PgmqError :> es) =>+  (IOE :> es, Error PgmqRuntimeError :> es) =>   Pool ->   Eff (Pgmq : es) a ->   Eff es a@@ -92,12 +154,12 @@  -- Internal helper runSession ::-  (IOE :> es, Error PgmqError :> es) =>+  (IOE :> es, Error PgmqRuntimeError :> es) =>   Pool ->   Hasql.Session.Session a ->   Eff es a runSession pool session = do   result <- Effectful.liftIO $ Pool.use pool session   case result of-    Left err -> throwError $ PgmqPoolError err+    Left err -> throwError $ fromUsageError err     Right a -> pure a

@@ -3,11 +3,46 @@ -- This module provides a traced version of the Pgmq interpreter that -- creates OpenTelemetry spans for all PGMQ operations. --+-- == OpenTelemetry Semantic Conventions+--+-- Spans emitted by this interpreter follow OpenTelemetry+-- [Semantic Conventions v1.24](https://github.com/open-telemetry/semantic-conventions/tree/v1.24.0)+-- for messaging clients and database clients.+--+-- For every operation the interpreter emits:+--+-- * @db.system = "postgresql"@+-- * @db.operation = "pgmq.<fn>"@ (for example @"pgmq.send"@, @"pgmq.read"@,+--   @"pgmq.archive"@).+--+-- For /messaging/ operations (the publish and receive families) the+-- interpreter additionally emits:+--+-- * @messaging.system = "pgmq"@+-- * @messaging.operation@ — one of @"publish"@ (every @send*@ variant,+--   including topic sends) or @"receive"@ (every @read*@ and @pop@+--   variant). The @"process"@ operation is intentionally not emitted+--   here — it belongs to the consumer, which should open its own+--   @process@ span around application-level handling of a received+--   message (see 'Pgmq.Effectful.Traced.readMessageWithContext').+-- * @messaging.destination.name@ — the queue name, or for topic sends+--   the routing key (which is the logical destination for pgmq topics).+--+-- == Span Names+--+-- Span names follow the v1.24 @"<operation> <destination>"@ form:+--+-- * Messaging spans: @"publish my-queue"@, @"receive my-queue"@.+-- * Lifecycle and observability spans: @"pgmq.archive my-queue"@,+--   @"pgmq.set_vt my-queue"@, @"pgmq.list_queues"@ (no destination).+-- -- == Span Kinds ----- * Producer spans: Queue creation, message send operations--- * Consumer spans: Message read operations--- * Internal spans: Message lifecycle (delete, archive, visibility timeout)+-- * 'OTel.Producer': every @send*@ / @send_topic*@ variant.+-- * 'OTel.Consumer': every @read*@ variant and @pop@.+-- * 'OTel.Internal': lifecycle (create, drop, archive, delete, set_vt,+--   purge, bind_topic, …) and observability (metrics, list_queues,+--   list_topic_bindings, validate_*). -- -- == Usage --@@ -22,7 +57,7 @@ --   pool <- ... -- --   runEff---     . runError @PgmqError+--     . runError @PgmqRuntimeError --     . runPgmqTraced pool tracer --     $ do --       createQueue "my-queue"@@ -40,18 +75,24 @@ where  import Control.Monad (when)+import Data.Int (Int64) import Data.Text (Text) import Data.Text qualified as T import Effectful (Eff, IOE, (:>)) import Effectful qualified import Effectful.Dispatch.Dynamic (interpret)-import Hasql.Pool (Pool, UsageError)+import Effectful.Error.Static (Error, throwError)+import Hasql.Errors qualified as HasqlErrors+import Hasql.Pool (Pool) import Hasql.Pool qualified as Pool import Hasql.Session qualified-import OpenTelemetry.Attributes (Attribute (..), PrimitiveAttribute (..))-import OpenTelemetry.Attributes.Map qualified as AttrMap+import OpenTelemetry.Attributes.Map (AttributeMap, insertByKey) import OpenTelemetry.Trace.Core qualified as OTel import Pgmq.Effectful.Effect (Pgmq (..))+import Pgmq.Effectful.Interpreter+  ( PgmqRuntimeError (..),+    fromUsageError,+  ) import Pgmq.Effectful.Telemetry import Pgmq.Hasql.Sessions qualified as Sessions import Pgmq.Hasql.Statements.Types qualified as Types@@ -76,9 +117,44 @@       includeMessageBodies = False     } +-- | Describes how a 'Pgmq' operation maps onto OpenTelemetry semantic+-- conventions v1.24.+data OpInfo = OpInfo+  { -- | The pgmq SQL function being invoked (@"pgmq.send"@ etc.).+    opDbFunction :: !Text,+    -- | The v1.24 @messaging.operation@ verb, if any. One of+    -- @"publish"@ or @"receive"@; @Nothing@ for lifecycle/observability+    -- operations.+    opMessagingKind :: !(Maybe Text),+    -- | The span kind.+    opSpanKind :: !OTel.SpanKind,+    -- | The logical destination (queue name, or the routing key for+    -- topic sends). @Nothing@ for operations without a per-queue scope+    -- (for example @list_queues@, @metrics_all@).+    opDestination :: !(Maybe Text),+    -- | A pre-known message id (for per-message operations like delete,+    -- archive, set_vt).+    opMessageId :: !(Maybe MessageId),+    -- | Batch size, for batch operations.+    opBatchCount :: !(Maybe Int)+  }++-- | An 'OpInfo' with no messaging verb, no destination, no message id,+-- and no batch count.+defaultOpInfo :: Text -> OTel.SpanKind -> OpInfo+defaultOpInfo fn kind =+  OpInfo+    { opDbFunction = fn,+      opMessagingKind = Nothing,+      opSpanKind = kind,+      opDestination = Nothing,+      opMessageId = Nothing,+      opBatchCount = Nothing+    }+ -- | Run the Pgmq effect with OpenTelemetry instrumentation. runPgmqTraced ::-  (IOE :> es) =>+  (IOE :> es, Error PgmqRuntimeError :> es) =>   Pool ->   OTel.Tracer ->   Eff (Pgmq : es) a ->@@ -87,323 +163,319 @@  -- | Run the Pgmq effect with custom tracing configuration. runPgmqTracedWith ::-  (IOE :> es) =>+  (IOE :> es, Error PgmqRuntimeError :> es) =>   Pool ->   TracingConfig ->   Eff (Pgmq : es) a ->   Eff es a runPgmqTracedWith pool config = interpret $ \_ -> \case-  -- Queue Management (Producer spans)+  -- Queue Management (Internal spans)   CreateQueue q ->-    withTracedSession config "pgmq create_queue" OTel.Producer q pool $+    withTracedOp config pool (queueOp "pgmq.create" OTel.Internal q) $       Sessions.createQueue q   DropQueue q ->-    withTracedSession config "pgmq drop_queue" OTel.Producer q pool $+    withTracedOp config pool (queueOp "pgmq.drop_queue" OTel.Internal q) $       Sessions.dropQueue q   CreatePartitionedQueue pq@(Types.CreatePartitionedQueue qn _ _) ->-    withTracedSession config "pgmq create_partitioned_queue" OTel.Producer qn pool $+    withTracedOp config pool (queueOp "pgmq.create_partitioned" OTel.Internal qn) $       Sessions.createPartitionedQueue pq   CreateUnloggedQueue q ->-    withTracedSession config "pgmq create_unlogged_queue" OTel.Producer q pool $+    withTracedOp config pool (queueOp "pgmq.create_unlogged" OTel.Internal q) $       Sessions.createUnloggedQueue q   DetachArchive _q ->     pure ()   EnableNotifyInsert cfg@(Types.EnableNotifyInsert qn _) ->-    withTracedSession config "pgmq enable_notify_insert" OTel.Internal qn pool $+    withTracedOp config pool (queueOp "pgmq.enable_notify_insert" OTel.Internal qn) $       Sessions.enableNotifyInsert cfg   DisableNotifyInsert q ->-    withTracedSession config "pgmq disable_notify_insert" OTel.Internal q pool $+    withTracedOp config pool (queueOp "pgmq.disable_notify_insert" OTel.Internal q) $       Sessions.disableNotifyInsert q   CreateFifoIndex q ->-    withTracedSession config "pgmq create_fifo_index" OTel.Internal q pool $+    withTracedOp config pool (queueOp "pgmq.create_fifo_index" OTel.Internal q) $       Sessions.createFifoIndex q   CreateFifoIndexesAll ->-    withTracedSessionNoQueue config "pgmq create_fifo_indexes_all" OTel.Internal pool $+    withTracedOp config pool (defaultOpInfo "pgmq.create_fifo_indexes_all" OTel.Internal) $       Sessions.createFifoIndexesAll-  -- Message Operations - Send (Producer spans)+  -- Message Operations - Send (Producer spans, messaging.operation=publish)   SendMessage msg@(Types.SendMessage qn _ _) ->-    withTracedSession config "pgmq send" OTel.Producer qn pool $+    withTracedOp config pool (publishOp "pgmq.send" qn) $       Sessions.sendMessage msg   SendMessageForLater msg@(Types.SendMessageForLater qn _ _) ->-    withTracedSession config "pgmq send" OTel.Producer qn pool $+    withTracedOp config pool (publishOp "pgmq.send" qn) $       Sessions.sendMessageForLater msg   BatchSendMessage msg@(Types.BatchSendMessage qn bodies _) ->-    withTracedSessionWithCount config "pgmq send_batch" OTel.Producer qn (length bodies) pool $+    withTracedOp config pool (publishBatchOp "pgmq.send_batch" qn (length bodies)) $       Sessions.batchSendMessage msg   BatchSendMessageForLater msg@(Types.BatchSendMessageForLater qn bodies _) ->-    withTracedSessionWithCount config "pgmq send_batch" OTel.Producer qn (length bodies) pool $+    withTracedOp config pool (publishBatchOp "pgmq.send_batch" qn (length bodies)) $       Sessions.batchSendMessageForLater msg   SendMessageWithHeaders msg@(Types.SendMessageWithHeaders qn _ _ _) ->-    withTracedSession config "pgmq send" OTel.Producer qn pool $+    withTracedOp config pool (publishOp "pgmq.send" qn) $       Sessions.sendMessageWithHeaders msg   SendMessageWithHeadersForLater msg@(Types.SendMessageWithHeadersForLater qn _ _ _) ->-    withTracedSession config "pgmq send" OTel.Producer qn pool $+    withTracedOp config pool (publishOp "pgmq.send" qn) $       Sessions.sendMessageWithHeadersForLater msg   BatchSendMessageWithHeaders msg@(Types.BatchSendMessageWithHeaders qn bodies _ _) ->-    withTracedSessionWithCount config "pgmq send_batch" OTel.Producer qn (length bodies) pool $+    withTracedOp config pool (publishBatchOp "pgmq.send_batch" qn (length bodies)) $       Sessions.batchSendMessageWithHeaders msg   BatchSendMessageWithHeadersForLater msg@(Types.BatchSendMessageWithHeadersForLater qn bodies _ _) ->-    withTracedSessionWithCount config "pgmq send_batch" OTel.Producer qn (length bodies) pool $+    withTracedOp config pool (publishBatchOp "pgmq.send_batch" qn (length bodies)) $       Sessions.batchSendMessageWithHeadersForLater msg-  -- Message Operations - Read (Consumer spans)+  -- Message Operations - Read (Consumer spans, messaging.operation=receive)   ReadMessage query@(Types.ReadMessage qn _ _ _) ->-    withTracedSession config "pgmq read" OTel.Consumer qn pool $+    withTracedOp config pool (receiveOp "pgmq.read" qn) $       Sessions.readMessage query   ReadWithPoll query@(Types.ReadWithPollMessage qn _ _ _ _ _) ->-    withTracedSession config "pgmq read_with_poll" OTel.Consumer qn pool $+    withTracedOp config pool (receiveOp "pgmq.read_with_poll" qn) $       Sessions.readWithPoll query   Pop query@(Types.PopMessage qn _) ->-    withTracedSession config "pgmq pop" OTel.Consumer qn pool $+    withTracedOp config pool (receiveOp "pgmq.pop" qn) $       Sessions.pop query   -- FIFO Read (Consumer spans)   ReadGrouped query@(Types.ReadGrouped qn _ _) ->-    withTracedSession config "pgmq read_grouped" OTel.Consumer qn pool $+    withTracedOp config pool (receiveOp "pgmq.read_grouped" qn) $       Sessions.readGrouped query   ReadGroupedWithPoll query@(Types.ReadGroupedWithPoll qn _ _ _ _) ->-    withTracedSession config "pgmq read_grouped_with_poll" OTel.Consumer qn pool $+    withTracedOp config pool (receiveOp "pgmq.read_grouped_with_poll" qn) $       Sessions.readGroupedWithPoll query   -- Round-robin FIFO Read (Consumer spans)   ReadGroupedRoundRobin query@(Types.ReadGrouped qn _ _) ->-    withTracedSession config "pgmq read_grouped_rr" OTel.Consumer qn pool $+    withTracedOp config pool (receiveOp "pgmq.read_grouped_rr" qn) $       Sessions.readGroupedRoundRobin query   ReadGroupedRoundRobinWithPoll query@(Types.ReadGroupedWithPoll qn _ _ _ _) ->-    withTracedSession config "pgmq read_grouped_rr_with_poll" OTel.Consumer qn pool $+    withTracedOp config pool (receiveOp "pgmq.read_grouped_rr_with_poll" qn) $       Sessions.readGroupedRoundRobinWithPoll query   -- Message Lifecycle (Internal spans)   DeleteMessage query@(Types.MessageQuery qn msgId) ->-    withTracedSessionWithMsgId config "pgmq delete" OTel.Internal qn msgId pool $+    withTracedOp config pool ((queueOp "pgmq.delete" OTel.Internal qn) {opMessageId = Just msgId}) $       Sessions.deleteMessage query   BatchDeleteMessages query@(Types.BatchMessageQuery qn msgIds) ->-    withTracedSessionWithCount config "pgmq delete_batch" OTel.Internal qn (length msgIds) pool $+    withTracedOp config pool ((queueOp "pgmq.delete" OTel.Internal qn) {opBatchCount = Just (length msgIds)}) $       Sessions.batchDeleteMessages query   ArchiveMessage query@(Types.MessageQuery qn msgId) ->-    withTracedSessionWithMsgId config "pgmq archive" OTel.Internal qn msgId pool $+    withTracedOp config pool ((queueOp "pgmq.archive" OTel.Internal qn) {opMessageId = Just msgId}) $       Sessions.archiveMessage query   BatchArchiveMessages query@(Types.BatchMessageQuery qn msgIds) ->-    withTracedSessionWithCount config "pgmq archive_batch" OTel.Internal qn (length msgIds) pool $+    withTracedOp config pool ((queueOp "pgmq.archive" OTel.Internal qn) {opBatchCount = Just (length msgIds)}) $       Sessions.batchArchiveMessages query   DeleteAllMessagesFromQueue q ->-    withTracedSession config "pgmq purge_queue" OTel.Internal q pool $+    withTracedOp config pool (queueOp "pgmq.purge_queue" OTel.Internal q) $       Sessions.deleteAllMessagesFromQueue q   ChangeVisibilityTimeout query@(Types.VisibilityTimeoutQuery qn msgId _) ->-    withTracedSessionWithMsgId config "pgmq set_vt" OTel.Internal qn msgId pool $+    withTracedOp config pool ((queueOp "pgmq.set_vt" OTel.Internal qn) {opMessageId = Just msgId}) $       Sessions.changeVisibilityTimeout query   BatchChangeVisibilityTimeout query@(Types.BatchVisibilityTimeoutQuery qn msgIds _) ->-    withTracedSessionWithCount config "pgmq set_vt_batch" OTel.Internal qn (length msgIds) pool $+    withTracedOp config pool ((queueOp "pgmq.set_vt" OTel.Internal qn) {opBatchCount = Just (length msgIds)}) $       Sessions.batchChangeVisibilityTimeout query   -- Timestamp-based VT (pgmq 1.10.0+)   SetVisibilityTimeoutAt query@(Types.VisibilityTimeoutAtQuery qn msgId _) ->-    withTracedSessionWithMsgId config "pgmq set_vt_at" OTel.Internal qn msgId pool $+    withTracedOp config pool ((queueOp "pgmq.set_vt" OTel.Internal qn) {opMessageId = Just msgId}) $       Sessions.setVisibilityTimeoutAt query   BatchSetVisibilityTimeoutAt query@(Types.BatchVisibilityTimeoutAtQuery qn msgIds _) ->-    withTracedSessionWithCount config "pgmq set_vt_at_batch" OTel.Internal qn (length msgIds) pool $+    withTracedOp config pool ((queueOp "pgmq.set_vt" OTel.Internal qn) {opBatchCount = Just (length msgIds)}) $       Sessions.batchSetVisibilityTimeoutAt query   -- Topic Management (Internal spans, pgmq 1.11.0+)   BindTopic params@(Types.BindTopic _ qn) ->-    withTracedSession config "pgmq bind_topic" OTel.Internal qn pool $+    withTracedOp config pool (queueOp "pgmq.bind_topic" OTel.Internal qn) $       Sessions.bindTopic params   UnbindTopic params@(Types.UnbindTopic _ qn) ->-    withTracedSession config "pgmq unbind_topic" OTel.Internal qn pool $+    withTracedOp config pool (queueOp "pgmq.unbind_topic" OTel.Internal qn) $       Sessions.unbindTopic params   ValidateRoutingKey key ->-    withTracedSessionWithRoutingKey config "pgmq validate_routing_key" OTel.Internal key pool $+    withTracedOp config pool (defaultOpInfo "pgmq.validate_routing_key" OTel.Internal) $       Sessions.validateRoutingKey key-  ValidateTopicPattern _pat ->-    withTracedSessionNoQueue config "pgmq validate_topic_pattern" OTel.Internal pool $-      Sessions.validateTopicPattern _pat+  ValidateTopicPattern pat ->+    withTracedOp config pool (defaultOpInfo "pgmq.validate_topic_pattern" OTel.Internal) $+      Sessions.validateTopicPattern pat   TestRouting key ->-    withTracedSessionWithRoutingKey config "pgmq test_routing" OTel.Internal key pool $+    withTracedOp config pool (defaultOpInfo "pgmq.test_routing" OTel.Internal) $       Sessions.testRouting key   ListTopicBindings ->-    withTracedSessionNoQueue config "pgmq list_topic_bindings" OTel.Internal pool $+    withTracedOp config pool (defaultOpInfo "pgmq.list_topic_bindings" OTel.Internal) $       Sessions.listTopicBindings   ListTopicBindingsForQueue q ->-    withTracedSession config "pgmq list_topic_bindings" OTel.Internal q pool $+    withTracedOp config pool (queueOp "pgmq.list_topic_bindings" OTel.Internal q) $       Sessions.listTopicBindingsForQueue q   -- Topic Sending (Producer spans, pgmq 1.11.0+)   SendTopic msg@(Types.SendTopic rk _ _) ->-    withTracedSessionWithRoutingKey config "pgmq send_topic" OTel.Producer rk pool $+    withTracedOp config pool (publishTopicOp "pgmq.send_topic" rk) $       Sessions.sendTopic msg   SendTopicWithHeaders msg@(Types.SendTopicWithHeaders rk _ _ _) ->-    withTracedSessionWithRoutingKey config "pgmq send_topic" OTel.Producer rk pool $+    withTracedOp config pool (publishTopicOp "pgmq.send_topic" rk) $       Sessions.sendTopicWithHeaders msg   BatchSendTopic msg@(Types.BatchSendTopic rk bodies _) ->-    withTracedSessionWithRoutingKeyAndCount config "pgmq send_batch_topic" OTel.Producer rk (length bodies) pool $+    withTracedOp config pool (publishTopicBatchOp "pgmq.send_batch_topic" rk (length bodies)) $       Sessions.batchSendTopic msg   BatchSendTopicForLater msg@(Types.BatchSendTopicForLater rk bodies _) ->-    withTracedSessionWithRoutingKeyAndCount config "pgmq send_batch_topic" OTel.Producer rk (length bodies) pool $+    withTracedOp config pool (publishTopicBatchOp "pgmq.send_batch_topic" rk (length bodies)) $       Sessions.batchSendTopicForLater msg   BatchSendTopicWithHeaders msg@(Types.BatchSendTopicWithHeaders rk bodies _ _) ->-    withTracedSessionWithRoutingKeyAndCount config "pgmq send_batch_topic" OTel.Producer rk (length bodies) pool $+    withTracedOp config pool (publishTopicBatchOp "pgmq.send_batch_topic" rk (length bodies)) $       Sessions.batchSendTopicWithHeaders msg   BatchSendTopicWithHeadersForLater msg@(Types.BatchSendTopicWithHeadersForLater rk bodies _ _) ->-    withTracedSessionWithRoutingKeyAndCount config "pgmq send_batch_topic" OTel.Producer rk (length bodies) pool $+    withTracedOp config pool (publishTopicBatchOp "pgmq.send_batch_topic" rk (length bodies)) $       Sessions.batchSendTopicWithHeadersForLater msg   -- Notification Management (Internal spans, pgmq 1.11.0+)   ListNotifyInsertThrottles ->-    withTracedSessionNoQueue config "pgmq list_notify_insert_throttles" OTel.Internal pool $+    withTracedOp config pool (defaultOpInfo "pgmq.list_notify_insert_throttles" OTel.Internal) $       Sessions.listNotifyInsertThrottles   UpdateNotifyInsert params@(Types.UpdateNotifyInsert qn _) ->-    withTracedSession config "pgmq update_notify_insert" OTel.Internal qn pool $+    withTracedOp config pool (queueOp "pgmq.update_notify_insert" OTel.Internal qn) $       Sessions.updateNotifyInsert params   -- Queue Observability (Internal spans)   ListQueues ->-    withTracedSessionNoQueue config "pgmq list_queues" OTel.Internal pool $+    withTracedOp config pool (defaultOpInfo "pgmq.list_queues" OTel.Internal) $       Sessions.listQueues   QueueMetrics q ->-    withTracedSession config "pgmq metrics" OTel.Internal q pool $+    withTracedOp config pool (queueOp "pgmq.metrics" OTel.Internal q) $       Sessions.queueMetrics q   AllQueueMetrics ->-    withTracedSessionNoQueue config "pgmq metrics_all" OTel.Internal pool $+    withTracedOp config pool (defaultOpInfo "pgmq.metrics_all" OTel.Internal) $       Sessions.allQueueMetrics --- Internal helpers+-- ---------------------------------------------------------------------+-- OpInfo constructors+-- --------------------------------------------------------------------- -withTracedSession ::-  (IOE :> es) =>-  TracingConfig ->-  Text ->-  OTel.SpanKind ->-  QueueName ->-  Pool ->-  Hasql.Session.Session a ->-  Eff es a-withTracedSession config spanName kind queueName pool session = do-  let args = OTel.defaultSpanArguments {OTel.kind = kind}-  Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do-    addQueueAttributes s queueName-    runSessionIO config s pool session+-- | 'OpInfo' for a non-messaging operation scoped to a single queue.+queueOp :: Text -> OTel.SpanKind -> QueueName -> OpInfo+queueOp fn kind qn = (defaultOpInfo fn kind) {opDestination = Just (queueNameToText qn)} -withTracedSessionNoQueue ::-  (IOE :> es) =>-  TracingConfig ->-  Text ->-  OTel.SpanKind ->-  Pool ->-  Hasql.Session.Session a ->-  Eff es a-withTracedSessionNoQueue config spanName kind pool session = do-  let args = OTel.defaultSpanArguments {OTel.kind = kind}-  Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do-    addBaseAttributes s-    runSessionIO config s pool session+-- | 'OpInfo' for a @publish@ (send) on a queue.+publishOp :: Text -> QueueName -> OpInfo+publishOp fn qn =+  (queueOp fn OTel.Producer qn) {opMessagingKind = Just "publish"} -withTracedSessionWithMsgId ::-  (IOE :> es) =>+-- | 'OpInfo' for a batch @publish@ on a queue.+publishBatchOp :: Text -> QueueName -> Int -> OpInfo+publishBatchOp fn qn n = (publishOp fn qn) {opBatchCount = Just n}++-- | 'OpInfo' for a topic send (publish to a routing key).+publishTopicOp :: Text -> RoutingKey -> OpInfo+publishTopicOp fn rk =+  (defaultOpInfo fn OTel.Producer)+    { opMessagingKind = Just "publish",+      opDestination = Just (routingKeyToText rk)+    }++-- | 'OpInfo' for a batch topic send.+publishTopicBatchOp :: Text -> RoutingKey -> Int -> OpInfo+publishTopicBatchOp fn rk n = (publishTopicOp fn rk) {opBatchCount = Just n}++-- | 'OpInfo' for a @receive@ (read, pop) on a queue.+receiveOp :: Text -> QueueName -> OpInfo+receiveOp fn qn =+  (queueOp fn OTel.Consumer qn) {opMessagingKind = Just "receive"}++-- ---------------------------------------------------------------------+-- Span assembly+-- ---------------------------------------------------------------------++-- | Span name: @"<messaging.operation> <destination>"@ for messaging+-- spans, else @"<db.operation> <destination>"@; if there is no+-- destination, the operation name alone.+spanNameFor :: OpInfo -> Text+spanNameFor info =+  let op = case info.opMessagingKind of+        Just mk -> mk+        Nothing -> info.opDbFunction+   in case info.opDestination of+        Just d -> op <> " " <> d+        Nothing -> op++operationAttributes :: OpInfo -> AttributeMap+operationAttributes info =+  let base =+        insertByKey db_system ("postgresql" :: Text)+          . insertByKey db_operation info.opDbFunction+      withMsgKind = case info.opMessagingKind of+        Just mk ->+          insertByKey messaging_system ("pgmq" :: Text)+            . insertByKey messaging_operation mk+        Nothing -> id+      withDest = case info.opDestination of+        Just d -> insertByKey messaging_destination_name d+        Nothing -> id+      withMsgId = case info.opMessageId of+        Just (MessageId mid) ->+          insertByKey messaging_message_id (T.pack (show mid))+        Nothing -> id+      withCount = case info.opBatchCount of+        Just n -> insertByKey messaging_batch_messageCount (fromIntegral n :: Int64)+        Nothing -> id+   in (withMsgId . withCount . withDest . withMsgKind . base) mempty++withTracedOp ::+  (IOE :> es, Error PgmqRuntimeError :> es) =>   TracingConfig ->-  Text ->-  OTel.SpanKind ->-  QueueName ->-  MessageId ->   Pool ->+  OpInfo ->   Hasql.Session.Session a ->   Eff es a-withTracedSessionWithMsgId config spanName kind queueName msgId pool session = do-  let args = OTel.defaultSpanArguments {OTel.kind = kind}-  Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do-    addQueueAttributes s queueName-    addMessageIdAttribute s msgId-    runSessionIO config s pool session+withTracedOp config pool info session = do+  let args =+        OTel.addAttributesToSpanArguments+          (operationAttributes info)+          OTel.defaultSpanArguments {OTel.kind = info.opSpanKind}+  result <-+    Effectful.liftIO $+      OTel.inSpan' config.tracer (spanNameFor info) args $ \s ->+        runSessionIO config s pool session+  throwOnLeft result -withTracedSessionWithCount ::-  (IOE :> es) =>-  TracingConfig ->-  Text ->-  OTel.SpanKind ->-  QueueName ->-  Int ->-  Pool ->-  Hasql.Session.Session a ->+-- ---------------------------------------------------------------------+-- Session runner and error recording+-- ---------------------------------------------------------------------++-- | Rethrow a runtime error via the 'Error' effect when present.+throwOnLeft ::+  (Error PgmqRuntimeError :> es) =>+  Either PgmqRuntimeError a ->   Eff es a-withTracedSessionWithCount config spanName kind queueName count pool session = do-  let args = OTel.defaultSpanArguments {OTel.kind = kind}-  Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do-    addQueueAttributes s queueName-    addBatchCountAttribute s count-    runSessionIO config s pool session+throwOnLeft = \case+  Left err -> throwError err+  Right a -> pure a  runSessionIO ::   TracingConfig ->   OTel.Span ->   Pool ->   Hasql.Session.Session a ->-  IO a+  IO (Either PgmqRuntimeError a) runSessionIO config s pool session = do   result <- Pool.use pool session   case result of     Left err -> do-      when config.recordExceptions $ recordUsageError s err-      OTel.setStatus s (OTel.Error $ T.pack $ show err)-      fail $ "PgmqPoolError: " <> show err+      let runtimeErr = fromUsageError err+      when config.recordExceptions $+        OTel.recordException s mempty Nothing err+      OTel.setStatus s (OTel.Error (errorStatusDescription runtimeErr))+      pure $ Left runtimeErr     Right a -> do       OTel.setStatus s OTel.Ok-      pure a--recordUsageError :: OTel.Span -> UsageError -> IO ()-recordUsageError s err = do-  OTel.addEvent s $-    OTel.NewEvent-      { OTel.newEventName = "exception",-        OTel.newEventTimestamp = Nothing,-        OTel.newEventAttributes =-          AttrMap.fromList-            [ ("exception.type", AttributeValue $ TextAttribute "UsageError"),-              ("exception.message", AttributeValue $ TextAttribute $ T.pack $ show err)-            ]-      }--addQueueAttributes :: OTel.Span -> QueueName -> IO ()-addQueueAttributes s queueName = do-  addBaseAttributes s-  OTel.addAttribute s messagingDestinationName (queueNameToText queueName)--addBaseAttributes :: OTel.Span -> IO ()-addBaseAttributes s = do-  OTel.addAttribute s messagingSystem ("pgmq" :: Text)-  OTel.addAttribute s dbSystem ("postgresql" :: Text)--addMessageIdAttribute :: OTel.Span -> MessageId -> IO ()-addMessageIdAttribute s (MessageId msgId) =-  OTel.addAttribute s messagingMessageId (T.pack $ show msgId)--addBatchCountAttribute :: OTel.Span -> Int -> IO ()-addBatchCountAttribute s count =-  OTel.addAttribute s messagingBatchMessageCount count--addRoutingKeyAttribute :: OTel.Span -> RoutingKey -> IO ()-addRoutingKeyAttribute s rk =-  OTel.addAttribute s messagingRoutingKey (routingKeyToText rk)--withTracedSessionWithRoutingKey ::-  (IOE :> es) =>-  TracingConfig ->-  Text ->-  OTel.SpanKind ->-  RoutingKey ->-  Pool ->-  Hasql.Session.Session a ->-  Eff es a-withTracedSessionWithRoutingKey config spanName kind routingKey pool session = do-  let args = OTel.defaultSpanArguments {OTel.kind = kind}-  Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do-    addBaseAttributes s-    addRoutingKeyAttribute s routingKey-    runSessionIO config s pool session+      pure $ Right a -withTracedSessionWithRoutingKeyAndCount ::-  (IOE :> es) =>-  TracingConfig ->-  Text ->-  OTel.SpanKind ->-  RoutingKey ->-  Int ->-  Pool ->-  Hasql.Session.Session a ->-  Eff es a-withTracedSessionWithRoutingKeyAndCount config spanName kind routingKey count pool session = do-  let args = OTel.defaultSpanArguments {OTel.kind = kind}-  Effectful.liftIO $ OTel.inSpan' config.tracer spanName args $ \s -> do-    addBaseAttributes s-    addRoutingKeyAttribute s routingKey-    addBatchCountAttribute s count-    runSessionIO config s pool session+-- | Short, non-PII span-status description derived from a+-- 'PgmqRuntimeError'. The detail lives inside the 'recordException'+-- event; the span status description is a coarse label that backends+-- can group on without risking query text or credentials leaking into+-- dashboards.+errorStatusDescription :: PgmqRuntimeError -> Text+errorStatusDescription = \case+  PgmqAcquisitionTimeout -> "pool.acquisition_timeout"+  PgmqConnectionError e -> "pool.connection." <> connectionLabel e+  PgmqSessionError e -> "pool.session." <> sessionLabel e+  where+    connectionLabel :: HasqlErrors.ConnectionError -> Text+    connectionLabel = \case+      HasqlErrors.NetworkingConnectionError _ -> "networking"+      HasqlErrors.AuthenticationConnectionError _ -> "authentication"+      HasqlErrors.CompatibilityConnectionError _ -> "compatibility"+      HasqlErrors.OtherConnectionError _ -> "other"+    sessionLabel :: HasqlErrors.SessionError -> Text+    sessionLabel = \case+      HasqlErrors.ConnectionSessionError _ -> "connection"+      HasqlErrors.StatementSessionError {} -> "statement"+      HasqlErrors.ScriptSessionError {} -> "script"+      HasqlErrors.MissingTypesSessionError _ -> "missing_types"+      HasqlErrors.DriverSessionError _ -> "driver"

@@ -2,114 +2,148 @@ -- -- This module provides: ----- * W3C Trace Context propagation via message headers--- * OpenTelemetry semantic conventions for messaging systems+-- * Trace-context propagation through pgmq message headers, using the+--   tracer provider's /configured/ propagator (W3C Trace Context by+--   default, but any propagator installed on the provider will work —+--   B3, Datadog, …). The pgmq header payload is a JSON object whose+--   keys are header names (lower-cased) and whose values are strings.+-- * Re-exports of typed 'AttributeKey' values from+--   "OpenTelemetry.SemanticConventions" for the v1.24 attributes used+--   by the pgmq instrumentation.+--+-- The semantic-conventions names mirror OpenTelemetry specification+-- v1.24 as generated into @hs-opentelemetry-semantic-conventions@+-- 0.1.0.0. module Pgmq.Effectful.Telemetry-  ( -- * Trace Context Operations+  ( -- * Trace Context Propagation     injectTraceContext,     extractTraceContext,+    traceHeadersToJson,+    jsonToTraceHeaders,     mergeTraceHeaders,     TraceHeaders, -    -- * Semantic Conventions-    messagingSystem,-    messagingOperationType,-    messagingDestinationName,-    messagingMessageId,-    messagingBatchMessageCount,-    messagingRoutingKey,-    dbSystem,-    dbOperationName,+    -- * Semantic Convention Keys (re-exported from++    -- "OpenTelemetry.SemanticConventions")+    messaging_system,+    messaging_operation,+    messaging_destination_name,+    messaging_message_id,+    messaging_batch_messageCount,+    db_system,+    db_operation,   ) where +import Control.Monad.IO.Class (MonadIO, liftIO) import Data.Aeson (Value (..)) import Data.Aeson.Key qualified as Key import Data.Aeson.KeyMap qualified as KM-import Data.ByteString (ByteString)-import Data.Text (Text)+import Data.CaseInsensitive qualified as CI+import Data.Text qualified as T import Data.Text.Encoding qualified as TE-import OpenTelemetry.Propagator.W3CTraceContext qualified as W3C-import OpenTelemetry.Trace.Core qualified as OTel---- | Headers for W3C Trace Context propagation.--- Contains traceparent and optionally tracestate.-type TraceHeaders = [(ByteString, ByteString)]---- | Inject trace context into message headers (for producers).--- Creates traceparent and tracestate headers from current span context.-injectTraceContext :: OTel.Span -> IO TraceHeaders-injectTraceContext otelSpan = do-  (traceparent, tracestate) <- W3C.encodeSpanContext otelSpan-  pure-    [ ("traceparent", traceparent),-      ("tracestate", tracestate)-    ]---- | Extract trace context from message headers (for consumers).--- Returns SpanContext that can be used as parent link.-extractTraceContext :: TraceHeaders -> Maybe OTel.SpanContext-extractTraceContext headers = do-  let traceparent = lookup "traceparent" headers-      tracestate = lookup "tracestate" headers-  W3C.decodeSpanContext traceparent tracestate---- | Merge trace headers into existing PGMQ message headers.--- Preserves existing headers while adding trace context.-mergeTraceHeaders :: TraceHeaders -> Maybe Value -> Value-mergeTraceHeaders traceHeaders existingHeaders =-  let traceObj =-        Object $-          KM.fromList-            [ (Key.fromText "traceparent", toJsonValue $ lookup "traceparent" traceHeaders),-              (Key.fromText "tracestate", toJsonValue $ lookup "tracestate" traceHeaders)-            ]-   in case existingHeaders of-        Just (Object obj) -> Object $ KM.union (toKeyMap traceObj) obj-        Just v -> Object $ KM.fromList [(Key.fromText "_original", v), (Key.fromText "_trace", traceObj)]-        Nothing -> traceObj-  where-    toJsonValue :: Maybe ByteString -> Value-    toJsonValue = maybe Null (String . TE.decodeUtf8)--    toKeyMap :: Value -> KM.KeyMap Value-    toKeyMap (Object o) = o-    toKeyMap _ = KM.empty---- OpenTelemetry Semantic Conventions for Messaging--- See: https://opentelemetry.io/docs/specs/semconv/messaging/+import Network.HTTP.Types (RequestHeaders)+import OpenTelemetry.Context qualified as Ctxt+import OpenTelemetry.Propagator (Propagator, extract, inject)+import OpenTelemetry.SemanticConventions+  ( db_operation,+    db_system,+    messaging_batch_messageCount,+    messaging_destination_name,+    messaging_message_id,+    messaging_operation,+    messaging_system,+  )+import OpenTelemetry.Trace.Core+  ( TracerProvider,+    getTracerProviderPropagators,+  ) --- | The messaging system identifier.--- Value: "pgmq"-messagingSystem :: Text-messagingSystem = "messaging.system"+-- | Carrier for propagated trace context. Same shape as+-- @http-types@' @RequestHeaders@ — @[(HeaderName, ByteString)]@, where+-- @HeaderName@ is case-insensitive — because that is the carrier type+-- used by every propagator in the @hs-opentelemetry@ ecosystem.+type TraceHeaders = RequestHeaders --- | The type of messaging operation.--- Values: "send", "receive", "process"-messagingOperationType :: Text-messagingOperationType = "messaging.operation.type"+-- | Inject trace context into carrier headers, using the tracer+-- provider's configured propagator.+--+-- Callers typically pass a 'Ctxt.Context' obtained from+-- 'OpenTelemetry.Context.ThreadLocal.getContext' (with any current+-- span inserted) so the propagator can write the current span's+-- context onto the carrier.+injectTraceContext ::+  (MonadIO m) =>+  TracerProvider ->+  Ctxt.Context ->+  m TraceHeaders+injectTraceContext provider ctxt =+  let propagator = getTracerProviderPropagators provider+   in injectWith propagator ctxt --- | The destination queue name.-messagingDestinationName :: Text-messagingDestinationName = "messaging.destination.name"+-- | Extract trace context from carrier headers, using the tracer+-- provider's configured propagator. Returns an updated 'Ctxt.Context'+-- carrying the extracted span context (if any) — suitable for starting+-- a child \"process\" span linked to the producer's trace.+extractTraceContext ::+  (MonadIO m) =>+  TracerProvider ->+  TraceHeaders ->+  Ctxt.Context ->+  m Ctxt.Context+extractTraceContext provider headers ctxt =+  let propagator = getTracerProviderPropagators provider+   in extract propagator headers ctxt --- | The unique message identifier.-messagingMessageId :: Text-messagingMessageId = "messaging.message.id"+-- | Propagator-generic inject wrapper (mainly useful for tests that+-- want to pin the propagator explicitly).+injectWith ::+  (MonadIO m) =>+  Propagator Ctxt.Context RequestHeaders RequestHeaders ->+  Ctxt.Context ->+  m TraceHeaders+injectWith propagator ctxt = liftIO $ inject propagator ctxt [] --- | Number of messages in a batch operation.-messagingBatchMessageCount :: Text-messagingBatchMessageCount = "messaging.batch.message_count"+-- | Encode carrier headers as a JSON object for storage in pgmq+-- message headers (jsonb). Header names are lower-cased (via+-- case-insensitive folding) so the on-wire representation is stable+-- regardless of the original case the propagator emitted.+traceHeadersToJson :: TraceHeaders -> Value+traceHeadersToJson = Object . traceHeadersKeyMap --- | The routing key for topic-based message routing.-messagingRoutingKey :: Text-messagingRoutingKey = "messaging.destination.routing_key"+-- | Decode a JSON object of trace headers back into the carrier+-- shape. Non-object inputs and non-string values are ignored.+jsonToTraceHeaders :: Value -> TraceHeaders+jsonToTraceHeaders = \case+  Object obj ->+    [ (CI.mk (TE.encodeUtf8 (Key.toText k)), TE.encodeUtf8 v)+    | (k, String v) <- KM.toList obj+    ]+  _ -> [] --- | The database system.--- Value: "postgresql"-dbSystem :: Text-dbSystem = "db.system"+-- | Merge propagated trace headers into an existing pgmq headers+-- value. Existing keys win — we never overwrite user-supplied headers.+mergeTraceHeaders :: TraceHeaders -> Maybe Value -> Value+mergeTraceHeaders traceHeaders existingHeaders =+  let traceObj = traceHeadersKeyMap traceHeaders+   in case existingHeaders of+        Just (Object obj) -> Object (KM.union obj traceObj)+        Just v ->+          Object+            ( KM.fromList+                [ (Key.fromText (T.pack "_original"), v),+                  (Key.fromText (T.pack "_trace"), Object traceObj)+                ]+            )+        Nothing -> Object traceObj --- | The database operation name.-dbOperationName :: Text-dbOperationName = "db.operation.name"+traceHeadersKeyMap :: TraceHeaders -> KM.KeyMap Value+traceHeadersKeyMap =+  KM.fromList+    . fmap+      ( \(name, value) ->+          ( Key.fromText (TE.decodeUtf8 (CI.foldedCase name)),+            String (TE.decodeUtf8 value)+          )+      )

@@ -1,20 +1,29 @@--- | Traced operations for PGMQ with automatic trace context propagation.+-- | Traced operations for PGMQ with automatic trace-context propagation. ----- This module provides higher-level operations that automatically inject--- W3C Trace Context into message headers (for producers) and extract--- trace context from received messages (for consumers).+-- Higher-level helpers layered on top of the Pgmq effect: --+-- * 'sendMessageTraced' writes the current trace context onto the+--   outgoing message headers using the tracer provider's /configured/+--   propagator (W3C Trace Context by default; swap in B3, Datadog, or+--   any other propagator by configuring the provider).+-- * 'readMessageWithContext' extracts the trace context from each+--   received message and returns an 'OTel.Context' suitable for+--   starting a child @process@ span that links back to the producer's+--   trace.+-- -- == Usage -- -- @--- -- Producer: Send with trace context--- sendMessageTraced tracer queueName body Nothing+-- -- Producer+-- provider <- OTel.getGlobalTracerProvider+-- sendMessageTraced provider queueName body Nothing ----- -- Consumer: Read with trace context extraction--- messagesWithCtx <- readMessageWithContext readQuery--- forM_ messagesWithCtx $ \(msg, maybeParentCtx) ->---   -- maybeParentCtx contains the SpanContext from the producer---   processMessage msg maybeParentCtx+-- -- Consumer+-- messagesWithCtx <- readMessageWithContext provider readQuery+-- forM_ messagesWithCtx $ \\(msg, parentCtx) ->+--   -- parentCtx carries the propagated parent context+--   OTel.inSpan tracer \"process\" OTel.defaultSpanArguments+--     (processMessage msg) -- @ module Pgmq.Effectful.Traced   ( -- * Traced Send Operations@@ -27,10 +36,6 @@ where  import Data.Aeson (Value (..))-import Data.Aeson.Key qualified as Key-import Data.Aeson.KeyMap qualified as KM-import Data.Text (Text)-import Data.Text.Encoding qualified as TE import Data.Vector (Vector) import Data.Vector qualified as V import Effectful (Eff, IOE, (:>))@@ -40,34 +45,39 @@ import OpenTelemetry.Trace.Core qualified as OTel import Pgmq.Effectful.Effect (Pgmq, readMessage, sendMessageWithHeaders) import Pgmq.Effectful.Telemetry+  ( extractTraceContext,+    injectTraceContext,+    jsonToTraceHeaders,+    mergeTraceHeaders,+  ) import Pgmq.Hasql.Statements.Types qualified as Types import Pgmq.Types --- | A message paired with its extracted trace context (if present).-type MessageWithContext = (Message, Maybe OTel.SpanContext)+-- | A message paired with the trace context extracted from its+-- headers. Use 'OpenTelemetry.Context.ThreadLocal.attachContext' (or+-- pass it to an @inSpan''@-shaped primitive that lets you set the+-- parent) to start a @process@ span linked to the producer's trace.+type MessageWithContext = (Message, Ctxt.Context) --- | Send a message with trace context automatically injected into headers.------ This creates traceparent and tracestate headers from the current span--- context, which can be extracted by consumers to link traces.+-- | Send a message with trace context injected into its headers. ----- If existing headers are provided, the trace headers are merged in.+-- The current 'Ctxt.Context' is fetched via+-- 'OpenTelemetry.Context.ThreadLocal.getContext' and handed to the+-- tracer provider's configured propagator, which writes the propagated+-- fields (traceparent/tracestate for W3C, x-b3-* for B3, etc.) onto+-- the message headers. Any user-supplied headers win against+-- propagator-written ones (the merge is additive). sendMessageTraced ::   (Pgmq :> es, IOE :> es) =>-  OTel.Tracer ->+  OTel.TracerProvider ->   QueueName ->   MessageBody ->   Maybe Value ->   Eff es MessageId-sendMessageTraced _tracer queueName body existingHeaders = do-  -- Get current span context and inject into headers+sendMessageTraced provider queueName body existingHeaders = do   ctx <- Effectful.liftIO CtxtLocal.getContext-  traceHeaders <- case Ctxt.lookupSpan ctx of-    Just s -> Effectful.liftIO $ injectTraceContext s-    Nothing -> pure []-+  traceHeaders <- injectTraceContext provider ctx   let mergedHeaders = MessageHeaders $ mergeTraceHeaders traceHeaders existingHeaders-   sendMessageWithHeaders $     Types.SendMessageWithHeaders       { queueName = queueName,@@ -76,51 +86,23 @@         delay = Nothing       } --- | Read messages and extract trace context from headers.------ Returns messages paired with their extracted SpanContext (if present).--- The SpanContext can be used to create a child span that links to the--- producer's trace.+-- | Read messages and extract trace context from their headers. ----- @--- messagesWithCtx <- readMessageWithContext readQuery--- forM_ messagesWithCtx $ \(msg, maybeParentCtx) ->---   inSpan' tracer "process" (linkToParent maybeParentCtx) $ do---     processMessage msg--- @+-- The returned 'Ctxt.Context' for each message is the+-- 'Ctxt.empty' context updated with whatever fields the tracer+-- provider's propagator was able to parse from the headers — so a+-- caller can open a child @process@ span linked to the producer's+-- trace without knowing which propagator is in use. readMessageWithContext ::-  (Pgmq :> es) =>+  (Pgmq :> es, IOE :> es) =>+  OTel.TracerProvider ->   Types.ReadMessage ->   Eff es (Vector MessageWithContext)-readMessageWithContext readQuery = do+readMessageWithContext provider readQuery = do   messages <- readMessage readQuery-  pure $ V.map extractContext messages-  where-    extractContext :: Message -> MessageWithContext-    extractContext msg =-      let ctx = extractTraceContextFromMessage msg-       in (msg, ctx)--    extractTraceContextFromMessage :: Message -> Maybe OTel.SpanContext-    extractTraceContextFromMessage msg = do-      hdrs <- msg.headers-      traceHeaders <- parseTraceHeaders hdrs-      extractTraceContext traceHeaders--    parseTraceHeaders :: Value -> Maybe TraceHeaders-    parseTraceHeaders (Object obj) = do-      traceparent <- KM.lookup (Key.fromText "traceparent") obj >>= asText-      let tracestate = KM.lookup (Key.fromText "tracestate") obj >>= asText-      pure $-        catMaybes-          [ Just ("traceparent", TE.encodeUtf8 traceparent),-            ("tracestate",) . TE.encodeUtf8 <$> tracestate-          ]-    parseTraceHeaders _ = Nothing--    asText :: Value -> Maybe Text-    asText (String t) = Just t-    asText _ = Nothing--    catMaybes :: [Maybe a] -> [a]-    catMaybes = foldr (\x acc -> maybe acc (: acc) x) []+  Effectful.liftIO $+    V.forM messages $ \msg -> do+      parentCtx <- case msg.headers of+        Just hdrs -> extractTraceContext provider (jsonToTraceHeaders hdrs) Ctxt.empty+        Nothing -> pure Ctxt.empty+      pure (msg, parentCtx)

@@ -0,0 +1,66 @@+{-# LANGUAGE OverloadedStrings #-}++module ClassificationSpec (tests) where++import Data.HashSet qualified as HashSet+import Hasql.Errors qualified as HasqlErrors+import Pgmq.Effectful+  ( PgmqRuntimeError (..),+    isTransient,+  )+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertBool, testCase)++tests :: TestTree+tests =+  testGroup+    "isTransient classification"+    [ testCase "acquisition timeout is transient" $+        assertBool "expected transient" (isTransient PgmqAcquisitionTimeout),+      testCase "networking connection error is transient" $+        assertBool+          "expected transient"+          ( isTransient+              (PgmqConnectionError (HasqlErrors.NetworkingConnectionError "refused"))+          ),+      testCase "authentication error is not transient" $+        assertBool+          "expected permanent"+          ( not $+              isTransient+                (PgmqConnectionError (HasqlErrors.AuthenticationConnectionError "bad password"))+          ),+      testCase "compatibility error is not transient" $+        assertBool+          "expected permanent"+          ( not $+              isTransient+                (PgmqConnectionError (HasqlErrors.CompatibilityConnectionError "version mismatch"))+          ),+      testCase "other connection error is treated as transient" $+        assertBool+          "expected transient"+          ( isTransient+              (PgmqConnectionError (HasqlErrors.OtherConnectionError "libpq says no"))+          ),+      testCase "connection-drop session error is transient" $+        assertBool+          "expected transient"+          ( isTransient+              (PgmqSessionError (HasqlErrors.ConnectionSessionError "dropped"))+          ),+      testCase "driver session error is not transient" $+        assertBool+          "expected permanent"+          ( not $+              isTransient+                (PgmqSessionError (HasqlErrors.DriverSessionError "bug"))+          ),+      testCase "missing-types session error is not transient" $+        assertBool+          "expected permanent"+          ( not $+              isTransient+                (PgmqSessionError (HasqlErrors.MissingTypesSessionError HashSet.empty))+          )+    ]

@@ -0,0 +1,44 @@+{-# LANGUAGE OverloadedStrings #-}++-- | Test database infrastructure using ephemeral-pg.+--+-- This is a copy of pgmq-hasql\'s EphemeralDb helper, adjusted for+-- pgmq-effectful\'s test suite. Keeping a local copy avoids cross-package+-- test-helper sharing (which would require a shared test-helper library+-- or fragile cross-package hs-source-dirs).+module EphemeralDb+  ( withPgmqDb,+    withPgmqPool,+    StartError,+  )+where++import EphemeralPg+  ( StartError,+    connectionSettings,+    withCached,+  )+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as PoolConfig+import Pgmq.Migration qualified as Migration++-- | Run an action with a temporary PostgreSQL database that has the+-- pgmq schema installed.+withPgmqDb :: (Pool.Pool -> IO a) -> IO (Either StartError a)+withPgmqDb action = withCached $ \db -> do+  let connSettings = connectionSettings db+      poolConfig =+        PoolConfig.settings+          [ PoolConfig.size 3,+            PoolConfig.staticConnectionSettings connSettings+          ]+  pool <- Pool.acquire poolConfig+  installResult <- Pool.use pool Migration.migrate+  case installResult of+    Left poolErr -> error $ "Failed to install pgmq schema: " <> show poolErr+    Right (Left migrationErr) -> error $ "Migration failed: " <> show migrationErr+    Right (Right ()) -> action pool++-- | Alias for 'withPgmqDb' kept for parallelism with pgmq-hasql.+withPgmqPool :: (Pool.Pool -> IO a) -> IO (Either StartError a)+withPgmqPool = withPgmqDb

@@ -0,0 +1,24 @@+{-# LANGUAGE OverloadedStrings #-}++module Main (main) where++import ClassificationSpec qualified+import EphemeralDb (withPgmqPool)+import PlainInterpreterSpec qualified+import Test.Tasty (defaultMain, testGroup)+import TracedInterpreterSpec qualified++main :: IO ()+main = do+  result <- withPgmqPool $ \pool -> do+    let tree =+          testGroup+            "pgmq-effectful"+            [ ClassificationSpec.tests,+              PlainInterpreterSpec.tests pool,+              TracedInterpreterSpec.tests pool+            ]+    defaultMain tree+  case result of+    Left err -> error $ "Failed to start temp database: " <> show err+    Right () -> pure ()

@@ -0,0 +1,68 @@+{-# LANGUAGE OverloadedStrings #-}++module PlainInterpreterSpec (tests) where++import Effectful (runEff)+import Effectful.Error.Static (runError)+import Hasql.Pool qualified as Pool+import Hasql.Pool.Config qualified as PoolConfig+import Pgmq.Effectful+  ( MessageId (..),+    MessageQuery (..),+    PgmqRuntimeError (..),+    deleteMessage,+    listQueues,+    parseQueueName,+    runPgmq,+  )+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertFailure, testCase)++tests :: Pool.Pool -> TestTree+tests pool =+  testGroup+    "Plain interpreter error propagation"+    [ testCase "statement error surfaces PgmqSessionError" $ do+        bogus <- case parseQueueName "queue_that_does_not_exist_xyz" of+          Right q -> pure q+          Left err -> assertFailure ("could not build test queue name: " <> show err) >> fail ""+        result <-+          runEff+            . runError @PgmqRuntimeError+            . runPgmq pool+            $ deleteMessage+              MessageQuery+                { queueName = bogus,+                  messageId = MessageId 1+                }+        case result of+          Left (_cs, PgmqSessionError _) -> pure ()+          Left (_cs, other) ->+            assertFailure $+              "Expected PgmqSessionError, got " <> show other+          Right _ ->+            assertFailure+              "Expected an error deleting from missing queue, got success",+      testCase "connection error surfaces PgmqConnectionError" $ do+        let badCfg =+              PoolConfig.settings+                [ PoolConfig.size 1,+                  PoolConfig.staticConnectionSettings+                    "host=127.0.0.1 port=1 user=nobody dbname=nonexistent connect_timeout=1"+                ]+        badPool <- Pool.acquire badCfg+        result <-+          runEff+            . runError @PgmqRuntimeError+            . runPgmq badPool+            $ listQueues+        Pool.release badPool+        case result of+          Left (_cs, PgmqConnectionError _) -> pure ()+          Left (_cs, other) ->+            assertFailure $+              "Expected PgmqConnectionError, got " <> show other+          Right _ ->+            assertFailure+              "Expected a connection error, got success"+    ]

@@ -0,0 +1,311 @@+{-# LANGUAGE OverloadedStrings #-}++module TracedInterpreterSpec (tests) where++import Control.Monad (unless)+import Control.Monad.IO.Class (liftIO)+import Data.HashMap.Strict qualified as HM+import Data.IORef (IORef, readIORef)+import Data.Text (Text)+import Data.Text qualified as T+import Data.Vector qualified as V+import Effectful (runEff)+import Effectful.Error.Static (runError)+import Hasql.Pool qualified as Pool+import OpenTelemetry.Attributes qualified as Attrs+import OpenTelemetry.Context qualified as Ctxt+import OpenTelemetry.Context.ThreadLocal qualified as CtxtLocal+import OpenTelemetry.Exporter.InMemory.Span qualified as InMemoryExporter+import OpenTelemetry.Propagator (extract)+import OpenTelemetry.Propagator.W3CTraceContext qualified as W3C+import OpenTelemetry.Trace.Core qualified as OTel+import OpenTelemetry.Trace.Id.Generator.Default (defaultIdGenerator)+import OpenTelemetry.Util (appendOnlyBoundedCollectionValues)+import Pgmq.Effectful+  ( MessageBody (..),+    MessageId (..),+    MessageQuery (..),+    PgmqRuntimeError (..),+    QueueName,+    ReadMessage (..),+    SendMessage (..),+    createQueue,+    deleteMessage,+    parseQueueName,+    queueNameToText,+    readMessage,+    runPgmqTraced,+    sendMessage,+    sendMessageTraced,+  )+import Pgmq.Effectful.Telemetry (jsonToTraceHeaders)+import Pgmq.Types qualified as Pgmq+import System.Random (randomRIO)+import Test.Tasty (TestTree, testGroup)+import Test.Tasty.HUnit (assertBool, assertEqual, assertFailure, testCase)++tests :: Pool.Pool -> TestTree+tests pool =+  testGroup+    "Traced interpreter"+    [ testGroup+        "Error propagation"+        [ testCase "statement error surfaces PgmqSessionError via Error channel" $ do+            (tracer, _provider, _spansRef) <- setupTracer+            bogus <- mkQueueName "queue_that_does_not_exist_xyz2"+            result <-+              runEff+                . runError @PgmqRuntimeError+                . runPgmqTraced pool tracer+                $ deleteMessage+                  MessageQuery+                    { queueName = bogus,+                      messageId = MessageId 1+                    }+            case result of+              Left (_cs, PgmqSessionError _) -> pure ()+              Left (_cs, other) ->+                assertFailure $+                  "Expected PgmqSessionError, got " <> show other+              Right _ ->+                assertFailure+                  "Expected an error, got success"+        ],+      testGroup+        "v1.24 semantic conventions"+        [ testCase "publish emits messaging.* + db.* attributes" $ do+            (tracer, _provider, spansRef) <- setupTracer+            queue <- mkUniqueQueue "publish_attrs"+            result <-+              runEff . runError @PgmqRuntimeError . runPgmqTraced pool tracer $ do+                createQueue queue+                sendMessage+                  SendMessage+                    { queueName = queue,+                      messageBody = MessageBody "hello",+                      delay = Nothing+                    }+            _ <- assertRight result+            spans <- readIORef spansRef+            let publishSpans = filter ((==) "publish" . firstWord . OTel.spanName) spans+            unless (length publishSpans == 1) $+              assertFailure $+                "Expected exactly one publish span, got "+                  <> show (map OTel.spanName spans)+            s <- singleSpan publishSpans "publish"+            assertEqual+              "span name carries destination"+              ("publish " <> queueNameToText queue)+              (OTel.spanName s)+            assertSpanKindProducer s+            assertAttrText s "messaging.system" "pgmq"+            assertAttrText s "messaging.operation" "publish"+            assertAttrText s "messaging.destination.name" (queueNameToText queue)+            assertAttrText s "db.system" "postgresql"+            assertAttrText s "db.operation" "pgmq.send",+          testCase "receive emits messaging.* + db.* attributes" $ do+            (tracer, _provider, spansRef) <- setupTracer+            queue <- mkUniqueQueue "receive_attrs"+            result <-+              runEff . runError @PgmqRuntimeError . runPgmqTraced pool tracer $ do+                createQueue queue+                _ <-+                  sendMessage+                    SendMessage+                      { queueName = queue,+                        messageBody = MessageBody "one",+                        delay = Nothing+                      }+                readMessage+                  ReadMessage+                    { queueName = queue,+                      delay = 30,+                      batchSize = Just 1,+                      conditional = Nothing+                    }+            _ <- assertRight result+            spans <- readIORef spansRef+            let receiveSpans = filter ((==) "receive" . firstWord . OTel.spanName) spans+            unless (length receiveSpans == 1) $+              assertFailure $+                "Expected exactly one receive span, got "+                  <> show (map OTel.spanName spans)+            s <- singleSpan receiveSpans "receive"+            assertEqual+              "span name carries destination"+              ("receive " <> queueNameToText queue)+              (OTel.spanName s)+            assertSpanKindConsumer s+            assertAttrText s "messaging.system" "pgmq"+            assertAttrText s "messaging.operation" "receive"+            assertAttrText s "messaging.destination.name" (queueNameToText queue)+            assertAttrText s "db.system" "postgresql"+            assertAttrText s "db.operation" "pgmq.read",+          testCase "error path records exception event and Error status" $ do+            (tracer, _provider, spansRef) <- setupTracer+            bogus <- mkQueueName "queue_that_does_not_exist_xyz3"+            _ <-+              runEff+                . runError @PgmqRuntimeError+                . runPgmqTraced pool tracer+                $ deleteMessage+                  MessageQuery {queueName = bogus, messageId = MessageId 1}+            spans <- readIORef spansRef+            s <- case spans of+              [only] -> pure only+              _ ->+                assertFailure+                  ( "Expected exactly one span, got "+                      <> show (map OTel.spanName spans)+                  )+                  >> fail ""+            let events =+                  V.toList+                    (appendOnlyBoundedCollectionValues (OTel.spanEvents s))+                hasExceptionEvent =+                  any ((==) "exception" . OTel.eventName) events+            assertBool "span has exception event" hasExceptionEvent+            case OTel.spanStatus s of+              OTel.Error desc ->+                assertBool+                  ( "span status description should be a short non-PII label, got "+                      <> T.unpack desc+                  )+                  ("pool.session.statement" `T.isPrefixOf` desc)+              other ->+                assertFailure $ "expected Error status, got " <> show other,+          testCase "W3C traceparent round-trips through message headers" $ do+            (tracer, provider, _spansRef) <- setupTracer+            queue <- mkUniqueQueue "w3c_roundtrip"+            parentSpan <-+              OTel.createSpan+                tracer+                Ctxt.empty+                "parent"+                OTel.defaultSpanArguments+            parentTraceId <- OTel.traceId <$> OTel.getSpanContext parentSpan+            _ <- CtxtLocal.attachContext (Ctxt.insertSpan parentSpan Ctxt.empty)+            result <-+              runEff . runError @PgmqRuntimeError . runPgmqTraced pool tracer $ do+                createQueue queue+                _ <- sendMessageTraced provider queue (MessageBody "payload") Nothing+                readMessage+                  ReadMessage+                    { queueName = queue,+                      delay = 30,+                      batchSize = Just 1,+                      conditional = Nothing+                    }+            OTel.endSpan parentSpan Nothing+            msgs <- assertRight result+            msg <- case V.toList msgs of+              [m] -> pure m+              other ->+                assertFailure ("expected 1 message, got " <> show (length other))+                  >> fail ""+            case Pgmq.headers msg of+              Nothing ->+                assertFailure+                  "expected trace headers on received message, got none"+              Just hdrs -> do+                let propagator = OTel.getTracerProviderPropagators provider+                ctx <-+                  liftIO $+                    extract propagator (jsonToTraceHeaders hdrs) Ctxt.empty+                case Ctxt.lookupSpan ctx of+                  Nothing ->+                    assertFailure $+                      "propagator did not extract a span from headers: " <> show hdrs+                  Just extracted -> do+                    extractedTraceId <- OTel.traceId <$> OTel.getSpanContext extracted+                    assertEqual+                      "trace id round-trips via message headers"+                      parentTraceId+                      extractedTraceId+        ]+    ]++-- ---------------------------------------------------------------------+-- Helpers+-- ---------------------------------------------------------------------++-- | Build a tracer, tracer provider (with in-memory list processor and+-- W3C Trace Context propagator), and a handle to the exported spans.+setupTracer :: IO (OTel.Tracer, OTel.TracerProvider, IORef [OTel.ImmutableSpan])+setupTracer = do+  (processor, spansRef) <- InMemoryExporter.inMemoryListExporter+  let opts =+        OTel.emptyTracerProviderOptions+          { OTel.tracerProviderOptionsIdGenerator = defaultIdGenerator,+            OTel.tracerProviderOptionsPropagators = W3C.w3cTraceContextPropagator+          }+  provider <- OTel.createTracerProvider [processor] opts+  let tracer =+        OTel.makeTracer provider "pgmq-effectful-test" OTel.tracerOptions+  pure (tracer, provider, spansRef)++mkQueueName :: String -> IO QueueName+mkQueueName raw =+  case parseQueueName (T.pack raw) of+    Right q -> pure q+    Left err ->+      assertFailure+        ("could not build queue name " <> show raw <> ": " <> show err)+        >> fail ""++mkUniqueQueue :: String -> IO QueueName+mkUniqueQueue prefix = do+  suffix <- randomRIO (100000 :: Int, 999999)+  mkQueueName (prefix <> "_" <> show suffix)++firstWord :: Text -> Text+firstWord = T.takeWhile (/= ' ')++assertRight :: (Show a) => Either a b -> IO b+assertRight = \case+  Right b -> pure b+  Left e ->+    assertFailure ("expected Right, got Left " <> show e) >> fail ""++singleSpan :: [OTel.ImmutableSpan] -> String -> IO OTel.ImmutableSpan+singleSpan spans label = case spans of+  [s] -> pure s+  _ ->+    assertFailure+      ( "expected exactly one "+          <> label+          <> " span, got "+          <> show (map OTel.spanName spans)+      )+      >> fail ""++-- | 'OTel.SpanKind' has no 'Eq' instance, so assert via pattern.+assertSpanKindProducer :: OTel.ImmutableSpan -> IO ()+assertSpanKindProducer s = case OTel.spanKind s of+  OTel.Producer -> pure ()+  other -> assertFailure ("expected Producer span kind, got " <> show other)++assertSpanKindConsumer :: OTel.ImmutableSpan -> IO ()+assertSpanKindConsumer s = case OTel.spanKind s of+  OTel.Consumer -> pure ()+  other -> assertFailure ("expected Consumer span kind, got " <> show other)++-- | Assert a span carries the given attribute as a 'TextAttribute' with+-- the given value. Produces a readable failure message with the full+-- attribute map when the assertion fails.+assertAttrText :: OTel.ImmutableSpan -> Text -> Text -> IO ()+assertAttrText s key expected = do+  let attrMap = Attrs.getAttributeMap (OTel.spanAttributes s)+  case HM.lookup key attrMap of+    Just (Attrs.AttributeValue (Attrs.TextAttribute actual)) ->+      assertEqual ("attribute " <> T.unpack key) expected actual+    other ->+      assertFailure $+        "expected text attribute "+          <> T.unpack key+          <> "="+          <> T.unpack expected+          <> "; got "+          <> show other+          <> " in "+          <> show attrMap