diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 --- /dev/null +++ b/CHANGELOG.md @@ -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. diff --git a/example/Main.hs b/example/Main.hs new file mode 100644 --- /dev/null +++ b/example/Main.hs @@ -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)) diff --git a/kiroku-metrics.cabal b/kiroku-metrics.cabal new file mode 100644 --- /dev/null +++ b/kiroku-metrics.cabal @@ -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 diff --git a/src/Kiroku/Metrics.hs b/src/Kiroku/Metrics.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Metrics.hs @@ -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 diff --git a/src/Kiroku/Metrics/Collector.hs b/src/Kiroku/Metrics/Collector.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Metrics/Collector.hs @@ -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 + } diff --git a/src/Kiroku/Metrics/Config.hs b/src/Kiroku/Metrics/Config.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Metrics/Config.hs @@ -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 + } diff --git a/src/Kiroku/Metrics/Health.hs b/src/Kiroku/Metrics/Health.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Metrics/Health.hs @@ -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))) diff --git a/src/Kiroku/Metrics/JSON.hs b/src/Kiroku/Metrics/JSON.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Metrics/JSON.hs @@ -0,0 +1,44 @@ +{- | JSON HTTP endpoints: @GET /metrics@ (the full snapshot) and +@GET /metrics/\@ (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/\@; 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")] diff --git a/src/Kiroku/Metrics/Prometheus.hs b/src/Kiroku/Metrics/Prometheus.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Metrics/Prometheus.hs @@ -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 diff --git a/src/Kiroku/Metrics/Server.hs b/src/Kiroku/Metrics/Server.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Metrics/Server.hs @@ -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/\@. +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)])) diff --git a/src/Kiroku/Metrics/Subscriptions.hs b/src/Kiroku/Metrics/Subscriptions.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Metrics/Subscriptions.hs @@ -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/\@ — 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)]))) diff --git a/src/Kiroku/Metrics/Types.hs b/src/Kiroku/Metrics/Types.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Metrics/Types.hs @@ -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 + ] diff --git a/src/Kiroku/Metrics/WebSocket.hs b/src/Kiroku/Metrics/WebSocket.hs new file mode 100644 --- /dev/null +++ b/src/Kiroku/Metrics/WebSocket.hs @@ -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 diff --git a/test/Main.hs b/test/Main.hs new file mode 100644 --- /dev/null +++ b/test/Main.hs @@ -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 diff --git a/test/Test/CollectorSpec.hs b/test/Test/CollectorSpec.hs new file mode 100644 --- /dev/null +++ b/test/Test/CollectorSpec.hs @@ -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 diff --git a/test/Test/IntegrationSpec.hs b/test/Test/IntegrationSpec.hs new file mode 100644 --- /dev/null +++ b/test/Test/IntegrationSpec.hs @@ -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) diff --git a/test/Test/ServerSpec.hs b/test/Test/ServerSpec.hs new file mode 100644 --- /dev/null +++ b/test/Test/ServerSpec.hs @@ -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) diff --git a/test/Test/SubscriptionsSpec.hs b/test/Test/SubscriptionsSpec.hs new file mode 100644 --- /dev/null +++ b/test/Test/SubscriptionsSpec.hs @@ -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 diff --git a/test/Test/WebSocketSpec.hs b/test/Test/WebSocketSpec.hs new file mode 100644 --- /dev/null +++ b/test/Test/WebSocketSpec.hs @@ -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)