kiroku-otel 0.2.0.12 → 0.3.0.0
raw patch · 5 files changed
+785/−758 lines, 5 filesdep ~kiroku-storePVP ok
version bump matches the API change (PVP)
Dependency ranges changed: kiroku-store
API changes (from Hackage documentation)
Files
- CHANGELOG.md +11/−0
- kiroku-otel.cabal +3/−3
- src/Kiroku/Otel/Subscription.hs +371/−365
- src/Kiroku/Otel/TraceContext.hs +55/−57
- test/Main.hs +345/−333
CHANGELOG.md view
@@ -1,5 +1,16 @@ # kiroku-otel changelog +## 0.3.0.0 — 2026-10-26++### Breaking Changes++- Require `kiroku-store ^>=0.12.0.0` for its expanded operational event surface.++### Bug Fixes++- Handle target-mismatch detail events without duplicating the subsequent+ terminal span, including startup refusals without `Started`.+ ## 0.2.0.12 — 2026-10-11 ### Other Changes
kiroku-otel.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.0 name: kiroku-otel-version: 0.2.0.12+version: 0.3.0.0 synopsis: OpenTelemetry W3C trace-context helpers for Kiroku event metadata @@ -51,7 +51,7 @@ , hs-opentelemetry-api ^>=1.0 , hs-opentelemetry-propagator-w3c ^>=1.0 , hs-opentelemetry-semantic-conventions ^>=1.40- , kiroku-store ^>=0.11.0.0+ , kiroku-store ^>=0.12.0.0 , lens >=5.2 && <5.4 , text >=2.0 && <2.2 , unordered-containers >=0.2 && <0.3@@ -77,7 +77,7 @@ , hs-opentelemetry-sdk ^>=1.0 , hspec >=2.10 && <2.12 , kiroku-otel- , kiroku-store ^>=0.11.0.0+ , kiroku-store ^>=0.12.0.0 , kiroku-test-support , lens >=5.2 && <5.4 , stm >=2.5 && <2.6
src/Kiroku/Otel/Subscription.hs view
@@ -1,96 +1,95 @@-{- | Turn a Kiroku subscription worker's lifecycle into OpenTelemetry spans.--A Kiroku /subscription/ is a long-lived worker that reads events in order and-feeds them to a handler, remembering progress in a durable checkpoint. The-worker is an explicit finite state machine — at any instant it is in exactly one-of @CatchingUp@, @Live@, @Paused@, @Reconnecting@, @Retrying@, or @Stopped@-(see @Kiroku.Store.Subscription.Fsm@) — and it announces every transition as a-structured 'KirokuEvent' delivered synchronously to the optional-@eventHandler :: Maybe (KirokuEvent -> IO ())@ callback an operator installs on-their connection settings.--This module bridges that event stream to OpenTelemetry traces. Call-'subscriptionTraceHandler' with a 'Tracer' to obtain a ready-made-@'KirokuEvent' -> IO ()@; install it as the subscription @eventHandler@ and-thereafter every subscription emits spans. On a trace timeline an operator can-then watch a subscription catch up, go live, pause under backpressure, reconnect-after a database outage, retry a poison event, and dead-letter it — each span-tagged with the subscription name, the consumer-group member, the checkpoint-position, the attempt counter, and batch sizes (the @kiroku.*@ attribute keys-exported below).--== Span model and the export-on-end constraint--A span is only exported to the backend when it /ends/ — the SDK's span-processors fire @onEnd@, never on a snapshot of an in-flight span (verified in-@hs-opentelemetry@: @endSpan@ → @tracerProviderOnEnd@ → @spanProcessorOnEnd@,-with no partial export). A single span held open for the worker's whole lifetime-would therefore be invisible while the worker runs and lost entirely on a crash.--So this module uses __short, promptly-ending spans__, never one lifetime span:--* __Episode spans__ open on a state-entry event and end on the matching exit- event, so a /completed/ episode shows its real duration:-- * @kiroku.subscription.catchup@ — opened on @Started@, ended on @CaughtUp@.- * @kiroku.subscription.paused@ — opened on @Paused@, ended on @Resumed@.- * @kiroku.subscription.reconnecting@ — opened on the first @Reconnecting@,- ended on the next @CaughtUp@; later reconnect attempts add a- @reconnect.attempt@ span event.- * @kiroku.subscription.retrying@ — keyed per poison event by its global- position; opened on the first @Retrying@ for that position, ended either- when a @DeadLettered@ for the same position arrives (status @Error@) or- when the worker moves on (a live @Delivered@/@CaughtUp@, status @Ok@).--* __Per-batch work spans__ open and end immediately on every delivered batch:-- * @kiroku.subscription.deliver@ — one per non-empty batch handed to the- handler, on __every__ target (@$all@, category, consumer group) and in- __both__ phases. It carries @kiroku.batch.rows@ and a- @kiroku.subscription.state@ of @"catchup"@ or @"live"@, so an @$all@- subscription's live phase is now traced (previously it emitted no- per-batch event and so went dark while live). These export continuously,- giving live visibility without a long-lived @Live@ span.- (@KirokuEventSubscriptionFetched@ — the DB-driven live-fetch-rate signal —- is intentionally __not__ traced; the deliver span subsumes it, so the- category\/consumer-group live path does not emit two spans per batch.)--* __Standalone spans__ for point events with no open episode:- @kiroku.subscription.dead_letter@ (an immediate dead-letter, no retry),- @kiroku.subscription.db_error@ (a DB error with no episode span open; when one- /is/ open the error is recorded as a @kiroku.db_error@ span event on it), and- @kiroku.subscription.stopped@ — __always__ emitted on @Stopped@, carrying the- @kiroku.subscription.stop_reason@ and the @kiroku.checkpoint.global_position@,- so the terminal state is present in the trace even for a healthy worker that- stops from @Live@ with no open episode span.--The honest limitation: an /in-progress/ (unresolved) episode does not appear in-the backend until it ends. Real-time "what state is this worker in right now" is-served instead by the @currentState@ subscription-handle accessor and the-'KirokuEvent' log stream (and, eventually, by a deferred state-gauge metric).--== Threading and the batch-processor requirement--The @eventHandler@ callback runs __synchronously on the worker's emit-site-thread__, and a consumer group runs one worker per member on separate threads.-The open-span bookkeeping is therefore held in a striped, lock-free registry-keyed by @(subscription name, member)@ so two members never collide: each key-has its own single-writer 'IORef' 'OpenState' and the outer registry is mutated-only when a key is first seen or removed, so the per-batch deliver-span work-never serializes workers on a shared lock. Opening and ending a span is cheap-and in-memory; it is the /export/ that may block. Configure the 'Tracer''s-'TracerProvider' with a __batch span processor__ (the SDK default) so export-happens on a background thread and the synchronous callback never stalls the-worker loop.--This module is a pure read-side consumer of the 'KirokuEvent' surface: it adds-no behavior to the worker, changes no core type, and adds no @hs-opentelemetry@-dependency to @kiroku-store@. The 'KirokuEvent' constructor set is /additive/,-so a future subscription constructor surfaces here as a-@-Wincomplete-patterns@ warning rather than a silent miss.--}-module Kiroku.Otel.Subscription (- -- * Handler factory+-- | Turn a Kiroku subscription worker's lifecycle into OpenTelemetry spans.+--+-- A Kiroku /subscription/ is a long-lived worker that reads events in order and+-- feeds them to a handler, remembering progress in a durable checkpoint. The+-- worker is an explicit finite state machine — at any instant it is in exactly one+-- of @CatchingUp@, @Live@, @Paused@, @Reconnecting@, @Retrying@, or @Stopped@+-- (see @Kiroku.Store.Subscription.Fsm@) — and it announces every transition as a+-- structured 'KirokuEvent' delivered synchronously to the optional+-- @eventHandler :: Maybe (KirokuEvent -> IO ())@ callback an operator installs on+-- their connection settings.+--+-- This module bridges that event stream to OpenTelemetry traces. Call+-- 'subscriptionTraceHandler' with a 'Tracer' to obtain a ready-made+-- @'KirokuEvent' -> IO ()@; install it as the subscription @eventHandler@ and+-- thereafter every subscription emits spans. On a trace timeline an operator can+-- then watch a subscription catch up, go live, pause under backpressure, reconnect+-- after a database outage, retry a poison event, and dead-letter it — each span+-- tagged with the subscription name, the consumer-group member, the checkpoint+-- position, the attempt counter, and batch sizes (the @kiroku.*@ attribute keys+-- exported below).+--+-- == Span model and the export-on-end constraint+--+-- A span is only exported to the backend when it /ends/ — the SDK's span+-- processors fire @onEnd@, never on a snapshot of an in-flight span (verified in+-- @hs-opentelemetry@: @endSpan@ → @tracerProviderOnEnd@ → @spanProcessorOnEnd@,+-- with no partial export). A single span held open for the worker's whole lifetime+-- would therefore be invisible while the worker runs and lost entirely on a crash.+--+-- So this module uses __short, promptly-ending spans__, never one lifetime span:+--+-- * __Episode spans__ open on a state-entry event and end on the matching exit+-- event, so a /completed/ episode shows its real duration:+--+-- * @kiroku.subscription.catchup@ — opened on @Started@, ended on @CaughtUp@.+-- * @kiroku.subscription.paused@ — opened on @Paused@, ended on @Resumed@.+-- * @kiroku.subscription.reconnecting@ — opened on the first @Reconnecting@,+-- ended on the next @CaughtUp@; later reconnect attempts add a+-- @reconnect.attempt@ span event.+-- * @kiroku.subscription.retrying@ — keyed per poison event by its global+-- position; opened on the first @Retrying@ for that position, ended either+-- when a @DeadLettered@ for the same position arrives (status @Error@) or+-- when the worker moves on (a live @Delivered@/@CaughtUp@, status @Ok@).+--+-- * __Per-batch work spans__ open and end immediately on every delivered batch:+--+-- * @kiroku.subscription.deliver@ — one per non-empty batch handed to the+-- handler, on __every__ target (@$all@, category, consumer group) and in+-- __both__ phases. It carries @kiroku.batch.rows@ and a+-- @kiroku.subscription.state@ of @"catchup"@ or @"live"@, so an @$all@+-- subscription's live phase is now traced (previously it emitted no+-- per-batch event and so went dark while live). These export continuously,+-- giving live visibility without a long-lived @Live@ span.+-- (@KirokuEventSubscriptionFetched@ — the DB-driven live-fetch-rate signal —+-- is intentionally __not__ traced; the deliver span subsumes it, so the+-- category\/consumer-group live path does not emit two spans per batch.)+--+-- * __Standalone spans__ for point events with no open episode:+-- @kiroku.subscription.dead_letter@ (an immediate dead-letter, no retry),+-- @kiroku.subscription.db_error@ (a DB error with no episode span open; when one+-- /is/ open the error is recorded as a @kiroku.db_error@ span event on it), and+-- @kiroku.subscription.stopped@ — __always__ emitted on @Stopped@, carrying the+-- @kiroku.subscription.stop_reason@ and the @kiroku.checkpoint.global_position@,+-- so the terminal state is present in the trace even for a healthy worker that+-- stops from @Live@ with no open episode span.+--+-- The honest limitation: an /in-progress/ (unresolved) episode does not appear in+-- the backend until it ends. Real-time "what state is this worker in right now" is+-- served instead by the @currentState@ subscription-handle accessor and the+-- 'KirokuEvent' log stream (and, eventually, by a deferred state-gauge metric).+--+-- == Threading and the batch-processor requirement+--+-- The @eventHandler@ callback runs __synchronously on the worker's emit-site+-- thread__, and a consumer group runs one worker per member on separate threads.+-- The open-span bookkeeping is therefore held in a striped, lock-free registry+-- keyed by @(subscription name, member)@ so two members never collide: each key+-- has its own single-writer 'IORef' 'OpenState' and the outer registry is mutated+-- only when a key is first seen or removed, so the per-batch deliver-span work+-- never serializes workers on a shared lock. Opening and ending a span is cheap+-- and in-memory; it is the /export/ that may block. Configure the 'Tracer''s+-- 'TracerProvider' with a __batch span processor__ (the SDK default) so export+-- happens on a background thread and the synchronous callback never stalls the+-- worker loop.+--+-- This module is a pure read-side consumer of the 'KirokuEvent' surface: it adds+-- no behavior to the worker, changes no core type, and adds no @hs-opentelemetry@+-- dependency to @kiroku-store@. The 'KirokuEvent' constructor set is /additive/,+-- so a future subscription constructor surfaces here as a+-- @-Wincomplete-patterns@ warning rather than a silent miss.+module Kiroku.Otel.Subscription+ ( -- * Handler factory subscriptionTraceHandler, -- * Span names@@ -121,7 +120,8 @@ attrMessagingBatchMessageCount, attrDbSystemName, attrDbOperationName,-) where+ )+where import Control.Applicative ((<|>)) import Data.HashMap.Strict qualified as HashMap@@ -132,19 +132,19 @@ import Data.Text (Text) import Data.Text qualified as T import GHC.Generics (Generic)-import Kiroku.Store.Observability (- KirokuEvent (..),+import Kiroku.Store.Observability+ ( KirokuEvent (..), SubscriptionDbPhase (..), SubscriptionDeliveryPhase (..), SubscriptionGroupContext (..),- )+ ) import Kiroku.Store.Subscription.Types (SubscriptionName (..)) import Kiroku.Store.Types (GlobalPosition (..)) import OpenTelemetry.Attributes (Attribute, AttributeKey (..), ToAttribute (toAttribute)) import OpenTelemetry.Context qualified as Context import OpenTelemetry.SemanticConventions qualified as Sem-import OpenTelemetry.Trace.Core (- NewEvent (..),+import OpenTelemetry.Trace.Core+ ( NewEvent (..), Span, SpanStatus (..), Tracer,@@ -154,38 +154,36 @@ defaultSpanArguments, endSpan, setStatus,- )--{- | Build a 'KirokuEvent' handler that turns subscription state into-OpenTelemetry spans, drawn from the given 'Tracer'.--@-import Kiroku.Otel.Subscription (subscriptionTraceHandler)--- tracer :: OpenTelemetry.Trace.Core.Tracer (from the app's TracerProvider)-handler <- subscriptionTraceHandler tracer--- install on the connection\/subscription settings:--- settings { eventHandler = Just handler }-@+ ) -The returned handler is thread-safe (a consumer group drives it from one thread-per member) and keeps its open-span state in a striped, lock-free per-key-registry keyed by @(subscription name, member)@: each key has its own-single-writer 'IORef' 'OpenState', and the outer registry is mutated only when a-key is first seen or removed. See the module documentation for the span model and-the batch-span-processor requirement.--}+-- | Build a 'KirokuEvent' handler that turns subscription state into+-- OpenTelemetry spans, drawn from the given 'Tracer'.+--+-- @+-- import Kiroku.Otel.Subscription (subscriptionTraceHandler)+-- -- tracer :: OpenTelemetry.Trace.Core.Tracer (from the app's TracerProvider)+-- handler <- subscriptionTraceHandler tracer+-- -- install on the connection\/subscription settings:+-- -- settings { eventHandler = Just handler }+-- @+--+-- The returned handler is thread-safe (a consumer group drives it from one thread+-- per member) and keeps its open-span state in a striped, lock-free per-key+-- registry keyed by @(subscription name, member)@: each key has its own+-- single-writer 'IORef' 'OpenState', and the outer registry is mutated only when a+-- key is first seen or removed. See the module documentation for the span model and+-- the batch-span-processor requirement. subscriptionTraceHandler :: Tracer -> IO (KirokuEvent -> IO ()) subscriptionTraceHandler tracer = do- cells <- newIORef Map.empty- pure (onEvent tracer cells)+ cells <- newIORef Map.empty+ pure (onEvent tracer cells) -{- | A per-key span-state registry. The outer 'IORef' is read-mostly — mutated-only when a key is first seen (@Started@) or removed (@Stopped@), via-'atomicModifyIORef''. Each key's inner 'IORef' 'OpenState' is single-writer (one-worker thread emits that key's events), so the per-batch hot path reads the outer-map lock-free and mutates only the key's own cell — no shared lock, no-cross-worker contention.--}+-- | A per-key span-state registry. The outer 'IORef' is read-mostly — mutated+-- only when a key is first seen (@Started@) or removed (@Stopped@), via+-- 'atomicModifyIORef''. Each key's inner 'IORef' 'OpenState' is single-writer (one+-- worker thread emits that key's events), so the per-batch hot path reads the outer+-- map lock-free and mutates only the key's own cell — no shared lock, no+-- cross-worker contention. type SpanCells = IORef (Map SpanKey (IORef OpenState)) -- | Identifies one worker: subscription name plus consumer-group member (if any).@@ -193,248 +191,244 @@ -- | The spans currently open for a single 'SpanKey'. data OpenState = OpenState- { catchup :: !(Maybe Span)- -- ^ Initial catch-up episode (opened on @Started@).- , reconnect :: !(Maybe Span)- -- ^ Reconnect episode (opened on the first @Reconnecting@).- , pause :: !(Maybe Span)- -- ^ Backpressure pause episode (opened on @Paused@).- , retries :: !(Map Int64 Span)- -- ^ Per-poison-event retry episodes, keyed by global position.- }- deriving stock (Generic)+ { -- | Initial catch-up episode (opened on @Started@).+ catchup :: !(Maybe Span),+ -- | Reconnect episode (opened on the first @Reconnecting@).+ reconnect :: !(Maybe Span),+ -- | Backpressure pause episode (opened on @Paused@).+ pause :: !(Maybe Span),+ -- | Per-poison-event retry episodes, keyed by global position.+ retries :: !(Map Int64 Span)+ }+ deriving stock (Generic) emptyOpenState :: OpenState emptyOpenState = OpenState Nothing Nothing Nothing Map.empty -{- | The most relevant single open episode span for a key, preferring an-in-flight reconnect, then catch-up, then pause. Used to attach DB-error span-events and the stop reason.--}+-- | The most relevant single open episode span for a key, preferring an+-- in-flight reconnect, then catch-up, then pause. Used to attach DB-error span+-- events and the stop reason. primaryEpisode :: OpenState -> Maybe Span primaryEpisode st = reconnect st <|> catchup st <|> pause st onEvent :: Tracer -> SpanCells -> KirokuEvent -> IO () onEvent tracer cell = \case- -- Resolution precedes Started and has no open episode to annotate. It is- -- retained as structured logging/metrics telemetry; tracing begins at the- -- Started catch-up span. A missing refusal is followed by Stopped, which- -- produces the terminal span.- KirokuEventSubscriptionCheckpointResolved{} -> pure ()- KirokuEventSubscriptionCheckpointMissing{} -> pure ()- KirokuEventSubscriptionTargetBound{} -> pure ()- KirokuEventSubscriptionGroupSizeMismatch{} -> pure ()- KirokuEventSubscriptionStarted name pos grp ->- withKey cell (keyOf name grp) $ \st -> do- -- Defensively close a catch-up span left open by a prior episode.- mapM_ closeSpan (catchup st)- sp <-- openSpan tracer spanCatchup $- baseAttrs name grp- ++ [(attrState, toAttribute ("catchup" :: Text)), (attrCheckpoint, posAttr pos)]- pure st{catchup = Just sp}- KirokuEventSubscriptionCaughtUp name pos grp ->- withKey cell (keyOf name grp) $ \st -> do- let endAttrs = [(attrCheckpoint, posAttr pos)]- -- A CaughtUp closes whichever of catch-up / reconnect is open.- mapM_ (`closeSpanWith` endAttrs) (catchup st)- mapM_ (\sp -> setStatus sp Ok >> closeSpanWith sp endAttrs) (reconnect st)- -- The worker has moved on, so any open retry succeeded.- mapM_ (\sp -> setStatus sp Ok >> closeSpan sp) (Map.elems (retries st))- pure st{catchup = Nothing, reconnect = Nothing, retries = Map.empty}- KirokuEventSubscriptionDelivered name count phase grp ->- withKey cell (keyOf name grp) $ \st -> do- let stateText = case phase of- DeliveredCatchUp -> "catchup" :: Text- DeliveredLive -> "live"- sp <-- openSpan tracer spanDeliver $- baseAttrs name grp- ++ [ (attrState, toAttribute stateText)- , (attrBatchRows, intAttr count)- , (attrMessagingOperationType, toAttribute ("process" :: Text))- , (attrMessagingBatchMessageCount, intAttr count)- ]- closeSpan sp- case phase of- -- A live batch means the worker advanced past any retried event,- -- so any still-open retry span succeeded: close them as Ok.- DeliveredLive -> do- mapM_ (\rsp -> setStatus rsp Ok >> closeSpan rsp) (Map.elems (retries st))- pure st{retries = Map.empty}- DeliveredCatchUp -> pure st- KirokuEventSubscriptionPaused name pos grp ->- withKey cell (keyOf name grp) $ \st -> do- mapM_ closeSpan (pause st) -- defensive- sp <-- openSpan tracer spanPaused $- baseAttrs name grp- ++ [(attrState, toAttribute ("paused" :: Text)), (attrCheckpoint, posAttr pos)]- pure st{pause = Just sp}- KirokuEventSubscriptionResumed name pos grp ->- withKey cell (keyOf name grp) $ \st -> do- -- Ignore a resume with no matching open pause span.- mapM_ (`closeSpanWith` [(attrCheckpoint, posAttr pos)]) (pause st)- pure st{pause = Nothing}- KirokuEventSubscriptionReconnecting name attempt grp ->- withKey cell (keyOf name grp) $ \st ->- case reconnect st of- Nothing -> do- sp <-- openSpan tracer spanReconnecting $- baseAttrs name grp- ++ [(attrState, toAttribute ("reconnecting" :: Text)), (attrAttempt, intAttr attempt)]- pure st{reconnect = Just sp}- Just sp -> do- spanEvent sp "reconnect.attempt" [(attrAttempt, intAttr attempt)]- setAttrs sp [(attrAttempt, intAttr attempt)]- pure st- KirokuEventSubscriptionRetrying name pos attempt grp ->- withKey cell (keyOf name grp) $ \st -> do- let GlobalPosition p = pos- case Map.lookup p (retries st) of- Nothing -> do- sp <-- openSpan tracer spanRetrying $- baseAttrs name grp- ++ [ (attrState, toAttribute ("retrying" :: Text))- , (attrEventPos, toAttribute p)- , (attrAttempt, intAttr attempt)- ]- pure st{retries = Map.insert p sp (retries st)}- Just sp -> do- spanEvent sp "retry.attempt" [(attrAttempt, intAttr attempt)]- setAttrs sp [(attrAttempt, intAttr attempt)]- pure st- KirokuEventSubscriptionDeadLettered name pos reason grp ->- withKey cell (keyOf name grp) $ \st -> do- let GlobalPosition p = pos- dlAttrs = [(attrDeadLetterReason, toAttribute (T.pack (show reason)))]- case Map.lookup p (retries st) of- Just sp -> do- -- A retry that exhausted: close the open retry span as dead-lettered.- setAttrs sp dlAttrs- spanEvent sp "dead_letter" dlAttrs- setStatus sp (Error "dead-lettered")- closeSpan sp- pure st{retries = Map.delete p (retries st)}- Nothing -> do- -- An immediate dead-letter (no retry): a short standalone span.- sp <-- openSpan tracer spanDeadLetter $- baseAttrs name grp ++ [(attrEventPos, toAttribute p)] ++ dlAttrs- closeSpan sp- pure st- KirokuEventSubscriptionDbError name phase _err grp ->- withKey cell (keyOf name grp) $ \st -> do- let phaseAttrs =- [ (attrDbPhase, toAttribute (T.pack (show phase)))- , (attrDbSystemName, toAttribute ("postgresql" :: Text))- , (attrDbOperationName, toAttribute (dbOperationName phase))- ]- case primaryEpisode st of- Just sp -> do- -- Annotate the open episode rather than spawn a competing span.- spanEvent sp "kiroku.db_error" phaseAttrs- pure st- Nothing -> do- sp <- openSpan tracer spanDbError (baseAttrs name grp ++ phaseAttrs)- closeSpan sp- pure st- KirokuEventSubscriptionStopped name pos reason grp ->- dropKey cell (keyOf name grp) $ \st -> do- let stopAttrs =- [ (attrStopReason, toAttribute (T.pack (show reason)))- , (attrCheckpoint, posAttr pos)- ]- -- Always emit a short standalone terminal span so the Stopped state is- -- present in the trace even for a healthy worker that stops from Live- -- with no open episode span.- term <- openSpan tracer spanStopped (baseAttrs name grp ++ stopAttrs)- closeSpan term- -- Also record the stop reason on the most relevant open episode (if any),- -- then end every open span so none leaks when the worker stops.- mapM_ (`setAttrs` stopAttrs) (primaryEpisode st)- mapM_ closeSpan (catchup st)- mapM_ closeSpan (reconnect st)- mapM_ closeSpan (pause st)- mapM_ closeSpan (Map.elems (retries st))- -- Fetched is now a no-op: the DB-driven live loops emit both Fetched and- -- Delivered per batch, and the deliver span (above) is keyed on Delivered so- -- the live path does not produce two spans per batch. Matched for exhaustiveness.- KirokuEventSubscriptionFetched{} -> pure ()- -- Non-subscription operational events are not traced here.- KirokuEventNotifierReconnecting{} -> pure ()- KirokuEventNotifierReconnected -> pure ()- KirokuEventPublisherPoolError{} -> pure ()- KirokuEventPublisherLoopError{} -> pure ()- KirokuEventPublisherDecodeFailed{} -> pure ()- KirokuEventSubscriptionHandlerStalled{} -> pure ()- KirokuEventHardDeleteIssued{} -> pure ()- KirokuEventHistoryRetentionLeaseAcquired{} -> pure ()- KirokuEventHistoryRetentionLeaseRenewed{} -> pure ()- KirokuEventHistoryRetentionLeaseReleased{} -> pure ()- KirokuEventHistoryRetentionLeasesPruned{} -> pure ()- KirokuEventHardDeleteHistoryRetentionConflict{} -> pure ()+ -- Resolution precedes Started and has no open episode to annotate. It is+ -- retained as structured logging/metrics telemetry; tracing begins at the+ -- Started catch-up span. A missing refusal is followed by Stopped, which+ -- produces the terminal span.+ KirokuEventSubscriptionCheckpointResolved {} -> pure ()+ KirokuEventSubscriptionCheckpointMissing {} -> pure ()+ KirokuEventSubscriptionTargetBound {} -> pure ()+ KirokuEventSubscriptionGroupSizeMismatch {} -> pure ()+ KirokuEventSubscriptionTargetMismatch {} -> pure ()+ KirokuEventSubscriptionStarted name pos grp ->+ withKey cell (keyOf name grp) $ \st -> do+ -- Defensively close a catch-up span left open by a prior episode.+ mapM_ closeSpan (catchup st)+ sp <-+ openSpan tracer spanCatchup $+ baseAttrs name grp+ ++ [(attrState, toAttribute ("catchup" :: Text)), (attrCheckpoint, posAttr pos)]+ pure st {catchup = Just sp}+ KirokuEventSubscriptionCaughtUp name pos grp ->+ withKey cell (keyOf name grp) $ \st -> do+ let endAttrs = [(attrCheckpoint, posAttr pos)]+ -- A CaughtUp closes whichever of catch-up / reconnect is open.+ mapM_ (`closeSpanWith` endAttrs) (catchup st)+ mapM_ (\sp -> setStatus sp Ok >> closeSpanWith sp endAttrs) (reconnect st)+ -- The worker has moved on, so any open retry succeeded.+ mapM_ (\sp -> setStatus sp Ok >> closeSpan sp) (Map.elems (retries st))+ pure st {catchup = Nothing, reconnect = Nothing, retries = Map.empty}+ KirokuEventSubscriptionDelivered name count phase grp ->+ withKey cell (keyOf name grp) $ \st -> do+ let stateText = case phase of+ DeliveredCatchUp -> "catchup" :: Text+ DeliveredLive -> "live"+ sp <-+ openSpan tracer spanDeliver $+ baseAttrs name grp+ ++ [ (attrState, toAttribute stateText),+ (attrBatchRows, intAttr count),+ (attrMessagingOperationType, toAttribute ("process" :: Text)),+ (attrMessagingBatchMessageCount, intAttr count)+ ]+ closeSpan sp+ case phase of+ -- A live batch means the worker advanced past any retried event,+ -- so any still-open retry span succeeded: close them as Ok.+ DeliveredLive -> do+ mapM_ (\rsp -> setStatus rsp Ok >> closeSpan rsp) (Map.elems (retries st))+ pure st {retries = Map.empty}+ DeliveredCatchUp -> pure st+ KirokuEventSubscriptionPaused name pos grp ->+ withKey cell (keyOf name grp) $ \st -> do+ mapM_ closeSpan (pause st) -- defensive+ sp <-+ openSpan tracer spanPaused $+ baseAttrs name grp+ ++ [(attrState, toAttribute ("paused" :: Text)), (attrCheckpoint, posAttr pos)]+ pure st {pause = Just sp}+ KirokuEventSubscriptionResumed name pos grp ->+ withKey cell (keyOf name grp) $ \st -> do+ -- Ignore a resume with no matching open pause span.+ mapM_ (`closeSpanWith` [(attrCheckpoint, posAttr pos)]) (pause st)+ pure st {pause = Nothing}+ KirokuEventSubscriptionReconnecting name attempt grp ->+ withKey cell (keyOf name grp) $ \st ->+ case reconnect st of+ Nothing -> do+ sp <-+ openSpan tracer spanReconnecting $+ baseAttrs name grp+ ++ [(attrState, toAttribute ("reconnecting" :: Text)), (attrAttempt, intAttr attempt)]+ pure st {reconnect = Just sp}+ Just sp -> do+ spanEvent sp "reconnect.attempt" [(attrAttempt, intAttr attempt)]+ setAttrs sp [(attrAttempt, intAttr attempt)]+ pure st+ KirokuEventSubscriptionRetrying name pos attempt grp ->+ withKey cell (keyOf name grp) $ \st -> do+ let GlobalPosition p = pos+ case Map.lookup p (retries st) of+ Nothing -> do+ sp <-+ openSpan tracer spanRetrying $+ baseAttrs name grp+ ++ [ (attrState, toAttribute ("retrying" :: Text)),+ (attrEventPos, toAttribute p),+ (attrAttempt, intAttr attempt)+ ]+ pure st {retries = Map.insert p sp (retries st)}+ Just sp -> do+ spanEvent sp "retry.attempt" [(attrAttempt, intAttr attempt)]+ setAttrs sp [(attrAttempt, intAttr attempt)]+ pure st+ KirokuEventSubscriptionDeadLettered name pos reason grp ->+ withKey cell (keyOf name grp) $ \st -> do+ let GlobalPosition p = pos+ dlAttrs = [(attrDeadLetterReason, toAttribute (T.pack (show reason)))]+ case Map.lookup p (retries st) of+ Just sp -> do+ -- A retry that exhausted: close the open retry span as dead-lettered.+ setAttrs sp dlAttrs+ spanEvent sp "dead_letter" dlAttrs+ setStatus sp (Error "dead-lettered")+ closeSpan sp+ pure st {retries = Map.delete p (retries st)}+ Nothing -> do+ -- An immediate dead-letter (no retry): a short standalone span.+ sp <-+ openSpan tracer spanDeadLetter $+ baseAttrs name grp ++ [(attrEventPos, toAttribute p)] ++ dlAttrs+ closeSpan sp+ pure st+ KirokuEventSubscriptionDbError name phase _err grp ->+ withKey cell (keyOf name grp) $ \st -> do+ let phaseAttrs =+ [ (attrDbPhase, toAttribute (T.pack (show phase))),+ (attrDbSystemName, toAttribute ("postgresql" :: Text)),+ (attrDbOperationName, toAttribute (dbOperationName phase))+ ]+ case primaryEpisode st of+ Just sp -> do+ -- Annotate the open episode rather than spawn a competing span.+ spanEvent sp "kiroku.db_error" phaseAttrs+ pure st+ Nothing -> do+ sp <- openSpan tracer spanDbError (baseAttrs name grp ++ phaseAttrs)+ closeSpan sp+ pure st+ KirokuEventSubscriptionStopped name pos reason grp ->+ dropKey cell (keyOf name grp) $ \st -> do+ let stopAttrs =+ [ (attrStopReason, toAttribute (T.pack (show reason))),+ (attrCheckpoint, posAttr pos)+ ]+ -- Always emit a short standalone terminal span so the Stopped state is+ -- present in the trace even for a healthy worker that stops from Live+ -- with no open episode span.+ term <- openSpan tracer spanStopped (baseAttrs name grp ++ stopAttrs)+ closeSpan term+ -- Also record the stop reason on the most relevant open episode (if any),+ -- then end every open span so none leaks when the worker stops.+ mapM_ (`setAttrs` stopAttrs) (primaryEpisode st)+ mapM_ closeSpan (catchup st)+ mapM_ closeSpan (reconnect st)+ mapM_ closeSpan (pause st)+ mapM_ closeSpan (Map.elems (retries st))+ -- Fetched is now a no-op: the DB-driven live loops emit both Fetched and+ -- Delivered per batch, and the deliver span (above) is keyed on Delivered so+ -- the live path does not produce two spans per batch. Matched for exhaustiveness.+ KirokuEventSubscriptionFetched {} -> pure ()+ -- Non-subscription operational events are not traced here.+ KirokuEventNotifierReconnecting {} -> pure ()+ KirokuEventNotifierReconnected -> pure ()+ KirokuEventPublisherPoolError {} -> pure ()+ KirokuEventPublisherLoopError {} -> pure ()+ KirokuEventPublisherDecodeFailed {} -> pure ()+ KirokuEventSubscriptionHandlerStalled {} -> pure ()+ KirokuEventHardDeleteIssued {} -> pure ()+ KirokuEventHistoryRetentionLeaseAcquired {} -> pure ()+ KirokuEventHistoryRetentionLeaseRenewed {} -> pure ()+ KirokuEventHistoryRetentionLeaseReleased {} -> pure ()+ KirokuEventHistoryRetentionLeasesPruned {} -> pure ()+ KirokuEventHardDeleteHistoryRetentionConflict {} -> pure () -- Span / state-cell helpers --------------------------------------------------- -{- | The 'IORef' holding a key's 'OpenState', creating an empty cell on the key's-first touch (its @Started@ event). The outer registry is read lock-free on the hot-path; it is written (via 'atomicModifyIORef'') only to insert a new key's cell. The-inner cell is single-writer, so reads/writes of it need no lock.--}+-- | The 'IORef' holding a key's 'OpenState', creating an empty cell on the key's+-- first touch (its @Started@ event). The outer registry is read lock-free on the hot+-- path; it is written (via 'atomicModifyIORef'') only to insert a new key's cell. The+-- inner cell is single-writer, so reads/writes of it need no lock. cellFor :: SpanCells -> SpanKey -> IO (IORef OpenState) cellFor cells key = do- m <- readIORef cells- case Map.lookup key m of- Just ref -> pure ref- Nothing -> do- fresh <- newIORef emptyOpenState- atomicModifyIORef' cells $ \m' ->- case Map.lookup key m' of- Just ref -> (m', ref) -- another key inserted meanwhile; reuse- Nothing -> (Map.insert key fresh m', fresh)--{- | Run a state-update against a key's own cell. The span IO in @f@ runs on the-key's single-writer 'IORef' with no shared lock held, so workers never serialize-on span work.+ m <- readIORef cells+ case Map.lookup key m of+ Just ref -> pure ref+ Nothing -> do+ fresh <- newIORef emptyOpenState+ atomicModifyIORef' cells $ \m' ->+ case Map.lookup key m' of+ Just ref -> (m', ref) -- another key inserted meanwhile; reuse+ Nothing -> (Map.insert key fresh m', fresh) -This is safe because each 'SpanKey' is __single-writer__: every-@(subscription name, member)@ is emitted by exactly one worker thread, and the-@eventHandler@ callback is synchronous on that thread, so a key's 'OpenState'-cannot change between the read and the write of its cell. The only cross-thread-structure is the outer registry, read lock-free on the hot path and mutated only-on the rare @Started@\/@Stopped@.--}+-- | Run a state-update against a key's own cell. The span IO in @f@ runs on the+-- key's single-writer 'IORef' with no shared lock held, so workers never serialize+-- on span work.+--+-- This is safe because each 'SpanKey' is __single-writer__: every+-- @(subscription name, member)@ is emitted by exactly one worker thread, and the+-- @eventHandler@ callback is synchronous on that thread, so a key's 'OpenState'+-- cannot change between the read and the write of its cell. The only cross-thread+-- structure is the outer registry, read lock-free on the hot path and mutated only+-- on the rare @Started@\/@Stopped@. withKey :: SpanCells -> SpanKey -> (OpenState -> IO OpenState) -> IO () withKey cells key f = do- ref <- cellFor cells key- st <- readIORef ref- st' <- f st- writeIORef ref st'+ ref <- cellFor cells key+ st <- readIORef ref+ st' <- f st+ writeIORef ref st' -{- | Run a finalizer against a key's 'OpenState', then drop the key from the outer-registry. @Stopped@ is a key's last event, so nothing touches the cell afterward.--}+-- | Run a finalizer against a key's 'OpenState', then drop the key from the outer+-- registry. @Stopped@ is a key's last event, so nothing touches the cell afterward. dropKey :: SpanCells -> SpanKey -> (OpenState -> IO ()) -> IO () dropKey cells key f = do- mref <- atomicModifyIORef' cells $ \m -> (Map.delete key m, Map.lookup key m)- st <- maybe (pure emptyOpenState) readIORef mref- f st+ mref <- atomicModifyIORef' cells $ \m -> (Map.delete key m, Map.lookup key m)+ st <- maybe (pure emptyOpenState) readIORef mref+ f st -- | Open a root span with the given name and initial attributes. openSpan :: Tracer -> Text -> [(Text, Attribute)] -> IO Span openSpan tracer name attrs = do- sp <- createSpan tracer Context.empty name defaultSpanArguments- setAttrs sp attrs- pure sp+ sp <- createSpan tracer Context.empty name defaultSpanArguments+ setAttrs sp attrs+ pure sp -{- | Set attributes on a span, overriding any existing value for the same key.-Uses the singular 'addAttribute' (an @insert@) rather than the bulk-@addAttributes@, whose left-biased union would keep a key's existing value and-silently drop the update (e.g. refreshing a checkpoint or attempt counter).--}+-- | Set attributes on a span, overriding any existing value for the same key.+-- Uses the singular 'addAttribute' (an @insert@) rather than the bulk+-- @addAttributes@, whose left-biased union would keep a key's existing value and+-- silently drop the update (e.g. refreshing a checkpoint or attempt counter). setAttrs :: Span -> [(Text, Attribute)] -> IO () setAttrs sp = mapM_ (\(k, v) -> addAttribute sp k v) @@ -445,19 +439,19 @@ -- | Add final attributes to a span and then end it. closeSpanWith :: Span -> [(Text, Attribute)] -> IO () closeSpanWith sp attrs = do- setAttrs sp attrs- endSpan sp Nothing+ setAttrs sp attrs+ endSpan sp Nothing -- | Add a timestamped span event with the given name and attributes. spanEvent :: Span -> Text -> [(Text, Attribute)] -> IO () spanEvent sp name attrs =- addEvent- sp- NewEvent- { newEventName = name- , newEventAttributes = HashMap.fromList attrs- , newEventTimestamp = Nothing- }+ addEvent+ sp+ NewEvent+ { newEventName = name,+ newEventAttributes = HashMap.fromList attrs,+ newEventTimestamp = Nothing+ } -- | Build the 'SpanKey' identifying the worker that emitted an event. keyOf :: SubscriptionName -> SubscriptionGroupContext -> SpanKey@@ -468,24 +462,23 @@ memberOf NonGroup = Nothing memberOf (GroupMember m _) = Just m -{- | The baseline attribute set set on every span: the subscription name and,-for a group member, its index and the group size.--}+-- | The baseline attribute set set on every span: the subscription name and,+-- for a group member, its index and the group size. baseAttrs :: SubscriptionName -> SubscriptionGroupContext -> [(Text, Attribute)] baseAttrs (SubscriptionName nm) grp =- (attrSubName, toAttribute nm)- : (attrMessagingSystem, toAttribute ("kiroku" :: Text))- : (attrMessagingDestinationName, toAttribute nm)- : case grp of- NonGroup -> []- GroupMember m sz -> [(attrGroupMember, intAttr m), (attrGroupSize, intAttr sz)]+ (attrSubName, toAttribute nm)+ : (attrMessagingSystem, toAttribute ("kiroku" :: Text))+ : (attrMessagingDestinationName, toAttribute nm)+ : case grp of+ NonGroup -> []+ GroupMember m sz -> [(attrGroupMember, intAttr m), (attrGroupSize, intAttr sz)] -- | A low-cardinality operation name for database-error telemetry. dbOperationName :: SubscriptionDbPhase -> Text dbOperationName = \case- LoadCheckpoint -> "load_checkpoint"- FetchBatch -> "fetch_batch"- SaveCheckpoint -> "save_checkpoint"+ LoadCheckpoint -> "load_checkpoint"+ FetchBatch -> "fetch_batch"+ SaveCheckpoint -> "save_checkpoint" -- | A 'GlobalPosition' as an 'Int64' attribute. posAttr :: GlobalPosition -> Attribute@@ -518,16 +511,29 @@ attrState = "kiroku.subscription.state" attrAttempt = "kiroku.subscription.attempt" attrCheckpoint = "kiroku.checkpoint.global_position"+ attrGroupMember = "kiroku.consumer_group.member"+ attrGroupSize = "kiroku.consumer_group.size"+ attrBatchRows = "kiroku.batch.rows"+ attrEventPos = "kiroku.event.global_position"+ attrDeadLetterReason = "kiroku.dead_letter.reason"+ attrStopReason = "kiroku.subscription.stop_reason"+ attrDbPhase = "kiroku.db.phase"+ attrMessagingSystem = unkey Sem.messaging_system+ attrMessagingDestinationName = unkey Sem.messaging_destination_name+ attrMessagingOperationType = unkey Sem.messaging_operation_type+ attrMessagingBatchMessageCount = unkey Sem.messaging_batch_messageCount+ attrDbSystemName = unkey Sem.db_system_name+ attrDbOperationName = unkey Sem.db_operation_name
src/Kiroku/Otel/TraceContext.hs view
@@ -1,19 +1,19 @@-{- | W3C trace-context helpers that read and write @traceparent@ /-@tracestate@ header strings inside Kiroku event metadata.--The on-the-wire JSON shape inside the event's @metadata@ JSONB column is:--> {-> "traceparent": "00-<32-hex traceId>-<16-hex spanId>-<2-hex flags>",-> "tracestate": "<vendor entries, optional>"-> }--Other keys in @metadata@ are preserved by 'injectTraceContext'.--}-module Kiroku.Otel.TraceContext (- injectTraceContext,+-- | W3C trace-context helpers that read and write @traceparent@ /+-- @tracestate@ header strings inside Kiroku event metadata.+--+-- The on-the-wire JSON shape inside the event's @metadata@ JSONB column is:+--+-- > {+-- > "traceparent": "00-<32-hex traceId>-<16-hex spanId>-<2-hex flags>",+-- > "tracestate": "<vendor entries, optional>"+-- > }+--+-- Other keys in @metadata@ are preserved by 'injectTraceContext'.+module Kiroku.Otel.TraceContext+ ( injectTraceContext, extractTraceContext,-) where+ )+where import Control.Lens ((&), (?~)) import Data.Aeson (Value (..))@@ -29,36 +29,34 @@ import OpenTelemetry.Propagator.W3CTraceContext (decodeSpanContext, encodeSpanContext) import OpenTelemetry.Trace.Core (SpanContext, wrapSpanContext) -{- | Encode a 'SpanContext' as W3C @traceparent@ / @tracestate@ strings and-merge them into the @metadata@ JSON object of an 'EventData'. Existing-keys in @metadata@ are preserved; existing @traceparent@ / @tracestate@-keys (if any) are overwritten — the W3C spec mandates exactly one of-each value per propagation.--If the input @metadata@ is 'Nothing', or is a non-object JSON value,-the helper starts from an empty object before merging.--This function is pure: it uses 'unsafePerformIO' to call-'encodeSpanContext' (OpenTelemetry 1.0 type:-@Span -> IO (ByteString, ByteString)@), which is observably pure on the-frozen span returned by 'wrapSpanContext' (no shared mutable state, no exceptions).-The \"unsafe\" annotation is mandatory to bridge the propagator's-@IO@-typed encoder to the pure interface this module exposes.--}+-- | Encode a 'SpanContext' as W3C @traceparent@ / @tracestate@ strings and+-- merge them into the @metadata@ JSON object of an 'EventData'. Existing+-- keys in @metadata@ are preserved; existing @traceparent@ / @tracestate@+-- keys (if any) are overwritten — the W3C spec mandates exactly one of+-- each value per propagation.+--+-- If the input @metadata@ is 'Nothing', or is a non-object JSON value,+-- the helper starts from an empty object before merging.+--+-- This function is pure: it uses 'unsafePerformIO' to call+-- 'encodeSpanContext' (OpenTelemetry 1.0 type:+-- @Span -> IO (ByteString, ByteString)@), which is observably pure on the+-- frozen span returned by 'wrapSpanContext' (no shared mutable state, no exceptions).+-- The \"unsafe\" annotation is mandatory to bridge the propagator's+-- @IO@-typed encoder to the pure interface this module exposes. {-# NOINLINE injectTraceContext #-} injectTraceContext :: SpanContext -> EventData -> EventData-injectTraceContext sc ed@EventData{metadata = oldMeta} =- ed & (#metadata ?~ Object (mergeTraceContext sc oldMeta))+injectTraceContext sc ed@EventData {metadata = oldMeta} =+ ed & (#metadata ?~ Object (mergeTraceContext sc oldMeta)) -{- | Pull a 'SpanContext' back out of a 'RecordedEvent'\'s @metadata@.-Returns 'Nothing' when @metadata@ is absent, is not a JSON object,-lacks a @traceparent@ key, or contains a @traceparent@ value that fails-W3C parsing. Never throws.--}+-- | Pull a 'SpanContext' back out of a 'RecordedEvent'\'s @metadata@.+-- Returns 'Nothing' when @metadata@ is absent, is not a JSON object,+-- lacks a @traceparent@ key, or contains a @traceparent@ value that fails+-- W3C parsing. Never throws. extractTraceContext :: RecordedEvent -> Maybe SpanContext extractTraceContext re = case re of- RecordedEvent{metadata = Just (Object o)} -> decodeFromObject o- _ -> Nothing+ RecordedEvent {metadata = Just (Object o)} -> decodeFromObject o+ _ -> Nothing -- --------------------------------------------------------------------------- -- Internal helpers@@ -66,23 +64,23 @@ mergeTraceContext :: SpanContext -> Maybe Value -> KeyMap Value mergeTraceContext sc mMeta =- let (tp, ts) = unsafePerformIO (encodeSpanContext (wrapSpanContext sc))- tpText = TE.decodeUtf8 tp- tsText = TE.decodeUtf8 ts- existing = case mMeta of- Just (Object o) -> o- _ -> KM.empty- withTp = KM.insert (Key.fromText (T.pack "traceparent")) (String tpText) existing- in if T.null tsText- then withTp- else KM.insert (Key.fromText (T.pack "tracestate")) (String tsText) withTp+ let (tp, ts) = unsafePerformIO (encodeSpanContext (wrapSpanContext sc))+ tpText = TE.decodeUtf8 tp+ tsText = TE.decodeUtf8 ts+ existing = case mMeta of+ Just (Object o) -> o+ _ -> KM.empty+ withTp = KM.insert (Key.fromText (T.pack "traceparent")) (String tpText) existing+ in if T.null tsText+ then withTp+ else KM.insert (Key.fromText (T.pack "tracestate")) (String tsText) withTp decodeFromObject :: KeyMap Value -> Maybe SpanContext decodeFromObject o = do- String tpText <- KM.lookup (Key.fromText (T.pack "traceparent")) o- let tpBs :: ByteString- tpBs = TE.encodeUtf8 tpText- tsBs = case KM.lookup (Key.fromText (T.pack "tracestate")) o of- Just (String tsText) -> Just (TE.encodeUtf8 tsText)- _ -> Nothing- decodeSpanContext (Just tpBs) tsBs+ String tpText <- KM.lookup (Key.fromText (T.pack "traceparent")) o+ let tpBs :: ByteString+ tpBs = TE.encodeUtf8 tpText+ tsBs = case KM.lookup (Key.fromText (T.pack "tracestate")) o of+ Just (String tsText) -> Just (TE.encodeUtf8 tsText)+ _ -> Nothing+ decodeSpanContext (Just tpBs) tsBs
test/Main.hs view
@@ -6,8 +6,9 @@ import Control.Concurrent.STM (TVar, atomically, newTVarIO, readTVar, registerDelay, writeTVar) import Control.Concurrent.STM qualified as STM+import Control.Exception (toException) import Control.Lens ((&), (.~))-import Control.Monad (unless)+import Control.Monad (forM_, unless) import Data.Aeson qualified as Aeson import Data.Aeson.Key qualified as Key import Data.Aeson.KeyMap qualified as KM@@ -22,8 +23,8 @@ import Data.Text qualified as T import Data.UUID qualified as UUID import Hasql.Pool (UsageError (..))-import Kiroku.Otel.Subscription (- attrAttempt,+import Kiroku.Otel.Subscription+ ( attrAttempt, attrBatchRows, attrCheckpoint, attrDbOperationName,@@ -47,10 +48,10 @@ spanRetrying, spanStopped, subscriptionTraceHandler,- )+ ) import Kiroku.Otel.TraceContext (extractTraceContext, injectTraceContext)-import Kiroku.Store (- ExpectedVersion (..),+import Kiroku.Store+ ( ExpectedVersion (..), KirokuStore, StreamName (..), SubscriptionResult (..),@@ -63,30 +64,30 @@ subscribe, wait, withStore,- )-import Kiroku.Store.Observability (- DeadLetterReason (..),+ )+import Kiroku.Store.Observability+ ( DeadLetterReason (..), KirokuEvent (..), SubscriptionDbPhase (..), SubscriptionDeliveryPhase (..), SubscriptionGroupContext (..), SubscriptionStopReason (..),- )-import Kiroku.Store.Subscription.Types (SubscriptionName (..))-import Kiroku.Store.Types (- EventData (..),+ )+import Kiroku.Store.Subscription.Types (InvalidHandlerStallWarnAfter (..), SubscriptionName (..), SubscriptionTargetMismatch (..))+import Kiroku.Store.Types+ ( EventData (..), EventId (..), EventType (..), GlobalPosition (..), RecordedEvent (..), StreamId (..), StreamVersion (..),- )+ ) import Kiroku.Test.Postgres (withMigratedTestDatabase, withSharedMigratedPostgres) import OpenTelemetry.Attributes (Attribute, getAttributeMap, toAttribute) import OpenTelemetry.Exporter.InMemory.Span (inMemoryListExporter)-import OpenTelemetry.Trace.Core (- Event (..),+import OpenTelemetry.Trace.Core+ ( Event (..), ImmutableSpan (..), SpanContext (..), SpanHot (..),@@ -96,306 +97,319 @@ makeTracer, traceFlagsFromWord8, tracerOptions,- )-import OpenTelemetry.Trace.Id (- Base (..),+ )+import OpenTelemetry.Trace.Id+ ( Base (..), baseEncodedToSpanId, baseEncodedToTraceId,- )+ ) import OpenTelemetry.Trace.TraceState qualified as TS import OpenTelemetry.Util (appendOnlyBoundedCollectionValues) import Test.Hspec main :: IO () main = withSharedMigratedPostgres $ hspec $ do- describe "TraceContext round-trip" $ do- it "encodes and decodes a SpanContext through metadata" $ do- let sc = mkTestSpanContext- ed0 = mkEmptyEventData- ed1 = injectTraceContext sc ed0- stub = mkStubRecorded (eventDataMetadata ed1)- case extractTraceContext stub of- Just sc' -> do- traceId (sc' :: SpanContext) `shouldBe` traceId (sc :: SpanContext)- spanId (sc' :: SpanContext) `shouldBe` spanId (sc :: SpanContext)- traceFlags (sc' :: SpanContext) `shouldBe` traceFlags (sc :: SpanContext)- Nothing -> expectationFailure "expected Just SpanContext"+ describe "TraceContext round-trip" $ do+ it "encodes and decodes a SpanContext through metadata" $ do+ let sc = mkTestSpanContext+ ed0 = mkEmptyEventData+ ed1 = injectTraceContext sc ed0+ stub = mkStubRecorded (eventDataMetadata ed1)+ case extractTraceContext stub of+ Just sc' -> do+ traceId (sc' :: SpanContext) `shouldBe` traceId (sc :: SpanContext)+ spanId (sc' :: SpanContext) `shouldBe` spanId (sc :: SpanContext)+ traceFlags (sc' :: SpanContext) `shouldBe` traceFlags (sc :: SpanContext)+ Nothing -> expectationFailure "expected Just SpanContext" - it "preserves existing metadata keys" $ do- let baseMeta =- Aeson.object- [ (Key.fromText (T.pack "tenant"), Aeson.String (T.pack "acme"))- ]- ed0 = mkEmptyEventDataWithMeta (Just baseMeta)- ed1 = injectTraceContext mkTestSpanContext ed0- case eventDataMetadata ed1 of- Just (Aeson.Object o) -> do- KM.lookup (Key.fromText (T.pack "tenant")) o- `shouldBe` Just (Aeson.String (T.pack "acme"))- KM.lookup (Key.fromText (T.pack "traceparent")) o- `shouldNotBe` Nothing- _ -> expectationFailure "metadata is not a JSON object"+ it "preserves existing metadata keys" $ do+ let baseMeta =+ Aeson.object+ [ (Key.fromText (T.pack "tenant"), Aeson.String (T.pack "acme"))+ ]+ ed0 = mkEmptyEventDataWithMeta (Just baseMeta)+ ed1 = injectTraceContext mkTestSpanContext ed0+ case eventDataMetadata ed1 of+ Just (Aeson.Object o) -> do+ KM.lookup (Key.fromText (T.pack "tenant")) o+ `shouldBe` Just (Aeson.String (T.pack "acme"))+ KM.lookup (Key.fromText (T.pack "traceparent")) o+ `shouldNotBe` Nothing+ _ -> expectationFailure "metadata is not a JSON object" - describe "extractTraceContext absence handling" $ do- it "returns Nothing when metadata is absent" $- extractTraceContext (mkStubRecorded Nothing) `shouldBe` Nothing+ describe "extractTraceContext absence handling" $ do+ it "returns Nothing when metadata is absent" $+ extractTraceContext (mkStubRecorded Nothing) `shouldBe` Nothing - it "returns Nothing when metadata is empty" $- extractTraceContext (mkStubRecorded (Just (Aeson.object [])))- `shouldBe` Nothing+ it "returns Nothing when metadata is empty" $+ extractTraceContext (mkStubRecorded (Just (Aeson.object [])))+ `shouldBe` Nothing - it "returns Nothing when traceparent is unparseable" $- extractTraceContext- ( mkStubRecorded- ( Just- ( Aeson.object- [ (Key.fromText (T.pack "traceparent"), Aeson.String (T.pack "garbage"))- ]- )- )+ it "returns Nothing when traceparent is unparseable" $+ extractTraceContext+ ( mkStubRecorded+ ( Just+ ( Aeson.object+ [ (Key.fromText (T.pack "traceparent"), Aeson.String (T.pack "garbage"))+ ] )- `shouldBe` Nothing+ )+ )+ `shouldBe` Nothing - describe "injectTraceContext overwrites prior trace keys" $ do- it "replaces an existing traceparent value" $ do- let preexisting =- Aeson.object- [ (Key.fromText (T.pack "traceparent"), Aeson.String (T.pack "00-aaaa-bbbb-00"))- , (Key.fromText (T.pack "tenant"), Aeson.String (T.pack "acme"))- ]- ed0 = mkEmptyEventDataWithMeta (Just preexisting)- ed1 = injectTraceContext mkTestSpanContext ed0- case eventDataMetadata ed1 of- Just (Aeson.Object o) -> do- KM.lookup (Key.fromText (T.pack "tenant")) o- `shouldBe` Just (Aeson.String (T.pack "acme"))- KM.lookup (Key.fromText (T.pack "traceparent")) o- `shouldNotBe` Just (Aeson.String (T.pack "00-aaaa-bbbb-00"))- _ -> expectationFailure "metadata is not a JSON object"+ describe "injectTraceContext overwrites prior trace keys" $ do+ it "replaces an existing traceparent value" $ do+ let preexisting =+ Aeson.object+ [ (Key.fromText (T.pack "traceparent"), Aeson.String (T.pack "00-aaaa-bbbb-00")),+ (Key.fromText (T.pack "tenant"), Aeson.String (T.pack "acme"))+ ]+ ed0 = mkEmptyEventDataWithMeta (Just preexisting)+ ed1 = injectTraceContext mkTestSpanContext ed0+ case eventDataMetadata ed1 of+ Just (Aeson.Object o) -> do+ KM.lookup (Key.fromText (T.pack "tenant")) o+ `shouldBe` Just (Aeson.String (T.pack "acme"))+ KM.lookup (Key.fromText (T.pack "traceparent")) o+ `shouldNotBe` Just (Aeson.String (T.pack "00-aaaa-bbbb-00"))+ _ -> expectationFailure "metadata is not a JSON object" - describe "subscriptionTraceHandler" $ do- it "catch-up then live yields an ended catchup span and a live deliver span" $ do- spans <-- runEvents- [ KirokuEventSubscriptionStarted subName (GlobalPosition 0) NonGroup- , KirokuEventSubscriptionCaughtUp subName (GlobalPosition 10) NonGroup- , KirokuEventSubscriptionDelivered subName 3 DeliveredLive NonGroup- ]- let catchups = spansNamed spanCatchup spans- delivers = spansNamed spanDeliver spans- length catchups `shouldBe` 1- length delivers `shouldBe` 1- -- the catch-up span carries the subscription name and its caught-up checkpoint- attrOf attrSubName (head catchups) `shouldBe` Just (toAttribute ("orders" :: Text))- attrOf attrCheckpoint (head catchups) `shouldBe` Just (i64 10)- -- the live deliver span carries the batch row count and state="live"- attrOf attrBatchRows (head delivers) `shouldBe` Just (i64 3)- attrOf attrState (head delivers) `shouldBe` Just (toAttribute ("live" :: Text))- -- standard OpenTelemetry messaging semantic-convention keys are emitted too- attrOf attrMessagingSystem (head delivers) `shouldBe` Just (toAttribute ("kiroku" :: Text))- attrOf attrMessagingDestinationName (head delivers) `shouldBe` Just (toAttribute ("orders" :: Text))- attrOf attrMessagingOperationType (head delivers) `shouldBe` Just (toAttribute ("process" :: Text))- attrOf attrMessagingBatchMessageCount (head delivers) `shouldBe` Just (i64 3)+ describe "subscriptionTraceHandler" $ do+ forM_ [False, True] $ \targetRefusal ->+ it ("emits one terminal span without Started for target refusal=" <> show targetRefusal) $ do+ let mismatch = SubscriptionTargetMismatch subName AllStreams (pure Nothing)+ err = if targetRefusal then toException mismatch else toException (InvalidHandlerStallWarnAfter 0)+ prefix = if targetRefusal then [KirokuEventSubscriptionTargetMismatch mismatch (GroupMember 1 2)] else []+ spans <- runEvents (prefix <> [KirokuEventSubscriptionStopped subName (GlobalPosition 0) (StopWorkerCrashed err) (GroupMember 1 2)])+ length spans `shouldBe` 1+ let stopped = spansNamed spanStopped spans+ length stopped `shouldBe` 1+ attrOf attrCheckpoint (head stopped) `shouldBe` Just (i64 0)+ attrOf attrStopReason (head stopped) `shouldBe` Just (toAttribute (T.pack (show (StopWorkerCrashed err))))+ attrOf attrGroupMember (head stopped) `shouldBe` Just (i64 1) - it "a catch-up delivery yields a deliver span tagged state=catchup" $ do- spans <-- runEvents- [ KirokuEventSubscriptionStarted subName (GlobalPosition 0) NonGroup- , KirokuEventSubscriptionDelivered subName 5 DeliveredCatchUp NonGroup- ]- let delivers = spansNamed spanDeliver spans- length delivers `shouldBe` 1- attrOf attrBatchRows (head delivers) `shouldBe` Just (i64 5)- attrOf attrState (head delivers) `shouldBe` Just (toAttribute ("catchup" :: Text))+ it "catch-up then live yields an ended catchup span and a live deliver span" $ do+ spans <-+ runEvents+ [ KirokuEventSubscriptionStarted subName (GlobalPosition 0) NonGroup,+ KirokuEventSubscriptionCaughtUp subName (GlobalPosition 10) NonGroup,+ KirokuEventSubscriptionDelivered subName 3 DeliveredLive NonGroup+ ]+ let catchups = spansNamed spanCatchup spans+ delivers = spansNamed spanDeliver spans+ length catchups `shouldBe` 1+ length delivers `shouldBe` 1+ -- the catch-up span carries the subscription name and its caught-up checkpoint+ attrOf attrSubName (head catchups) `shouldBe` Just (toAttribute ("orders" :: Text))+ attrOf attrCheckpoint (head catchups) `shouldBe` Just (i64 10)+ -- the live deliver span carries the batch row count and state="live"+ attrOf attrBatchRows (head delivers) `shouldBe` Just (i64 3)+ attrOf attrState (head delivers) `shouldBe` Just (toAttribute ("live" :: Text))+ -- standard OpenTelemetry messaging semantic-convention keys are emitted too+ attrOf attrMessagingSystem (head delivers) `shouldBe` Just (toAttribute ("kiroku" :: Text))+ attrOf attrMessagingDestinationName (head delivers) `shouldBe` Just (toAttribute ("orders" :: Text))+ attrOf attrMessagingOperationType (head delivers) `shouldBe` Just (toAttribute ("process" :: Text))+ attrOf attrMessagingBatchMessageCount (head delivers) `shouldBe` Just (i64 3) - it "a clean stop from live always yields a standalone stopped span" $ do- spans <-- runEvents- [ KirokuEventSubscriptionStarted subName (GlobalPosition 0) NonGroup- , KirokuEventSubscriptionCaughtUp subName (GlobalPosition 10) NonGroup- , KirokuEventSubscriptionDelivered subName 2 DeliveredLive NonGroup- , KirokuEventSubscriptionStopped subName (GlobalPosition 12) StopHandlerRequested NonGroup- ]- let stops = spansNamed spanStopped spans- length stops `shouldBe` 1- attrOf attrStopReason (head stops)- `shouldBe` Just (toAttribute (T.pack (show StopHandlerRequested)))- attrOf attrCheckpoint (head stops) `shouldBe` Just (i64 12)+ it "a catch-up delivery yields a deliver span tagged state=catchup" $ do+ spans <-+ runEvents+ [ KirokuEventSubscriptionStarted subName (GlobalPosition 0) NonGroup,+ KirokuEventSubscriptionDelivered subName 5 DeliveredCatchUp NonGroup+ ]+ let delivers = spansNamed spanDeliver spans+ length delivers `shouldBe` 1+ attrOf attrBatchRows (head delivers) `shouldBe` Just (i64 5)+ attrOf attrState (head delivers) `shouldBe` Just (toAttribute ("catchup" :: Text)) - it "pause then resume yields an ended paused span" $ do- spans <-- runEvents- [ KirokuEventSubscriptionPaused subName (GlobalPosition 5) NonGroup- , KirokuEventSubscriptionResumed subName (GlobalPosition 5) NonGroup- ]- -- present in the exported set ⇒ the episode closed and is observable- length (spansNamed spanPaused spans) `shouldBe` 1+ it "a clean stop from live always yields a standalone stopped span" $ do+ spans <-+ runEvents+ [ KirokuEventSubscriptionStarted subName (GlobalPosition 0) NonGroup,+ KirokuEventSubscriptionCaughtUp subName (GlobalPosition 10) NonGroup,+ KirokuEventSubscriptionDelivered subName 2 DeliveredLive NonGroup,+ KirokuEventSubscriptionStopped subName (GlobalPosition 12) StopHandlerRequested NonGroup+ ]+ let stops = spansNamed spanStopped spans+ length stops `shouldBe` 1+ attrOf attrStopReason (head stops)+ `shouldBe` Just (toAttribute (T.pack (show StopHandlerRequested)))+ attrOf attrCheckpoint (head stops) `shouldBe` Just (i64 12) - it "reconnect yields one span with a reconnect.attempt event and attempt=2" $ do- spans <-- runEvents- [ KirokuEventSubscriptionReconnecting subName 1 NonGroup- , KirokuEventSubscriptionReconnecting subName 2 NonGroup- , KirokuEventSubscriptionCaughtUp subName (GlobalPosition 7) NonGroup- ]- let reconnects = spansNamed spanReconnecting spans- length reconnects `shouldBe` 1- eventNamesOf (head reconnects) `shouldContain` ["reconnect.attempt"]- attrOf attrAttempt (head reconnects) `shouldBe` Just (i64 2)+ it "pause then resume yields an ended paused span" $ do+ spans <-+ runEvents+ [ KirokuEventSubscriptionPaused subName (GlobalPosition 5) NonGroup,+ KirokuEventSubscriptionResumed subName (GlobalPosition 5) NonGroup+ ]+ -- present in the exported set ⇒ the episode closed and is observable+ length (spansNamed spanPaused spans) `shouldBe` 1 - it "retry then dead-letter yields one retrying span with position and reason" $ do- let reason = DeadLetterPoison "bad event"- spans <-- runEvents- [ KirokuEventSubscriptionRetrying subName (GlobalPosition 42) 1 NonGroup- , KirokuEventSubscriptionRetrying subName (GlobalPosition 42) 2 NonGroup- , KirokuEventSubscriptionDeadLettered subName (GlobalPosition 42) reason NonGroup- ]- let retries = spansNamed spanRetrying spans- length retries `shouldBe` 1- attrOf attrEventPos (head retries) `shouldBe` Just (i64 42)- attrOf attrDeadLetterReason (head retries)- `shouldBe` Just (toAttribute (T.pack (show reason)))- -- the dead-letter closed the open retry span; no standalone span- length (spansNamed spanDeadLetter spans) `shouldBe` 0+ it "reconnect yields one span with a reconnect.attempt event and attempt=2" $ do+ spans <-+ runEvents+ [ KirokuEventSubscriptionReconnecting subName 1 NonGroup,+ KirokuEventSubscriptionReconnecting subName 2 NonGroup,+ KirokuEventSubscriptionCaughtUp subName (GlobalPosition 7) NonGroup+ ]+ let reconnects = spansNamed spanReconnecting spans+ length reconnects `shouldBe` 1+ eventNamesOf (head reconnects) `shouldContain` ["reconnect.attempt"]+ attrOf attrAttempt (head reconnects) `shouldBe` Just (i64 2) - it "an immediate dead-letter (no retry) yields a standalone dead_letter span" $ do- spans <-- runEvents- [KirokuEventSubscriptionDeadLettered subName (GlobalPosition 9) (DeadLetterPoison "x") NonGroup]- let dls = spansNamed spanDeadLetter spans- length dls `shouldBe` 1- attrOf attrEventPos (head dls) `shouldBe` Just (i64 9)+ it "retry then dead-letter yields one retrying span with position and reason" $ do+ let reason = DeadLetterPoison "bad event"+ spans <-+ runEvents+ [ KirokuEventSubscriptionRetrying subName (GlobalPosition 42) 1 NonGroup,+ KirokuEventSubscriptionRetrying subName (GlobalPosition 42) 2 NonGroup,+ KirokuEventSubscriptionDeadLettered subName (GlobalPosition 42) reason NonGroup+ ]+ let retries = spansNamed spanRetrying spans+ length retries `shouldBe` 1+ attrOf attrEventPos (head retries) `shouldBe` Just (i64 42)+ attrOf attrDeadLetterReason (head retries)+ `shouldBe` Just (toAttribute (T.pack (show reason)))+ -- the dead-letter closed the open retry span; no standalone span+ length (spansNamed spanDeadLetter spans) `shouldBe` 0 - it "consumer-group members produce separate, correctly tagged spans" $ do- spans <-- runEvents- [ KirokuEventSubscriptionStarted subName (GlobalPosition 0) (GroupMember 0 2)- , KirokuEventSubscriptionStarted subName (GlobalPosition 0) (GroupMember 1 2)- , KirokuEventSubscriptionCaughtUp subName (GlobalPosition 4) (GroupMember 0 2)- , KirokuEventSubscriptionCaughtUp subName (GlobalPosition 6) (GroupMember 1 2)- ]- let catchups = spansNamed spanCatchup spans- length catchups `shouldBe` 2- sort (mapMaybe (attrOf attrGroupMember) catchups)- `shouldBe` sort [i64 0, i64 1]+ it "an immediate dead-letter (no retry) yields a standalone dead_letter span" $ do+ spans <-+ runEvents+ [KirokuEventSubscriptionDeadLettered subName (GlobalPosition 9) (DeadLetterPoison "x") NonGroup]+ let dls = spansNamed spanDeadLetter spans+ length dls `shouldBe` 1+ attrOf attrEventPos (head dls) `shouldBe` Just (i64 9) - it "stop ends an open pause span so no span leaks" $ do- spans <-- runEvents- [ KirokuEventSubscriptionPaused subName (GlobalPosition 5) NonGroup- , KirokuEventSubscriptionStopped subName (GlobalPosition 5) StopCancelled NonGroup- ]- length (spansNamed spanPaused spans) `shouldBe` 1+ it "consumer-group members produce separate, correctly tagged spans" $ do+ spans <-+ runEvents+ [ KirokuEventSubscriptionStarted subName (GlobalPosition 0) (GroupMember 0 2),+ KirokuEventSubscriptionStarted subName (GlobalPosition 0) (GroupMember 1 2),+ KirokuEventSubscriptionCaughtUp subName (GlobalPosition 4) (GroupMember 0 2),+ KirokuEventSubscriptionCaughtUp subName (GlobalPosition 6) (GroupMember 1 2)+ ]+ let catchups = spansNamed spanCatchup spans+ length catchups `shouldBe` 2+ sort (mapMaybe (attrOf attrGroupMember) catchups)+ `shouldBe` sort [i64 0, i64 1] - it "a standalone database error span uses stable database semantic-convention keys" $ do- spans <-- runEvents- [KirokuEventSubscriptionDbError subName FetchBatch AcquisitionTimeoutUsageError NonGroup]- let dbErrors = spansNamed spanDbError spans- length dbErrors `shouldBe` 1- attrOf attrDbSystemName (head dbErrors) `shouldBe` Just (toAttribute ("postgresql" :: Text))- attrOf attrDbOperationName (head dbErrors) `shouldBe` Just (toAttribute ("fetch_batch" :: Text))+ it "stop ends an open pause span so no span leaks" $ do+ spans <-+ runEvents+ [ KirokuEventSubscriptionPaused subName (GlobalPosition 5) NonGroup,+ KirokuEventSubscriptionStopped subName (GlobalPosition 5) StopCancelled NonGroup+ ]+ length (spansNamed spanPaused spans) `shouldBe` 1 - describe "subscriptionTraceHandler end-to-end (real $all worker)" $ do- it "a real AllStreams worker emits catchup, a live deliver span, and a stopped span" $- withMigratedTestDatabase $ \connStr -> do- -- 1. In-memory exporter + provider + tracer + the trace handler.- (processor, spansRef) <- inMemoryListExporter- tp <- createTracerProvider [processor] emptyTracerProviderOptions- let tracer = makeTracer tp "kiroku-otel-e2e" tracerOptions- traceHandler <- subscriptionTraceHandler tracer- caughtUp <- newTVarIO False- let handler event = do- traceHandler event- case event of- KirokuEventSubscriptionCaughtUp{} -> atomically (writeTVar caughtUp True)- _ -> pure ()- -- 2. A store whose eventHandler IS the tracer.- let settings = defaultConnectionSettings connStr & #eventHandler .~ Just handler- withStore settings $ \store -> do- -- 3. Seed some history so the worker has a real catch-up phase.- appendStoreEvents store "e2e-catchup" 5- -- 4. A Continue handler that counts deliveries, so we can wait- -- for the worker to actually go live and deliver live events.- delivered <- newTVarIO (0 :: Int)- let cfg =- defaultSubscriptionConfig (SubscriptionName "otel-e2e") AllStreams $ \_event -> do- atomically (modifyTVarCount delivered)- pure Continue- handle <- subscribe store cfg- -- 5. Wait for catch-up to drain the 5 seeded events.- waitForCount delivered 5 10_000_000- waitForFlag caughtUp 10_000_000- -- 6. Append MORE events now that the worker is live; these flow- -- through the publisher's bounded queue into the (Nothing,- -- AllStreams) Live branch -> DeliverBatch -> processEvents ->- -- KirokuEventSubscriptionDelivered DeliveredLive.- appendStoreEvents store "e2e-live" 3- waitForCount delivered 8 10_000_000- -- 7. Stop the worker (cancel) and wait for the Stopped event.- cancel handle- _ <- wait handle- -- 8. Flush the exporter and read the collected, ended spans.- _ <- forceFlushTracerProvider tp Nothing- rawSpans <- readIORef spansRef- spans <- mapM captureSpan rawSpans- -- 9. Assertions: the two gaps are closed.- let catchups = spansNamed spanCatchup spans- delivers = spansNamed spanDeliver spans- liveDelivers =- filter ((== Just (toAttribute ("live" :: Text))) . attrOf attrState) delivers- stops = spansNamed spanStopped spans- length catchups `shouldSatisfy` (>= 1)- -- The gap-1 proof: at least one LIVE deliver span exists.- length liveDelivers `shouldSatisfy` (>= 1)- -- The gap-2 proof: a terminal stopped span exists.- length stops `shouldBe` 1+ it "a standalone database error span uses stable database semantic-convention keys" $ do+ spans <-+ runEvents+ [KirokuEventSubscriptionDbError subName FetchBatch AcquisitionTimeoutUsageError NonGroup]+ let dbErrors = spansNamed spanDbError spans+ length dbErrors `shouldBe` 1+ attrOf attrDbSystemName (head dbErrors) `shouldBe` Just (toAttribute ("postgresql" :: Text))+ attrOf attrDbOperationName (head dbErrors) `shouldBe` Just (toAttribute ("fetch_batch" :: Text)) + describe "subscriptionTraceHandler end-to-end (real $all worker)" $ do+ it "a real AllStreams worker emits catchup, a live deliver span, and a stopped span" $+ withMigratedTestDatabase $ \connStr -> do+ -- 1. In-memory exporter + provider + tracer + the trace handler.+ (processor, spansRef) <- inMemoryListExporter+ tp <- createTracerProvider [processor] emptyTracerProviderOptions+ let tracer = makeTracer tp "kiroku-otel-e2e" tracerOptions+ traceHandler <- subscriptionTraceHandler tracer+ caughtUp <- newTVarIO False+ let handler event = do+ traceHandler event+ case event of+ KirokuEventSubscriptionCaughtUp {} -> atomically (writeTVar caughtUp True)+ _ -> pure ()+ -- 2. A store whose eventHandler IS the tracer.+ let settings = defaultConnectionSettings connStr & #eventHandler .~ Just handler+ withStore settings $ \store -> do+ -- 3. Seed some history so the worker has a real catch-up phase.+ appendStoreEvents store "e2e-catchup" 5+ -- 4. A Continue handler that counts deliveries, so we can wait+ -- for the worker to actually go live and deliver live events.+ delivered <- newTVarIO (0 :: Int)+ let cfg =+ defaultSubscriptionConfig (SubscriptionName "otel-e2e") AllStreams $ \_event -> do+ atomically (modifyTVarCount delivered)+ pure Continue+ handle <- subscribe store cfg+ -- 5. Wait for catch-up to drain the 5 seeded events.+ waitForCount delivered 5 10_000_000+ waitForFlag caughtUp 10_000_000+ -- 6. Append MORE events now that the worker is live; these flow+ -- through the publisher's bounded queue into the (Nothing,+ -- AllStreams) Live branch -> DeliverBatch -> processEvents ->+ -- KirokuEventSubscriptionDelivered DeliveredLive.+ appendStoreEvents store "e2e-live" 3+ waitForCount delivered 8 10_000_000+ -- 7. Stop the worker (cancel) and wait for the Stopped event.+ cancel handle+ _ <- wait handle+ -- 8. Flush the exporter and read the collected, ended spans.+ _ <- forceFlushTracerProvider tp Nothing+ rawSpans <- readIORef spansRef+ spans <- mapM captureSpan rawSpans+ -- 9. Assertions: the two gaps are closed.+ let catchups = spansNamed spanCatchup spans+ delivers = spansNamed spanDeliver spans+ liveDelivers =+ filter ((== Just (toAttribute ("live" :: Text))) . attrOf attrState) delivers+ stops = spansNamed spanStopped spans+ length catchups `shouldSatisfy` (>= 1)+ -- The gap-1 proof: at least one LIVE deliver span exists.+ length liveDelivers `shouldSatisfy` (>= 1)+ -- The gap-2 proof: a terminal stopped span exists.+ length stops `shouldBe` 1+ mkEmptyEventData :: EventData mkEmptyEventData = mkEmptyEventDataWithMeta Nothing mkEmptyEventDataWithMeta :: Maybe Aeson.Value -> EventData mkEmptyEventDataWithMeta meta =- EventData- { eventId = Nothing- , eventType = EventType (T.pack "X")- , payload = Aeson.Null- , metadata = meta- , causationId = Nothing- , correlationId = Nothing- }+ EventData+ { eventId = Nothing,+ eventType = EventType (T.pack "X"),+ payload = Aeson.Null,+ metadata = meta,+ causationId = Nothing,+ correlationId = Nothing+ } eventDataMetadata :: EventData -> Maybe Aeson.Value-eventDataMetadata EventData{metadata = m} = m+eventDataMetadata EventData {metadata = m} = m mkStubRecorded :: Maybe Aeson.Value -> RecordedEvent mkStubRecorded meta =- RecordedEvent- { eventId = EventId UUID.nil- , eventType = EventType (T.pack "X")- , streamVersion = StreamVersion 1- , globalPosition = GlobalPosition 1- , originalStreamId = StreamId 1- , originalVersion = StreamVersion 1- , payload = Aeson.Null- , metadata = meta- , causationId = Nothing- , correlationId = Nothing- , createdAt = read "2026-05-14 00:00:00 UTC"- }+ RecordedEvent+ { eventId = EventId UUID.nil,+ eventType = EventType (T.pack "X"),+ streamVersion = StreamVersion 1,+ globalPosition = GlobalPosition 1,+ originalStreamId = StreamId 1,+ originalVersion = StreamVersion 1,+ payload = Aeson.Null,+ metadata = meta,+ causationId = Nothing,+ correlationId = Nothing,+ createdAt = read "2026-05-14 00:00:00 UTC"+ } mkTestSpanContext :: SpanContext mkTestSpanContext =- SpanContext- { traceFlags = traceFlagsFromWord8 0x01- , isRemote = True- , traceId =- either error id (baseEncodedToTraceId Base16 "4bf92f3577b34da6a3ce929d0e0e4736")- , spanId =- either error id (baseEncodedToSpanId Base16 "00f067aa0ba902b7")- , traceState = TS.empty- }+ SpanContext+ { traceFlags = traceFlagsFromWord8 0x01,+ isRemote = True,+ traceId =+ either error id (baseEncodedToTraceId Base16 "4bf92f3577b34da6a3ce929d0e0e4736"),+ spanId =+ either error id (baseEncodedToSpanId Base16 "00f067aa0ba902b7"),+ traceState = TS.empty+ } -- Subscription-tracing helpers ------------------------------------------------ @@ -403,38 +417,37 @@ subName :: SubscriptionName subName = SubscriptionName "orders" -{- | Drive a synthetic 'KirokuEvent' sequence through a fresh-'subscriptionTraceHandler' wired to an in-memory span exporter, and return the-exported (ended) spans. Because the in-memory exporter only ever receives ended-spans, a span appearing in the result is proof its episode closed.--}+-- | Drive a synthetic 'KirokuEvent' sequence through a fresh+-- 'subscriptionTraceHandler' wired to an in-memory span exporter, and return the+-- exported (ended) spans. Because the in-memory exporter only ever receives ended+-- spans, a span appearing in the result is proof its episode closed. runEvents :: [KirokuEvent] -> IO [CapturedSpan] runEvents evs = do- (processor, ref) <- inMemoryListExporter- tp <- createTracerProvider [processor] emptyTracerProviderOptions- let tracer = makeTracer tp "kiroku-otel-test" tracerOptions- handler <- subscriptionTraceHandler tracer- mapM_ handler evs- _ <- forceFlushTracerProvider tp Nothing- readIORef ref >>= mapM captureSpan+ (processor, ref) <- inMemoryListExporter+ tp <- createTracerProvider [processor] emptyTracerProviderOptions+ let tracer = makeTracer tp "kiroku-otel-test" tracerOptions+ handler <- subscriptionTraceHandler tracer+ mapM_ handler evs+ _ <- forceFlushTracerProvider tp Nothing+ readIORef ref >>= mapM captureSpan data CapturedSpan = CapturedSpan- { capturedName :: !Text- , capturedAttributes :: !AttributeMap- , capturedEvents :: ![Text]- }+ { capturedName :: !Text,+ capturedAttributes :: !AttributeMap,+ capturedEvents :: ![Text]+ } type AttributeMap = HashMap.HashMap Text Attribute captureSpan :: ImmutableSpan -> IO CapturedSpan captureSpan s = do- hot <- readIORef (spanHot s)- pure- CapturedSpan- { capturedName = hotName hot- , capturedAttributes = getAttributeMap (hotAttributes hot)- , capturedEvents = map eventName (toList (appendOnlyBoundedCollectionValues (hotEvents hot)))- }+ hot <- readIORef (spanHot s)+ pure+ CapturedSpan+ { capturedName = hotName hot,+ capturedAttributes = getAttributeMap (hotAttributes hot),+ capturedEvents = map eventName (toList (appendOnlyBoundedCollectionValues (hotEvents hot)))+ } -- | The spans with the given name. spansNamed :: Text -> [CapturedSpan] -> [CapturedSpan]@@ -458,46 +471,45 @@ modifyTVarCount :: TVar Int -> STM.STM () modifyTVarCount v = readTVar v >>= \c -> writeTVar v (c + 1) -{- | Append @n@ trivial events to a fresh stream via the native in-IO store-interpreter ('runStoreIO' + 'appendToStream'), mirroring @kiroku-store@'s own-tests. Each event is a minimal 'EventData' with a unique type tag.--}+-- | Append @n@ trivial events to a fresh stream via the native in-IO store+-- interpreter ('runStoreIO' + 'appendToStream'), mirroring @kiroku-store@'s own+-- tests. Each event is a minimal 'EventData' with a unique type tag. appendStoreEvents :: KirokuStore -> Text -> Int -> IO () appendStoreEvents store streamPrefix n = do- let events =- [ EventData- { eventId = Nothing- , eventType = EventType ("E" <> T.pack (show i))- , payload = Aeson.Null- , metadata = Nothing- , causationId = Nothing- , correlationId = Nothing- }- | i <- [1 .. n]- ]- result <- runStoreIO store (appendToStream (StreamName streamPrefix) NoStream events)- case result of- Right _ -> pure ()- Left err -> expectationFailure ("appendStoreEvents failed: " <> show err)+ let events =+ [ EventData+ { eventId = Nothing,+ eventType = EventType ("E" <> T.pack (show i)),+ payload = Aeson.Null,+ metadata = Nothing,+ causationId = Nothing,+ correlationId = Nothing+ }+ | i <- [1 .. n]+ ]+ result <- runStoreIO store (appendToStream (StreamName streamPrefix) NoStream events)+ case result of+ Right _ -> pure ()+ Left err -> expectationFailure ("appendStoreEvents failed: " <> show err) -- | Wait until the 'TVar' reaches @target@ or the timeout (micros) fires; fail on timeout. waitForCount :: TVar Int -> Int -> Int -> IO () waitForCount countVar target timeoutMicros = do- timeoutVar <- registerDelay timeoutMicros- ok <-- atomically $- (do c <- readTVar countVar; STM.check (c >= target); pure True)- `STM.orElse` (do t <- readTVar timeoutVar; STM.check t; pure False)- unless ok $ do- actual <- atomically (readTVar countVar)- expectationFailure ("Timed out waiting for " <> show target <> ", got " <> show actual)+ timeoutVar <- registerDelay timeoutMicros+ ok <-+ atomically $+ (do c <- readTVar countVar; STM.check (c >= target); pure True)+ `STM.orElse` (do t <- readTVar timeoutVar; STM.check t; pure False)+ unless ok $ do+ actual <- atomically (readTVar countVar)+ expectationFailure ("Timed out waiting for " <> show target <> ", got " <> show actual) -- | Wait until the 'TVar' flag is true or the timeout (micros) fires; fail on timeout. waitForFlag :: TVar Bool -> Int -> IO () waitForFlag flagVar timeoutMicros = do- timeoutVar <- registerDelay timeoutMicros- ok <-- atomically $- (do flag <- readTVar flagVar; STM.check flag; pure True)- `STM.orElse` (do t <- readTVar timeoutVar; STM.check t; pure False)- unless ok $ expectationFailure "Timed out waiting for subscription to catch up"+ timeoutVar <- registerDelay timeoutMicros+ ok <-+ atomically $+ (do flag <- readTVar flagVar; STM.check flag; pure True)+ `STM.orElse` (do t <- readTVar timeoutVar; STM.check t; pure False)+ unless ok $ expectationFailure "Timed out waiting for subscription to catch up"