diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/kiroku-otel.cabal b/kiroku-otel.cabal --- a/kiroku-otel.cabal +++ b/kiroku-otel.cabal @@ -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 diff --git a/src/Kiroku/Otel/Subscription.hs b/src/Kiroku/Otel/Subscription.hs --- a/src/Kiroku/Otel/Subscription.hs +++ b/src/Kiroku/Otel/Subscription.hs @@ -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 diff --git a/src/Kiroku/Otel/TraceContext.hs b/src/Kiroku/Otel/TraceContext.hs --- a/src/Kiroku/Otel/TraceContext.hs +++ b/src/Kiroku/Otel/TraceContext.hs @@ -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": "" -> } - -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": "" +-- > } +-- +-- 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 diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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"