diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,81 @@ +# kiroku-otel changelog + +## Unreleased + +## 0.2.0.0 — 2026-05-31 + +### Breaking Changes + +#### Complete FSM-state span coverage (`Kiroku.Otel.Subscription`, MasterPlan 7 EP-2) + +- The per-batch span is now keyed on the new `KirokuEventSubscriptionDelivered` + event and named `kiroku.subscription.deliver` (replacing the `spanFetch` / + `kiroku.subscription.fetch` span). It is emitted **once per delivered batch on + every target** (`$all`, category, consumer group) and in **both** phases, + carrying `kiroku.batch.rows` and a `kiroku.subscription.state` of `"catchup"` + or `"live"`. This closes the gap where an `$all` subscription's live phase + produced no per-batch span. `KirokuEventSubscriptionFetched` is now ignored by + the tracer (matched for exhaustiveness) so the DB-driven live path does not emit + two spans per batch. +- `kiroku.subscription.stopped` is **always** emitted on `Stopped` (carrying + `kiroku.subscription.stop_reason` and `kiroku.checkpoint.global_position`), so + the terminal state appears in the trace even for a healthy worker that stops + from `Live` with no open episode span. +- Internal: the per-key span state is striped from one shared `MVar` into a + lock-free `IORef (Map SpanKey (IORef OpenState))` (each key single-writer; the + outer registry mutated only on `Started`/`Stopped`), so the now-per-batch span + work never serializes workers on a shared lock. `base`-only, no public API + change, behavior-identical. +- New exported span-name constants `spanDeliver` / `spanStopped` (replacing + `spanFetch`). A database-backed end-to-end test now runs a real `$all` worker + with the tracer installed and asserts the catch-up, live `deliver`, and + `stopped` spans all appear. + +### New Features + +#### Subscription-state tracing (`Kiroku.Otel.Subscription`, MasterPlan 6 EP-5) + +- `subscriptionTraceHandler :: Tracer -> IO (KirokuEvent -> IO ())` — a ready-made + `eventHandler` callback that turns the subscription worker's finite-state + lifecycle into OpenTelemetry spans. Install it on `ConnectionSettings` + (or a per-subscription) `eventHandler` and every subscription emits spans. +- Short, promptly-ending spans (the SDK only exports a span on `endSpan`, so + there is no worker-lifetime span): per-episode `kiroku.subscription.catchup` / + `paused` / `reconnecting` / `retrying`, per-batch `kiroku.subscription.fetch`, + and standalone `kiroku.subscription.dead_letter` / `db_error`. Each carries a + `kiroku.*` attribute set (subscription name, state, consumer-group member/size, + checkpoint position, attempt, batch rows, event position, dead-letter/stop + reason); the span-name and attribute-key constants are exported. +- The library remains separate from `kiroku-store`, so `kiroku-store` gains no + OpenTelemetry dependency. The handler keeps its open spans in a thread-safe + `MVar` keyed by `(subscription name, member)`. Requires a **batch span + processor** so the synchronous callback never blocks the worker on export. + +### Other Changes + +#### OpenTelemetry 1.0 and current semantic-convention keys + +- `kiroku-otel` now depends on the OpenTelemetry 1.0 package family: + `hs-opentelemetry-api ^>=1.0`, `hs-opentelemetry-propagator-w3c ^>=1.0`, + and `hs-opentelemetry-semantic-conventions ^>=1.40`. +- Subscription spans now emit generated OpenTelemetry semantic-convention + attribute keys alongside the existing `kiroku.*` keys. Delivery spans include + `messaging.system = "kiroku"`, `messaging.destination.name`, + `messaging.operation.type = "process"`, and + `messaging.batch.message_count`; database-error spans include + `db.system.name = "postgresql"` and `db.operation.name`. + +## 0.1.0.0 — 2026-05-23 + +### New Features + +- Initial release. Exposes `Kiroku.Otel.TraceContext` with two helpers: + - `injectTraceContext :: SpanContext -> EventData -> EventData` — encodes + the supplied `SpanContext` to W3C `traceparent` / `tracestate` strings + and merges them into the event's `metadata` JSON object. Existing keys + in `metadata` are preserved. + - `extractTraceContext :: RecordedEvent -> Maybe SpanContext` — reads + the same JSON keys back out and decodes them through + `OpenTelemetry.Propagator.W3CTraceContext.decodeSpanContext`. Returns + `Nothing` when `metadata` is absent, is not a JSON object, lacks a + `traceparent` key, or carries an unparseable value. Never throws. diff --git a/kiroku-otel.cabal b/kiroku-otel.cabal new file mode 100644 --- /dev/null +++ b/kiroku-otel.cabal @@ -0,0 +1,86 @@ +cabal-version: 3.0 +name: kiroku-otel +version: 0.2.0.0 +synopsis: + OpenTelemetry W3C trace-context helpers for Kiroku event metadata + +description: + Pure helpers to inject and extract W3C trace-context (@traceparent@ and + @tracestate@ header strings, per the W3C trace-context specification) + into and out of the @metadata@ JSONB column carried by every Kiroku + event. Provided as a sister package to keep @kiroku-store@ free of any + @hs-opentelemetry@ dependency; consumers opt in by depending on this + package directly. + +author: Nadeem Bitar +maintainer: nadeem@gmail.com +license: BSD-3-Clause +build-type: Simple +category: Database, Eventing, Observability +extra-doc-files: CHANGELOG.md +homepage: https://github.com/shinzui/kiroku +bug-reports: https://github.com/shinzui/kiroku/issues + +source-repository head + type: git + location: https://github.com/shinzui/kiroku.git + +common common + default-language: GHC2024 + default-extensions: + DeriveAnyClass + DuplicateRecordFields + OverloadedLabels + OverloadedStrings + + ghc-options: -Wall + +library + import: common + exposed-modules: + Kiroku.Otel.Subscription + Kiroku.Otel.TraceContext + + build-depends: + , aeson >=2.1 && <2.3 + , base >=4.18 && <5 + , bytestring >=0.11 && <0.13 + , containers >=0.6 && <0.8 + , generic-lens >=2.2 && <2.4 + , hs-opentelemetry-api ^>=1.0 + , hs-opentelemetry-propagator-w3c ^>=1.0 + , hs-opentelemetry-semantic-conventions ^>=1.40 + , kiroku-store ^>=0.2 + , lens >=5.2 && <5.4 + , text >=2.0 && <2.2 + , unordered-containers >=0.2 && <0.3 + + hs-source-dirs: src + +test-suite kiroku-otel-test + import: common + type: exitcode-stdio-1.0 + main-is: Main.hs + hs-source-dirs: test + ghc-options: -threaded -rtsopts -with-rtsopts=-N + build-depends: + , aeson >=2.1 && <2.3 + , base >=4.18 && <5 + , bytestring >=0.11 && <0.13 + , containers >=0.6 && <0.8 + , ephemeral-pg >=0.2 && <0.3 + , generic-lens >=2.2 && <2.4 + , hasql-pool >=1.2 && <1.5 + , hs-opentelemetry-api + , hs-opentelemetry-exporter-in-memory ^>=1.0 + , hs-opentelemetry-sdk ^>=1.0 + , hspec >=2.10 && <2.12 + , kiroku-otel + , kiroku-store ^>=0.2 + , kiroku-test-support + , lens >=5.2 && <5.4 + , stm >=2.5 && <2.6 + , text >=2.0 && <2.2 + , time >=1.12 && <1.15 + , unordered-containers >=0.2 && <0.3 + , uuid >=1.3 && <1.4 diff --git a/src/Kiroku/Otel/Subscription.hs b/src/Kiroku/Otel/Subscription.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Otel/Subscription.hs @@ -0,0 +1,517 @@ +{- | 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 + spanCatchup, + spanDeliver, + spanPaused, + spanReconnecting, + spanRetrying, + spanDeadLetter, + spanDbError, + spanStopped, + + -- * Attribute keys + attrSubName, + attrState, + attrAttempt, + attrCheckpoint, + attrGroupMember, + attrGroupSize, + attrBatchRows, + attrEventPos, + attrDeadLetterReason, + attrStopReason, + attrDbPhase, + attrMessagingSystem, + attrMessagingDestinationName, + attrMessagingOperationType, + attrMessagingBatchMessageCount, + attrDbSystemName, + attrDbOperationName, +) where + +import Control.Applicative ((<|>)) +import Data.HashMap.Strict qualified as HashMap +import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef, writeIORef) +import Data.Int (Int32, Int64) +import Data.Map.Strict (Map) +import Data.Map.Strict qualified as Map +import Data.Text (Text) +import Data.Text qualified as T +import GHC.Generics (Generic) +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 (..), + Span, + SpanStatus (..), + Tracer, + addAttribute, + addEvent, + createSpan, + 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. +-} +subscriptionTraceHandler :: Tracer -> IO (KirokuEvent -> IO ()) +subscriptionTraceHandler tracer = do + 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. +-} +type SpanCells = IORef (Map SpanKey (IORef OpenState)) + +-- | Identifies one worker: subscription name plus consumer-group member (if any). +type SpanKey = (Text, Maybe Int32) + +-- | 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) + +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. +-} +primaryEpisode :: OpenState -> Maybe Span +primaryEpisode st = reconnect st <|> catchup st <|> pause st + +onEvent :: Tracer -> SpanCells -> KirokuEvent -> IO () +onEvent tracer cell = \case + 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 () + KirokuEventHardDeleteIssued{} -> 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. +-} +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. + +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' + +{- | 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 + +-- | 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 + +{- | 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) + +-- | End a span at the current time. +closeSpan :: Span -> IO () +closeSpan sp = endSpan sp Nothing + +-- | 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 + +-- | 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 + } + +-- | Build the 'SpanKey' identifying the worker that emitted an event. +keyOf :: SubscriptionName -> SubscriptionGroupContext -> SpanKey +keyOf (SubscriptionName nm) grp = (nm, memberOf grp) + +-- | The member index of a group context (Nothing for a non-grouped worker). +memberOf :: SubscriptionGroupContext -> Maybe Int32 +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. +-} +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)] + +-- | A low-cardinality operation name for database-error telemetry. +dbOperationName :: SubscriptionDbPhase -> Text +dbOperationName = \case + LoadCheckpoint -> "load_checkpoint" + FetchBatch -> "fetch_batch" + SaveCheckpoint -> "save_checkpoint" + +-- | A 'GlobalPosition' as an 'Int64' attribute. +posAttr :: GlobalPosition -> Attribute +posAttr (GlobalPosition p) = toAttribute p + +-- | Any small integral count as an 'Int64' attribute. +intAttr :: (Integral a) => a -> Attribute +intAttr n = toAttribute (fromIntegral n :: Int64) + +-- Span name constants --------------------------------------------------------- + +spanCatchup, spanDeliver, spanPaused, spanReconnecting, spanRetrying, spanDeadLetter, spanDbError, spanStopped :: Text +spanCatchup = "kiroku.subscription.catchup" +spanDeliver = "kiroku.subscription.deliver" +spanPaused = "kiroku.subscription.paused" +spanReconnecting = "kiroku.subscription.reconnecting" +spanRetrying = "kiroku.subscription.retrying" +spanDeadLetter = "kiroku.subscription.dead_letter" +spanDbError = "kiroku.subscription.db_error" +spanStopped = "kiroku.subscription.stopped" + +-- Attribute key constants ----------------------------------------------------- + +attrSubName, attrState, attrAttempt, attrCheckpoint :: Text +attrGroupMember, attrGroupSize, attrBatchRows, attrEventPos :: Text +attrDeadLetterReason, attrStopReason, attrDbPhase :: Text +attrMessagingSystem, attrMessagingDestinationName, attrMessagingOperationType, attrMessagingBatchMessageCount :: Text +attrDbSystemName, attrDbOperationName :: Text +attrSubName = "kiroku.subscription.name" +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 new file mode 100644 --- /dev/null +++ b/src/Kiroku/Otel/TraceContext.hs @@ -0,0 +1,88 @@ +{- | 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 + +import Control.Lens ((&), (?~)) +import Data.Aeson (Value (..)) +import Data.Aeson.Key qualified as Key +import Data.Aeson.KeyMap (KeyMap) +import Data.Aeson.KeyMap qualified as KM +import Data.ByteString (ByteString) +import Data.Generics.Labels () +import Data.Text qualified as T +import Data.Text.Encoding qualified as TE +import GHC.IO (unsafePerformIO) +import Kiroku.Store.Types (EventData (..), RecordedEvent (..)) +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. +-} +{-# NOINLINE injectTraceContext #-} +injectTraceContext :: SpanContext -> EventData -> EventData +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. +-} +extractTraceContext :: RecordedEvent -> Maybe SpanContext +extractTraceContext re = case re of + RecordedEvent{metadata = Just (Object o)} -> decodeFromObject o + _ -> Nothing + +-- --------------------------------------------------------------------------- +-- Internal helpers +-- --------------------------------------------------------------------------- + +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 + +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 diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -0,0 +1,503 @@ +-- 'head' is used in the tracing tests immediately after asserting the list has +-- exactly one element, so the partiality is guarded by the preceding assertion. +{-# OPTIONS_GHC -Wno-x-partial #-} + +module Main where + +import Control.Concurrent.STM (TVar, atomically, newTVarIO, readTVar, registerDelay, writeTVar) +import Control.Concurrent.STM qualified as STM +import Control.Lens ((&), (.~)) +import Control.Monad (unless) +import Data.Aeson qualified as Aeson +import Data.Aeson.Key qualified as Key +import Data.Aeson.KeyMap qualified as KM +import Data.Foldable (toList) +import Data.Generics.Labels () +import Data.HashMap.Strict qualified as HashMap +import Data.IORef (readIORef) +import Data.Int (Int64) +import Data.List (sort) +import Data.Maybe (mapMaybe) +import Data.Text (Text) +import Data.Text qualified as T +import Data.UUID qualified as UUID +import Hasql.Pool (UsageError (..)) +import Kiroku.Otel.Subscription ( + attrAttempt, + attrBatchRows, + attrCheckpoint, + attrDbOperationName, + attrDbSystemName, + attrDeadLetterReason, + attrEventPos, + attrGroupMember, + attrMessagingBatchMessageCount, + attrMessagingDestinationName, + attrMessagingOperationType, + attrMessagingSystem, + attrState, + attrStopReason, + attrSubName, + spanCatchup, + spanDbError, + spanDeadLetter, + spanDeliver, + spanPaused, + spanReconnecting, + spanRetrying, + spanStopped, + subscriptionTraceHandler, + ) +import Kiroku.Otel.TraceContext (extractTraceContext, injectTraceContext) +import Kiroku.Store ( + ExpectedVersion (..), + KirokuStore, + StreamName (..), + SubscriptionResult (..), + SubscriptionTarget (..), + appendToStream, + cancel, + defaultConnectionSettings, + defaultSubscriptionConfig, + runStoreIO, + subscribe, + wait, + withStore, + ) +import Kiroku.Store.Observability ( + DeadLetterReason (..), + KirokuEvent (..), + SubscriptionDbPhase (..), + SubscriptionDeliveryPhase (..), + SubscriptionGroupContext (..), + SubscriptionStopReason (..), + ) +import Kiroku.Store.Subscription.Types (SubscriptionName (..)) +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 (..), + ImmutableSpan (..), + SpanContext (..), + SpanHot (..), + createTracerProvider, + emptyTracerProviderOptions, + forceFlushTracerProvider, + makeTracer, + traceFlagsFromWord8, + tracerOptions, + ) +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" + + 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 + + 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")) + ] + ) + ) + ) + `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 "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) + + 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 "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 "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 "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 "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 "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 "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 "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 "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 + } + +eventDataMetadata :: EventData -> Maybe Aeson.Value +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" + } + +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 + } + +-- Subscription-tracing helpers ------------------------------------------------ + +-- | A fixed subscription name used across the tracing tests. +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. +-} +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 + +data CapturedSpan = CapturedSpan + { 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))) + } + +-- | The spans with the given name. +spansNamed :: Text -> [CapturedSpan] -> [CapturedSpan] +spansNamed n = filter ((== n) . capturedName) + +-- | Look up an attribute on a span. +attrOf :: Text -> CapturedSpan -> Maybe Attribute +attrOf k s = HashMap.lookup k (capturedAttributes s) + +-- | The names of the span events recorded on a span. +eventNamesOf :: CapturedSpan -> [Text] +eventNamesOf = capturedEvents + +-- | An 'Int64'-valued attribute (the encoding the tracer uses for counts/positions). +i64 :: Int64 -> Attribute +i64 = toAttribute + +-- End-to-end (real worker) helpers -------------------------------------------- + +-- | Increment a 'TVar' 'Int' counter by one within STM. +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. +-} +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) + +-- | 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) + +-- | 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"