kiroku-metrics (empty) → 0.1.0.0
raw patch · 19 files changed
Files
- CHANGELOG.md +17/−0
- example/Main.hs +227/−0
- kiroku-metrics.cabal +148/−0
- src/Kiroku/Metrics.hs +22/−0
- src/Kiroku/Metrics/Collector.hs +310/−0
- src/Kiroku/Metrics/Config.hs +55/−0
- src/Kiroku/Metrics/Health.hs +144/−0
- src/Kiroku/Metrics/JSON.hs +44/−0
- src/Kiroku/Metrics/Prometheus.hs +148/−0
- src/Kiroku/Metrics/Server.hs +222/−0
- src/Kiroku/Metrics/Subscriptions.hs +61/−0
- src/Kiroku/Metrics/Types.hs +159/−0
- src/Kiroku/Metrics/WebSocket.hs +479/−0
- test/Main.hs +18/−0
- test/Test/CollectorSpec.hs +197/−0
- test/Test/IntegrationSpec.hs +157/−0
- test/Test/ServerSpec.hs +209/−0
- test/Test/SubscriptionsSpec.hs +181/−0
- test/Test/WebSocketSpec.hs +357/−0
+ CHANGELOG.md view
@@ -0,0 +1,17 @@+# Revision history for kiroku-metrics++## 0.1.0.0 -- 2026-06-15++First release. A sister package to `kiroku-store` that exposes operational+metrics and event streams over HTTP without pulling a web framework into the+core library.++### New Features++* In-process metrics collector and JSON-encodable snapshot type.+* HTTP endpoints serving metrics as JSON, Prometheus exposition format, and a+ health check.+* WebSocket channel for streaming live metrics and events out of a running+ store.+* Live subscription-status endpoint over HTTP, with a CLI remote client.+* Runnable, self-verifying `kiroku-metrics-example` and a user guide.
+ example/Main.hs view
@@ -0,0 +1,227 @@+{-# LANGUAGE OverloadedLabels #-}+{-# LANGUAGE ScopedTypeVariables #-}++{- | A self-verifying runnable demo of @kiroku-metrics@ (EP-4).++Boots an ephemeral, migrated PostgreSQL, opens a store with the metrics collector+wired into its callbacks /before/ @withStore@, starts the combined HTTP+WebSocket+server, appends a few events, then checks its own endpoints over real HTTP and a+real WebSocket. Prints a step transcript and exits non-zero if any check fails, so+running it is a test that the documented behavior holds:++@+cabal run kiroku-metrics-example+@+-}+module Main (main) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (async, wait)+import Control.Concurrent.STM (STM, TVar, atomically, newTVarIO, readTVar, writeTVar)+import Control.Lens ((&), (.~))+import Control.Monad (unless)+import Data.Aeson (Value (..), decode, encode, object, (.=))+import Data.Aeson.Key qualified as Key+import Data.Aeson.KeyMap qualified as KM+import Data.ByteString.Lazy (ByteString)+import Data.ByteString.Lazy qualified as LBS+import Data.IntMap.Strict qualified as IntMap+import Data.Scientific (toRealFloat)+import Data.Text (Text)+import Data.Text qualified as T+import Data.Text.Encoding qualified as TE+import Network.HTTP.Client (+ Manager,+ defaultManagerSettings,+ httpLbs,+ newManager,+ parseRequest,+ responseBody,+ responseStatus,+ )+import Network.HTTP.Types (statusCode)+import Network.WebSockets qualified as WS+import System.Exit (exitFailure)+import System.IO (hPutStrLn, stderr)+import System.Timeout (timeout)++import Kiroku.Metrics (+ MetricsServer (..),+ MetricsSnapshot (..),+ StoreGauges (..),+ defaultConfig,+ metricsEventHandler,+ metricsObservationHandler,+ newKirokuMetricsWith,+ postgresPing,+ snapshotMetrics,+ withMetricsServerWithStore,+ )+import Kiroku.Metrics.Config (MetricsServerConfig (..))+import Kiroku.Store (+ EventData (..),+ EventType (..),+ ExpectedVersion (..),+ KirokuStore (..),+ StreamName (..),+ appendToStream,+ defaultConnectionSettings,+ runStoreIO,+ withStore,+ )+import Kiroku.Store.Subscription.EventPublisher (EventPublisher (..), publisherPosition)+import Kiroku.Store.Types (GlobalPosition (..))+import Kiroku.Test.Postgres (withMigratedTestDatabase)++main :: IO ()+main = withMigratedTestDatabase $ \connStr -> do+ step "[1/6] ephemeral postgres ready"++ -- The collector must observe events from the first append, so its callbacks+ -- go on ConnectionSettings BEFORE withStore. But snapshots read store-level+ -- gauges from the live handle — so the collector is built from STM readers+ -- over a TVar that holds the store once it is open (the deferral pattern).+ storeVar <- newTVarIO Nothing+ metrics <- newKirokuMetricsWith (readPosition storeVar) (readSubscribers storeVar)+ let settings =+ defaultConnectionSettings connStr+ & #eventHandler .~ Just (metricsEventHandler metrics Nothing)+ & #observationHandler .~ Just (metricsObservationHandler metrics Nothing)++ withStore settings $ \store -> do+ atomically (writeTVar storeVar (Just store))+ withMetricsServerWithStore (defaultConfig{port = 0}) metrics store [postgresPing store] $ \srv -> do+ threadDelay 300_000+ let port = srv.serverPort+ base = "http://127.0.0.1:" <> show port+ step ("[2/6] store + collector + metrics server on port " <> show port)++ appendEvents store (StreamName "orders-1") ["OrderCreated", "OrderPaid", "OrderShipped"]+ step "[3/6] appended 3 events to orders-1"++ mgr <- newManager defaultManagerSettings++ (sMetrics, bMetrics) <- httpGet mgr (base <> "/metrics")+ check "GET /metrics is 200" (sMetrics == 200)+ let gpos = globalPositionOf bMetrics+ check ("GET /metrics store.global_position >= 3 (got " <> show gpos <> ")") (gpos >= 3)++ (sProm, bProm) <- httpGet mgr (base <> "/metrics/prometheus")+ check "GET /metrics/prometheus is 200" (sProm == 200)+ check+ "Prometheus body contains kiroku_events_appended_total"+ (T.isInfixOf "kiroku_events_appended_total" (bodyText bProm))++ (sLive, _) <- httpGet mgr (base <> "/health/live")+ check "GET /health/live is 200" (sLive == 200)+ (sReady, _) <- httpGet mgr (base <> "/health/ready")+ check "GET /health/ready is 200" (sReady == 200)+ step "[4/6] HTTP /metrics, /prometheus, /health/live, /health/ready all OK"++ -- WebSocket: subscribe to the live event tail, then append one more+ -- event and assert it arrives over the socket as a JSON event message.+ evType <-+ requireJust "WebSocket /ws/events round-trip timed out" $+ timeout 15_000_000 $+ WS.runClient "127.0.0.1" port "/ws/events" $ \conn -> do+ WS.sendTextData conn (encode (object ["type" .= ("subscribe_events" :: Text)]))+ _ <- waitForType conn "event_stream_started"+ appendThread <- async (appendEvents store (StreamName "orders-2") ["OrderRefunded"])+ ev <- waitForType conn "event"+ wait appendThread+ pure (eventTypeOf ev)+ check ("WebSocket event eventType == OrderRefunded (got " <> T.unpack evType <> ")") (evType == "OrderRefunded")+ step ("[5/6] WebSocket /ws/events received event eventType=" <> T.unpack evType)++ snap <- snapshotMetrics metrics+ step ("[6/6] kiroku-metrics-example: all checks passed (snapshot global position = " <> show snap.store.globalPosition <> ")")++--------------------------------------------------------------------------------+-- Helpers+--------------------------------------------------------------------------------++-- | Print a transcript step.+step :: String -> IO ()+step = putStrLn++-- | Assert a condition; on failure print to stderr and exit non-zero.+check :: String -> Bool -> IO ()+check label ok =+ unless ok $ do+ hPutStrLn stderr ("kiroku-metrics-example: FAILED: " <> label)+ exitFailure++-- | Require a timed action to have produced a result; else fail the example.+requireJust :: String -> IO (Maybe a) -> IO a+requireJust label act =+ act >>= \case+ Just a -> pure a+ Nothing -> do+ hPutStrLn stderr ("kiroku-metrics-example: FAILED: " <> label)+ exitFailure++-- | GET a URL, returning the status code and the body (no throwing on non-2xx).+httpGet :: Manager -> String -> IO (Int, ByteString)+httpGet mgr url = do+ req <- parseRequest url+ resp <- httpLbs req mgr+ pure (statusCode (responseStatus resp), responseBody resp)++-- | Read server messages until one with the given @"type"@ arrives; return it.+waitForType :: WS.Connection -> Text -> IO Value+waitForType conn want = go+ where+ go = do+ raw <- WS.receiveData conn :: IO ByteString+ case decode raw of+ Just v | lookKey ["type"] v == Just (String want) -> pure v+ _ -> go++-- | Extract @event.eventType@ from an @event@ message ("" if absent).+eventTypeOf :: Value -> Text+eventTypeOf v = case lookKey ["event", "eventType"] v of+ Just (String t) -> t+ _ -> ""++-- | Extract @store.global_position@ from a @/metrics@ JSON body.+globalPositionOf :: ByteString -> Int+globalPositionOf body = case decode body of+ Just v | Just (Number n) <- lookKey ["store", "global_position"] v -> truncate (toRealFloat n :: Double)+ _ -> -1++-- | Navigate nested object keys.+lookKey :: [Text] -> Value -> Maybe Value+lookKey [] v = Just v+lookKey (k : ks) (Object o) = KM.lookup (Key.fromText k) o >>= lookKey ks+lookKey _ _ = Nothing++bodyText :: ByteString -> Text+bodyText = TE.decodeUtf8 . LBS.toStrict++appendEvents :: KirokuStore -> StreamName -> [Text] -> IO ()+appendEvents store stream types = do+ let events =+ [ EventData+ { eventId = Nothing+ , eventType = EventType t+ , payload = Null+ , metadata = Nothing+ , causationId = Nothing+ , correlationId = Nothing+ }+ | t <- types+ ]+ result <- runStoreIO store (appendToStream stream NoStream events)+ case result of+ Right _ -> pure ()+ Left err -> do+ hPutStrLn stderr ("kiroku-metrics-example: append failed: " <> show err)+ exitFailure++readPosition :: TVar (Maybe KirokuStore) -> STM GlobalPosition+readPosition storeVar =+ readTVar storeVar >>= maybe (pure (GlobalPosition 0)) (publisherPosition . (.publisher))++readSubscribers :: TVar (Maybe KirokuStore) -> STM Int+readSubscribers storeVar =+ readTVar storeVar >>= maybe (pure 0) (\s -> IntMap.size <$> readTVar (subscribers s.publisher))
+ kiroku-metrics.cabal view
@@ -0,0 +1,148 @@+cabal-version: 3.0+name: kiroku-metrics+version: 0.1.0.0+synopsis:+ Metrics, health, and event-streaming HTTP endpoints for Kiroku++description:+ HTTP/JSON, Prometheus, and WebSocket endpoints exposing operational metrics for+ a running Kiroku event store, plus a WebSocket channel for streaming events out+ of the store. A sister package to @kiroku-store@; the core library never depends+ on a web framework.++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+ DerivingStrategies+ DuplicateRecordFields+ LambdaCase+ OverloadedRecordDot+ OverloadedStrings+ RecordWildCards++ ghc-options: -Wall++library+ import: common+ exposed-modules:+ Kiroku.Metrics+ Kiroku.Metrics.Collector+ Kiroku.Metrics.Config+ Kiroku.Metrics.Health+ Kiroku.Metrics.JSON+ Kiroku.Metrics.Prometheus+ Kiroku.Metrics.Server+ Kiroku.Metrics.Subscriptions+ Kiroku.Metrics.Types+ Kiroku.Metrics.WebSocket++ build-depends:+ , aeson >=2.1 && <2.3+ , async >=2.2 && <2.3+ , base >=4.18 && <5+ , bytestring >=0.11 && <0.13+ , containers >=0.6 && <0.8+ , hasql >=1.10 && <1.11+ , hasql-pool >=1.2 && <1.5+ , http-types >=0.12 && <0.13+ , kiroku-cli ^>=0.1+ , kiroku-store ^>=0.2+ , stm >=2.5 && <2.6+ , text >=2.0 && <2.2+ , time >=1.12 && <1.15+ , uuid >=1.3 && <1.4+ , vector >=0.13 && <0.14+ , wai >=3.2 && <3.3+ , wai-websockets >=3.0 && <3.1+ , warp >=3.4 && <3.5+ , websockets >=0.13 && <0.14++ hs-source-dirs: src++flag example+ description:+ Build the self-verifying example executable. Off by default because it+ depends on @kiroku-test-support@/@ephemeral-pg@, which are not published to+ Hackage (so a default build of the released package would fail to resolve+ them) and have no buildable source in the pinned Nix Haskell set. Enable+ with @-fexample@ to run @cabal run kiroku-metrics-example@ in the dev shell.++ default: False+ manual: True++executable kiroku-metrics-example+ import: common+ main-is: Main.hs+ hs-source-dirs: example+ ghc-options: -threaded -rtsopts -with-rtsopts=-N++ if !flag(example)+ buildable: False++ build-depends:+ , aeson >=2.1 && <2.3+ , async >=2.2 && <2.3+ , base >=4.18 && <5+ , bytestring >=0.11 && <0.13+ , containers >=0.6 && <0.8+ , http-client >=0.7 && <0.8+ , http-types >=0.12 && <0.13+ , kiroku-metrics+ , kiroku-store ^>=0.2+ , kiroku-test-support+ , lens >=5.2 && <5.4+ , scientific >=0.3 && <0.4+ , stm >=2.5 && <2.6+ , text >=2.0 && <2.2+ , websockets >=0.13 && <0.14++test-suite kiroku-metrics-test+ import: common+ type: exitcode-stdio-1.0+ main-is: Main.hs+ other-modules:+ Test.CollectorSpec+ Test.IntegrationSpec+ Test.ServerSpec+ Test.SubscriptionsSpec+ Test.WebSocketSpec++ hs-source-dirs: test+ ghc-options: -threaded -rtsopts -with-rtsopts=-N+ build-depends:+ , aeson >=2.1 && <2.3+ , async >=2.2 && <2.3+ , base >=4.18 && <5+ , bytestring >=0.11 && <0.13+ , containers >=0.6 && <0.8+ , generic-lens >=2.2 && <2.4+ , hasql >=1.10 && <1.11+ , hasql-pool >=1.2 && <1.5+ , hspec >=2.10 && <2.12+ , http-client >=0.7 && <0.8+ , http-types >=0.12 && <0.13+ , kiroku-cli ^>=0.1+ , kiroku-metrics+ , kiroku-store ^>=0.2+ , kiroku-test-support+ , lens >=5.2 && <5.4+ , scientific >=0.3 && <0.4+ , stm >=2.5 && <2.6+ , text >=2.0 && <2.2+ , uuid >=1.3 && <1.4+ , warp >=3.4 && <3.5+ , websockets >=0.13 && <0.14
+ src/Kiroku/Metrics.hs view
@@ -0,0 +1,22 @@+{- | Umbrella re-export for the @kiroku-metrics@ package. Importing+@Kiroku.Metrics@ brings the in-process metrics collector, the JSON-encodable+snapshot type, the server configuration and lifecycle, and the health-check+types into scope.+-}+module Kiroku.Metrics (+ module Kiroku.Metrics.Types,+ module Kiroku.Metrics.Collector,+ module Kiroku.Metrics.Config,+ module Kiroku.Metrics.Health,+ module Kiroku.Metrics.Server,+ module Kiroku.Metrics.Subscriptions,+ module Kiroku.Metrics.WebSocket,+) where++import Kiroku.Metrics.Collector+import Kiroku.Metrics.Config+import Kiroku.Metrics.Health+import Kiroku.Metrics.Server+import Kiroku.Metrics.Subscriptions+import Kiroku.Metrics.Types+import Kiroku.Metrics.WebSocket
+ src/Kiroku/Metrics/Collector.hs view
@@ -0,0 +1,310 @@+{- | In-process metrics collector for a running Kiroku store.++A 'KirokuMetrics' is an opaque handle holding STM counters, a pool+connection-status map, and a per-subscription accumulation map. It is driven+entirely by the store's existing public callback seams — wrap+'metricsEventHandler' and 'metricsObservationHandler' into the+'Kiroku.Store.Connection.ConnectionSettings' @eventHandler@/@observationHandler@+fields /before/ opening the store, and the collector will see every event. Call+'snapshotMetrics' at any time for an immutable, JSON-encodable+'MetricsSnapshot'.++All callback updates are plain non-blocking STM ('modifyTVar''); they perform no+I/O. This matters because the store invokes the callbacks synchronously on the+emit-site thread (notifier loop, publisher loop, subscription worker, store+interpreter), so a slow callback would stall that loop.+-}+module Kiroku.Metrics.Collector (+ KirokuMetrics,+ newKirokuMetrics,+ newKirokuMetricsWith,+ metricsEventHandler,+ metricsObservationHandler,+ snapshotMetrics,+) where++import Control.Concurrent.STM (STM, TVar, atomically, modifyTVar', newTVarIO, readTVar)+import Data.Foldable (for_)+import Data.Int (Int64)+import Data.IntMap.Strict qualified as IntMap+import Data.Map.Strict (Map)+import Data.Map.Strict qualified as Map+import Data.Text (Text)+import Data.UUID (UUID)+import Kiroku.Metrics.Types (+ LifecycleCounters (..),+ MetricsSnapshot (..),+ StoreGauges (..),+ SubscriptionMetrics (..),+ )+import Kiroku.Store (+ ConnectionReadyForUseReason (..),+ ConnectionStatus (..),+ KirokuEvent (..),+ KirokuStore (..),+ Observation (..),+ SubscriptionDbPhase (..),+ SubscriptionStopReason (..),+ )+import Kiroku.Store.Subscription.EventPublisher (+ EventPublisher (..),+ publisherPosition,+ )+import Kiroku.Store.Subscription.Types (SubscriptionName (..))+import Kiroku.Store.Types (GlobalPosition (..))++-- | Opaque collector handle. Construct with 'newKirokuMetrics'.+data KirokuMetrics = KirokuMetrics+ { kmCounters :: !(TVar LifecycleCounters)+ , kmPool :: !(TVar PoolMutable)+ , kmSubs :: !(TVar (Map Text SubMutable))+ , kmPosition :: !(STM GlobalPosition)+ -- ^ Reads the store's global position at snapshot time.+ , kmSubscribers :: !(STM Int)+ -- ^ Reads the store's active-subscriber count at snapshot time.+ }++-- | Mutable pool state: per-connection status plus cumulative lifecycle counts.+data PoolMutable = PoolMutable+ { pmStatuses :: !(Map UUID ConnectionStatus)+ , pmEstablishedTotal :: !Int64+ , pmTerminatedTotal :: !Int64+ }++-- | Mutable per-subscription accumulation.+data SubMutable = SubMutable+ { smLastKnownPosition :: !Int64+ , smDbErrorCount :: !Int64+ , smLastStopReason :: !(Maybe Text)+ }++emptyCounters :: LifecycleCounters+emptyCounters =+ LifecycleCounters+ { notifierReconnecting = 0+ , notifierReconnected = 0+ , publisherPoolErrors = 0+ , publisherLoopErrors = 0+ , subscriptionDbErrorsLoad = 0+ , subscriptionDbErrorsFetch = 0+ , subscriptionDbErrorsSave = 0+ , subscriptionsStarted = 0+ , subscriptionsCaughtUp = 0+ , subscriptionsPaused = 0+ , subscriptionsResumed = 0+ , subscriptionsReconnecting = 0+ , subscriptionsRetrying = 0+ , subscriptionsDeadLettered = 0+ , subscriptionsStoppedHandler = 0+ , subscriptionsStoppedCancelled = 0+ , subscriptionsStoppedOverflow = 0+ , subscriptionsStoppedCrashed = 0+ , liveFetches = 0+ , batchesDelivered = 0+ , eventsDelivered = 0+ , hardDeletesIssued = 0+ }++emptySub :: SubMutable+emptySub = SubMutable{smLastKnownPosition = 0, smDbErrorCount = 0, smLastStopReason = Nothing}++{- | Construct a collector for a live store. The two store-level gauges+(global position and active-subscriber count) are read from the store's+'EventPublisher' at snapshot time.+-}+newKirokuMetrics :: KirokuStore -> IO KirokuMetrics+newKirokuMetrics store =+ newKirokuMetricsWith+ (publisherPosition store.publisher)+ (IntMap.size <$> readTVar (subscribers store.publisher))++{- | Construct a collector from explicit STM readers for the two store-level+gauges. This is the test seam: a unit test passes @pure (GlobalPosition n)@ and+@pure k@ to avoid needing a real store, and an integration test can defer the+store with a @TVar (Maybe KirokuStore)@ (STM readers cannot read an 'IORef').+-}+newKirokuMetricsWith :: STM GlobalPosition -> STM Int -> IO KirokuMetrics+newKirokuMetricsWith readPosition readSubscribers = do+ kmCounters <- newTVarIO emptyCounters+ kmPool <- newTVarIO (PoolMutable Map.empty 0 0)+ kmSubs <- newTVarIO Map.empty+ pure+ KirokuMetrics+ { kmCounters+ , kmPool+ , kmSubs+ , kmPosition = readPosition+ , kmSubscribers = readSubscribers+ }++{- | Wrap the collector into the store's @eventHandler@ seam, composing with an+optional user passthrough. The collector's STM update runs first, then the+passthrough is invoked with the same event.+-}+metricsEventHandler :: KirokuMetrics -> Maybe (KirokuEvent -> IO ()) -> (KirokuEvent -> IO ())+metricsEventHandler km mPassthrough event = do+ atomically (applyEvent km event)+ for_ mPassthrough ($ event)++{- | Wrap the collector into the store's @observationHandler@ seam, composing+with an optional user passthrough. The collector's STM update runs first.+-}+metricsObservationHandler :: KirokuMetrics -> Maybe (Observation -> IO ()) -> (Observation -> IO ())+metricsObservationHandler km mPassthrough obs = do+ atomically (applyObservation km obs)+ for_ mPassthrough ($ obs)++-- | Apply a single 'KirokuEvent' to the collector state.+applyEvent :: KirokuMetrics -> KirokuEvent -> STM ()+applyEvent km = \case+ KirokuEventNotifierReconnecting _ _ ->+ bumpCounters km (\c -> c{notifierReconnecting = c.notifierReconnecting + 1})+ KirokuEventNotifierReconnected ->+ bumpCounters km (\c -> c{notifierReconnected = c.notifierReconnected + 1})+ KirokuEventPublisherPoolError _ ->+ bumpCounters km (\c -> c{publisherPoolErrors = c.publisherPoolErrors + 1})+ KirokuEventPublisherLoopError _ ->+ bumpCounters km (\c -> c{publisherLoopErrors = c.publisherLoopErrors + 1})+ KirokuEventSubscriptionDbError name phase _ _ -> do+ bumpCounters km (bumpDbPhase phase)+ touchSub km name (\s -> s{smDbErrorCount = s.smDbErrorCount + 1})+ KirokuEventSubscriptionStarted name pos _ -> do+ bumpCounters km (\c -> c{subscriptionsStarted = c.subscriptionsStarted + 1})+ recordPosition km name pos+ KirokuEventSubscriptionCaughtUp name pos _ -> do+ bumpCounters km (\c -> c{subscriptionsCaughtUp = c.subscriptionsCaughtUp + 1})+ recordPosition km name pos+ KirokuEventSubscriptionStopped name pos reason _ -> do+ bumpCounters km (bumpStopReason reason)+ recordPosition km name pos+ touchSub km name (\s -> s{smLastStopReason = Just (stopReasonText reason)})+ KirokuEventSubscriptionPaused name pos _ -> do+ bumpCounters km (\c -> c{subscriptionsPaused = c.subscriptionsPaused + 1})+ recordPosition km name pos+ KirokuEventSubscriptionResumed name pos _ -> do+ bumpCounters km (\c -> c{subscriptionsResumed = c.subscriptionsResumed + 1})+ recordPosition km name pos+ KirokuEventSubscriptionReconnecting name _ _ -> do+ bumpCounters km (\c -> c{subscriptionsReconnecting = c.subscriptionsReconnecting + 1})+ touchSub km name id+ KirokuEventSubscriptionFetched name _ _ -> do+ bumpCounters km (\c -> c{liveFetches = c.liveFetches + 1})+ touchSub km name id+ KirokuEventSubscriptionDelivered name n _ _ -> do+ bumpCounters+ km+ ( \c ->+ c+ { batchesDelivered = c.batchesDelivered + 1+ , eventsDelivered = c.eventsDelivered + fromIntegral n+ }+ )+ touchSub km name id+ KirokuEventSubscriptionRetrying name pos _ _ -> do+ bumpCounters km (\c -> c{subscriptionsRetrying = c.subscriptionsRetrying + 1})+ recordPosition km name pos+ KirokuEventSubscriptionDeadLettered name pos _ _ -> do+ bumpCounters km (\c -> c{subscriptionsDeadLettered = c.subscriptionsDeadLettered + 1})+ recordPosition km name pos+ KirokuEventHardDeleteIssued _ _ ->+ bumpCounters km (\c -> c{hardDeletesIssued = c.hardDeletesIssued + 1})++bumpDbPhase :: SubscriptionDbPhase -> LifecycleCounters -> LifecycleCounters+bumpDbPhase phase c = case phase of+ LoadCheckpoint -> c{subscriptionDbErrorsLoad = c.subscriptionDbErrorsLoad + 1}+ FetchBatch -> c{subscriptionDbErrorsFetch = c.subscriptionDbErrorsFetch + 1}+ SaveCheckpoint -> c{subscriptionDbErrorsSave = c.subscriptionDbErrorsSave + 1}++bumpStopReason :: SubscriptionStopReason -> LifecycleCounters -> LifecycleCounters+bumpStopReason reason c = case reason of+ StopHandlerRequested -> c{subscriptionsStoppedHandler = c.subscriptionsStoppedHandler + 1}+ StopCancelled -> c{subscriptionsStoppedCancelled = c.subscriptionsStoppedCancelled + 1}+ StopOverflowed -> c{subscriptionsStoppedOverflow = c.subscriptionsStoppedOverflow + 1}+ StopWorkerCrashed _ -> c{subscriptionsStoppedCrashed = c.subscriptionsStoppedCrashed + 1}++stopReasonText :: SubscriptionStopReason -> Text+stopReasonText = \case+ StopHandlerRequested -> "handler"+ StopCancelled -> "cancelled"+ StopOverflowed -> "overflow"+ StopWorkerCrashed _ -> "crashed"++-- | Apply a single pool 'Observation' to the collector state.+applyObservation :: KirokuMetrics -> Observation -> STM ()+applyObservation km (ConnectionObservation connId status) =+ modifyTVar' (kmPool km) update+ where+ update p = case status of+ TerminatedConnectionStatus _ ->+ p+ { pmStatuses = Map.delete connId p.pmStatuses+ , pmTerminatedTotal = p.pmTerminatedTotal + 1+ }+ ReadyForUseConnectionStatus EstablishedConnectionReadyForUseReason ->+ p+ { pmStatuses = Map.insert connId status p.pmStatuses+ , pmEstablishedTotal = p.pmEstablishedTotal + 1+ }+ _ -> p{pmStatuses = Map.insert connId status p.pmStatuses}++bumpCounters :: KirokuMetrics -> (LifecycleCounters -> LifecycleCounters) -> STM ()+bumpCounters km f = modifyTVar' (kmCounters km) f++-- | Update a subscription's position to the max of its current and the new one.+recordPosition :: KirokuMetrics -> SubscriptionName -> GlobalPosition -> STM ()+recordPosition km name (GlobalPosition pos) =+ touchSub km name (\s -> s{smLastKnownPosition = max s.smLastKnownPosition pos})++-- | Apply @f@ to the named subscription's mutable entry, creating it if absent.+touchSub :: KirokuMetrics -> SubscriptionName -> (SubMutable -> SubMutable) -> STM ()+touchSub km (SubscriptionName name) f =+ modifyTVar' (kmSubs km) (Map.insertWith (\_new old -> f old) name (f emptySub))++{- | Read an atomic, immutable snapshot. Reads the collector's own @TVar@s /and/+the two store readers in a single STM transaction so the result is a coherent+point-in-time view.+-}+snapshotMetrics :: KirokuMetrics -> IO MetricsSnapshot+snapshotMetrics km = atomically $ do+ counters <- readTVar (kmCounters km)+ pool <- readTVar (kmPool km)+ subsMap <- readTVar (kmSubs km)+ GlobalPosition gpos <- kmPosition km+ nSubscribers <- kmSubscribers km+ let gauges =+ StoreGauges+ { globalPosition = gpos+ , activeSubscribers = nSubscribers+ , poolConnecting = countStatuses isConnecting pool+ , poolReady = countStatuses isReady pool+ , poolInUse = countStatuses isInUse pool+ , poolEstablishedTotal = pool.pmEstablishedTotal+ , poolTerminatedTotal = pool.pmTerminatedTotal+ }+ pure+ MetricsSnapshot+ { store = gauges+ , counters+ , subscriptions = Map.map (toSubMetrics gpos) subsMap+ }++countStatuses :: (ConnectionStatus -> Bool) -> PoolMutable -> Int+countStatuses p pool = Map.size (Map.filter p pool.pmStatuses)++isConnecting :: ConnectionStatus -> Bool+isConnecting = \case ConnectingConnectionStatus -> True; _ -> False++isReady :: ConnectionStatus -> Bool+isReady = \case ReadyForUseConnectionStatus _ -> True; _ -> False++isInUse :: ConnectionStatus -> Bool+isInUse = \case InUseConnectionStatus -> True; _ -> False++toSubMetrics :: Int64 -> SubMutable -> SubscriptionMetrics+toSubMetrics gpos s =+ SubscriptionMetrics+ { lastKnownPosition = s.smLastKnownPosition+ , lag = max 0 (gpos - s.smLastKnownPosition)+ , dbErrorCount = s.smDbErrorCount+ , lastStopReason = s.smLastStopReason+ }
+ src/Kiroku/Metrics/Config.hs view
@@ -0,0 +1,55 @@+-- | Configuration for the Kiroku metrics web server.+module Kiroku.Metrics.Config (+ MetricsServerConfig (..),+ defaultConfig,+) where++import Data.Int (Int64)+import Numeric.Natural (Natural)++{- | Configuration for the metrics web server. The @ws*@ fields are consumed by+the WebSocket endpoint (EP-3); they exist here so that plan needs no config+change. @readinessMaxLag@ is the Kiroku analogue of Marten's @maxEventLag@; it+defaults higher than Marten's 100 because Kiroku's lag is an /upper bound/ (see+the collector's lag limitation).+-}+data MetricsServerConfig = MetricsServerConfig+ { port :: !Int+ -- ^ Port to listen on (default: 9091).+ , enableJSON :: !Bool+ -- ^ Enable the JSON metrics and health endpoints (default: True).+ , enablePrometheus :: !Bool+ -- ^ Enable the Prometheus text-exposition endpoint (default: True).+ , enableWebSocket :: !Bool+ -- ^ Enable the WebSocket upgrade path (default: True; EP-3 makes it functional).+ , wsPushIntervalUs :: !Int+ -- ^ WebSocket live-push interval in microseconds (default: 1_000_000 = 1s).+ , wsMaxConnections :: !Int+ -- ^ Maximum concurrent WebSocket connections (default: 100).+ , wsEventQueueCap :: !Natural+ {- ^ Per-connection broadcast queue capacity, in batches, for the event+ stream tail (default: 256). Additive field introduced by EP-3; under+ 'Kiroku.Store.Subscription.Types.DropOldest' a slow client drops the+ oldest undelivered batches rather than stalling the publisher.+ -}+ , readinessMaxLag :: !Int64+ -- ^ A subscription lagging beyond this fails readiness (default: 10_000).+ , livenessTimeoutUs :: !Int+ -- ^ Timeout for the liveness snapshot in microseconds (default: 1_000_000 = 1s).+ }+ deriving stock (Eq, Show)++-- | Default configuration: port 9091, all endpoints enabled.+defaultConfig :: MetricsServerConfig+defaultConfig =+ MetricsServerConfig+ { port = 9091+ , enableJSON = True+ , enablePrometheus = True+ , enableWebSocket = True+ , wsPushIntervalUs = 1_000_000+ , wsMaxConnections = 100+ , wsEventQueueCap = 256+ , readinessMaxLag = 10_000+ , livenessTimeoutUs = 1_000_000+ }
+ src/Kiroku/Metrics/Health.hs view
@@ -0,0 +1,144 @@+{- | Kubernetes-style health checks derived from the metrics snapshot.++* Liveness — can we read a snapshot within a time budget? (Is the process and+ collector responsive?)+* Readiness — is the store ready to serve: no subscription overflow-stopped, no+ subscription lagging beyond 'Kiroku.Metrics.Config.readinessMaxLag', and all+ configured 'DependencyCheck's passing?+* Detailed health — the readiness verdict plus the full snapshot, for humans.++The built-in 'postgresPing' dependency check issues @SELECT 1@ through the+store's pool. Other dependency checks are arbitrary @IO 'DependencyStatus'@+actions the caller supplies.++The lag signal is an /upper bound/: the collector observes a subscription's+position only at lifecycle callback points (see the collector's limitation), so+readiness may briefly report a caught-up subscription as lagging until its next+lifecycle event.+-}+module Kiroku.Metrics.Health (+ DependencyCheck,+ DependencyStatus (..),+ LivenessStatus (..),+ ReadinessStatus (..),+ checkLiveness,+ checkReadiness,+ checkDetailedHealth,+ postgresPing,+) where++import Data.Aeson (ToJSON (..), object, (.=))+import Data.Map.Strict qualified as Map+import Data.Maybe (isJust)+import Data.Text (Text)+import Data.Text qualified as T+import GHC.Clock (getMonotonicTimeNSec)+import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import System.Timeout (timeout)++import Kiroku.Metrics.Collector (KirokuMetrics, snapshotMetrics)+import Kiroku.Metrics.Config (MetricsServerConfig (..))+import Kiroku.Metrics.Types (+ MetricsSnapshot (..),+ SubscriptionMetrics (..),+ )+import Kiroku.Store (KirokuStore (..))++-- | A dependency check is an IO action returning that dependency's status.+type DependencyCheck = IO DependencyStatus++-- | Status of one external dependency.+data DependencyStatus = DependencyStatus+ { name :: !Text+ , healthy :: !Bool+ , latencyMs :: !(Maybe Int)+ , errorMsg :: !(Maybe Text)+ }+ deriving stock (Eq, Show)++instance ToJSON DependencyStatus where+ toJSON ds =+ object+ [ "name" .= ds.name+ , "healthy" .= ds.healthy+ , "latency_ms" .= ds.latencyMs+ , "error" .= ds.errorMsg+ ]++-- | Liveness verdict.+newtype LivenessStatus = LivenessStatus {alive :: Bool}+ deriving stock (Eq, Show)++instance ToJSON LivenessStatus where+ toJSON s = object ["alive" .= s.alive]++-- | Readiness verdict and its components.+data ReadinessStatus = ReadinessStatus+ { ready :: !Bool+ , lagOk :: !Bool+ , noOverflow :: !Bool+ , dependencies :: ![DependencyStatus]+ }+ deriving stock (Eq, Show)++instance ToJSON ReadinessStatus where+ toJSON s =+ object+ [ "ready" .= s.ready+ , "lag_ok" .= s.lagOk+ , "no_overflow" .= s.noOverflow+ , "dependencies" .= s.dependencies+ ]++{- | Liveness: can we take a snapshot within 'livenessTimeoutUs'? Proves the+collector and store @TVar@s are reachable within the budget.+-}+checkLiveness :: MetricsServerConfig -> KirokuMetrics -> IO LivenessStatus+checkLiveness cfg m = do+ result <- timeout cfg.livenessTimeoutUs (snapshotMetrics m)+ pure (LivenessStatus{alive = isJust result})++{- | Readiness: no subscription overflow-stopped, none lagging beyond+'readinessMaxLag', and all dependency checks healthy.+-}+checkReadiness :: MetricsServerConfig -> KirokuMetrics -> [DependencyCheck] -> IO ReadinessStatus+checkReadiness cfg m depChecks = do+ snap <- snapshotMetrics m+ deps <- sequence depChecks+ let subs = Map.elems snap.subscriptions+ noOverflow = not (any ((== Just "overflow") . (.lastStopReason)) subs)+ lagOk = all ((<= cfg.readinessMaxLag) . (.lag)) subs+ depsOk = all (.healthy) deps+ pure+ ReadinessStatus+ { ready = noOverflow && lagOk && depsOk+ , lagOk+ , noOverflow+ , dependencies = deps+ }++-- | Detailed health: the readiness verdict plus the full snapshot.+checkDetailedHealth ::+ MetricsServerConfig ->+ KirokuMetrics ->+ [DependencyCheck] ->+ IO (ReadinessStatus, MetricsSnapshot)+checkDetailedHealth cfg m depChecks = do+ readiness <- checkReadiness cfg m depChecks+ snap <- snapshotMetrics m+ pure (readiness, snap)++{- | Built-in dependency check: time a @SELECT 1@ through the store's pool. A+@Right@ result is healthy; a @Left@ 'Pool.UsageError' is unhealthy with the+error rendered into 'errorMsg'.+-}+postgresPing :: KirokuStore -> DependencyCheck+postgresPing store = do+ t0 <- getMonotonicTimeNSec+ result <- Pool.use store.pool (Session.script "SELECT 1")+ t1 <- getMonotonicTimeNSec+ let ms = fromIntegral ((t1 - t0) `div` 1_000_000)+ pure $ case result of+ Right () -> DependencyStatus "postgres" True (Just ms) Nothing+ Left e -> DependencyStatus "postgres" False (Just ms) (Just (T.pack (show e)))
+ src/Kiroku/Metrics/JSON.hs view
@@ -0,0 +1,44 @@+{- | JSON HTTP endpoints: @GET /metrics@ (the full snapshot) and+@GET /metrics/\<name\>@ (one subscription's metrics, or 404).+-}+module Kiroku.Metrics.JSON (+ jsonApp,+ jsonResponse,+) where++import Data.Aeson (encode, object, (.=))+import Data.ByteString.Lazy qualified as LBS+import Data.Map.Strict qualified as Map+import Data.Text (Text)+import Network.HTTP.Types (Status, hContentType, status200, status404)+import Network.Wai (Application, Response, pathInfo, responseLBS)++import Kiroku.Metrics.Collector (KirokuMetrics, snapshotMetrics)+import Kiroku.Metrics.Types (MetricsSnapshot (..))++{- | WAI application for the JSON metrics endpoints. Routes @/metrics@ and+@/metrics/\<name\>@; any other path returns a 404 JSON body.+-}+jsonApp :: KirokuMetrics -> Application+jsonApp m req respond = do+ resp <- case pathInfo req of+ ["metrics"] -> do+ snap <- snapshotMetrics m+ pure (jsonResponse status200 (encode snap))+ ["metrics", name] -> do+ snap <- snapshotMetrics m+ pure $ case Map.lookup name snap.subscriptions of+ Just sm -> jsonResponse status200 (encode sm)+ Nothing ->+ jsonResponse status404 $+ encode $+ object+ [ "error" .= ("subscription not found" :: Text)+ , "subscription" .= name+ ]+ _ -> pure (jsonResponse status404 (encode (object ["error" .= ("Not found" :: Text)])))+ respond resp++-- | Build an @application/json@ response with the given status and body.+jsonResponse :: Status -> LBS.ByteString -> Response+jsonResponse status = responseLBS status [(hContentType, "application/json")]
+ src/Kiroku/Metrics/Prometheus.hs view
@@ -0,0 +1,148 @@+{- | Prometheus text-exposition endpoint (@GET /metrics/prometheus@).++Hand-rolled (no @prometheus-client@ dependency, no global registry): the+snapshot is the single source of truth and is rendered straight to the+text-exposition format that @promtool check metrics@ accepts. Metric names are a+public contract for dashboards — keep them stable once shipped (EP-4 documents+the full list).+-}+module Kiroku.Metrics.Prometheus (+ prometheusApp,+ renderPrometheus,+) where++import Data.ByteString.Builder qualified as B+import Data.ByteString.Lazy qualified as LBS+import Data.Int (Int64)+import Data.Map.Strict (Map)+import Data.Map.Strict qualified as Map+import Data.Text (Text)+import Data.Text qualified as T+import Data.Text.Encoding qualified as T+import Network.HTTP.Types (hContentType, status200)+import Network.Wai (Application, responseLBS)++import Kiroku.Metrics.Collector (KirokuMetrics, snapshotMetrics)+import Kiroku.Metrics.Types (+ LifecycleCounters (..),+ MetricsSnapshot (..),+ StoreGauges (..),+ SubscriptionMetrics (..),+ )++-- | WAI application for the Prometheus endpoint.+prometheusApp :: KirokuMetrics -> Application+prometheusApp m _req respond = do+ snap <- snapshotMetrics m+ respond $+ responseLBS+ status200+ [(hContentType, "text/plain; version=0.0.4; charset=utf-8")]+ (renderPrometheus snap)++-- | Render a snapshot as Prometheus text-exposition format.+renderPrometheus :: MetricsSnapshot -> LBS.ByteString+renderPrometheus snap = B.toLazyByteString (storeSection snap.store <> counterSection snap.counters <> subSection snap.subscriptions)++storeSection :: StoreGauges -> B.Builder+storeSection g =+ mconcat+ [ metric "kiroku_events_appended_total" "counter" "Total events appended store-wide (gap-free global position)." (g.globalPosition)+ , metricI "kiroku_active_subscribers" "gauge" "Currently registered subscribers." g.activeSubscribers+ , help "kiroku_pool_connections" "Pool connections by state."+ , typ "kiroku_pool_connections" "gauge"+ , labelledI "kiroku_pool_connections" "state" "connecting" g.poolConnecting+ , labelledI "kiroku_pool_connections" "state" "ready" g.poolReady+ , labelledI "kiroku_pool_connections" "state" "in_use" g.poolInUse+ , metric "kiroku_pool_established_total" "counter" "Pool connections established." (g.poolEstablishedTotal)+ , metric "kiroku_pool_terminated_total" "counter" "Pool connections terminated." (g.poolTerminatedTotal)+ ]++counterSection :: LifecycleCounters -> B.Builder+counterSection c =+ mconcat+ [ metric "kiroku_notifier_reconnecting_total" "counter" "Notifier reconnection attempts started." c.notifierReconnecting+ , metric "kiroku_notifier_reconnected_total" "counter" "Notifier reconnections completed." c.notifierReconnected+ , metric "kiroku_publisher_pool_errors_total" "counter" "EventPublisher read-query pool errors." c.publisherPoolErrors+ , metric "kiroku_publisher_loop_errors_total" "counter" "EventPublisher callback or broadcast loop errors." c.publisherLoopErrors+ , help "kiroku_subscription_db_errors_by_phase_total" "Subscription database errors by phase."+ , typ "kiroku_subscription_db_errors_by_phase_total" "counter"+ , labelled "kiroku_subscription_db_errors_by_phase_total" "phase" "load" c.subscriptionDbErrorsLoad+ , labelled "kiroku_subscription_db_errors_by_phase_total" "phase" "fetch" c.subscriptionDbErrorsFetch+ , labelled "kiroku_subscription_db_errors_by_phase_total" "phase" "save" c.subscriptionDbErrorsSave+ , metric "kiroku_subscriptions_started_total" "counter" "Subscription workers started." c.subscriptionsStarted+ , metric "kiroku_subscriptions_caught_up_total" "counter" "Subscriptions that reached live mode." c.subscriptionsCaughtUp+ , metric "kiroku_subscriptions_paused_total" "counter" "Subscription pauses (backpressure)." c.subscriptionsPaused+ , metric "kiroku_subscriptions_resumed_total" "counter" "Subscription resumes after pause." c.subscriptionsResumed+ , metric "kiroku_subscriptions_reconnecting_total" "counter" "Subscription live-fetch reconnects." c.subscriptionsReconnecting+ , metric "kiroku_subscriptions_retrying_total" "counter" "Subscription event redeliveries." c.subscriptionsRetrying+ , metric "kiroku_subscriptions_dead_lettered_total" "counter" "Events written to dead letters." c.subscriptionsDeadLettered+ , help "kiroku_subscriptions_stopped_total" "Subscription stops by reason."+ , typ "kiroku_subscriptions_stopped_total" "counter"+ , labelled "kiroku_subscriptions_stopped_total" "reason" "handler" c.subscriptionsStoppedHandler+ , labelled "kiroku_subscriptions_stopped_total" "reason" "cancelled" c.subscriptionsStoppedCancelled+ , labelled "kiroku_subscriptions_stopped_total" "reason" "overflow" c.subscriptionsStoppedOverflow+ , labelled "kiroku_subscriptions_stopped_total" "reason" "crashed" c.subscriptionsStoppedCrashed+ , metric "kiroku_live_fetches_total" "counter" "Live-mode database fetches." c.liveFetches+ , metric "kiroku_batches_delivered_total" "counter" "Non-empty batches delivered to handlers." c.batchesDelivered+ , metric "kiroku_events_delivered_total" "counter" "Events delivered to handlers." c.eventsDelivered+ , metric "kiroku_hard_deletes_total" "counter" "Hard-delete transactions issued." c.hardDeletesIssued+ ]++subSection :: Map Text SubscriptionMetrics -> B.Builder+subSection subs =+ mconcat+ [ help "kiroku_subscription_position" "Last-known global position per subscription."+ , typ "kiroku_subscription_position" "gauge"+ , Map.foldMapWithKey (\n sm -> labelled "kiroku_subscription_position" "subscription" n sm.lastKnownPosition) subs+ , help "kiroku_subscription_lag" "Lag behind the global position per subscription (upper bound)."+ , typ "kiroku_subscription_lag" "gauge"+ , Map.foldMapWithKey (\n sm -> labelled "kiroku_subscription_lag" "subscription" n sm.lag) subs+ , help "kiroku_subscription_db_errors_total" "Database errors per subscription."+ , typ "kiroku_subscription_db_errors_total" "counter"+ , Map.foldMapWithKey (\n sm -> labelled "kiroku_subscription_db_errors_total" "subscription" n sm.dbErrorCount) subs+ ]++-- | A single unlabelled metric: HELP, TYPE, and one sample (Int64-valued).+metric :: Text -> Text -> Text -> Int64 -> B.Builder+metric n t h v = help n h <> typ n t <> name n <> B.char8 ' ' <> B.int64Dec v <> B.char8 '\n'++-- | A single unlabelled metric with an 'Int'-valued sample.+metricI :: Text -> Text -> Text -> Int -> B.Builder+metricI n t h v = metric n t h (fromIntegral v)++help :: Text -> Text -> B.Builder+help n h = "# HELP " <> name n <> B.char8 ' ' <> text h <> B.char8 '\n'++typ :: Text -> Text -> B.Builder+typ n t = "# TYPE " <> name n <> B.char8 ' ' <> text t <> B.char8 '\n'++-- | A labelled sample line (Int64-valued).+labelled :: Text -> Text -> Text -> Int64 -> B.Builder+labelled n labelKey labelVal v =+ name n+ <> B.char8 '{'+ <> text labelKey+ <> "=\""+ <> text (escapeLabel labelVal)+ <> "\"} "+ <> B.int64Dec v+ <> B.char8 '\n'++-- | A labelled sample line (Int-valued).+labelledI :: Text -> Text -> Text -> Int -> B.Builder+labelledI n labelKey labelVal v = labelled n labelKey labelVal (fromIntegral v)++name :: Text -> B.Builder+name = text++text :: Text -> B.Builder+text = B.byteString . T.encodeUtf8++-- | Escape a Prometheus label value: backslash, double-quote, and newline.+escapeLabel :: Text -> Text+escapeLabel = T.concatMap $ \case+ '\\' -> "\\\\"+ '"' -> "\\\""+ '\n' -> "\\n"+ ch -> T.singleton ch
+ src/Kiroku/Metrics/Server.hs view
@@ -0,0 +1,222 @@+{- | The combined metrics web server.++Builds a single WAI 'Application' with 'WaiWS.websocketsOr': WebSocket upgrades+go to a 'WS.ServerApp' seam, everything else to the HTTP router. EP-2 supplies a+rejecting stub for the seam ('stubWebSocketApp'); EP-3 replaces it with the real+event-streaming app via 'startMetricsServerWith' without changing this module.++The server takes 'KirokuMetrics' plus a list of 'DependencyCheck's and a+'WS.ServerApp' — it does /not/ take the 'KirokuStore' directly. Everything+store-specific is captured in caller-built closures ('postgresPing' over the+pool, and EP-3's WebSocket app over the store), keeping the server store-agnostic.+-}+module Kiroku.Metrics.Server (+ MetricsServer (..),+ startMetricsServer,+ startMetricsServerWith,+ startMetricsServerWith',+ startMetricsServerWithStore,+ stopMetricsServer,+ withMetricsServer,+ withMetricsServerWithStore,+ withMetricsServerSubscriptions,+ combinedApp,+ httpApp,+ stubWebSocketApp,+) where++import Control.Concurrent.Async (Async, async, cancel)+import Control.Exception (bracket)+import Data.Aeson (encode, object, (.=))+import Data.Text (Text)+import Network.HTTP.Types (status200, status404, status503)+import Network.Wai (Application, pathInfo)+import Network.Wai.Handler.Warp qualified as Warp+import Network.Wai.Handler.WebSockets qualified as WaiWS+import Network.WebSockets qualified as WS++import Kiroku.Metrics.Collector (KirokuMetrics)+import Kiroku.Metrics.Config (MetricsServerConfig (..))+import Kiroku.Metrics.Health (+ DependencyCheck,+ LivenessStatus (..),+ ReadinessStatus (..),+ checkDetailedHealth,+ checkLiveness,+ checkReadiness,+ )+import Kiroku.Metrics.JSON (jsonApp, jsonResponse)+import Kiroku.Metrics.Prometheus (prometheusApp)+import Kiroku.Metrics.Subscriptions (SubscriptionStatusProvider, subscriptionsApp)+import Kiroku.Metrics.WebSocket (newWebSocketState, websocketApp)+import Kiroku.Store (KirokuStore)++-- | A running metrics server: the Warp thread and the port it bound.+data MetricsServer = MetricsServer+ { serverThread :: !(Async ())+ , serverPort :: !Int+ }++{- | Start the server with the rejecting WebSocket stub. Use this until EP-3's+real WebSocket app is wired via 'startMetricsServerWith'.+-}+startMetricsServer :: MetricsServerConfig -> KirokuMetrics -> [DependencyCheck] -> IO MetricsServer+startMetricsServer cfg m deps = startMetricsServerWith' cfg m deps Nothing stubWebSocketApp++{- | Start the server with an explicit WebSocket app (the IP-3 seam) and no+subscription-status provider. When @cfg.port == 0@ an OS-assigned free port is+used and reported in 'serverPort'.+-}+startMetricsServerWith ::+ MetricsServerConfig ->+ KirokuMetrics ->+ [DependencyCheck] ->+ WS.ServerApp ->+ IO MetricsServer+startMetricsServerWith cfg m deps = startMetricsServerWith' cfg m deps Nothing++{- | Start the server with an explicit WebSocket app (the IP-3 seam) /and/ an+optional subscription-status provider (the IP-5 seam, EP-5). The provider, when+@Just@, serves @GET /subscriptions@; when @Nothing@, that route returns a+configured-404. All EP-2/EP-3 starters delegate here with @Nothing@.+-}+startMetricsServerWith' ::+ MetricsServerConfig ->+ KirokuMetrics ->+ [DependencyCheck] ->+ Maybe SubscriptionStatusProvider ->+ WS.ServerApp ->+ IO MetricsServer+startMetricsServerWith' cfg m deps mProvider wsApp = do+ let app = combinedApp cfg m deps mProvider wsApp+ if cfg.port == 0+ then do+ (actualPort, sock) <- Warp.openFreePort+ let settings = Warp.setPort actualPort Warp.defaultSettings+ thread <- async (Warp.runSettingsSocket settings sock app)+ pure (MetricsServer thread actualPort)+ else do+ let settings = Warp.setHost "*" (Warp.setPort cfg.port Warp.defaultSettings)+ thread <- async (Warp.runSettings settings app)+ pure (MetricsServer thread cfg.port)++{- | Start the server with the real WebSocket app (EP-3), which streams live+metrics and events out of the given 'KirokuStore'. This is the recommended entry+point once event streaming is wanted: it allocates one shared connection-limiting+state (bounded by @cfg.wsMaxConnections@) and wires+'Kiroku.Metrics.WebSocket.websocketApp'. EP-2's 'startMetricsServer' (stub) is+unchanged for callers who do not want the WebSocket.+-}+startMetricsServerWithStore ::+ MetricsServerConfig ->+ KirokuMetrics ->+ KirokuStore ->+ [DependencyCheck] ->+ IO MetricsServer+startMetricsServerWithStore cfg m store deps = do+ wsState <- newWebSocketState cfg.wsMaxConnections+ startMetricsServerWith cfg m deps (websocketApp cfg m store wsState)++-- | Stop the server by cancelling its Warp thread.+stopMetricsServer :: MetricsServer -> IO ()+stopMetricsServer server = cancel server.serverThread++-- | Run an action with a running server, tearing it down afterwards.+withMetricsServer ::+ MetricsServerConfig ->+ KirokuMetrics ->+ [DependencyCheck] ->+ (MetricsServer -> IO a) ->+ IO a+withMetricsServer cfg m deps =+ bracket (startMetricsServer cfg m deps) stopMetricsServer++{- | Run an action with a running store-aware server (EP-3 WebSocket), tearing+it down afterwards.+-}+withMetricsServerWithStore ::+ MetricsServerConfig ->+ KirokuMetrics ->+ KirokuStore ->+ [DependencyCheck] ->+ (MetricsServer -> IO a) ->+ IO a+withMetricsServerWithStore cfg m store deps =+ bracket (startMetricsServerWithStore cfg m store deps) stopMetricsServer++{- | Run an action with a server that serves @GET /subscriptions@ from the given+provider (EP-5), using the rejecting WebSocket stub. The common case for a worker+that wants remote subscription introspection but not the event-streaming socket.+-}+withMetricsServerSubscriptions ::+ MetricsServerConfig ->+ KirokuMetrics ->+ [DependencyCheck] ->+ SubscriptionStatusProvider ->+ (MetricsServer -> IO a) ->+ IO a+withMetricsServerSubscriptions cfg m deps provider =+ bracket+ (startMetricsServerWith' cfg m deps (Just provider) stubWebSocketApp)+ stopMetricsServer++-- | The combined WAI app: WebSocket upgrades to @wsApp@, everything else to the HTTP router.+combinedApp ::+ MetricsServerConfig ->+ KirokuMetrics ->+ [DependencyCheck] ->+ Maybe SubscriptionStatusProvider ->+ WS.ServerApp ->+ Application+combinedApp cfg m deps mProvider wsApp =+ WaiWS.websocketsOr WS.defaultConnectionOptions wsApp (httpApp cfg m deps mProvider)++{- | The EP-2 WebSocket stub: reject the upgrade with a clear message. EP-3+replaces this with the real event-streaming app.+-}+stubWebSocketApp :: WS.ServerApp+stubWebSocketApp pending = WS.rejectRequest pending "WebSocket endpoint not yet implemented"++-- | HTTP router. Matches @/metrics/prometheus@ before @/metrics/\<name\>@.+httpApp ::+ MetricsServerConfig ->+ KirokuMetrics ->+ [DependencyCheck] ->+ Maybe SubscriptionStatusProvider ->+ Application+httpApp cfg m deps mProvider req respond =+ case pathInfo req of+ ["metrics", "prometheus"] | cfg.enablePrometheus -> prometheusApp m req respond+ ["metrics"] | cfg.enableJSON -> jsonApp m req respond+ ["metrics", _] | cfg.enableJSON -> jsonApp m req respond+ ["subscriptions"] -> subscriptionsRoute+ ["subscriptions", _] -> subscriptionsRoute+ ["health"] | cfg.enableJSON -> do+ (readiness, snap) <- checkDetailedHealth cfg m deps+ respond $+ jsonResponse+ (statusFor readiness.ready)+ (encode (object ["status" .= readiness, "metrics" .= snap]))+ ["health", "live"] | cfg.enableJSON -> do+ liveness <- checkLiveness cfg m+ respond (jsonResponse (statusFor liveness.alive) (encode liveness))+ ["health", "ready"] | cfg.enableJSON -> do+ readiness <- checkReadiness cfg m deps+ respond (jsonResponse (statusFor readiness.ready) (encode readiness))+ ["ws"]+ | cfg.enableWebSocket ->+ respond $+ jsonResponse+ status404+ (encode (object ["error" .= ("WebSocket endpoint - use ws:// protocol" :: Text)]))+ _ ->+ respond (jsonResponse status404 (encode (object ["error" .= ("Not found" :: Text)])))+ where+ statusFor ok = if ok then status200 else status503+ subscriptionsRoute = case mProvider of+ Just provider -> subscriptionsApp provider req respond+ Nothing ->+ respond $+ jsonResponse+ status404+ (encode (object ["error" .= ("subscription status not configured" :: Text)]))
+ src/Kiroku/Metrics/Subscriptions.hs view
@@ -0,0 +1,61 @@+{- | The @GET /subscriptions@ HTTP endpoint (EP-5 / IP-5).++Serves a worker's __live subscription registry__ as JSON — each running+subscription's name, consumer-group member, current FSM phase, and current global+cursor. The data comes from a caller-supplied 'SubscriptionStatusProvider' closure+(the server stays store-agnostic, exactly as EP-2 keeps 'KirokuStore' out of its+signature). The canonical provider, 'storeSubscriptionStatus', reads the live+registry through the public 'Kiroku.Store.Subscription.subscriptionStates'.++The wire shape is the CLI's 'SubscriptionStatusRow' JSON (a JSON array of+@{subscription, member, phase, global_position}@), reused via a @build-depends@ on+@kiroku-cli@ so the server encoder and the CLI remote-client decoder share one+codec — the IP-5 contract.+-}+module Kiroku.Metrics.Subscriptions (+ SubscriptionStatusProvider,+ storeSubscriptionStatus,+ subscriptionsApp,+) where++import Data.Aeson (encode, object, (.=))+import Data.Text (Text)+import Network.HTTP.Types (status200, status404)+import Network.Wai (Application, pathInfo)++import Kiroku.Cli.Subscription.Status (SubscriptionStatusRow (..), subscriptionStatusRows)+import Kiroku.Metrics.JSON (jsonResponse)+import Kiroku.Store (KirokuStore)+import Kiroku.Store.Subscription (subscriptionStates)++{- | A closure the caller supplies; it reads the live subscription registry on+demand and returns the rows to serve.+-}+type SubscriptionStatusProvider = IO [SubscriptionStatusRow]++{- | The canonical provider, built by a caller who owns the 'KirokuStore':+reads the live registry through 'subscriptionStates' and maps it to rows.+-}+storeSubscriptionStatus :: KirokuStore -> SubscriptionStatusProvider+storeSubscriptionStatus store = subscriptionStatusRows <$> subscriptionStates store++{- | WAI app for the subscription-status endpoint. Routes:++ * @GET /subscriptions@ — 200, JSON array of all rows.+ * @GET /subscriptions/\<name\>@ — 200, JSON array of rows for that name+ (possibly empty).++Any other path returns a 404 with body @{"error":"Not found"}@. This app is mounted by+'Kiroku.Metrics.Server.httpApp' only when a provider is configured.+-}+subscriptionsApp :: SubscriptionStatusProvider -> Application+subscriptionsApp provider req respond =+ case pathInfo req of+ ["subscriptions"] -> do+ rows <- provider+ respond (jsonResponse status200 (encode rows))+ ["subscriptions", name] -> do+ rows <- provider+ let filtered = filter (\row -> row.subscription == name) rows+ respond (jsonResponse status200 (encode filtered))+ _ -> respond (jsonResponse status404 (encode (object ["error" .= ("Not found" :: Text)])))
+ src/Kiroku/Metrics/Types.hs view
@@ -0,0 +1,159 @@+{- | Immutable, JSON-encodable snapshot of a running Kiroku store's operational+metrics.++A 'MetricsSnapshot' is produced by+'Kiroku.Metrics.Collector.snapshotMetrics' and is the single source of truth+every endpoint renders. It carries three parts:++* 'StoreGauges' — point-in-time gauges read from the live store handle at+ snapshot time: the gap-free global position (which doubles as the total+ number of events appended store-wide and the high-water mark), the active+ subscriber count, and pool connection gauges/counters.+* 'LifecycleCounters' — monotonic counters accumulated from the store's+ 'Kiroku.Store.Observability.KirokuEvent' callback stream.+* a per-subscription map ('SubscriptionMetrics') keyed by subscription name,+ carrying each subscription's last-known position, derived lag, database-error+ count, and last stop reason.++The 'ToJSON' instances are written by hand (not derived) so the wire shape is+stable and documented for downstream consumers (EP-2's JSON/Prometheus+renderers and EP-4's user guide).+-}+module Kiroku.Metrics.Types (+ MetricsSnapshot (..),+ StoreGauges (..),+ LifecycleCounters (..),+ SubscriptionMetrics (..),+) where++import Data.Aeson (ToJSON (..), object, (.=))+import Data.Int (Int64)+import Data.Map.Strict (Map)+import Data.Text (Text)++-- | A coherent point-in-time view of the store's metrics.+data MetricsSnapshot = MetricsSnapshot+ { store :: !StoreGauges+ , counters :: !LifecycleCounters+ , subscriptions :: !(Map Text SubscriptionMetrics)+ -- ^ Keyed by 'Kiroku.Store.Subscription.Types.SubscriptionName' text.+ }+ deriving stock (Eq, Show)++-- | Gauges read from the live store handle at snapshot time.+data StoreGauges = StoreGauges+ { globalPosition :: !Int64+ -- ^ Total events appended store-wide (gap-free) == high-water mark.+ , activeSubscribers :: !Int+ , poolConnecting :: !Int+ , poolReady :: !Int+ , poolInUse :: !Int+ , poolEstablishedTotal :: !Int64+ , poolTerminatedTotal :: !Int64+ }+ deriving stock (Eq, Show)++{- | Monotonic counters accumulated from the 'KirokuEvent' callback stream.++The constructor set the store emits is richer than counters here name one-to-one:+subscription lifecycle events ('Started', 'CaughtUp', 'Paused', 'Resumed',+'Reconnecting', 'Retrying', 'DeadLettered', 'Stopped' by reason), per-batch+delivery, live fetches, per-phase database errors, notifier reconnects,+publisher pool and loop errors, and hard deletes.+-}+data LifecycleCounters = LifecycleCounters+ { notifierReconnecting :: !Int64+ , notifierReconnected :: !Int64+ , publisherPoolErrors :: !Int64+ , publisherLoopErrors :: !Int64+ , subscriptionDbErrorsLoad :: !Int64+ , subscriptionDbErrorsFetch :: !Int64+ , subscriptionDbErrorsSave :: !Int64+ , subscriptionsStarted :: !Int64+ , subscriptionsCaughtUp :: !Int64+ , subscriptionsPaused :: !Int64+ , subscriptionsResumed :: !Int64+ , subscriptionsReconnecting :: !Int64+ , subscriptionsRetrying :: !Int64+ , subscriptionsDeadLettered :: !Int64+ , subscriptionsStoppedHandler :: !Int64+ , subscriptionsStoppedCancelled :: !Int64+ , subscriptionsStoppedOverflow :: !Int64+ , subscriptionsStoppedCrashed :: !Int64+ , liveFetches :: !Int64+ -- ^ Count of live-mode DB fetches ('KirokuEventSubscriptionFetched').+ , batchesDelivered :: !Int64+ -- ^ Count of non-empty batches delivered ('KirokuEventSubscriptionDelivered').+ , eventsDelivered :: !Int64+ -- ^ Sum of batch row counts across all deliveries.+ , hardDeletesIssued :: !Int64+ }+ deriving stock (Eq, Show)++-- | Per-subscription accumulated metrics.+data SubscriptionMetrics = SubscriptionMetrics+ { lastKnownPosition :: !Int64+ -- ^ Most recent position seen at a lifecycle event (lower bound on true position).+ , lag :: !Int64+ -- ^ @max 0 (globalPosition - lastKnownPosition)@ (upper bound on true lag).+ , dbErrorCount :: !Int64+ , lastStopReason :: !(Maybe Text)+ -- ^ @"handler" | "cancelled" | "overflow" | "crashed"@.+ }+ deriving stock (Eq, Show)++instance ToJSON MetricsSnapshot where+ toJSON s =+ object+ [ "store" .= s.store+ , "counters" .= s.counters+ , "subscriptions" .= s.subscriptions+ ]++instance ToJSON StoreGauges where+ toJSON g =+ object+ [ "global_position" .= g.globalPosition+ , "active_subscribers" .= g.activeSubscribers+ , "pool_connecting" .= g.poolConnecting+ , "pool_ready" .= g.poolReady+ , "pool_in_use" .= g.poolInUse+ , "pool_established_total" .= g.poolEstablishedTotal+ , "pool_terminated_total" .= g.poolTerminatedTotal+ ]++instance ToJSON LifecycleCounters where+ toJSON c =+ object+ [ "notifier_reconnecting" .= c.notifierReconnecting+ , "notifier_reconnected" .= c.notifierReconnected+ , "publisher_pool_errors" .= c.publisherPoolErrors+ , "publisher_loop_errors" .= c.publisherLoopErrors+ , "subscription_db_errors_load" .= c.subscriptionDbErrorsLoad+ , "subscription_db_errors_fetch" .= c.subscriptionDbErrorsFetch+ , "subscription_db_errors_save" .= c.subscriptionDbErrorsSave+ , "subscriptions_started" .= c.subscriptionsStarted+ , "subscriptions_caught_up" .= c.subscriptionsCaughtUp+ , "subscriptions_paused" .= c.subscriptionsPaused+ , "subscriptions_resumed" .= c.subscriptionsResumed+ , "subscriptions_reconnecting" .= c.subscriptionsReconnecting+ , "subscriptions_retrying" .= c.subscriptionsRetrying+ , "subscriptions_dead_lettered" .= c.subscriptionsDeadLettered+ , "subscriptions_stopped_handler" .= c.subscriptionsStoppedHandler+ , "subscriptions_stopped_cancelled" .= c.subscriptionsStoppedCancelled+ , "subscriptions_stopped_overflow" .= c.subscriptionsStoppedOverflow+ , "subscriptions_stopped_crashed" .= c.subscriptionsStoppedCrashed+ , "live_fetches" .= c.liveFetches+ , "batches_delivered" .= c.batchesDelivered+ , "events_delivered" .= c.eventsDelivered+ , "hard_deletes_issued" .= c.hardDeletesIssued+ ]++instance ToJSON SubscriptionMetrics where+ toJSON m =+ object+ [ "last_known_position" .= m.lastKnownPosition+ , "lag" .= m.lag+ , "db_error_count" .= m.dbErrorCount+ , "last_stop_reason" .= m.lastStopReason+ ]
+ src/Kiroku/Metrics/WebSocket.hs view
@@ -0,0 +1,479 @@+{-# LANGUAGE ScopedTypeVariables #-}++{- | The WebSocket endpoint for the Kiroku metrics server.++Fills the IP-3 seam EP-2 left stubbed ('Kiroku.Metrics.Server.stubWebSocketApp')+with a real 'WS.ServerApp' that dispatches on the request path:++ * @\/ws\/metrics@ — the /metrics channel/: a 'MetricsSnapshot' on connect,+ then a fresh snapshot every @wsPushIntervalUs@ microseconds; @ping@ → @pong@.+ * @\/ws\/events@ — the /event channel/: after a @subscribe_events@ message, a+ JSON message per appended 'RecordedEvent' in global-position order, live.+ Optionally replays history from a chosen @from_position@ and/or restricts to+ a single @category@.++The event tail is built on the /public/ 'EventPublisher' broadcast+('subscribePublisher') for live "from-now" delivery and the public effectful+reads ('readAllForward' / 'readCategory' via 'runStoreIO') for replay and+category filtering. It creates /no persistent subscription/ and writes nothing to+the @subscriptions@ checkpoint table — transient watchers leave no trace. See the+plan's Decision Log.++This module owns IP-4: the 'ClientMessage' / 'ServerMessage' protocol and the+explicit 'recordedEventToJSON' encoder (an orphan-free function, not a @ToJSON@+instance — 'RecordedEvent' has none and @kiroku-store@ keeps its @Types@ module+instance-light).+-}+module Kiroku.Metrics.WebSocket (+ -- * Protocol+ ClientMessage (..),+ ServerMessage (..),+ recordedEventToJSON,++ -- * Connection limiting+ WebSocketState (..),+ newWebSocketState,++ -- * The WebSocket application (the IP-3 seam)+ websocketApp,+) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.Async (async, cancel, link)+import Control.Concurrent.STM (+ STM,+ TBQueue,+ TVar,+ atomically,+ check,+ modifyTVar',+ newTVarIO,+ readTBQueue,+ readTVar,+ writeTVar,+ )+import Control.Exception (catch, finally)+import Control.Monad (forever)+import Data.Aeson (+ FromJSON (..),+ ToJSON (..),+ Value,+ eitherDecode',+ encode,+ object,+ withObject,+ (.:),+ (.:?),+ (.=),+ )+import Data.ByteString qualified as BS+import Data.ByteString.Lazy qualified as LBS+import Data.Foldable (for_)+import Data.Int (Int64)+import Data.Text (Text)+import Data.Text qualified as T+import Data.Vector (Vector)+import Data.Vector qualified as V+import Network.WebSockets qualified as WS++import Kiroku.Metrics.Collector (KirokuMetrics, snapshotMetrics)+import Kiroku.Metrics.Config (MetricsServerConfig (..))+import Kiroku.Metrics.Types (MetricsSnapshot)+import Kiroku.Store (+ CategoryName (..),+ GlobalPosition (..),+ KirokuStore (..),+ RecordedEvent (..),+ readAllForward,+ readCategory,+ runStoreIO,+ )+import Kiroku.Store.Subscription.EventPublisher (+ SubscriberStatus (..),+ publisherPosition,+ subscribePublisher,+ )+import Kiroku.Store.Subscription.Types (OverflowPolicy (..))+import Kiroku.Store.Types (+ EventId (..),+ EventType (..),+ StreamId (..),+ StreamVersion (..),+ )++--------------------------------------------------------------------------------+-- Protocol (IP-4)+--------------------------------------------------------------------------------++{- | Messages from a WebSocket client to the server (tagged on a @"type"@ field).++The constructors are field-less (positional) on purpose: 'SubscribeEvents' and+'ServerMessage'\'s 'EventStreamStarted' would otherwise both define a+@fromPosition@ record selector, which collides when @Kiroku.Metrics@ re-exports+both @(..)@ lists.+-}+data ClientMessage+ = -- | Keepalive request; answered with 'Pong'.+ Ping+ | -- | (Metrics channel) request a fresh snapshot now.+ SubscribeMetrics+ | {- | (Event channel) start streaming. The first field is @from_position@+ ('Nothing' = "from now"); the second is @category@ ('Nothing' = all streams).+ -}+ SubscribeEvents !(Maybe Int64) !(Maybe Text)+ | -- | (Event channel) stop the current tail.+ UnsubscribeEvents+ deriving stock (Eq, Show)++-- | Messages from the server to a WebSocket client (tagged on a @"type"@ field).+data ServerMessage+ = -- | Answer to 'Ping'.+ Pong+ | -- | (Metrics channel) a metrics snapshot, embedded under @"metrics"@.+ Snapshot !MetricsSnapshot+ | -- | (Event channel) one appended event, embedded under @"event"@.+ Event !Value+ | {- | (Event channel) acknowledgement that streaming has begun from a+ given global position (the @from_position@ field on the wire).+ -}+ EventStreamStarted !Int64+ | -- | The connection is being torn down.+ Goodbye+ | -- | A non-fatal error message for the client.+ ErrorMsg !Text+ deriving stock (Eq, Show)++instance FromJSON ClientMessage where+ parseJSON = withObject "ClientMessage" $ \v -> do+ msgType <- v .: "type"+ case msgType :: Text of+ "ping" -> pure Ping+ "subscribe_metrics" -> pure SubscribeMetrics+ "subscribe_events" ->+ SubscribeEvents <$> v .:? "from_position" <*> v .:? "category"+ "unsubscribe_events" -> pure UnsubscribeEvents+ other -> fail ("Unknown client message type: " <> T.unpack other)++instance ToJSON ServerMessage where+ toJSON Pong = object ["type" .= ("pong" :: Text)]+ toJSON (Snapshot snap) = object ["type" .= ("snapshot" :: Text), "metrics" .= snap]+ toJSON (Event ev) = object ["type" .= ("event" :: Text), "event" .= ev]+ toJSON (EventStreamStarted p) =+ object ["type" .= ("event_stream_started" :: Text), "from_position" .= p]+ toJSON Goodbye = object ["type" .= ("goodbye" :: Text)]+ toJSON (ErrorMsg msg) = object ["type" .= ("error" :: Text), "message" .= msg]++{- | Encode a 'RecordedEvent' to a JSON 'Value' (IP-4). An explicit function+rather than a @ToJSON@ instance: 'RecordedEvent' has no instance today and a+library-level orphan is undesirable. EP-4's user guide documents this shape.+-}+recordedEventToJSON :: RecordedEvent -> Value+recordedEventToJSON e =+ object+ [ "eventId" .= unEventId e.eventId+ , "eventType" .= unEventType e.eventType+ , "streamVersion" .= unStreamVersion e.streamVersion+ , "globalPosition" .= unGlobalPosition e.globalPosition+ , "originalStreamId" .= unStreamId e.originalStreamId+ , "originalVersion" .= unStreamVersion e.originalVersion+ , "payload" .= e.payload+ , "metadata" .= e.metadata+ , "causationId" .= e.causationId+ , "correlationId" .= e.correlationId+ , "createdAt" .= e.createdAt+ ]+ where+ unEventId (EventId u) = u+ unEventType (EventType t) = t+ unStreamVersion (StreamVersion n) = n+ unGlobalPosition (GlobalPosition n) = n+ unStreamId (StreamId n) = n++--------------------------------------------------------------------------------+-- Connection limiting (mirrors shibuya-metrics)+--------------------------------------------------------------------------------++{- | Shared state bounding the number of concurrent WebSocket connections.+Allocated once per server in 'Kiroku.Metrics.Server.startMetricsServerWithStore'+and captured by 'websocketApp', so the bound is shared across connections.+-}+data WebSocketState = WebSocketState+ { connectionCount :: !(TVar Int)+ , maxConnections :: !Int+ }++-- | Create a 'WebSocketState' bounding connections at @maxConns@.+newWebSocketState :: Int -> IO WebSocketState+newWebSocketState maxConns = do+ countVar <- newTVarIO 0+ pure WebSocketState{connectionCount = countVar, maxConnections = maxConns}++-- | Try to claim a connection slot; 'True' on success.+acquireConnection :: WebSocketState -> STM Bool+acquireConnection st = do+ count <- readTVar st.connectionCount+ if count < st.maxConnections+ then writeTVar st.connectionCount (count + 1) >> pure True+ else pure False++-- | Release a connection slot.+releaseConnection :: WebSocketState -> STM ()+releaseConnection st = modifyTVar' st.connectionCount (\c -> max 0 (c - 1))++--------------------------------------------------------------------------------+-- Path dispatch+--------------------------------------------------------------------------------++data WsPath = MetricsPath | EventsPath | UnknownPath++-- | Dispatch on the request path, ignoring any query string.+dispatchPath :: BS.ByteString -> WsPath+dispatchPath raw =+ case BS.takeWhile (/= 0x3f) raw of -- 0x3f = '?'+ p+ | p == "/ws/metrics" -> MetricsPath+ | p == "/ws/events" -> EventsPath+ | otherwise -> UnknownPath++--------------------------------------------------------------------------------+-- The WebSocket application+--------------------------------------------------------------------------------++{- | The real WebSocket app filling the IP-3 seam. Closes over the config, the+collector, the store (for event streaming — EP-2 deliberately kept the store out+of the /server/ signature, so it enters here), and the shared connection-limiting+state. Each upgrade is dispatched by path; an over-capacity upgrade is rejected.+-}+websocketApp ::+ MetricsServerConfig ->+ KirokuMetrics ->+ KirokuStore ->+ WebSocketState ->+ WS.ServerApp+websocketApp cfg m store st pending =+ case dispatchPath (WS.requestPath (WS.pendingRequest pending)) of+ MetricsPath -> guarded (handleMetrics cfg m)+ EventsPath -> guarded (handleEvents cfg store)+ UnknownPath ->+ WS.rejectRequest pending "Unknown WebSocket path; use /ws/metrics or /ws/events"+ where+ guarded run = do+ acquired <- atomically (acquireConnection st)+ if not acquired+ then WS.rejectRequest pending "Too many connections"+ else+ (run pending `catch` ignoreClosed)+ `finally` atomically (releaseConnection st)+ -- A normal client disconnect surfaces as a 'WS.ConnectionException' from the+ -- receive loop; treat it as a clean end-of-connection rather than letting it+ -- escape to the server's exception reporter.+ ignoreClosed (_ :: WS.ConnectionException) = pure ()++--------------------------------------------------------------------------------+-- Metrics channel+--------------------------------------------------------------------------------++-- | Handle a @/ws/metrics@ connection: snapshot on connect, periodic push, ping/pong.+handleMetrics :: MetricsServerConfig -> KirokuMetrics -> WS.PendingConnection -> IO ()+handleMetrics cfg m pending = do+ conn <- WS.acceptRequest pending+ WS.withPingThread conn 30 (pure ()) $ do+ sendMsg conn . Snapshot =<< snapshotMetrics m+ pushThread <- async (metricsPushLoop cfg m conn)+ link pushThread+ finally+ (metricsReceiveLoop m conn)+ (cancel pushThread >> sendMsg conn Goodbye)++-- | Periodically push a fresh snapshot every @wsPushIntervalUs@.+metricsPushLoop :: MetricsServerConfig -> KirokuMetrics -> WS.Connection -> IO ()+metricsPushLoop cfg m conn = forever $ do+ threadDelay cfg.wsPushIntervalUs+ sendMsg conn . Snapshot =<< snapshotMetrics m++-- | Answer @ping@ with @pong@ and @subscribe_metrics@ with a fresh snapshot.+metricsReceiveLoop :: KirokuMetrics -> WS.Connection -> IO ()+metricsReceiveLoop m conn = forever $ do+ cmd <- recvMsg conn+ case cmd of+ Just Ping -> sendMsg conn Pong+ Just SubscribeMetrics -> sendMsg conn . Snapshot =<< snapshotMetrics m+ _ -> pure ()++--------------------------------------------------------------------------------+-- Event channel+--------------------------------------------------------------------------------++{- | Handle a @/ws/events@ connection. The connection starts idle; a+@subscribe_events@ message starts (or restarts) a tail, @unsubscribe_events@+stops it, and @ping@ → @pong@. The tail runs in a child thread tracked in a+'TVar' so the receive loop can cancel and replace it; disconnect tears it down.+-}+handleEvents :: MetricsServerConfig -> KirokuStore -> WS.PendingConnection -> IO ()+handleEvents cfg store pending = do+ conn <- WS.acceptRequest pending+ WS.withPingThread conn 30 (pure ()) $ do+ tailVar <- newTVarIO Nothing+ let stopTail = do+ mt <- atomically (readTVar tailVar)+ for_ mt cancel+ atomically (writeTVar tailVar Nothing)+ startTail from cat = do+ stopTail+ t <- async (eventTail cfg store conn from cat)+ atomically (writeTVar tailVar (Just t))+ finally+ ( forever $ do+ cmd <- recvMsg conn+ case cmd of+ Just Ping -> sendMsg conn Pong+ Just (SubscribeEvents from cat) -> startTail from cat+ Just UnsubscribeEvents -> stopTail+ _ -> pure ()+ )+ (stopTail >> sendMsg conn Goodbye)++-- | A reasonable replay/category page size.+eventReadLimit :: Int+eventReadLimit = 500++{- | Stream events to the client. Dispatches on the request shape:++ * no @from@, no @category@: live "from-now" tail via the broadcast.+ * @from@, no @category@: replay history from @from@, tracking the highest+ delivered position, then live with the broadcast filtered to positions above+ that covered boundary. A replay read error sends an @error@ frame and ends+ the tail instead of entering live mode with a gap.+ * any @category@: a DB-driven loop over 'readCategory' gated on the publisher+ position (the broadcast carries no stream names, so it cannot be filtered by+ category in-process — see the plan).+-}+eventTail ::+ MetricsServerConfig ->+ KirokuStore ->+ WS.Connection ->+ Maybe Int64 ->+ Maybe Text ->+ IO ()+eventTail cfg store conn mFrom mCategory =+ case mCategory of+ Just cat -> do+ start <- case mFrom of+ Just p -> pure p+ Nothing -> unGP <$> atomically (publisherPosition store.publisher)+ sendMsg conn (EventStreamStarted start)+ categoryLoop store conn (CategoryName cat) start+ Nothing -> do+ (queue, statusVar, unsubscribe) <-+ atomically (subscribePublisher store.publisher cfg.wsEventQueueCap DropOldest)+ attachPos <- unGP <$> atomically (publisherPosition store.publisher)+ flip finally unsubscribe $+ case mFrom of+ Nothing -> do+ sendMsg conn (EventStreamStarted attachPos)+ broadcastLoop conn queue statusVar (const True)+ Just p -> do+ sendMsg conn (EventStreamStarted p)+ mCovered <- replayHistory store conn p attachPos+ for_ mCovered $ \covered ->+ broadcastLoop conn queue statusVar (\e -> unGP e.globalPosition > covered)++{- | Page history from the requested position up to @attachPos@ with+'readAllForward'. Returns @Just covered@, the highest global position the+client is now guaranteed to have received (at least @attachPos@; more when the+final page read past it), or @Nothing@ after a read error, which has already+been surfaced to the client as an 'ErrorMsg'. The caller must terminate the+tail on @Nothing@ rather than continue live with a gap.+-}+replayHistory :: KirokuStore -> WS.Connection -> Int64 -> Int64 -> IO (Maybe Int64)+replayHistory store conn from attachPos = go from attachPos+ where+ go cursor covered+ | cursor >= attachPos = pure (Just covered)+ | otherwise = do+ res <- runStoreIO store (readAllForward (GlobalPosition cursor) (fromIntegral eventReadLimit))+ case res of+ Left err -> do+ sendMsg conn (ErrorMsg (T.pack ("replay error: " <> show err)))+ pure Nothing+ Right evs+ | V.null evs -> pure (Just covered)+ | otherwise -> do+ sendEvents conn evs+ let lastPos = unGP (V.last evs).globalPosition+ go lastPos (max covered lastPos)++{- | Drain the broadcast queue forever, sending each kept event. Defensively+surfaces an @Overflowed@ status (not set under 'DropOldest', but handled).+-}+broadcastLoop ::+ WS.Connection ->+ -- | broadcast queue+ TBQueue (Vector RecordedEvent) ->+ TVar SubscriberStatus ->+ (RecordedEvent -> Bool) ->+ IO ()+broadcastLoop conn queue statusVar keep = forever $ do+ batch <- atomically (readTBQueue queue)+ sendEvents conn (V.filter keep batch)+ status <- atomically (readTVar statusVar)+ case status of+ Overflowed -> sendMsg conn (ErrorMsg "event stream overflowed; some events dropped")+ _ -> pure ()++{- | DB-driven category live loop. Mirrors the subscription worker's+@liveLoopDbDriven@: gate on the publisher advancing past the /last observed/+position (not the cursor) so an unmatched category does not busy-spin, then drain+the category to empty before waiting again.+-}+categoryLoop :: KirokuStore -> WS.Connection -> CategoryName -> Int64 -> IO ()+categoryLoop store conn cat startPos = go startPos 0+ where+ go cursor waitFrom = do+ pubPos <- atomically $ do+ GlobalPosition p <- publisherPosition store.publisher+ check (p > waitFrom)+ pure p+ drained <- drainTo cursor+ case drained of+ Nothing -> pure () -- a DB error already surfaced; stop the tail+ Just cursor' -> go cursor' pubPos+ drainTo cursor = do+ res <- runStoreIO store (readCategory cat (GlobalPosition cursor) (fromIntegral eventReadLimit))+ case res of+ Left err -> do+ sendMsg conn (ErrorMsg (T.pack ("category read error: " <> show err)))+ pure Nothing+ Right evs+ | V.null evs -> pure (Just cursor)+ | otherwise -> do+ sendEvents conn evs+ drainTo (unGP (V.last evs).globalPosition)++--------------------------------------------------------------------------------+-- Send / receive helpers+--------------------------------------------------------------------------------++-- | Send each event in a batch as an 'Event' message.+sendEvents :: WS.Connection -> Vector RecordedEvent -> IO ()+sendEvents conn = V.mapM_ (sendMsg conn . Event . recordedEventToJSON)++{- | Send a 'ServerMessage', swallowing a closed-connection exception so cleanup+in a @finally@ never re-throws on an already-dead socket.+-}+sendMsg :: WS.Connection -> ServerMessage -> IO ()+sendMsg conn msg =+ WS.sendTextData conn (encode msg)+ `catch` \(_ :: WS.ConnectionException) -> pure ()++{- | Receive and decode one 'ClientMessage'. 'Nothing' on an undecodable frame+(ignored by the caller).+-}+recvMsg :: WS.Connection -> IO (Maybe ClientMessage)+recvMsg conn = do+ raw <- WS.receiveData conn :: IO LBS.ByteString+ pure (either (const Nothing) Just (eitherDecode' raw))++-- | Unwrap a 'GlobalPosition' to its underlying 'Int64'.+unGP :: GlobalPosition -> Int64+unGP (GlobalPosition n) = n
+ test/Main.hs view
@@ -0,0 +1,18 @@+module Main (main) where++import Test.Hspec (hspec)++import Kiroku.Test.Postgres (withSharedMigratedPostgres)+import Test.CollectorSpec qualified as CollectorSpec+import Test.IntegrationSpec qualified as IntegrationSpec+import Test.ServerSpec qualified as ServerSpec+import Test.SubscriptionsSpec qualified as SubscriptionsSpec+import Test.WebSocketSpec qualified as WebSocketSpec++main :: IO ()+main = withSharedMigratedPostgres $ hspec $ do+ CollectorSpec.spec+ IntegrationSpec.spec+ ServerSpec.spec+ WebSocketSpec.spec+ SubscriptionsSpec.spec
+ test/Test/CollectorSpec.hs view
@@ -0,0 +1,197 @@+{- | Pure unit tests for the metrics collector: feed scripted 'KirokuEvent' and+'Observation' values through the callback wrappers and assert the resulting+snapshot. No store is opened — the two store-level gauges are supplied by fake+STM readers via 'newKirokuMetricsWith'.+-}+module Test.CollectorSpec (spec) where++import Control.Concurrent.STM (STM)+import Control.Exception (SomeException, toException)+import Control.Monad (forM_)+import Data.Map.Strict qualified as Map+import Data.UUID (UUID, fromWords)+import Hasql.Pool (UsageError (..))+import Test.Hspec++import Kiroku.Metrics (+ LifecycleCounters (..),+ MetricsSnapshot (..),+ StoreGauges (..),+ SubscriptionMetrics (..),+ metricsEventHandler,+ metricsObservationHandler,+ newKirokuMetricsWith,+ snapshotMetrics,+ )+import Kiroku.Store (+ ConnectionReadyForUseReason (..),+ ConnectionStatus (..),+ ConnectionTerminationReason (..),+ KirokuEvent (..),+ Observation (..),+ SubscriptionDbPhase (..),+ SubscriptionGroupContext (..),+ SubscriptionStopReason (..),+ )+import Kiroku.Store.Observability (SubscriptionDeliveryPhase (..))+import Kiroku.Store.Subscription.Types (SubscriptionName (..))+import Kiroku.Store.Types (GlobalPosition (..))++{- | A collector whose fake store readers report the given global position and+active-subscriber count, fed the given scripted events and observations.+-}+runScript ::+ STM GlobalPosition ->+ STM Int ->+ [KirokuEvent] ->+ [Observation] ->+ IO MetricsSnapshot+runScript readPos readSubs events obs = do+ km <- newKirokuMetricsWith readPos readSubs+ forM_ events (metricsEventHandler km Nothing)+ forM_ obs (metricsObservationHandler km Nothing)+ snapshotMetrics km++sub :: SubscriptionName+sub = SubscriptionName "p"++uuidN :: Word -> UUID+uuidN n = fromWords 0 0 0 (fromIntegral n)++-- | A dummy exception for the strict 'SomeException' field of notifier events.+someExc :: SomeException+someExc = toException (userError "test")++-- | A dummy 'UsageError' for the strict field of database-error events.+dbErr :: UsageError+dbErr = AcquisitionTimeoutUsageError++spec :: Spec+spec = describe "Kiroku.Metrics.Collector" $ do+ it "counts notifier reconnect events" $ do+ snap <-+ runScript+ (pure (GlobalPosition 0))+ (pure 0)+ [ KirokuEventNotifierReconnecting 1 someExc+ , KirokuEventNotifierReconnecting 2 someExc+ , KirokuEventNotifierReconnected+ ]+ []+ snap.counters.notifierReconnecting `shouldBe` 2+ snap.counters.notifierReconnected `shouldBe` 1++ it "counts publisher pool and loop errors separately" $ do+ snap <-+ runScript+ (pure (GlobalPosition 0))+ (pure 0)+ [ KirokuEventPublisherPoolError dbErr+ , KirokuEventPublisherLoopError someExc+ , KirokuEventPublisherLoopError someExc+ ]+ []+ snap.counters.publisherPoolErrors `shouldBe` 1+ snap.counters.publisherLoopErrors `shouldBe` 2++ it "records subscription position and derives lag from the global position" $ do+ snap <-+ runScript+ (pure (GlobalPosition 12))+ (pure 3)+ [KirokuEventSubscriptionStarted sub (GlobalPosition 5) NonGroup]+ []+ snap.store.globalPosition `shouldBe` 12+ snap.store.activeSubscribers `shouldBe` 3+ snap.counters.subscriptionsStarted `shouldBe` 1+ case Map.lookup "p" snap.subscriptions of+ Nothing -> expectationFailure "expected subscription \"p\" in snapshot"+ Just m -> do+ m.lastKnownPosition `shouldBe` 5+ m.lag `shouldBe` 7++ it "advances last-known position monotonically and never reports negative lag" $ do+ snap <-+ runScript+ (pure (GlobalPosition 8))+ (pure 1)+ [ KirokuEventSubscriptionStarted sub (GlobalPosition 5) NonGroup+ , KirokuEventSubscriptionCaughtUp sub (GlobalPosition 10) NonGroup+ ]+ []+ case Map.lookup "p" snap.subscriptions of+ Nothing -> expectationFailure "expected subscription \"p\""+ Just m -> do+ m.lastKnownPosition `shouldBe` 10+ m.lag `shouldBe` 0 -- max 0 (8 - 10)+ it "tallies stop reasons and records the last one per subscription" $ do+ snap <-+ runScript+ (pure (GlobalPosition 20))+ (pure 0)+ [ KirokuEventSubscriptionStarted sub (GlobalPosition 0) NonGroup+ , KirokuEventSubscriptionStopped sub (GlobalPosition 20) StopHandlerRequested NonGroup+ ]+ []+ snap.counters.subscriptionsStoppedHandler `shouldBe` 1+ (Map.lookup "p" snap.subscriptions >>= (.lastStopReason)) `shouldBe` Just "handler"++ it "tallies per-phase database errors and the per-subscription error count" $ do+ snap <-+ runScript+ (pure (GlobalPosition 0))+ (pure 0)+ [ KirokuEventSubscriptionDbError sub LoadCheckpoint dbErr NonGroup+ , KirokuEventSubscriptionDbError sub FetchBatch dbErr NonGroup+ , KirokuEventSubscriptionDbError sub FetchBatch dbErr NonGroup+ ]+ []+ snap.counters.subscriptionDbErrorsLoad `shouldBe` 1+ snap.counters.subscriptionDbErrorsFetch `shouldBe` 2+ (Map.lookup "p" snap.subscriptions >>= Just . (.dbErrorCount)) `shouldBe` Just 3++ it "counts delivered batches and sums delivered events" $ do+ snap <-+ runScript+ (pure (GlobalPosition 0))+ (pure 0)+ [ KirokuEventSubscriptionDelivered sub 3 DeliveredCatchUp NonGroup+ , KirokuEventSubscriptionDelivered sub 2 DeliveredLive NonGroup+ , KirokuEventSubscriptionFetched sub 2 NonGroup+ ]+ []+ snap.counters.batchesDelivered `shouldBe` 2+ snap.counters.eventsDelivered `shouldBe` 5+ snap.counters.liveFetches `shouldBe` 1++ it "tracks pool connection gauges and cumulative counters from observations" $ do+ let u = uuidN 1+ snap <-+ runScript+ (pure (GlobalPosition 0))+ (pure 0)+ []+ [ ConnectionObservation u ConnectingConnectionStatus+ , ConnectionObservation u (ReadyForUseConnectionStatus EstablishedConnectionReadyForUseReason)+ , ConnectionObservation u InUseConnectionStatus+ ]+ snap.store.poolConnecting `shouldBe` 0+ snap.store.poolReady `shouldBe` 0+ snap.store.poolInUse `shouldBe` 1+ snap.store.poolEstablishedTotal `shouldBe` 1+ snap.store.poolTerminatedTotal `shouldBe` 0++ it "removes terminated connections from the gauge and bumps the terminated counter" $ do+ let u = uuidN 2+ snap <-+ runScript+ (pure (GlobalPosition 0))+ (pure 0)+ []+ [ ConnectionObservation u (ReadyForUseConnectionStatus EstablishedConnectionReadyForUseReason)+ , ConnectionObservation u (TerminatedConnectionStatus ReleaseConnectionTerminationReason)+ ]+ snap.store.poolReady `shouldBe` 0+ snap.store.poolInUse `shouldBe` 0+ snap.store.poolEstablishedTotal `shouldBe` 1+ snap.store.poolTerminatedTotal `shouldBe` 1
+ test/Test/IntegrationSpec.hs view
@@ -0,0 +1,157 @@+{-# LANGUAGE OverloadedLabels #-}++{- | Integration test: wire the collector into a real ephemeral+PostgreSQL-backed store, run a live @$all@ subscription, append events, and+assert the snapshot reflects real store activity (not just scripted inputs).++The ordering wrinkle is that 'newKirokuMetrics' needs the 'KirokuStore' but the+callbacks must be installed on 'ConnectionSettings' before 'withStore' creates+it. We resolve it with a @TVar (Maybe KirokuStore)@ the snapshot-time STM+readers consult; it is filled inside 'withStore'. (A 'TVar', not an 'IORef',+because the readers are 'STM'.)+-}+module Test.IntegrationSpec (spec) where++import Control.Concurrent.STM (+ STM,+ TVar,+ atomically,+ check,+ newTVarIO,+ orElse,+ readTVar,+ registerDelay,+ writeTVar,+ )+import Control.Lens ((&), (.~))+import Control.Monad (unless)+import Data.Aeson qualified as Aeson+import Data.Generics.Labels ()+import Data.IntMap.Strict qualified as IntMap+import Data.Map.Strict qualified as Map+import Data.Text qualified as T+import Test.Hspec++import Kiroku.Metrics (+ LifecycleCounters (..),+ MetricsSnapshot (..),+ StoreGauges (..),+ SubscriptionMetrics (..),+ metricsEventHandler,+ metricsObservationHandler,+ newKirokuMetricsWith,+ snapshotMetrics,+ )+import Kiroku.Store (+ EventData (..),+ EventType (..),+ ExpectedVersion (..),+ KirokuStore (..),+ StreamName (..),+ SubscriptionResult (..),+ SubscriptionTarget (..),+ appendToStream,+ cancel,+ defaultConnectionSettings,+ defaultSubscriptionConfig,+ runStoreIO,+ subscribe,+ wait,+ withStore,+ )+import Kiroku.Store.Subscription.EventPublisher (EventPublisher (..), publisherPosition)+import Kiroku.Store.Subscription.Types (SubscriptionName (..))+import Kiroku.Store.Types (GlobalPosition (..))+import Kiroku.Test.Postgres (withMigratedTestDatabase)++subName :: SubscriptionName+subName = SubscriptionName "metrics-it"++spec :: Spec+spec = describe "Kiroku.Metrics.Collector (integration)" $ do+ it "observes real store activity: appends, a running subscription, and lag" $+ withMigratedTestDatabase $ \connStr -> do+ -- Deferred store handle for the snapshot-time STM readers.+ storeVar <- newTVarIO Nothing+ km <-+ newKirokuMetricsWith+ (readPosition storeVar)+ (readSubscribers storeVar)+ let settings =+ defaultConnectionSettings connStr+ & #eventHandler .~ Just (metricsEventHandler km Nothing)+ & #observationHandler .~ Just (metricsObservationHandler km Nothing)+ withStore settings $ \store -> do+ atomically (writeTVar storeVar (Just store))++ -- Subscribe to $all on an empty store: the worker goes live+ -- immediately, so the 5 events we append next flow through the+ -- live publisher path and advance publisherPosition.+ delivered <- newTVarIO (0 :: Int)+ let cfg =+ defaultSubscriptionConfig subName AllStreams $ \_event -> do+ atomically (modifyCount delivered)+ pure Continue+ handle <- subscribe store cfg++ appendStoreEvents store (StreamName "metrics-it-stream") 5+ waitForCount delivered 5 10_000_000++ cancel handle+ _ <- wait handle++ snap <- snapshotMetrics km+ snap.store.globalPosition `shouldSatisfy` (>= 5)+ snap.counters.subscriptionsStarted `shouldBe` 1+ snap.counters.subscriptionsStoppedCancelled `shouldBe` 1+ snap.counters.eventsDelivered `shouldSatisfy` (>= 5)+ case Map.lookup "metrics-it" snap.subscriptions of+ Nothing -> expectationFailure "expected subscription \"metrics-it\" in snapshot"+ Just m -> m.lag `shouldSatisfy` (>= 0)++{- | Snapshot-time reader for the store global position, defaulting to 0 until+the store is filled in.+-}+readPosition :: TVar (Maybe KirokuStore) -> STM GlobalPosition+readPosition storeVar =+ readTVar storeVar >>= maybe (pure (GlobalPosition 0)) (publisherPosition . (.publisher))++-- | Snapshot-time reader for the active-subscriber count.+readSubscribers :: TVar (Maybe KirokuStore) -> STM Int+readSubscribers storeVar =+ readTVar storeVar+ >>= maybe (pure 0) (\s -> IntMap.size <$> readTVar (subscribers s.publisher))++modifyCount :: TVar Int -> STM ()+modifyCount v = readTVar v >>= \c -> writeTVar v (c + 1)++-- | Append @n@ trivial events to a fresh stream via the IO store interpreter.+appendStoreEvents :: KirokuStore -> StreamName -> Int -> IO ()+appendStoreEvents store stream 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 stream NoStream events)+ case result of+ Right _ -> pure ()+ Left err -> expectationFailure ("appendStoreEvents failed: " <> show err)++-- | Wait until the counter reaches @target@ or the timeout (micros) fires.+waitForCount :: TVar Int -> Int -> Int -> IO ()+waitForCount countVar target timeoutMicros = do+ timeoutVar <- registerDelay timeoutMicros+ ok <-+ atomically $+ (do c <- readTVar countVar; check (c >= target); pure True)+ `orElse` (do t <- readTVar timeoutVar; check t; pure False)+ unless ok $ do+ actual <- atomically (readTVar countVar)+ expectationFailure ("Timed out waiting for " <> show target <> ", got " <> show actual)
+ test/Test/ServerSpec.hs view
@@ -0,0 +1,209 @@+{-# LANGUAGE OverloadedLabels #-}++{- | End-to-end test of the HTTP endpoints: boot a real store with the collector+wired in, run a live subscription, start the server on an OS-assigned port, and+hit every endpoint with an HTTP client asserting status codes and bodies.+-}+module Test.ServerSpec (spec) where++import Control.Concurrent (threadDelay)+import Control.Concurrent.STM (+ STM,+ TVar,+ atomically,+ check,+ newTVarIO,+ orElse,+ readTVar,+ registerDelay,+ writeTVar,+ )+import Control.Lens ((&), (.~))+import Control.Monad (unless)+import Data.Aeson (Value (..), decode)+import Data.Aeson.Key qualified as Key+import Data.Aeson.KeyMap qualified as KM+import Data.ByteString.Lazy (ByteString)+import Data.ByteString.Lazy qualified as LBS+import Data.IntMap.Strict qualified as IntMap+import Data.Scientific (toRealFloat)+import Data.Text qualified as T+import Data.Text.Encoding qualified as TE+import Network.HTTP.Client (+ Manager,+ defaultManagerSettings,+ httpLbs,+ newManager,+ parseRequest,+ responseBody,+ responseStatus,+ )+import Network.HTTP.Types (statusCode)+import Test.Hspec++import Kiroku.Metrics (+ DependencyStatus (..),+ MetricsServer (..),+ defaultConfig,+ metricsEventHandler,+ metricsObservationHandler,+ newKirokuMetricsWith,+ startMetricsServer,+ stopMetricsServer,+ )+import Kiroku.Metrics.Config (MetricsServerConfig (..))+import Kiroku.Store (+ EventData (..),+ EventType (..),+ ExpectedVersion (..),+ KirokuStore (..),+ StreamName (..),+ SubscriptionResult (..),+ SubscriptionTarget (..),+ appendToStream,+ cancel,+ defaultConnectionSettings,+ defaultSubscriptionConfig,+ runStoreIO,+ subscribe,+ wait,+ withStore,+ )+import Kiroku.Store.Subscription.EventPublisher (EventPublisher (..), publisherPosition)+import Kiroku.Store.Subscription.Types (SubscriptionName (..))+import Kiroku.Store.Types (GlobalPosition (..))+import Kiroku.Test.Postgres (withMigratedTestDatabase)++subName :: SubscriptionName+subName = SubscriptionName "metrics-server-it"++spec :: Spec+spec = describe "Kiroku.Metrics.Server (endpoints)" $ do+ it "serves JSON, Prometheus, and health endpoints over HTTP" $+ withMigratedTestDatabase $ \connStr -> do+ storeVar <- newTVarIO Nothing+ km <- newKirokuMetricsWith (readPosition storeVar) (readSubscribers storeVar)+ let settings =+ defaultConnectionSettings connStr+ & #eventHandler .~ Just (metricsEventHandler km Nothing)+ & #observationHandler .~ Just (metricsObservationHandler km Nothing)+ withStore settings $ \store -> do+ atomically (writeTVar storeVar (Just store))+ delivered <- newTVarIO (0 :: Int)+ let cfg =+ defaultSubscriptionConfig subName AllStreams $ \_event -> do+ atomically (modifyCount delivered)+ pure Continue+ handle <- subscribe store cfg+ appendStoreEvents store (StreamName "metrics-server-stream") 3+ waitForCount delivered 3 10_000_000++ mgr <- newManager defaultManagerSettings++ -- Healthy server (no dependency checks).+ let serverCfg = defaultConfig{port = 0}+ srv <- startMetricsServer serverCfg km []+ threadDelay 200_000+ let base = "http://127.0.0.1:" <> show srv.serverPort++ (sMetrics, bMetrics) <- get mgr (base <> "/metrics")+ sMetrics `shouldBe` 200+ globalPositionOf bMetrics `shouldSatisfy` (>= 3)++ (sOne, _) <- get mgr (base <> "/metrics/metrics-server-it")+ sOne `shouldBe` 200++ (sUnknown, _) <- get mgr (base <> "/metrics/does-not-exist")+ sUnknown `shouldBe` 404++ (sProm, bProm) <- get mgr (base <> "/metrics/prometheus")+ sProm `shouldBe` 200+ bodyText bProm `shouldSatisfy` T.isInfixOf "kiroku_events_appended_total"++ (sLive, _) <- get mgr (base <> "/health/live")+ sLive `shouldBe` 200++ (sReady, _) <- get mgr (base <> "/health/ready")+ sReady `shouldBe` 200++ (sWs, _) <- get mgr (base <> "/ws")+ sWs `shouldBe` 404++ (sNope, _) <- get mgr (base <> "/nope")+ sNope `shouldBe` 404++ stopMetricsServer srv++ -- A failing dependency check flips readiness to 503.+ let failing = pure (DependencyStatus "fake" False Nothing (Just "down"))+ srv2 <- startMetricsServer serverCfg km [failing]+ threadDelay 200_000+ let base2 = "http://127.0.0.1:" <> show srv2.serverPort+ (sReady2, _) <- get mgr (base2 <> "/health/ready")+ sReady2 `shouldBe` 503+ stopMetricsServer srv2++ cancel handle+ _ <- wait handle+ pure ()++-- | GET a URL, returning the status code and the body, without throwing on non-2xx.+get :: Manager -> String -> IO (Int, ByteString)+get mgr url = do+ req <- parseRequest url+ resp <- httpLbs req mgr+ pure (statusCode (responseStatus resp), responseBody resp)++-- | Extract @store.global_position@ from a @/metrics@ JSON body.+globalPositionOf :: ByteString -> Int+globalPositionOf body =+ case decode body of+ Just (Object o)+ | Just (Object store) <- KM.lookup (Key.fromText "store") o+ , Just (Number n) <- KM.lookup (Key.fromText "global_position") store ->+ truncate (toRealFloat n :: Double)+ _ -> -1++bodyText :: ByteString -> T.Text+bodyText = TE.decodeUtf8 . LBS.toStrict++readPosition :: TVar (Maybe KirokuStore) -> STM GlobalPosition+readPosition storeVar =+ readTVar storeVar >>= maybe (pure (GlobalPosition 0)) (publisherPosition . (.publisher))++readSubscribers :: TVar (Maybe KirokuStore) -> STM Int+readSubscribers storeVar =+ readTVar storeVar+ >>= maybe (pure 0) (\s -> IntMap.size <$> readTVar (subscribers s.publisher))++modifyCount :: TVar Int -> STM ()+modifyCount v = readTVar v >>= \c -> writeTVar v (c + 1)++appendStoreEvents :: KirokuStore -> StreamName -> Int -> IO ()+appendStoreEvents store stream n = do+ let events =+ [ EventData+ { eventId = Nothing+ , eventType = EventType ("E" <> T.pack (show i))+ , payload = Null+ , metadata = Nothing+ , causationId = Nothing+ , correlationId = Nothing+ }+ | i <- [1 .. n]+ ]+ result <- runStoreIO store (appendToStream stream NoStream events)+ case result of+ Right _ -> pure ()+ Left err -> expectationFailure ("appendStoreEvents failed: " <> show err)++waitForCount :: TVar Int -> Int -> Int -> IO ()+waitForCount countVar target timeoutMicros = do+ timeoutVar <- registerDelay timeoutMicros+ ok <-+ atomically $+ (do c <- readTVar countVar; check (c >= target); pure True)+ `orElse` (do t <- readTVar timeoutVar; check t; pure False)+ unless ok $ do+ actual <- atomically (readTVar countVar)+ expectationFailure ("Timed out waiting for " <> show target <> ", got " <> show actual)
+ test/Test/SubscriptionsSpec.hs view
@@ -0,0 +1,181 @@+{- | EP-5 tests for the @GET /subscriptions@ endpoint and the CLI remote client.++ * Cross-package shape: rows produced by the server-side mapping+ ('subscriptionStatusRows') encode and decode back through the shared+ @kiroku-cli@ codec — guarding against either side re-encoding locally.+ * End-to-end: boot a real store with a live subscription, serve+ @\/subscriptions@ from 'storeSubscriptionStatus', and assert the real CLI+ client ('fetchRemoteSubscriptionStatusRows') reports the live subscription+ with a sane phase and the expected cursor. A raw GET checks the by-name route+ (known → one row, unknown → @[]@), and a server with no provider yields the+ configured-404.+-}+module Test.SubscriptionsSpec (spec) where++import Control.Concurrent (threadDelay)+import Data.Aeson qualified as Aeson+import Data.Int (Int32, Int64)+import Data.Map.Strict qualified as Map+import Data.Text (Text)+import Data.Text qualified as T+import Network.HTTP.Client (+ defaultManagerSettings,+ httpLbs,+ newManager,+ parseRequest,+ responseBody,+ responseStatus,+ )+import Network.HTTP.Types.Status (statusCode)+import Network.Wai.Handler.Warp qualified as Warp+import Test.Hspec++import Kiroku.Cli.Command (RemoteEndpoint (..))+import Kiroku.Cli.Subscription.Status (+ SubscriptionStatusRow (..),+ fetchRemoteSubscriptionStatusRows,+ subscriptionStatusRows,+ )+import Kiroku.Metrics (+ MetricsServer (..),+ defaultConfig,+ newKirokuMetrics,+ startMetricsServer,+ stopMetricsServer,+ storeSubscriptionStatus,+ subscriptionsApp,+ withMetricsServerSubscriptions,+ )+import Kiroku.Metrics.Config (MetricsServerConfig (..))+import Kiroku.Store (+ EventData (..),+ EventType (..),+ ExpectedVersion (..),+ KirokuStore,+ StreamName (..),+ SubscriptionResult (..),+ SubscriptionTarget (..),+ appendToStream,+ cancel,+ defaultConnectionSettings,+ defaultSubscriptionConfig,+ runStoreIO,+ subscribe,+ withStore,+ )+import Kiroku.Store.Subscription (SubscriptionStateView (..), subscriptionStates)+import Kiroku.Store.Subscription.Fsm (SubscriptionState (..))+import Kiroku.Store.Subscription.Types (SubscriptionName (..))+import Kiroku.Store.Types (GlobalPosition (..))+import Kiroku.Test.Postgres (withMigratedTestDatabase)++spec :: Spec+spec = describe "Kiroku.Metrics.Subscriptions (/subscriptions)" $ do+ it "encodes server-side rows that decode back through the shared CLI codec" $ do+ let views =+ Map.fromList+ [ ((SubscriptionName "beta", 1), mkView "beta" 1 "live" 42)+ , ((SubscriptionName "alpha", 0), mkView "alpha" 0 "catching_up" 7)+ ]+ rows = subscriptionStatusRows views+ Aeson.eitherDecode (Aeson.encode rows) `shouldBe` Right rows++ it "reports a live subscription to the CLI remote client end to end" $+ withMigratedTestDatabase $ \connStr ->+ withStore (defaultConnectionSettings connStr) $ \store -> do+ m <- newKirokuMetrics store+ let name = SubscriptionName "subs-endpoint-it"+ Right _ <-+ runStoreIO store $+ appendToStream (StreamName "subs-endpoint-1") NoStream [ev "A", ev "B", ev "C"]+ handle <- subscribe store (defaultSubscriptionConfig name AllStreams (\_ -> pure Continue))+ live <- waitUntilPhase 10_000_000 store (name, 0) "live"+ live `shouldBe` True++ withMetricsServerSubscriptions (defaultConfig{port = 0}) m [] (storeSubscriptionStatus store) $ \srv -> do+ threadDelay 200_000+ let base = "http://127.0.0.1:" <> show srv.serverPort++ -- The real CLI client hits GET /subscriptions and decodes the rows.+ rowsResult <- fetchRemoteSubscriptionStatusRows (RemoteEndpoint (T.pack base))+ case rowsResult of+ Left err -> expectationFailure ("expected rows, got error: " <> T.unpack err)+ Right rows ->+ case filter (\r -> r.subscription == "subs-endpoint-it") rows of+ [row] -> do+ row.phase `shouldBe` "live"+ row.globalPosition `shouldBe` 3+ other -> expectationFailure ("expected exactly one matching row, got " <> show other)++ -- By-name route: known name → one row, unknown name → [].+ known <- getRows (base <> "/subscriptions/subs-endpoint-it")+ map (.subscription) known `shouldBe` ["subs-endpoint-it"]+ unknown <- getRows (base <> "/subscriptions/does-not-exist")+ unknown `shouldBe` []++ cancel handle++ it "returns a configured-404 when no provider is wired" $+ withMigratedTestDatabase $ \connStr ->+ withStore (defaultConnectionSettings connStr) $ \store -> do+ m <- newKirokuMetrics store+ srv <- startMetricsServer (defaultConfig{port = 0}) m []+ threadDelay 200_000+ result <-+ fetchRemoteSubscriptionStatusRows+ (RemoteEndpoint (T.pack ("http://127.0.0.1:" <> show srv.serverPort)))+ stopMetricsServer srv+ case result of+ Left err -> err `shouldSatisfy` T.isInfixOf "404"+ Right rows -> expectationFailure ("expected a 404 error, got rows: " <> show rows)++ it "returns the documented 404 JSON body for unknown paths when mounted standalone" $+ Warp.testWithApplication (pure (subscriptionsApp (pure []))) $ \port -> do+ mgr <- newManager defaultManagerSettings+ req <- parseRequest ("http://127.0.0.1:" <> show port <> "/definitely/not/a/route")+ resp <- httpLbs req mgr+ statusCode (responseStatus resp) `shouldBe` 404+ Aeson.decode (responseBody resp)+ `shouldBe` Just (Aeson.object ["error" Aeson..= ("Not found" :: Text)])++-- | GET a URL and decode the body as @[SubscriptionStatusRow]@ (via the shared codec).+getRows :: String -> IO [SubscriptionStatusRow]+getRows url = do+ mgr <- newManager defaultManagerSettings+ req <- parseRequest url+ resp <- httpLbs req mgr+ case Aeson.eitherDecode (responseBody resp) of+ Right rows -> pure rows+ Left err -> expectationFailure ("could not decode " <> url <> ": " <> err) >> pure []++mkView :: Text -> Int32 -> Text -> Int64 -> SubscriptionStateView+mkView name member phase position =+ SubscriptionStateView+ { subscriptionName = SubscriptionName name+ , member = member+ , state = Live (GlobalPosition position)+ , statePhase = phase+ , cursor = GlobalPosition position+ }++ev :: Text -> EventData+ev typ =+ EventData+ { eventId = Nothing+ , eventType = EventType typ+ , payload = Aeson.Null+ , metadata = Nothing+ , causationId = Nothing+ , correlationId = Nothing+ }++waitUntilPhase :: Int -> KirokuStore -> (SubscriptionName, Int32) -> Text -> IO Bool+waitUntilPhase budget store key phase+ | budget <= 0 = pure False+ | otherwise = do+ snapshot <- subscriptionStates store+ case Map.lookup key snapshot of+ Just status | statePhase status == phase -> pure True+ _ -> do+ threadDelay 20_000+ waitUntilPhase (budget - 20_000) store key phase
+ test/Test/WebSocketSpec.hs view
@@ -0,0 +1,357 @@+{-# LANGUAGE OverloadedLabels #-}+{-# LANGUAGE ScopedTypeVariables #-}++{- | End-to-end test of the WebSocket endpoints (EP-3): boot a real store with the+collector wired in, start the store-aware server on an OS-assigned port, then+drive both channels over a real socket with the @websockets@ client.++ * @/ws/events@: subscribe, append three events from the test thread, assert+ three @event@ messages arrive in global-position order with the expected+ @eventType@s. After the client disconnects, the publisher's transient+ broadcast subscriber must be cleaned up (count returns to baseline) — proof+ the tail's @finally@ wiring deregisters and that it leaves no trace.+ * @/ws/metrics@: connect, receive a @snapshot@, assert @store.global_position@+ reflects the appended events.+-}+module Test.WebSocketSpec (spec) where++import Control.Concurrent (myThreadId, threadDelay)+import Control.Concurrent.Async (async, asyncThreadId, wait)+import Control.Concurrent.STM (+ STM,+ TVar,+ atomically,+ check,+ newTVarIO,+ orElse,+ readTVar,+ readTVarIO,+ registerDelay,+ writeTVar,+ )+import Control.Exception (finally)+import Control.Lens ((&), (.~))+import Control.Monad (replicateM, unless, when)+import Data.Aeson (Value (..), decode, encode, object, (.=))+import Data.Aeson.Key qualified as Key+import Data.Aeson.KeyMap qualified as KM+import Data.ByteString.Lazy qualified as LBS+import Data.Foldable (for_)+import Data.IntMap.Strict qualified as IntMap+import Data.Scientific (toRealFloat)+import Data.Text (Text)+import Data.Text qualified as T+import Hasql.Pool qualified as Pool+import Hasql.Session qualified as Session+import Network.WebSockets qualified as WS+import System.Timeout (timeout)+import Test.Hspec++import Kiroku.Metrics (+ MetricsServer (..),+ defaultConfig,+ metricsEventHandler,+ metricsObservationHandler,+ newKirokuMetricsWith,+ startMetricsServerWithStore,+ stopMetricsServer,+ )+import Kiroku.Metrics.Config (MetricsServerConfig (..))+import Kiroku.Store (+ EventData (..),+ EventType (..),+ ExpectedVersion (..),+ KirokuStore (..),+ StreamName (..),+ appendToStream,+ defaultConnectionSettings,+ runStoreIO,+ withStore,+ )+import Kiroku.Store.Subscription.EventPublisher (EventPublisher (..), publisherPosition, subscribePublisher)+import Kiroku.Store.Subscription.Types (OverflowPolicy (..))+import Kiroku.Store.Types (GlobalPosition (..))+import Kiroku.Test.Postgres (withMigratedTestDatabase)++spec :: Spec+spec = describe "Kiroku.Metrics.WebSocket (endpoints)" $ do+ it "streams appended events over /ws/events and a snapshot over /ws/metrics" $+ withMigratedTestDatabase $ \connStr -> do+ storeVar <- newTVarIO Nothing+ km <- newKirokuMetricsWith (readPosition storeVar) (readSubscribers storeVar)+ let settings =+ defaultConnectionSettings connStr+ & #eventHandler .~ Just (metricsEventHandler km Nothing)+ & #observationHandler .~ Just (metricsObservationHandler km Nothing)+ withStore settings $ \store -> do+ atomically (writeTVar storeVar (Just store))+ let serverCfg = defaultConfig{port = 0}+ srv <- startMetricsServerWithStore serverCfg km store []+ threadDelay 300_000+ let port = srv.serverPort+ stream = StreamName "ws-events-stream"++ -- /ws/events: subscribe, then append three events and read them back.+ received <-+ requireJust "ws/events client timed out" $+ timeout 15_000_000 $+ WS.runClient "127.0.0.1" port "/ws/events" $ \conn -> do+ sendJSON conn (object ["type" .= ("subscribe_events" :: Text)])+ _ <- waitForType conn "event_stream_started"+ appendThread <- async (appendStoreEvents store stream 3)+ evs <- replicateM 3 (readEventType conn)+ wait appendThread+ pure evs+ received `shouldBe` ["E1", "E2", "E3"]++ -- The tail uses the public broadcast and unsubscribes on disconnect;+ -- the publisher's subscriber count must return to baseline (0).+ waitForSubscriberCount store 0 5_000_000++ -- /ws/metrics: the snapshot reflects the three appended events.+ gpos <-+ requireJust "ws/metrics client timed out" $+ timeout 15_000_000 $+ WS.runClient "127.0.0.1" port "/ws/metrics" $ \conn -> do+ snap <- waitForType conn "snapshot"+ pure (globalPositionOf snap)+ gpos `shouldSatisfy` (>= 3)++ stopMetricsServer srv++ it "replays history from a position then continues live without duplicating the boundary" $+ withMigratedTestDatabase $ \connStr -> do+ storeVar <- newTVarIO Nothing+ km <- newKirokuMetricsWith (readPosition storeVar) (readSubscribers storeVar)+ let settings =+ defaultConnectionSettings connStr+ & #eventHandler .~ Just (metricsEventHandler km Nothing)+ & #observationHandler .~ Just (metricsObservationHandler km Nothing)+ withStore settings $ \store -> do+ atomically (writeTVar storeVar (Just store))+ let serverCfg = defaultConfig{port = 0}+ srv <- startMetricsServerWithStore serverCfg km store []+ threadDelay 300_000+ let port = srv.serverPort++ -- Two events exist before the client connects (stream A); they must+ -- be replayed. The live event lands on a second stream (B) so the+ -- @NoStream@ append helper does not conflict.+ appendStoreEvents store (StreamName "ws-replay-a") 2+ received <-+ requireJust "ws/events replay client timed out" $+ timeout 15_000_000 $+ WS.runClient "127.0.0.1" port "/ws/events" $ \conn -> do+ sendJSON+ conn+ (object ["type" .= ("subscribe_events" :: Text), "from_position" .= (0 :: Int)])+ _ <- waitForType conn "event_stream_started"+ -- Two replayed (A's E1,E2), then one appended live+ -- (B's E1) — three in global-position order, with no+ -- duplicate at the replay/live boundary.+ replayed <- replicateM 2 (readEventType conn)+ liveThread <- async (appendStoreEvents store (StreamName "ws-replay-b") 1)+ live <- readEventType conn+ wait liveThread+ pure (replayed <> [live])+ received `shouldBe` ["E1", "E2", "E1"]++ stopMetricsServer srv++ it "delivers each global position exactly once when the replay overlaps live appends" $+ withMigratedTestDatabase $ \connStr -> do+ gateVar <- newTVarIO True+ storeVar <- newTVarIO Nothing+ let publisherGate e = do+ mStore <- readTVarIO storeVar+ for_ mStore $ \s -> do+ tid <- myThreadId+ when (tid == asyncThreadId (publisherThread s.publisher)) $+ atomically (readTVar gateVar >>= check)+ pure e+ settings =+ defaultConnectionSettings connStr+ & #storeSettings . #decodeHook .~ Just publisherGate+ km <- newKirokuMetricsWith (readPosition storeVar) (readSubscribers storeVar)+ withStore settings $ \store -> do+ atomically (writeTVar storeVar (Just store))+ let serverCfg = defaultConfig{port = 0}+ srv <- startMetricsServerWithStore serverCfg km store []+ threadDelay 300_000+ let port = srv.serverPort++ appendStoreEvents store (StreamName "ws-race-a") 2+ waitForPublisherPosition store 2 5_000_000++ (_, _, unsubscribeDummy) <- atomically (subscribePublisher store.publisher 16 DropOldest)+ received <-+ flip finally unsubscribeDummy $+ do+ atomically (writeTVar gateVar False)+ appendStoreEvents store (StreamName "ws-race-b") 3++ requireJust "ws/events exactly-once client timed out" $+ timeout 15_000_000 $+ WS.runClient "127.0.0.1" port "/ws/events" $ \conn -> do+ sendJSON+ conn+ (object ["type" .= ("subscribe_events" :: Text), "from_position" .= (0 :: Int)])+ _ <- waitForType conn "event_stream_started"+ replayed <- replicateM 5 (readEventPosition conn)++ atomically (writeTVar gateVar True)+ appendStoreEvents store (StreamName "ws-race-c") 1+ live <- readEventPosition conn+ quiet <- timeout 500_000 (WS.receiveData conn :: IO LBS.ByteString)+ quiet `shouldBe` Nothing+ pure (replayed <> [live])+ received `shouldBe` [1, 2, 3, 4, 5, 6]++ stopMetricsServer srv++ it "terminates the tail after a replay error instead of streaming with a gap" $+ withMigratedTestDatabase $ \connStr -> do+ storeVar <- newTVarIO Nothing+ km <- newKirokuMetricsWith (readPosition storeVar) (readSubscribers storeVar)+ let settings =+ defaultConnectionSettings connStr+ & #eventHandler .~ Just (metricsEventHandler km Nothing)+ & #observationHandler .~ Just (metricsObservationHandler km Nothing)+ withStore settings $ \store -> do+ atomically (writeTVar storeVar (Just store))+ let serverCfg = defaultConfig{port = 0}+ srv <- startMetricsServerWithStore serverCfg km store []+ threadDelay 300_000+ let port = srv.serverPort++ appendStoreEvents store (StreamName "ws-err-a") 2+ waitForPublisherPosition store 2 5_000_000+ renamed <- Pool.use store.pool (Session.script "ALTER TABLE events RENAME TO events_hidden")+ renamed `shouldBe` Right ()++ requireJust "ws/events replay-error client timed out" $+ timeout 15_000_000 $+ WS.runClient "127.0.0.1" port "/ws/events" $ \conn -> do+ sendJSON+ conn+ (object ["type" .= ("subscribe_events" :: Text), "from_position" .= (0 :: Int)])+ _ <- waitForType conn "event_stream_started"+ err <- waitForType conn "error"+ case look ["message"] err of+ Just (String msg) -> msg `shouldSatisfy` T.isInfixOf "replay error"+ other -> expectationFailure ("error without message: " <> show other)+ waitForSubscriberCount store 0 5_000_000++ stopMetricsServer srv++-- | Fail the example with a message if the timed action returned 'Nothing'.+requireJust :: String -> IO (Maybe a) -> IO a+requireJust msg act = act >>= maybe (expectationFailure msg >> error msg) pure++sendJSON :: WS.Connection -> Value -> IO ()+sendJSON conn = WS.sendTextData conn . encode++recvValue :: WS.Connection -> IO Value+recvValue conn = do+ raw <- WS.receiveData conn :: IO LBS.ByteString+ case decode raw of+ Just v -> pure v+ Nothing -> expectationFailure ("undecodable frame: " <> show raw) >> error "undecodable"++-- | Read server messages until one with the given @"type"@ arrives; return it.+waitForType :: WS.Connection -> Text -> IO Value+waitForType conn want = go+ where+ go = do+ v <- recvValue conn+ if look ["type"] v == Just (String want) then pure v else go++-- | Read until an @event@ message, returning its @event.eventType@.+readEventType :: WS.Connection -> IO Text+readEventType conn = do+ v <- waitForType conn "event"+ case look ["event", "eventType"] v of+ Just (String t) -> pure t+ other -> expectationFailure ("event without eventType: " <> show other) >> error "no eventType"++-- | Read until an @event@ message, returning its @event.globalPosition@.+readEventPosition :: WS.Connection -> IO Int+readEventPosition conn = do+ v <- waitForType conn "event"+ case look ["event", "globalPosition"] v of+ Just (Number n) -> pure (truncate (toRealFloat n :: Double))+ other -> expectationFailure ("event without globalPosition: " <> show other) >> error "no position"++-- | Navigate nested object keys.+look :: [Text] -> Value -> Maybe Value+look [] v = Just v+look (k : ks) (Object o) = KM.lookup (Key.fromText k) o >>= look ks+look _ _ = Nothing++-- | Extract @metrics.store.global_position@ from a @snapshot@ message.+globalPositionOf :: Value -> Int+globalPositionOf v =+ case look ["metrics", "store", "global_position"] v of+ Just (Number n) -> truncate (toRealFloat n :: Double)+ _ -> -1++readPosition :: TVar (Maybe KirokuStore) -> STM GlobalPosition+readPosition storeVar =+ readTVar storeVar >>= maybe (pure (GlobalPosition 0)) (publisherPosition . (.publisher))++readSubscribers :: TVar (Maybe KirokuStore) -> STM Int+readSubscribers storeVar =+ readTVar storeVar+ >>= maybe (pure 0) (\s -> IntMap.size <$> readTVar (subscribers s.publisher))++-- | Block until the publisher's subscriber count reaches @target@ or time out.+waitForSubscriberCount :: KirokuStore -> Int -> Int -> IO ()+waitForSubscriberCount store target timeoutMicros = do+ timeoutVar <- registerDelay timeoutMicros+ ok <-+ atomically $+ ( do+ c <- IntMap.size <$> readTVar (subscribers store.publisher)+ check (c == target)+ pure True+ )+ `orElse` (readTVar timeoutVar >>= \t -> check t >> pure False)+ unless ok $ do+ actual <- atomically (IntMap.size <$> readTVar (subscribers store.publisher))+ expectationFailure+ ("subscriber count did not reach " <> show target <> "; still " <> show actual)++-- | Block until the publisher cursor reaches @target@ or time out.+waitForPublisherPosition :: KirokuStore -> Int -> Int -> IO ()+waitForPublisherPosition store target timeoutMicros = do+ timeoutVar <- registerDelay timeoutMicros+ ok <-+ atomically $+ ( do+ GlobalPosition p <- publisherPosition store.publisher+ check (p >= fromIntegral target)+ pure True+ )+ `orElse` (readTVar timeoutVar >>= \t -> check t >> pure False)+ unless ok $ do+ GlobalPosition actual <- atomically (publisherPosition store.publisher)+ expectationFailure+ ("publisher position did not reach " <> show target <> "; still " <> show actual)++appendStoreEvents :: KirokuStore -> StreamName -> Int -> IO ()+appendStoreEvents store stream n = do+ let events =+ [ EventData+ { eventId = Nothing+ , eventType = EventType ("E" <> T.pack (show i))+ , payload = Null+ , metadata = Nothing+ , causationId = Nothing+ , correlationId = Nothing+ }+ | i <- [1 .. n]+ ]+ result <- runStoreIO store (appendToStream stream NoStream events)+ case result of+ Right _ -> pure ()+ Left err -> expectationFailure ("appendStoreEvents failed: " <> show err)