diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 ` 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 diff --git a/keiro-pgmq.cabal b/keiro-pgmq.cabal --- a/keiro-pgmq.cabal +++ b/keiro-pgmq.cabal @@ -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 diff --git a/src/Keiro/PGMQ/Job.hs b/src/Keiro/PGMQ/Job.hs --- a/src/Keiro/PGMQ/Job.hs +++ b/src/Keiro/PGMQ/Job.hs @@ -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 @\ process@, and carries +@messaging.system=shibuya@, @messaging.destination.name=\@, +@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 \@, @receive \@, 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 @\ 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 +@\ 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) => diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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 @\ 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