kiroku-otel (empty) → 0.2.0.0
raw patch · 5 files changed
Files
- CHANGELOG.md +81/−0
- kiroku-otel.cabal +86/−0
- src/Kiroku/Otel/Subscription.hs +517/−0
- src/Kiroku/Otel/TraceContext.hs +88/−0
- test/Main.hs +503/−0
+ CHANGELOG.md view
@@ -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.
+ kiroku-otel.cabal view
@@ -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
+ src/Kiroku/Otel/Subscription.hs view
@@ -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
+ src/Kiroku/Otel/TraceContext.hs view
@@ -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": "<vendor entries, optional>"+> }++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
+ test/Main.hs view
@@ -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"