keiro-pgmq 0.3.0.0 → 0.4.0.1
raw patch · 4 files changed
+542/−67 lines, 4 filesdep +hs-opentelemetry-exporter-in-memorydep ~aesondep ~effectful-coredep ~hasqlPVP ok
version bump matches the API change (PVP)
Dependencies added: hs-opentelemetry-exporter-in-memory
Dependency ranges changed: aeson, effectful-core, hasql, hasql-pool, keiro-core, streamly-core, text, time
API changes (from Hackage documentation)
Files
- CHANGELOG.md +30/−1
- keiro-pgmq.cabal +27/−26
- src/Keiro/PGMQ/Job.hs +201/−34
- test/Main.hs +284/−6
CHANGELOG.md view
@@ -6,7 +6,36 @@ ## [Unreleased] -_No unreleased changes._+## 0.4.0.1 — 2026-07-28++### Other Changes++- Adds PVP upper bounds to every dependency that previously carried a lower+ bound only, so `cabal check` reports no packaging warnings. No API or+ behaviour change from 0.4.0.0, which was tagged but never published.+++## 0.4.0.0 — 2026-07-28++### Bug Fixes++- One-shot job processing (`runJobOnce` and `runJobOnceWithContext`) now continues the+ producer's trace instead of leaving a hole between the enqueue and settlement spans.+ Each claimed message extracts the W3C `traceparent` that `enqueueTraced` stored in the+ PGMQ `headers` column, installs it as a remote parent, and runs the handler inside one+ Consumer-kind `<jobName> process` span. The span carries the same common surface as the+ continuous `runJobWorkers` path — `messaging.system`, `messaging.destination.name`,+ `messaging.operation.type`, `messaging.message.id`, `shibuya.partition` for FIFO+ deliveries, and `shibuya.ack.decision` recorded only after the finalizing PGMQ statement+ returns, with `OK` for `Done`/retry and `ERROR` for dead-lettering. A handler that throws+ is recorded as an exception with an `ERROR` status and no acknowledgement attribute.+ Deliveries carrying no usable trace header still get exactly one span.++ Deliberately absent: the `shibuya.inflight.*` gauges the continuous path reports. A+ bounded drain has no shibuya inbox and no concurrency meter to describe.++ No public API, function signature, delivery semantic, queue behavior, PGMQ header shape,+ or wire payload changed — downstream applications gain trace continuity by rebuilding. ## 0.3.0.0 — 2026-07-14
keiro-pgmq.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.0 name: keiro-pgmq-version: 0.3.0.0+version: 0.4.0.1 synopsis: PostgreSQL job-queue (PGMQ) integration for Keiro description: A typed background-job queue for Keiro applications on top of PGMQ (the@@ -52,22 +52,22 @@ hs-source-dirs: src build-depends:- , aeson >=2.2+ , aeson >=2.2 && <2.3 , base >=4.21 && <5- , effectful-core >=2.6- , hasql >=1.10- , hasql-pool >=1.2+ , effectful-core >=2.6 && <2.7+ , hasql >=1.10 && <1.11+ , hasql-pool >=1.2 && <1.5 , hs-opentelemetry-api >=1.0 && <1.1- , keiro-core ^>=0.3.0.0+ , keiro-core ^>=0.4.0.1 , pgmq-config >=0.4 && <0.5 , pgmq-core >=0.4 && <0.5 , pgmq-effectful >=0.4 && <0.5 , pgmq-hasql >=0.4 && <0.5 , shibuya-core >=0.8.0.1 && <0.9 , shibuya-pgmq-adapter >=0.12 && <0.13- , streamly-core >=0.3- , text >=2.1- , time >=1.12+ , streamly-core >=0.3 && <0.4+ , text >=2.1 && <2.2+ , time >=1.12 && <1.15 test-suite keiro-pgmq-test import: warnings, shared@@ -76,22 +76,23 @@ main-is: Main.hs ghc-options: -threaded -rtsopts -with-rtsopts=-N build-depends:- , aeson >=2.2- , base >=4.21 && <5- , effectful-core >=2.6- , hasql >=1.10- , hasql-pool >=1.2- , hs-opentelemetry-api >=1.0 && <1.1- , hs-opentelemetry-propagator-w3c >=1.0 && <1.1- , hs-opentelemetry-sdk >=1.0 && <1.1- , hspec >=2.11- , keiro-core ^>=0.3.0.0+ , aeson >=2.2 && <2.3+ , base >=4.21 && <5+ , effectful-core >=2.6 && <2.7+ , hasql >=1.10 && <1.11+ , hasql-pool >=1.2 && <1.5+ , hs-opentelemetry-api >=1.0 && <1.1+ , hs-opentelemetry-exporter-in-memory >=1.0 && <1.1+ , hs-opentelemetry-propagator-w3c >=1.0 && <1.1+ , hs-opentelemetry-sdk >=1.0 && <1.1+ , hspec >=2.11+ , keiro-core ^>=0.4.0.1 , keiro-pgmq , keiro-test-support- , pgmq-config >=0.4 && <0.5- , pgmq-core >=0.4 && <0.5- , pgmq-effectful >=0.4 && <0.5- , pgmq-migration >=0.4 && <0.5- , shibuya-core >=0.8.0.1 && <0.9- , shibuya-pgmq-adapter >=0.12 && <0.13- , text >=2.1+ , pgmq-config >=0.4 && <0.5+ , pgmq-core >=0.4 && <0.5+ , pgmq-effectful >=0.4 && <0.5+ , pgmq-migration >=0.4 && <0.5+ , shibuya-core >=0.8.0.1 && <0.9+ , shibuya-pgmq-adapter >=0.12 && <0.13+ , text >=2.1 && <2.2
src/Keiro/PGMQ/Job.hs view
@@ -42,6 +42,32 @@ Transient database errors during PGMQ polling are retried by the adapter, and a polling failure that exhausts that retry policy is propagated visibly through shibuya supervision rather than completing the worker silently.++== Tracing++Both execution shapes propagate W3C trace context and emit the same common+per-message span. A message enqueued with 'enqueueTraced' carries @traceparent@+(and optional @tracestate@) in PGMQ's JSONB @headers@ column; at consumption+time that context is extracted and installed as the parent, so the handler's+span continues the producer's trace even across processes. The span is+Consumer-kind, named @\<jobName\> process@, and carries+@messaging.system=shibuya@, @messaging.destination.name=\<jobName\>@,+@messaging.operation.type=process@, @messaging.message.id@,+@shibuya.partition@ for FIFO deliveries, and @shibuya.ack.decision@ once the+message has actually been finalized. @AckOk@ and @AckRetry@ end the span @OK@;+dead-lettering and halting end it @ERROR@ with the reason.++The continuous 'runJobWorkers' path gets this from shibuya's supervised runner+and additionally reports @shibuya.inflight.count@ and @shibuya.inflight.max@.+The bounded 'runJobOnce' \/ 'runJobOnceWithContext' path opens the span itself+and deliberately omits those two: a direct drain has no shibuya inbox and no+concurrency meter to describe. Lower-level PGMQ operation spans+(@publish \<queue\>@, @receive \<queue\>@, deletes, visibility changes, DLQ+sends) come from the traced @pgmq-effectful@ interpreter and are unaffected.++Tracing is opt-in: with no tracer wired into the runtime (see+'Keiro.PGMQ.Runtime.withJobRuntime'), every span operation is a no-op and+processing behavior is identical. -} module Keiro.PGMQ.Job ( -- * Job declaration@@ -143,11 +169,45 @@ mkProcessor, runApp, )-import "shibuya-core" Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), RetryDelay (..))+import "shibuya-core" Shibuya.Core.Ack (+ AckDecision (..),+ DeadLetterReason (..),+ HaltReason (..),+ RetryDelay (..),+ ) import "shibuya-core" Shibuya.Core.Ingested qualified as Shibuya import "shibuya-core" Shibuya.Core.Lease (Lease (..)) import "shibuya-core" Shibuya.Core.Types (Attempt (..), Envelope (..))-import "shibuya-core" Shibuya.Telemetry.Effect (Tracing)++-- Qualified only for 'unMessageId': shibuya's @MessageId@ type name would+-- otherwise collide with @Pgmq.Effectful@'s, which the producer signatures use.+import "shibuya-core" Shibuya.Core.Types qualified as ShibuyaTypes+import "shibuya-core" Shibuya.Telemetry.Effect (+ Span,+ SpanStatus (..),+ Tracing,+ addAttribute,+ addEvent,+ recordException,+ setStatus,+ toAttribute,+ withExtractedContext,+ withSpan',+ )+import "shibuya-core" Shibuya.Telemetry.Propagation (extractTraceContext)+import "shibuya-core" Shibuya.Telemetry.Semantic (+ attrMessagingDestinationName,+ attrMessagingMessageId,+ attrMessagingOperation,+ attrMessagingSystem,+ attrShibuyaAckDecision,+ attrShibuyaPartition,+ consumerSpanArgs,+ eventHandlerCompleted,+ eventHandlerStarted,+ mkEvent,+ processSpanName,+ ) import "shibuya-pgmq-adapter" Shibuya.Adapter.Pgmq ( FifoConfig (..), FifoReadStrategy (..),@@ -164,6 +224,7 @@ pgmqMessageToEnvelope, ) import "text" Data.Text (Text)+import "text" Data.Text qualified as Text import "time" Data.Time (NominalDiffTime, nominalDiffTimeToSeconds) -- | What a job handler decides. Never exposes shibuya/PGMQ wire types to the caller.@@ -724,6 +785,13 @@ {- | Continuous, multi-processor run (the @rei@ cadence): run a supervised app over several processors built with 'jobProcessor'. Returns the app handle; the caller decides whether to block on it. The inbox size is clamped to at least 1.++Shibuya's supervised runner opens the per-message @\<jobName\> process@ span+described in the module's tracing section, continuing the producer's trace from+the message's @traceparent@. Because this path owns an inbox and a concurrency+limit, its spans additionally carry @shibuya.inflight.count@ and+@shibuya.inflight.max@, which the bounded 'runJobOnceWithContext' path has no+equivalent for. -} runJobWorkers :: (Pgmq :> es, Reader PgmqAdapterEnv :> es, IOE :> es, Tracing :> es) =>@@ -735,6 +803,77 @@ ps <- sequence procs runApp AppConfig{strategy = strategy, inboxSize = max 1 inboxSize} ps +{- | Open the one-shot equivalent of shibuya's per-message processing span.++The continuous worker path gets this from shibuya's supervised runner; the+direct drain has no runner, so it opens the same span itself. The trace context+that 'enqueueTraced' wrote into the PGMQ @headers@ column (and that+'pgmqMessageToEnvelope' projects onto @Envelope.traceContext@) is installed as+the parent for the dynamic extent of this one delivery, so the span continues+the producer's trace across processes rather than starting a new one. Deliveries+without a usable @traceparent@ fall back to whatever local context is active,+and still get exactly one span.++The attribute set is deliberately the subset the two execution shapes agree on:+the OTel @messaging.*@ quartet plus @shibuya.partition@ for FIFO deliveries. The+@shibuya.inflight.*@ gauges are omitted because the direct drain has no shibuya+inbox and no concurrency meter to report.+-}+withOneShotProcessSpan ::+ (IOE :> es, Tracing :> es) =>+ Job p ->+ Envelope Value ->+ (Span -> Eff es a) ->+ Eff es a+withOneShotProcessSpan job envelope act =+ withExtractedContext (envelope.traceContext >>= extractTraceContext) $+ withSpan' (processSpanName job.jobName) consumerSpanArgs $ \traceSpan -> do+ let ShibuyaTypes.MessageId messageIdText = envelope.messageId+ addAttribute traceSpan attrMessagingSystem ("shibuya" :: Text)+ addAttribute traceSpan attrMessagingDestinationName job.jobName+ addAttribute traceSpan attrMessagingOperation ("process" :: Text)+ addAttribute traceSpan attrMessagingMessageId messageIdText+ case envelope.partition of+ Just partition -> addAttribute traceSpan attrShibuyaPartition partition+ Nothing -> pure ()+ act traceSpan++{- | Record a finalization that already succeeded on the process span, using the+same decision text and status mapping as shibuya's continuous runner. Call this+only /after/ the corresponding PGMQ statement returned, so the attribute never+claims an acknowledgement that did not happen.+-}+recordAckOnSpan ::+ (IOE :> es, Tracing :> es) => Span -> AckDecision -> Eff es ()+recordAckOnSpan traceSpan decision = do+ let decisionText = ackDecisionText decision+ addEvent traceSpan $+ mkEvent eventHandlerCompleted [(attrShibuyaAckDecision, toAttribute decisionText)]+ addAttribute traceSpan attrShibuyaAckDecision decisionText+ setStatus traceSpan $ case decision of+ AckOk -> Ok+ AckRetry _ -> Ok+ AckDeadLetter reason -> Error (deadLetterReasonText reason)+ AckHalt reason -> Error (haltReasonText reason)++-- | The @shibuya.ack.decision@ value for a decision, matching shibuya's runner.+ackDecisionText :: AckDecision -> Text+ackDecisionText AckOk = "ack_ok"+ackDecisionText (AckRetry _) = "ack_retry"+ackDecisionText (AckDeadLetter _) = "ack_dead_letter"+ackDecisionText (AckHalt _) = "ack_halt"++-- | The @ERROR@ status description for a dead-letter, matching shibuya's runner.+deadLetterReasonText :: DeadLetterReason -> Text+deadLetterReasonText (PoisonPill t) = "poison_pill: " <> t+deadLetterReasonText (InvalidPayload t) = "invalid_payload: " <> t+deadLetterReasonText MaxRetriesExceeded = "max_retries_exceeded"++-- | The @ERROR@ status description for a halt, matching shibuya's runner.+haltReasonText :: HaltReason -> Text+haltReasonText (HaltOrderedStream t) = "halt_ordered_stream: " <> t+haltReasonText (HaltFatal t) = "halt_fatal: " <> t+ {- | One-shot drain of up to @n@ messages with explicit tuning and a context-aware handler. This reads directly from PGMQ and returns when the queue is empty or @n@ messages have been acknowledged/retried/dead-lettered,@@ -743,6 +882,19 @@ If a handler throws, the message is left on the main queue and remains invisible until the active visibility timeout expires; the drain keeps processing the rest of the batch and does not count that message in the returned total.++Each claimed message is processed inside one Consumer-kind+@\<jobName\> process@ span that continues the producer's trace when the message+carries a W3C @traceparent@ (see 'enqueueTraced'), exactly as the continuous+'runJobWorkers' path does. The span carries @messaging.system@,+@messaging.destination.name@, @messaging.operation.type@,+@messaging.message.id@, @shibuya.partition@ for FIFO deliveries, and — once the+finalizing PGMQ statement has returned — @shibuya.ack.decision@ with a matching+span status. A handler that throws is recorded as an exception with an @ERROR@+status and no acknowledgement attribute, because the direct drain deliberately+issues no finalizer call and leaves the row for visibility-timeout redelivery.+Unlike the continuous path this span has no @shibuya.inflight.*@ attributes:+there is no shibuya inbox or concurrency meter behind a bounded drain. -} runJobOnceWithContext :: (Pgmq :> es, IOE :> es, Tracing :> es) =>@@ -791,51 +943,64 @@ nextBatchSize remaining = fromIntegral (min remaining (fromIntegral tuning.batchSize :: Int)) + -- One conversion point per delivery: the envelope supplies the payload, the+ -- attempt number, the FIFO partition, the message id, and the trace context. step count message = do- disposed <- processMessage message+ let envelope = pgmqMessageToEnvelope message+ disposed <-+ withOneShotProcessSpan job envelope (processMessage message envelope) pure $ if disposed then count + 1 else count - processMessage message- | message.readCount > job.jobPolicy.maxRetries = do- ackMessage message (AckDeadLetter MaxRetriesExceeded)- pure True+ -- Settle exactly as before; the span only observes what already happened.+ -- 'recordAckOnSpan' runs after 'ackMessage' returns, so a failed+ -- finalization propagates without leaving a false acknowledgement behind.+ processMessage message envelope traceSpan+ | message.readCount > job.jobPolicy.maxRetries =+ settle message traceSpan (AckDeadLetter MaxRetriesExceeded) | otherwise =- case decodeJob job.jobCodec (messagePayload message) of- Left (JobPayloadFromFuture _payloadVersion _workerVersion) -> do- ackMessage message (AckRetry job.jobPolicy.defaultRetryDelay)- pure True- Left (JobPayloadMalformed err) -> do- ackMessage message (AckDeadLetter (InvalidPayload err))- pure True+ case decodeJob job.jobCodec envelope.payload of+ Left (JobPayloadFromFuture _payloadVersion _workerVersion) ->+ settle message traceSpan (AckRetry job.jobPolicy.defaultRetryDelay)+ Left (JobPayloadMalformed err) ->+ settle message traceSpan (AckDeadLetter (InvalidPayload err)) Right p -> do- outcome <- EffException.try @SomeException (handle (contextFor message) p)+ addEvent traceSpan (mkEvent eventHandlerStarted [])+ outcome <-+ EffException.try @SomeException (handle (contextFor message envelope) p) case outcome of- Left _handlerException ->+ Left handlerException -> do+ -- No finalizer call: the row stays invisible until its+ -- visibility timeout expires, so there is no ack to claim.+ recordException traceSpan handlerException+ setStatus traceSpan (Error (handlerExceptionText handlerException)) pure False- Right jobOutcome -> do- ackMessage message (outcomeToAck jobOutcome)- pure True+ Right jobOutcome ->+ settle message traceSpan (outcomeToAck jobOutcome) - messagePayload = (.payload) . pgmqMessageToEnvelope+ settle message traceSpan decision = do+ ackMessage message decision+ recordAckOnSpan traceSpan decision+ pure True - contextFor message =- let envelope = pgmqMessageToEnvelope message- in JobContext- { extendLease = \duration ->- void $- Pgmq.changeVisibilityTimeout- VisibilityTimeoutQuery- { queueName = job.jobQueue.physicalName- , messageId = message.messageId- , visibilityTimeoutOffset = nominalToSeconds duration- }- , attempt = fmap (.unAttempt) envelope.attempt- , headers = message.headers- }+ handlerExceptionText ex = "handler exception: " <> Text.pack (show (ex :: SomeException)) + contextFor message envelope =+ JobContext+ { extendLease = \duration ->+ void $+ Pgmq.changeVisibilityTimeout+ VisibilityTimeoutQuery+ { queueName = job.jobQueue.physicalName+ , messageId = message.messageId+ , visibilityTimeoutOffset = nominalToSeconds duration+ }+ , attempt = fmap (.unAttempt) envelope.attempt+ , headers = message.headers+ }+ outcomeToAck Done = AckOk outcomeToAck (Retry d) = AckRetry d outcomeToAck RetryDefault = AckRetry job.jobPolicy.defaultRetryDelay@@ -904,6 +1069,8 @@ {- | One-shot drain of up to @n@ messages (the @hospital-capacity@ cadence): read directly from PGMQ with 'defaultJobTuning', run the handler on each available message, and return promptly when the queue is empty.++Each delivery is traced exactly as 'runJobOnceWithContext' describes. -} runJobOnce :: (Pgmq :> es, IOE :> es, Tracing :> es) =>
test/Main.hs view
@@ -26,7 +26,7 @@ import Data.Aeson.Types (parseEither) import Data.Either (isRight) import Data.Foldable (toList, traverse_)-import Data.IORef (modifyIORef', newIORef, readIORef, writeIORef)+import Data.IORef (IORef, modifyIORef', newIORef, readIORef, writeIORef) import Data.Int (Int32, Int64) import Data.List (find) import Data.List.NonEmpty (NonEmpty (..))@@ -48,11 +48,16 @@ import Keiro.Codec qualified as CoreCodec import Keiro.PGMQ import Keiro.Test.Postgres qualified as Postgres+import OpenTelemetry.Attributes (Attribute (..), Attributes, PrimitiveAttribute (..), lookupAttribute) import OpenTelemetry.Context qualified as Ctxt import OpenTelemetry.Context.ThreadLocal qualified as CtxtLocal+import OpenTelemetry.Exporter.InMemory.Span (inMemoryListExporter)+import OpenTelemetry.Processor.Span (SpanProcessor) import OpenTelemetry.Propagator.W3CTraceContext qualified as W3C+import OpenTelemetry.Trace.Core (Event (..), ImmutableSpan (..), SpanHot (..)) import OpenTelemetry.Trace.Core qualified as OTel import OpenTelemetry.Trace.Id.Generator.Default (defaultIdGenerator)+import OpenTelemetry.Util (appendOnlyBoundedCollectionValues) import Pgmq.Config.Types qualified as Config import Pgmq.Effectful (Message (..), MessageBody (..), Pgmq, QueueMetrics (..), ReadMessage (..), SendMessage (..)) import Pgmq.Effectful qualified as Pgmq@@ -205,18 +210,114 @@ {- | A real tracer provider with the W3C Trace Context propagator and a non-dummy id generator, so an active span produces a @traceparent@ on injection.-No span processors are needed — the test inspects propagated headers, not-exported spans.+Dummy ids cannot encode a valid @traceparent@, which is why the default id+generator is wired in explicitly. -}-setupW3CProvider :: IO OTel.TracerProvider-setupW3CProvider =+mkW3CProvider :: [SpanProcessor] -> IO OTel.TracerProvider+mkW3CProvider processors = OTel.createTracerProvider- []+ processors OTel.emptyTracerProviderOptions { OTel.tracerProviderOptionsIdGenerator = defaultIdGenerator , OTel.tracerProviderOptionsPropagators = W3C.w3cTraceContextPropagator } +{- | A W3C provider with no span processors: enough to inspect propagated+headers, not enough to inspect exported spans.+-}+setupW3CProvider :: IO OTel.TracerProvider+setupW3CProvider = mkW3CProvider []++{- | A W3C provider whose ended spans are collected in memory. Shut the provider+down (see 'capturedSpans') before reading the reference so every span that was+still open has been flushed.+-}+setupCapturingProvider :: IO (OTel.TracerProvider, IORef [ImmutableSpan])+setupCapturingProvider = do+ (processor, spansRef) <- inMemoryListExporter+ provider <- mkW3CProvider [processor]+ pure (provider, spansRef)++{- | Shut the provider down (ending and exporting everything still buffered) and+return a frozen snapshot of every captured span.+-}+capturedSpans :: OTel.TracerProvider -> IORef [ImmutableSpan] -> IO [CapturedSpan]+capturedSpans provider spansRef = do+ _ <- OTel.shutdownTracerProvider provider Nothing+ traverse captureSpan =<< readIORef spansRef++{- | A frozen snapshot of an 'ImmutableSpan'. In hs-opentelemetry 1.0 the mutable+span fields (name, attributes, status) live behind the @spanHot :: IORef SpanHot@+field rather than directly on 'ImmutableSpan', so the tests read that reference+once after the span ends and assert on this flat record.+-}+data CapturedSpan = CapturedSpan+ { csName :: Text+ , csKind :: OTel.SpanKind+ , csAttributes :: Attributes+ , csStatus :: OTel.SpanStatus+ , csContext :: OTel.SpanContext+ , csParent :: Maybe OTel.Span+ , csEventNames :: [Text]+ }++captureSpan :: ImmutableSpan -> IO CapturedSpan+captureSpan sp = do+ hot <- readIORef (spanHot sp)+ pure+ CapturedSpan+ { csName = hotName hot+ , csKind = spanKind sp+ , csAttributes = hotAttributes hot+ , csStatus = hotStatus hot+ , csContext = spanContext sp+ , csParent = spanParent sp+ , csEventNames =+ map eventName (toList (appendOnlyBoundedCollectionValues (hotEvents hot)))+ }++textAttr :: Attributes -> Text -> Maybe Text+textAttr attrs name = case lookupAttribute attrs name of+ Just (AttributeValue (TextAttribute t)) -> Just t+ _ -> Nothing++-- | Every captured span whose name matches exactly.+spansNamed :: Text -> [CapturedSpan] -> [CapturedSpan]+spansNamed name = filter ((== name) . csName)++-- | The 'OTel.SpanContext' of a captured span's parent, if it had one.+parentSpanContext :: CapturedSpan -> IO (Maybe OTel.SpanContext)+parentSpanContext = traverse OTel.getSpanContext . csParent++{- | The one @\<jobName\> process@ span a single one-shot delivery must produce.+Anything other than exactly one is a failure that names every captured span, so+a duplicate wrapper is diagnosed rather than silently accepted by taking the+head of the list.+-}+theProcessSpan :: Text -> [CapturedSpan] -> IO CapturedSpan+theProcessSpan jobName spans =+ case spansNamed (jobName <> " process") spans of+ [only] -> pure only+ other ->+ fail+ ( "expected exactly one "+ <> show (jobName <> " process")+ <> " span, got "+ <> show (length other)+ <> "; all captured spans: "+ <> show (map csName spans)+ )++{- | Run a @Stack@ action against a fresh 'JobRuntime' wired to @tracer@, so both+the shibuya 'Tracing' effect and the @pgmq@ interpreter emit spans. Fails the+test on any PGMQ runtime error, exactly like 'runDb'.+-}+runDbTraced :: Text -> OTel.Tracer -> Eff Stack a -> IO a+runDbTraced connStr tracer act =+ withJobRuntime connStr (Just tracer) $ \rt -> do+ res <- runJobEff rt act+ either (\e -> fail ("PGMQ runtime error: " <> show e)) pure res+ stopAppQuickly :: (IOE :> es) => AppHandle es -> Eff es () stopAppQuickly app = do _ <- stopAppGracefully ShutdownConfig{drainTimeout = 1} app@@ -778,6 +879,183 @@ headerKey "traceparent" captured `shouldSatisfy` \case Just (String _) -> True _ -> False++ -- EP-111 M1: the captured-span fixture itself, proven against the spans the+ -- traced pgmq interpreter already emits.+ it "captured tracing fixture sees PGMQ publish and receive spans" $ \connStr -> do+ (provider, spansRef) <- setupCapturingProvider+ let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions+ job = mkJob "keiro_pgmq_test.fixture_spans"+ queue = queueNameToText job.jobQueue.physicalName+ runDbTraced connStr tracer $ do+ ensureJobQueue job+ _ <- enqueue job (Ping "fixture" 1)+ _ <- readMessages job.jobQueue.physicalName 1+ pure ()+ spans <- capturedSpans provider spansRef+ map csName spans `shouldSatisfy` elem ("publish " <> queue)+ map csName spans `shouldSatisfy` elem ("receive " <> queue)++ -- EP-111 M2: the central proof — the one-shot process span continues the+ -- producer's trace using only what the PGMQ headers carry.+ it "one-shot process span continues the enqueued W3C parent" $ \connStr -> do+ (provider, spansRef) <- setupCapturingProvider+ let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions+ job = mkJob "keiro_pgmq_test.one_shot_parent"+ producerSpan <- OTel.createSpan tracer Ctxt.empty "enqueue" OTel.defaultSpanArguments+ producerCtx <- OTel.getSpanContext producerSpan+ -- Attach the producer span only for the enqueue, then detach it. The+ -- drain therefore has no local parent to inherit: the only path from+ -- producer to consumer is the traceparent stored in the PGMQ headers.+ token <- CtxtLocal.attachContext (Ctxt.insertSpan producerSpan Ctxt.empty)+ runDbTraced connStr tracer $ do+ ensureJobQueue job+ _ <- enqueueTraced provider job (MessageHeaders (object [])) (Ping "traced" 1)+ pure ()+ CtxtLocal.detachContext token+ OTel.endSpan producerSpan Nothing++ drained <-+ runDbTraced connStr tracer $+ runJobOnceWithContext defaultJobTuning 1 job \_ctx _payload -> pure Done+ drained `shouldBe` 1++ spans <- capturedSpans provider spansRef+ processSpan <- theProcessSpan job.jobName spans+ csKind processSpan `shouldBe` OTel.Consumer+ OTel.traceId (csContext processSpan) `shouldBe` OTel.traceId producerCtx+ parent <- parentSpanContext processSpan+ fmap OTel.spanId parent `shouldBe` Just (OTel.spanId producerCtx)+ textAttr (csAttributes processSpan) "messaging.system" `shouldBe` Just "shibuya"+ textAttr (csAttributes processSpan) "messaging.destination.name"+ `shouldBe` Just job.jobName+ textAttr (csAttributes processSpan) "messaging.operation.type" `shouldBe` Just "process"+ textAttr (csAttributes processSpan) "messaging.message.id" `shouldSatisfy` \case+ Just _ -> True+ Nothing -> False+ textAttr (csAttributes processSpan) "shibuya.ack.decision" `shouldBe` Just "ack_ok"+ csStatus processSpan `shouldBe` OTel.Ok++ -- EP-111 M3: the branches whose telemetry meaning differs from plain success.+ it "one-shot Retry reports ack_retry with an OK span and hides the row" $ \connStr -> do+ (provider, spansRef) <- setupCapturingProvider+ let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions+ job = mkJob "keiro_pgmq_test.span_retry"+ (drained, len, hidden) <-+ runDbTraced connStr tracer $ do+ ensureJobQueue job+ _ <- enqueue job (Ping "r" 1)+ drained <-+ runJobOnceWithContext defaultJobTuning 1 job \_ctx _payload ->+ pure (Retry (RetryDelay 30))+ len <- queueLen job.jobQueue.physicalName+ hidden <- readOneIsEmpty job.jobQueue.physicalName+ pure (drained, len, hidden)+ drained `shouldBe` 1+ len `shouldBe` 1+ hidden `shouldBe` True+ processSpan <- theProcessSpan job.jobName =<< capturedSpans provider spansRef+ textAttr (csAttributes processSpan) "shibuya.ack.decision" `shouldBe` Just "ack_retry"+ csStatus processSpan `shouldBe` OTel.Ok++ it "one-shot Dead reports ack_dead_letter with an ERROR span" $ \connStr -> do+ (provider, spansRef) <- setupCapturingProvider+ let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions+ job = mkJob "keiro_pgmq_test.span_dead"+ (mainLen, dlqLen) <-+ runDbTraced connStr tracer $ do+ ensureJobQueue job+ _ <- enqueue job (Ping "poison" 1)+ runJobOnce 1 job (\_ -> pure (Dead "bad"))+ mainLen <- queueLen job.jobQueue.physicalName+ dlqLen <- queueLen job.jobQueue.dlqName+ pure (mainLen, dlqLen)+ mainLen `shouldBe` 0+ dlqLen `shouldBe` 1+ processSpan <- theProcessSpan job.jobName =<< capturedSpans provider spansRef+ textAttr (csAttributes processSpan) "shibuya.ack.decision"+ `shouldBe` Just "ack_dead_letter"+ csStatus processSpan `shouldBe` OTel.Error "poison_pill: bad"++ it "an undecodable payload reports ack_dead_letter without a handler-started event" $ \connStr -> do+ (provider, spansRef) <- setupCapturingProvider+ let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions+ job = mkJob "keiro_pgmq_test.span_malformed"+ dlqLen <-+ runDbTraced connStr tracer $ do+ ensureJobQueue job+ _ <-+ Pgmq.sendMessage+ SendMessage+ { queueName = job.jobQueue.physicalName+ , messageBody = MessageBody (String "not a ping")+ , delay = Nothing+ }+ runJobOnce 1 job (\_ -> pure Done)+ queueLen job.jobQueue.dlqName+ dlqLen `shouldBe` 1+ processSpan <- theProcessSpan job.jobName =<< capturedSpans provider spansRef+ textAttr (csAttributes processSpan) "shibuya.ack.decision"+ `shouldBe` Just "ack_dead_letter"+ csStatus processSpan `shouldSatisfy` \case+ OTel.Error reason -> "invalid_payload: " `Text.isPrefixOf` reason+ _ -> False+ csEventNames processSpan `shouldNotSatisfy` elem "shibuya.handler.started"++ it "a thrown handler records an exception and claims no acknowledgement" $ \connStr -> do+ (provider, spansRef) <- setupCapturingProvider+ let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions+ job = mkJob "keiro_pgmq_test.span_throw"+ (drained, len, hidden) <-+ runDbTraced connStr tracer $ do+ ensureJobQueue job+ _ <- enqueue job (Ping "boom" 1)+ drained <-+ runJobOnceWithContext defaultJobTuning 1 job \_ctx _payload ->+ liftIO (throwIO (userError "handler exploded"))+ len <- queueLen job.jobQueue.physicalName+ hidden <- readOneIsEmpty job.jobQueue.physicalName+ pure (drained, len, hidden)+ drained `shouldBe` 0+ len `shouldBe` 1+ hidden `shouldBe` True+ processSpan <- theProcessSpan job.jobName =<< capturedSpans provider spansRef+ csEventNames processSpan `shouldSatisfy` elem "shibuya.handler.started"+ csEventNames processSpan `shouldSatisfy` elem "exception"+ csEventNames processSpan `shouldNotSatisfy` elem "shibuya.handler.completed"+ textAttr (csAttributes processSpan) "shibuya.ack.decision" `shouldBe` Nothing+ csStatus processSpan `shouldSatisfy` \case+ OTel.Error reason -> "handler exception: " `Text.isPrefixOf` reason+ _ -> False++ it "a message with no trace headers still gets exactly one process span" $ \connStr -> do+ (provider, spansRef) <- setupCapturingProvider+ let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions+ job = mkJob "keiro_pgmq_test.span_no_parent"+ -- Plain 'enqueue' writes no headers at all, so there is no traceparent+ -- to extract and shibuya falls back to the ambient local context.+ runDbTraced connStr tracer $ do+ ensureJobQueue job+ _ <- enqueue job (Ping "plain" 1)+ runJobOnce 1 job (\_ -> pure Done)+ processSpan <- theProcessSpan job.jobName =<< capturedSpans provider spansRef+ csKind processSpan `shouldBe` OTel.Consumer+ textAttr (csAttributes processSpan) "shibuya.ack.decision" `shouldBe` Just "ack_ok"+ csStatus processSpan `shouldBe` OTel.Ok++ it "a FIFO delivery carries shibuya.partition on its process span" $ \connStr -> do+ (provider, spansRef) <- setupCapturingProvider+ let tracer = OTel.makeTracer provider "keiro-pgmq-test" OTel.tracerOptions+ job = mkJob "keiro_pgmq_test.span_partition"+ drained <-+ runDbTraced connStr tracer $ do+ ensureOrderedJobQueue job+ _ <- enqueueToGroup job "g1" (Ping "grouped" 1)+ runJobOnceWithContext (withOrdering FifoThroughput defaultJobTuning) 1 job \_ctx _p ->+ pure Done+ drained `shouldBe` 1+ processSpan <- theProcessSpan job.jobName =<< capturedSpans provider spansRef+ textAttr (csAttributes processSpan) "shibuya.partition" `shouldBe` Just "g1" -- EP-2 M1: unlogged vs standard provisioning. it "ensureJobQueueWith unlogged creates an unlogged queue" $ \connStr -> do