packages feed

keiro-ops-http 0.23.0.0 → 0.24.0.0

raw patch · 11 files changed

+508/−81 lines, 11 filesdep ~keirodep ~keiro-migrationsdep ~keiro-opsPVP ok

version bump matches the API change (PVP)

Dependency ranges changed: keiro, keiro-migrations, keiro-ops, keiro-pgmq, keiro-test-support

API changes (from Hackage documentation)

+ Keiro.Ops.Http.Host: HostOptions :: !Maybe Text -> !OpsHttpConfig -> !OpsHttpServerConfig -> HostOptions
+ Keiro.Ops.Http.Host: [databaseUrl] :: HostOptions -> !Maybe Text
+ Keiro.Ops.Http.Host: [http] :: HostOptions -> !OpsHttpConfig
+ Keiro.Ops.Http.Host: [server] :: HostOptions -> !OpsHttpServerConfig
+ Keiro.Ops.Http.Host: data HostOptions
+ Keiro.Ops.Http.Host: hostOptionsParser :: Parser HostOptions
+ Keiro.Ops.Http.Host: mainWithHooks :: AppHooks -> IO ()
+ Keiro.Ops.Http.Host: withHost :: HostOptions -> AppHooks -> (OpsHttpServer -> IO a) -> IO a

Files

CHANGELOG.md view
@@ -2,6 +2,22 @@  ## [Unreleased] +## 0.24.0.0 — 2026-11-05++### New Features++- Expose embedded `GET /projection/consumers?category=NAME` through the shared grammar and existing HTTP budgets.+- Bound dedicated inspection reads to one PostgreSQL backend per query, preserving+  admission/deadline limits and ordinary writer settings.+- Expose `GET /wf/inspect/counts` through the shared command grammar, schema checks and inspection limits.+- `/projection/status` and `projection_groups` feed items include sorted+  `feeding_subscriptions`, shared with the CLI encoder.++- Export `Keiro.Ops.Http.Host` (`HostOptions`, `hostOptionsParser`, `withHost`,+  `mainWithHooks`) and re-export it from `Keiro.Ops.Http`. The standalone+  executable calls `mainWithHooks emptyAppHooks`; application hosts share its+  flags, schema verification, origin policy, mutation gate, and feed lifecycle.+ ## 0.23.0.0 — 2026-10-30  ### Breaking Changes
app/Main.hs view
@@ -1,50 +1,7 @@ module Main (main) where -import Control.Concurrent.Async (wait)-import Data.Foldable (traverse_)-import Data.Text (Text)-import Data.Text qualified as Text-import Data.Text.IO qualified as Text-import Hasql.Connection.Settings qualified as Settings-import Keiro.Migrations.SchemaCheck import Keiro.Ops (emptyAppHooks)-import Keiro.Ops.Env (resolveConnectionString)-import Keiro.Ops.Http-import Keiro.Ops.Http.Config qualified as Config-import Kiroku.Store.Connection (ConnectionSettingsM (..), KirokuStore (..), defaultConnectionSettings, withStore)-import Options.Applicative-import System.IO (stderr)--data Options = Options !(Maybe Text) !OpsHttpConfig !OpsHttpServerConfig--options :: Parser Options-options =-  Options-    <$> optional (textOption (long "database-url" <> metavar "URL"))-    <*> ( (\enabled drift origins -> defaultOpsHttpConfig {mutations = if enabled then MutationsEnabled else MutationsDisabled, schemaDrift = if drift then AllowDrift else RefuseOnDrift, Config.allowedOrigins = origins})-            <$> switch (long "enable-mutations")-            <*> switch (long "allow-schema-drift")-            <*> many (textOption (long "allow-origin" <> metavar "ORIGIN"))-        )-    <*> (OpsHttpServerConfig <$> textOption (long "bind" <> value "127.0.0.1") <*> option auto (long "port" <> value 9092) <*> option auto (long "request-timeout" <> value 300))-  where-    textOption modifiers = Text.pack <$> strOption modifiers+import Keiro.Ops.Http (mainWithHooks)  main :: IO ()-main = do-  Options database config serverConfig <- execParser (info (options <**> helper) (fullDesc <> progDesc "Serve Keiro inspection on a trusted network or behind an authenticating proxy"))-  connectionString <- resolveConnectionString database-  verified <- verifyExpectedSchema (Settings.connectionString connectionString)-  case verified of-    Left err -> ioError (userError (show err))-    Right drift -> traverse_ (Text.hPutStrLn stderr . ("warning: " <>) . renderSchemaDrift) drift-  -- This store belongs only to inspection. All its pooled connections inherit-  -- the SQL deadline; command execution never borrows the runtime's pool.-  let settings = (defaultConnectionSettings connectionString) {poolSize = config.inspectionLimits.maxConcurrentReads, statementTimeout = Just 2}-  withStore settings $ \baseStore -> withOpsInspectionPool config.inspectionLimits settings $ \pool -> do-    let store = baseStore {pool}-    withFeedHandle config defaultFeedConfig store $ \feeds -> do-      app <- opsApplicationWithFeeds config feeds emptyAppHooks store-      withOpsHttpServer serverConfig app $ \server -> do-        Text.hPutStrLn stderr ("keiro-ops-http: listening on " <> serverConfig.bindHost <> ":" <> Text.pack (show server.serverPort) <> " (mutations " <> (if config.mutations == MutationsEnabled then "enabled" else "disabled") <> ")")-        wait server.serverThread+main = mainWithHooks emptyAppHooks
bench/InspectionClients.hs view
@@ -18,11 +18,18 @@ import Network.HTTP.Client qualified as HTTP import Network.HTTP.Types (statusCode) import Network.WebSockets qualified as WS+import System.Environment (lookupEnv) import System.IO (hFlush, hPutStrLn, stderr, stdout) import System.Timeout (timeout)  runClients :: [String] -> IO () runClients [portArg, clientsArg, _, secondsArg, mode]+  | mode `elem` ["wf-counts", "wf-counts-many"] =+      runCountClients (read portArg) (read clientsArg) (read secondsArg)+runClients [portArg, clientsArg, _, secondsArg, mode]+  | mode `elem` ["projection-poll", "projection-feed"] =+      runProjectionClients (read portArg) (read clientsArg) (read secondsArg) (mode == "projection-feed")+runClients [portArg, clientsArg, _, secondsArg, mode]   | mode `elem` ["pm-types", "pm-types-many"] =       runManagerClients (read portArg) (read clientsArg) (read secondsArg) (mode == "pm-types-many") runClients [portArg, clientsArg, viewsArg, secondsArg, mode] = do@@ -139,3 +146,127 @@     unless (good == clients * seconds) (fail "manager polling load disappeared")     hPutStrLn stderr ("manager discovery clients: successes=" <> show good <> " failures=0")     Char8.putStrLn (encode (good, 0 :: Int))++-- Matched projection-only load. Client encoding and validation live in a separate+-- process so writer RTS counters measure the mounted runtime, not its clients.+runProjectionClients :: Int -> Int -> Int -> Bool -> IO ()+runProjectionClients port clients seconds feed = do+  expectNames <- (== Just "1") <$> lookupEnv "KEIRO_PROJECTION_EXPECT_SUBSCRIPTIONS"+  let validateRow (Object fields) = case KeyMap.lookup "group_id" fields of+        Just (String group) -> do+          let expected = if "0" `Text.isSuffixOf` group then [group <> "-sub"] else []+          unless+            (not expectNames || KeyMap.lookup "feeding_subscriptions" fields == Just (Aeson.toJSON expected))+            (fail "incorrect feeding subscriptions")+        _ -> fail "missing projection group identity"+      validateRow _ = fail "invalid projection row"+      validateItems (Object fields) = case KeyMap.lookup "items" fields of+        Just (Array rows) | length rows == 100 -> mapM_ validateRow rows+        _ -> fail "expected 100 projection items"+      validateItems _ = fail "invalid projection page"+  go <- newEmptyMVar+  successes <- newIORef (0 :: Int)+  let count = atomicModifyIORef' successes (\n -> (n + 1, ()))+      ready =+        putStrLn "ready" >> hFlush stdout >> getLine >>= \line ->+          if line == "go" then putMVar go () else fail "missing measurement control"+  if feed+    then do+      let worker signal = WS.runClient "127.0.0.1" port "/keiro/ws" $ \connection -> do+            WS.sendTextData connection (encode (object ["type" .= ("subscribe" :: Text), "feed" .= ("projection_groups" :: Text), "limit" .= (100 :: Int)]))+            raw <- WS.receiveData connection :: IO Lazy.ByteString+            value <- either fail pure (Aeson.eitherDecode raw)+            validateItems value+            count+            putMVar signal ()+            readMVar go+            forever $ do+              frame <- WS.receiveData connection :: IO Lazy.ByteString+              case Aeson.eitherDecode frame of+                Right (Object fields) | KeyMap.lookup "type" fields == Just (String "update") ->+                  case KeyMap.lookup "changed" fields of+                    Just (Array rows) -> mapM_ validateRow rows >> count+                    _ -> fail "invalid projection update"+                _ -> fail "projection feed failed"+          start = do+            signal <- newEmptyMVar+            task <- async (worker signal)+            outcome <- race (wait task) (takeMVar signal)+            case outcome of Right () -> pure task; _ -> cancel task >> fail "projection feed startup failed"+      bracket (forM [1 .. clients] (const start)) (mapM_ cancel) $ \tasks -> do+        ready+        race_ (threadDelay (seconds * 1000000)) (mapConcurrently_ wait tasks)+    else bracket (HTTP.newManager HTTP.defaultManagerSettings {HTTP.managerConnCount = clients}) HTTP.closeManager $ \manager -> do+      request <- HTTP.parseRequest ("http://127.0.0.1:" <> show port <> "/keiro/projection/status?limit=100")+      let fetch = do+            response <- HTTP.httpLbs request manager+            unless (statusCode (HTTP.responseStatus response) == 200) (fail "projection polling refused")+            either fail validateItems (Aeson.eitherDecode (HTTP.responseBody response))+          worker index = do+            readMVar go+            threadDelay (index * 1000000 `div` clients)+            forM_ [1 .. seconds] $ \second -> withAsync (threadDelay 1000000) $ \tick -> do+              fetch+              count+              unless (second == seconds) (wait tick)+      fetch+      bracket (forM [0 .. clients - 1] (async . worker)) (mapM_ cancel) $ \tasks -> do+        ready+        outcome <- timeout ((seconds + 5) * 1000000) (mapConcurrently_ wait tasks)+        unless (outcome == Just ()) (fail "lost projection polling load")+  good <- readIORef successes+  unless (if feed then good >= clients else good == clients * seconds) (fail "lost projection clients")+  Char8.putStrLn (encode (good, 0 :: Int))++-- Fixed one-second ticks; all eight clients start aligned to exercise admission.+-- Keep raw outcomes on stderr in the controller's retained leg log.+runCountClients :: Int -> Int -> Int -> IO ()+runCountClients port clients seconds =+  bracket (HTTP.newManager HTTP.defaultManagerSettings {HTTP.managerConnCount = clients}) HTTP.closeManager $ \manager -> do+    counters <- newIORef Map.empty+    go <- newEmptyMVar+    let count key = atomicModifyIORef' counters (\m -> (Map.insertWith (+) key (1 :: Int) m, ()))+        names = ["", "?workflow_name=seed-common", "?workflow_name=seed-rare", "?workflow_name=absent"]+        fetch index = do+          request <- HTTP.parseRequest ("http://127.0.0.1:" <> show port <> "/keiro/wf/inspect/counts" <> if clients == 1 then "" else names !! (index `mod` 4))+          result <- try (HTTP.httpLbs request manager) :: IO (Either HTTP.HttpException (HTTP.Response Lazy.ByteString))+          case result of+            Left err -> count ("exception:" <> show err)+            Right response -> do+              let code = statusCode (HTTP.responseStatus response)+              count ("http_" <> show code)+              whenSuccess code (HTTP.responseBody response)+        whenSuccess code raw =+          if code /= 200+            then pure ()+            else case Aeson.eitherDecode raw of+              Right (Object fields)+                | Just (Object counts) <- KeyMap.lookup "counts" fields,+                  Just (String _) <- KeyMap.lookup "observed_at" fields,+                  KeyMap.size counts == 7,+                  all (\key -> case KeyMap.lookup key counts of Just (Number n) -> n >= 0 && fromInteger (round n) == n; _ -> False) ["running", "retrying", "sleeping", "awaiting", "completed", "cancelled", "failed"] ->+                    pure ()+              _ -> fail "invalid workflow counts response"+        worker index = do+          readMVar go+          forM_ [1 .. seconds] $ \second -> withAsync (threadDelay 1000000) $ \tick -> do+            fetch index+            unless (second == seconds) (wait tick)+    bracket (forM [0 .. clients - 1] (async . worker)) (mapM_ cancel) $ \tasks -> do+      putStrLn "ready"+      hFlush stdout+      instruction <- getLine+      unless (instruction == "go") (fail "missing measurement control")+      putMVar go ()+      finished <- timeout ((seconds + 5) * 1000000) (mapConcurrently_ wait tasks)+      unless (finished == Just ()) (fail "lost count polling schedule")+    recovery <- HTTP.parseRequest ("http://127.0.0.1:" <> show port <> "/keiro/wf/inspect/counts") >>= (\request -> HTTP.httpLbs request manager)+    unless (statusCode (HTTP.responseStatus recovery) == 200) (fail "count admission did not recover")+    outcomes <- readIORef counters+    hPutStrLn stderr ("workflow count outcomes: " <> show outcomes)+    let good = Map.findWithDefault 0 "http_200" outcomes+        bad = sum (Map.elems outcomes) - good+    unless (good > 0 && good + bad == clients * seconds) (fail "count workload disappeared")+    unless (clients /= 1 || bad == 0) (fail "single count client failed")+    unless (all (\key -> key `elem` ["http_200", "http_503"]) (Map.keys outcomes)) (fail "unexpected count failure")+    Char8.putStrLn (encode (good, bad))
bench/InspectionInterference.hs view
@@ -116,15 +116,27 @@   let duration = maybe (maybe 20 id (configuredSeconds >>= readMaybe)) id retentionSeconds       sampleCount = maybe 1000 (* duration) offeredRate       modes = maybe ["disabled"] (const selectedModes) offeredRate+  workflowRows <- maybe 100000 read <$> lookupEnv "KEIRO_INSPECTION_WORKFLOW_ROWS"+  countsWorkload <- (== Just "1") <$> lookupEnv "KEIRO_COUNTS_WORKLOAD"+  unless (workflowRows > 0 && duration > 0) (fail "positive fixture size and duration required")   enabled <- getRTSStatsEnabled   unless enabled (fail "run with +RTS -T")   createDirectoryIfMissing True output   withMigratedBenchmarkSuite $ \fixture ->     forM_ [1 .. trialCount] $ \trial ->       forM_ modes $ \mode -> withFreshDatabase fixture $ \url -> do-        execute url "INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status) SELECT 'seed-'||n,'seed-'||(n%31),'running' FROM generate_series(1,100000) n; ANALYZE keiro.keiro_workflows"+        execute+          url+          ( if countsWorkload+              then "INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status,attempts,wake_after,completed_at) SELECT 'seed-'||n,CASE WHEN n%997=0 THEN 'seed-rare' WHEN n%10<>0 THEN 'seed-common' ELSE 'seed-'||(n%31) END,CASE n%7 WHEN 0 THEN 'running' WHEN 1 THEN 'running' WHEN 2 THEN 'suspended' WHEN 3 THEN 'suspended' WHEN 4 THEN 'completed' WHEN 5 THEN 'cancelled' ELSE 'failed' END,CASE WHEN n%7=1 THEN 2 ELSE 0 END,CASE WHEN n%7=2 THEN '2100-01-01'::timestamptz END,CASE WHEN n%7>=4 THEN now() END FROM generate_series(1," <> number workflowRows <> ") n; ANALYZE keiro.keiro_workflows"+              else "INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status) SELECT 'seed-'||n,'seed-'||(n%31),'running' FROM generate_series(1," <> number workflowRows <> ") n; ANALYZE keiro.keiro_workflows"+          )         execute url "UPDATE keiro.keiro_workflows SET last_error=repeat('x',65536) WHERE workflow_id=(SELECT workflow_id FROM keiro.keiro_workflows WHERE workflow_name='seed-0' ORDER BY workflow_id LIMIT 1)"         execute url "INSERT INTO keiro.keiro_timers(timer_id,process_manager_name,correlation_id,fire_at,payload,status) SELECT ('00000000-0000-0000-0000-'||lpad(n::text,12,'0'))::uuid,'seed-timer','corr-'||(n%100),'2100-01-01'::timestamptz+n*interval '1 second',CASE WHEN n=2 THEN to_jsonb(repeat('x',65536)) ELSE jsonb_build_object('n',n) END,CASE WHEN n%4=0 THEN 'fired' WHEN n%8=1 THEN 'firing' ELSE 'scheduled' END FROM generate_series(1,100000) n; ANALYZE keiro.keiro_timers"+        when ("projection-" `Text.isPrefixOf` mode) $ do+          execute url "INSERT INTO keiro.keiro_projection_rebuild_groups(group_id,slice_fingerprint,status) SELECT 'group-'||lpad(n::text,8,'0'),'fixture','live' FROM generate_series(1,10000) n"+          execute url "INSERT INTO keiro.keiro_projection_group_cursors(group_id,position_basis,subscription_names) SELECT group_id,CASE WHEN right(group_id,1)='0' THEN 'checkpoint' ELSE 'append' END,CASE WHEN right(group_id,1)='0' THEN ARRAY[group_id||'-sub'] ELSE ARRAY[]::text[] END FROM keiro.keiro_projection_rebuild_groups"+          execute url "INSERT INTO keiro.keiro_read_models(name,version,shape_hash,status,rebuild_group_id) SELECT group_id||'-model-'||n,1,'shape','live',group_id FROM keiro.keiro_projection_rebuild_groups CROSS JOIN generate_series(1,3) n; ANALYZE keiro.keiro_projection_rebuild_groups; ANALYZE keiro.keiro_projection_group_cursors; ANALYZE keiro.keiro_read_models"         withStore (defaultConnectionSettings url) {poolSize = 1} $ \store -> withMount store url mode $ \app -> do           _ <- runStoreIO store (registerReadModel "inspection-benchmark" 1 "shape") >>= right           settings <- use store (Session.statement () (Statement.preparable "SELECT jsonb_build_object('fsync',current_setting('fsync'),'synchronous_commit',current_setting('synchronous_commit'),'full_page_writes',current_setting('full_page_writes'),'autovacuum',current_setting('autovacuum'),'server_version',current_setting('server_version'))" E.noParams (D.singleRow (D.column (D.nonNullable D.jsonb)))))@@ -134,7 +146,7 @@           -- Both package sets execute this same explicit checkpoint; normal           -- durable settings and autovacuum stay enabled during measurement.           use store (Session.script "CHECKPOINT")-          forM_ [1 .. 50] $ \n -> operation store ("warm-" <> number n) >> pure ()+          forM_ [1 .. 50] $ \n -> operation countsWorkload store ("warm-" <> number n) >> pure ()           lifecycleBefore <- if retentionSeconds /= Nothing then Just <$> heapSample "mounted-before-pollers" 0 else pure Nothing           ((samples, statsBefore, started, before, writerFinished), pollSuccess, pollFailure) <- withPollers app mode duration $ \startPolling -> do             -- Let warm writer notifications settle in every mode, including@@ -151,11 +163,19 @@                   let due = maybe now (\rate -> started + fromIntegral (n - 1) * div 1000000000 (fromIntegral rate)) offeredRate                   when (due > now) (threadDelay (fromIntegral (div (due - now) 1000)))                   admitted <- getMonotonicTimeNSec-                  (schedule, claim, resume, timer, allocation) <- operation store ("measured-" <> Text.justifyRight 9 '0' (number (sampleCount - n)))+                  (schedule, claim, resume, timer, allocation) <- operation countsWorkload store ("measured-" <> Text.justifyRight 9 '0' (number n))                   let waiting = if offeredRate /= Nothing && admitted > due then fromIntegral (admitted - due) / 1000 else 0                   pure (waiting, schedule, claim, resume, timer, allocation)             samples <- case retentionSeconds of-              Nothing -> forM [1 .. sampleCount] measured+              Nothing -> case offeredRate of+                Just _ -> forM [1 .. sampleCount] measured+                Nothing -> do+                  let loop n acc = do+                        now <- getMonotonicTimeNSec+                        if now >= started + fromIntegral duration * 1000000000+                          then pure (reverse acc)+                          else do sample <- measured n; loop (n + 1) (sample : acc)+                  loop 1 []               Just _ -> withFile (output </> (show trial <> "-" <> Text.unpack mode <> "-heap.jsonl")) WriteMode $ \handle -> do                 let emit phase n = do                       value <- heapSample phase n@@ -191,9 +211,10 @@           let elapsed = fromIntegral (writerFinished - started) / 1e9 :: Double               cleanupSeconds = fromIntegral (finished - writerFinished) / 1e9 :: Double               durations = map (\(q, a, b, c, d, _) -> q + a + b + c + d) samples-              percentile p = if null durations then Nothing else Just (sort durations !! min (sampleCount - 1) (floor (p * fromIntegral sampleCount)))+              completedCount = if retentionSeconds == Nothing then length samples else sampleCount+              percentile p = if null durations then Nothing else Just (sort durations !! min (completedCount - 1) (floor (p * fromIntegral completedCount)))               (views, clients) = pollingShape mode-              artifact = object ["resources_after_clients" .= resourcesAfter, "trial" .= trial, "retention_seconds" .= retentionSeconds, "lifecycle_before" .= lifecycleBefore, "lifecycle_after" .= lifecycleAfter, "mode" .= mode, "views" .= views, "clients" .= clients, "poll_success" .= pollSuccess, "poll_failure" .= pollFailure, "writer_thread_allocated_bytes_per_operation" .= (if null samples then Nothing else Just (sum [allocated | (_, _, _, _, _, allocated) <- samples] / fromIntegral sampleCount)), "allocated_bytes" .= (statsAfter.allocated_bytes - statsBefore.allocated_bytes), "cpu_ns" .= (statsAfter.cpu_ns - statsBefore.cpu_ns), "settings" .= (settings :: Value), "offered_ops_per_second" .= offeredRate, "elapsed_seconds" .= elapsed, "client_cleanup_seconds" .= cleanupSeconds, "minor_and_major_gcs" .= (statsAfter.gcs - statsBefore.gcs), "major_gcs" .= (statsAfter.major_gcs - statsBefore.major_gcs), "gc_cpu_ns" .= (statsAfter.gc_cpu_ns - statsBefore.gc_cpu_ns), "gc_elapsed_ns" .= (statsAfter.gc_elapsed_ns - statsBefore.gc_elapsed_ns), "throughput" .= (fromIntegral sampleCount / elapsed), "p50_us" .= percentile 0.5, "p95_us" .= percentile 0.95, "p99_us" .= percentile 0.99, "before" .= before, "after" .= after, "samples_wait_schedule_claim_resume_timer_us_writer_bytes" .= samples]+              artifact = object ["workflow_rows" .= workflowRows, "counts_workload" .= countsWorkload, "measurement_seconds" .= duration, "completed_operations" .= completedCount, "resources_after_clients" .= resourcesAfter, "trial" .= trial, "retention_seconds" .= retentionSeconds, "lifecycle_before" .= lifecycleBefore, "lifecycle_after" .= lifecycleAfter, "mode" .= mode, "views" .= views, "clients" .= clients, "poll_success" .= pollSuccess, "poll_failure" .= pollFailure, "writer_thread_allocated_bytes_per_operation" .= (if null samples then Nothing else Just (sum [allocated | (_, _, _, _, _, allocated) <- samples] / fromIntegral completedCount)), "allocated_bytes" .= (statsAfter.allocated_bytes - statsBefore.allocated_bytes), "cpu_ns" .= (statsAfter.cpu_ns - statsBefore.cpu_ns), "settings" .= (settings :: Value), "offered_ops_per_second" .= offeredRate, "elapsed_seconds" .= elapsed, "client_cleanup_seconds" .= cleanupSeconds, "minor_and_major_gcs" .= (statsAfter.gcs - statsBefore.gcs), "major_gcs" .= (statsAfter.major_gcs - statsBefore.major_gcs), "gc_cpu_ns" .= (statsAfter.gc_cpu_ns - statsBefore.gc_cpu_ns), "gc_elapsed_ns" .= (statsAfter.gc_elapsed_ns - statsBefore.gc_elapsed_ns), "throughput" .= (fromIntegral completedCount / elapsed), "p50_us" .= percentile 0.5, "p95_us" .= percentile 0.95, "p99_us" .= percentile 0.99, "before" .= before, "after" .= after, "samples_wait_schedule_claim_resume_timer_us_writer_bytes" .= samples]               path = output </> (show trial <> "-" <> Text.unpack mode <> ".json")           Lazy.writeFile path (Aeson.encode artifact)           putStrLn path@@ -212,8 +233,8 @@  -- Returns schedule/claim/resume phase durations. The resume performs a real -- named step and completion, then releases its advance lease.-operation :: KirokuStore -> Text -> IO (Double, Double, Double, Double, Double)-operation store wid = do+operation :: Bool -> KirokuStore -> Text -> IO (Double, Double, Double, Double, Double)+operation countsWorkload store wid = do   let name = WorkflowName "index-evidence"       identity = WorkflowId wid       checked :: Eff '[Store, Error StoreError, IOE] a -> IO a@@ -221,6 +242,12 @@   allocationBefore <- getAllocationCounter   a <- getMonotonicTimeNSec   checked (runTransaction (Instance.upsertInstanceTx wid "index-evidence" 0 Instance.WfRunning Nothing))+  when countsWorkload $ do+    crash <- checked (runTransaction (Instance.recordCrashTx wid "index-evidence" "benchmark retry"))+    unless (crash == Just 1) (fail "missing retry transition")+    checked (Instance.resetInstanceAttempts name identity)+    checked (Instance.markInstanceSuspendedAwaiting name identity 0 "benchmark-await")+    checked (runTransaction (Instance.upsertInstanceTx wid "index-evidence" 0 Instance.WfRunning Nothing))   b <- getMonotonicTimeNSec   claim <- checked (Instance.claimInstance "index-evidence" 30 name identity)   unless (claim == Instance.ClaimAcquired) (fail (show claim))@@ -326,7 +353,7 @@     (views, clients) = pollingShape mode  inspectionModes :: [Text]-inspectionModes = ["disabled", "idle", "active-shared", "active-distinct", "slow-reader", "churn", "pm-types-idle", "pm-types", "pm-types-many-idle", "pm-types-many"]+inspectionModes = ["disabled", "idle", "active-shared", "active-distinct", "slow-reader", "churn", "pm-types-idle", "pm-types", "pm-types-many-idle", "pm-types-many", "projection-idle", "projection-poll", "projection-feed", "wf-counts", "wf-counts-many"]  #endif @@ -334,6 +361,10 @@ -- poll all views, while many clients can share one view's authoritative route. pollingShape :: Text -> (Int, Int) pollingShape = \case+  "wf-counts" -> (0, 1)+  "wf-counts-many" -> (0, 8)+  "projection-poll" -> (0, 4)+  "projection-feed" -> (1, 4)   "pm-types" -> (0, 4)   "pm-types-many" -> (0, 4)   "active-shared" -> (1, 1)
keiro-ops-http.cabal view
@@ -1,6 +1,6 @@ cabal-version: 3.0 name: keiro-ops-http-version: 0.23.0.0+version: 0.24.0.0 synopsis:   Bounded HTTP transport for the Keiro operational command tree @@ -57,6 +57,7 @@     Keiro.Ops.Http.Config     Keiro.Ops.Http.Cors     Keiro.Ops.Http.Feed+    Keiro.Ops.Http.Host     Keiro.Ops.Http.Pool     Keiro.Ops.Http.Protocol     Keiro.Ops.Http.Route@@ -78,9 +79,9 @@     hasql >=1.10 && <1.11,     hasql-pool >=1.2 && <1.5,     http-types >=0.12 && <0.13,-    keiro ^>=0.23.0.0,-    keiro-migrations ^>=0.23.0.0,-    keiro-ops ^>=0.23.0.0,+    keiro ^>=0.24.0.0,+    keiro-migrations ^>=0.24.0.0,+    keiro-ops ^>=0.24.0.0,     kiroku-store ^>=0.13.0.1,     network >=3.1 && <3.3,     network-uri >=2.6.4.2 && <2.7,@@ -104,15 +105,9 @@     -with-rtsopts=-N    build-depends:-    async >=2.2 && <2.3,     base >=4.21 && <5,-    hasql >=1.10 && <1.11,-    keiro-migrations ^>=0.23.0.0,-    keiro-ops ^>=0.23.0.0,+    keiro-ops ^>=0.24.0.0,     keiro-ops-http,-    kiroku-store ^>=0.13.0.1,-    optparse-applicative >=0.19 && <0.20,-    text >=2.1 && <2.2,  flag compose-example   description: Build the composed inspection example with the sibling metrics surface.@@ -133,7 +128,7 @@   build-depends:     async >=2.2 && <2.3,     base >=4.21 && <5,-    keiro-ops ^>=0.23.0.0,+    keiro-ops ^>=0.24.0.0,     keiro-ops-http,     kiroku-metrics ^>=0.4.0.1,     kiroku-store ^>=0.13.0.1,@@ -165,10 +160,10 @@     http-client >=0.7 && <0.8,     http-types >=0.12 && <0.13,     jitsurei,-    keiro ^>=0.23.0.0,-    keiro-ops ^>=0.23.0.0,+    keiro ^>=0.24.0.0,+    keiro-ops ^>=0.24.0.0,     keiro-ops-http,-    keiro-test-support ^>=0.23.0.0,+    keiro-test-support ^>=0.24.0.0,     kiroku-store ^>=0.13.0.1,     process >=1.6 && <1.7,     text >=2.1 && <2.2,@@ -205,12 +200,12 @@     http-client >=0.7 && <0.8,     http-types >=0.12 && <0.13,     jitsurei,-    keiro ^>=0.23.0.0,-    keiro-migrations ^>=0.23.0.0,-    keiro-ops ^>=0.23.0.0,+    keiro ^>=0.24.0.0,+    keiro-migrations ^>=0.24.0.0,+    keiro-ops ^>=0.24.0.0,     keiro-ops-http,-    keiro-pgmq ^>=0.23.0.0,-    keiro-test-support ^>=0.23.0.0,+    keiro-pgmq ^>=0.24.0.0,+    keiro-test-support ^>=0.24.0.0,     kiroku-metrics ^>=0.4.0.1,     kiroku-store ^>=0.13.0.1,     network >=3.1 && <3.3,
src/Keiro/Ops/Http.hs view
@@ -7,6 +7,7 @@     errorResponse,     module Keiro.Ops.Http.Config,     module Keiro.Ops.Http.Server,+    module Keiro.Ops.Http.Host,     withOpsInspectionPool,     corsMiddleware,   )@@ -16,6 +17,7 @@ import Keiro.Ops.Http.Compose import Keiro.Ops.Http.Config import Keiro.Ops.Http.Cors+import Keiro.Ops.Http.Host import Keiro.Ops.Http.Pool import Keiro.Ops.Http.Server
+ src/Keiro/Ops/Http/Host.hs view
@@ -0,0 +1,78 @@+-- | Shared lifecycle and command-line options for standalone inspection hosts.+module Keiro.Ops.Http.Host+  ( HostOptions (..),+    hostOptionsParser,+    withHost,+    mainWithHooks,+  )+where++import Control.Concurrent.Async (wait)+import Data.Foldable (traverse_)+import Data.Text (Text)+import Data.Text qualified as Text+import Data.Text.IO qualified as Text+import Hasql.Connection.Settings qualified as Settings+import Keiro.Migrations.SchemaCheck+import Keiro.Ops (AppHooks)+import Keiro.Ops.Env (resolveConnectionString)+import Keiro.Ops.Http.Config+import Keiro.Ops.Http.Config qualified as Config+import Keiro.Ops.Http.Feed+import Keiro.Ops.Http.Pool+import Keiro.Ops.Http.Server+import Keiro.Ops.Http.WebSocket (opsApplicationWithFeeds)+import Kiroku.Store.Connection (ConnectionSettingsM (..), KirokuStore (..), defaultConnectionSettings, withStore)+import Options.Applicative+import System.Environment (getProgName)+import System.IO (stderr)++-- | Database selection, request policy, and listener settings for one host.+data HostOptions = HostOptions+  { databaseUrl :: !(Maybe Text),+    http :: !OpsHttpConfig,+    server :: !OpsHttpServerConfig+  }++-- | The common flag surface used by database-only and application hosts.+hostOptionsParser :: Parser HostOptions+hostOptionsParser =+  HostOptions+    <$> optional (textOption (long "database-url" <> metavar "URL"))+    <*> ( (\enabled drift origins -> defaultOpsHttpConfig {mutations = if enabled then MutationsEnabled else MutationsDisabled, schemaDrift = if drift then AllowDrift else RefuseOnDrift, Config.allowedOrigins = origins})+            <$> switch (long "enable-mutations")+            <*> switch (long "allow-schema-drift")+            <*> many (textOption (long "allow-origin" <> metavar "ORIGIN"))+        )+    <*> (OpsHttpServerConfig <$> textOption (long "bind" <> value "127.0.0.1") <*> option auto (long "port" <> value 9092) <*> option auto (long "request-timeout" <> value 300))+  where+    textOption modifiers = Text.pack <$> strOption modifiers++-- | Run the command-line host with the application's operational hooks.+mainWithHooks :: AppHooks -> IO ()+mainWithHooks hooks = do+  opts <- execParser (info (hostOptionsParser <**> helper) (fullDesc <> progDesc "Serve Keiro inspection on a trusted network or behind an authenticating proxy"))+  name <- Text.pack <$> getProgName+  withHost opts hooks $ \running -> do+    Text.hPutStrLn stderr (name <> ": listening on " <> opts.server.bindHost <> ":" <> Text.pack (show running.serverPort) <> " (mutations " <> (if opts.http.mutations == MutationsEnabled then "enabled" else "disabled") <> ")")+    wait running.serverThread++-- | Allocate the inspection store, feeds, and listener for the continuation.+-- All resources are released before returning, including on exceptions.+-- Port zero selects a free port reported by 'OpsHttpServer'.+withHost :: HostOptions -> AppHooks -> (OpsHttpServer -> IO a) -> IO a+withHost opts hooks action = do+  let config = opts.http+  connectionString <- resolveConnectionString opts.databaseUrl+  verified <- verifyExpectedSchema (Settings.connectionString connectionString)+  case verified of+    Left err -> ioError (userError (show err))+    Right drift -> traverse_ (Text.hPutStrLn stderr . ("warning: " <>) . renderSchemaDrift) drift+  -- This store belongs only to inspection. All its pooled connections inherit+  -- the SQL deadline; command execution never borrows the runtime's pool.+  let settings = (defaultConnectionSettings connectionString) {poolSize = config.inspectionLimits.maxConcurrentReads, statementTimeout = Just 2}+  withStore settings $ \baseStore -> withOpsInspectionPool config.inspectionLimits settings $ \pool -> do+    let store = baseStore {pool}+    withFeedHandle config defaultFeedConfig store $ \feeds -> do+      app <- opsApplicationWithFeeds config feeds hooks store+      withOpsHttpServer opts.server app action
src/Keiro/Ops/Http/Pool.hs view
@@ -15,7 +15,8 @@ -- | Create an inspection-owned pool, preserving the host's connection string -- and search path. Substitute this pool on the store passed to the HTTP/feed -- surface. Every connection gets the same statement timeout, including after--- cancellation/reconnection. No writer connection is modified.+-- cancellation/reconnection. Queries use one backend each so concurrent reads+-- do not multiply into parallel worker groups. No writer connection is modified. withOpsInspectionPool :: InspectionLimits -> ConnectionSettings -> (Pool -> IO a) -> IO a withOpsInspectionPool limits settings action = do   _ <- newInspectionBudget limits@@ -25,7 +26,7 @@     initScript =       "SET search_path TO "         <> Text.intercalate ", " (map quote (settings.schema : settings.extraSearchPath))-        <> ", pg_catalog; SET statement_timeout = '"+        <> ", pg_catalog; SET max_parallel_workers_per_gather = 0; SET statement_timeout = '"         <> Text.pack (show limits.statementTimeoutMillis)         <> "ms'; SET idle_in_transaction_session_timeout = '"         <> Text.pack (show settings.idleInTransactionTimeout)
src/Keiro/Ops/Http/Views.hs view
@@ -26,7 +26,7 @@ import Effectful.Error.Static (Error) import Effectful.Exception (throwIO) import Keiro.Ops.Http.Protocol (FeedName (..), feedNameText)-import Keiro.Ops.Projection (projectionGroupStatusJson)+import Keiro.Ops.Projection (projectionGroupStatusRowJson) import Keiro.Projection.Catalog (rebuildGroupIdText) import Keiro.ReadModel (storeHeadPosition) import Keiro.ReadModel.Rebuild@@ -132,7 +132,7 @@         GlobalPosition visibleHead <- storeHeadPosition         pure           ( page.observedAt,-            [let groupId = rebuildGroupIdText row.groupId in Item groupId (object ["group_id" .= groupId]) (projectionGroupStatusJson row) | row <- page.items],+            [let groupId = rebuildGroupIdText row.status.groupId in Item groupId (object ["group_id" .= groupId]) (projectionGroupStatusRowJson row) | row <- page.items],             Just (object ["visible_store_head" .= visibleHead])           )     )
test/Main.hs view
@@ -29,6 +29,8 @@ import Hasql.Session qualified as Session import Hasql.Statement qualified as Statement import Jitsurei.Inspection (jitsureiProcessManagerInspectors)+import Jitsurei.ReadModels (orderCatalogOperations)+import Keiro.Inspection.Consumers qualified as Consumers import Keiro.Ops qualified as Ops import Keiro.Ops.Env import Keiro.Ops.Http@@ -47,9 +49,11 @@ import Keiro.PGMQ import Keiro.Projection.Catalog qualified as Catalog import Keiro.Projection.Catalog.Operations qualified as CatalogOperations+import Keiro.Snapshot.Schema qualified as Snapshot import Keiro.Test.Postgres import Keiro.Timer qualified as Timer import Keiro.Workflow (WorkflowId (..), WorkflowJournalEvent (..), WorkflowName (..), appendJournalEntry, continueAsNew, runWorkflow)+import Keiro.Workflow.Inspection qualified as Inspection import Keiro.Workflow.Resume (defaultWorkflowResumeOptions) import Kiroku.Store.Append (appendToStream) import Kiroku.Store.Connection@@ -162,6 +166,144 @@       forM_ [0, -1, maxBound] $ \seconds ->         newInspectionBudget defaultInspectionLimits {deliveryDeadlineSeconds = seconds} `shouldThrow` anyIOException   around (withFreshStore fixture) $ do+    describe "raw snapshot routes" $ do+      it "returns null for absent rows, preserves arbitrary incompatible JSON and reads without mutations" $ \store -> do+        app <- opsApplication defaultOpsHttpConfig Ops.emptyAppHooks store+        Right appended <- runStoreIO store (appendToStream (StreamName "snapshot-http") AnyVersion [EventData Nothing (EventType "test") Null Nothing Nothing Nothing])+        forM_ ["missing", "snapshot-http"] $ \name -> do+          absent <- call app methodGet ("/snapshot/show?stream=" <> name) ""+          simpleStatus absent `shouldBe` status200+          body absent `shouldBe` Null+        let payload = object ["custom" .= [object ["balance" .= (25 :: Int), "active" .= False]], "legacy" .= True]+        Right () <- runStoreIO store (Snapshot.writeSnapshotRow (Snapshot.SnapshotWrite appended.streamId appended.streamVersion payload 999 "old-registers" "old-fold"))+        Right before <- runStoreIO store (Snapshot.lookupSnapshotRow appended.streamId)+        Right streamBefore <- runStoreIO store (getStream (StreamName "snapshot-http"))+        shown <- call app methodGet "/snapshot/show?stream=snapshot-http" ""+        simpleStatus shown `shouldBe` status200+        field "state" (body shown) `shouldBe` Just payload+        field "state_codec_version" (body shown) `shouldBe` Just (Aeson.toJSON (999 :: Int))+        field "regfile_shape_hash" (body shown) `shouldBe` Just (String "old-registers")+        field "state_shape_hash" (body shown) `shouldBe` Just (String "old-fold")+        field "compatibility" (body shown) `shouldBe` Nothing+        refused <- call app methodPost "/snapshot/show?stream=snapshot-http" ""+        assertError refused status405 "method_not_allowed"+        runStoreIO store (Snapshot.lookupSnapshotRow appended.streamId) `shouldReturn` Right before+        runStoreIO store (getStream (StreamName "snapshot-http")) `shouldReturn` Right streamBefore++      it "bounds stored payloads and releases the only admission slot after oversized encoding" $ \store -> do+        Right appended <- runStoreIO store (appendToStream (StreamName "snapshot-large") AnyVersion [EventData Nothing (EventType "test") Null Nothing Nothing Nothing])+        Right () <- runStoreIO store (Snapshot.writeSnapshotRow (Snapshot.SnapshotWrite appended.streamId appended.streamVersion (String (Text.replicate 4096 "x")) 1 "regs" "fold"))+        Right before <- runStoreIO store (Snapshot.lookupSnapshotRow appended.streamId)+        Right streamBefore <- runStoreIO store (getStream (StreamName "snapshot-large"))+        let config = defaultOpsHttpConfig {inspectionLimits = defaultInspectionLimits {maxResponseBytes = 1024, maxConcurrentReads = 1}}+        app <- opsApplication config Ops.emptyAppHooks store+        oversized <- call app methodGet "/snapshot/show?stream=snapshot-large" ""+        assertError oversized status413 "inspection_result_too_large"+        health <- call app methodGet "/health" ""+        simpleStatus health `shouldBe` status200+        small <- call app methodGet "/snapshot/show?stream=missing" ""+        simpleStatus small `shouldBe` status200+        body small `shouldBe` Null+        runStoreIO store (Snapshot.lookupSnapshotRow appended.streamId) `shouldReturn` Right before+        runStoreIO store (getStream (StreamName "snapshot-large")) `shouldReturn` Right streamBefore++    describe "category consumers HTTP" $ do+      it "serves the real catalog before checkpoints and agrees with the CLI" $ \store -> do+        let mounted = Ops.emptyAppHooks {Ops.projectionCatalog = Just orderCatalogOperations}+        app <- opsApplication defaultOpsHttpConfig mounted store+        forM_ ["order", "creditLimit", "ir59-unused-category"] $ \category -> do+          response <- call app methodGet ("/projection/consumers?category=" <> category) ""+          simpleStatus response `shouldBe` status200+          case parseArguments mounted ["projection", "consumers", "--category", category] of+            Parsed command -> do+              Succeeded result <- Ops.runCommand mounted (OpsEnv store Json False [] False) command+              body response `shouldBe` result.jsonValue+            _ -> expectationFailure "CLI consumers did not parse"+          field "coverage" (body response) `shouldBe` Just (object ["projection_catalog" .= True, "durable_subscribers" .= False])+        matching <- call app methodGet "/projection/consumers?category=order" ""+        let source = object ["kind" .= ("category" :: Text), "category" .= ("order" :: Text)]+            binding projection handler subscription =+              object+                [ "projection_id" .= (projection :: Text),+                  "handler_name" .= (handler :: Text),+                  "delivery" .= (if isJust subscription then "subscription" else "inline" :: Text),+                  "subscription_id" .= subscription,+                  "subscription_name" .= (subscription :: Maybe Text),+                  "source" .= source+                ]+            expectedGroup =+              object+                [ "kind" .= ("projection_group" :: Text),+                  "name" .= ("jitsurei-order-reporting" :: Text),+                  "group_id" .= ("jitsurei-order-reporting" :: Text),+                  "bindings" .= [binding "jitsurei-order-audit" "jitsurei-order-audit-async" (Just "jitsurei-order-audit-subscription"), binding "jitsurei-order-summary" "jitsurei-order-summary-inline" Nothing]+                ]+        field "items" (body matching) `shouldBe` Just (Aeson.toJSON [expectedGroup])+        empty <- call app methodGet "/projection/consumers?category=creditLimit" ""+        field "items" (body empty) `shouldBe` Just (Aeson.toJSON ([] :: [Value]))+        post <- call app methodPost "/projection/consumers?category=order" ""+        assertError post status405 "method_not_allowed"+      it "validates categories, limits and cursors and distinguishes absent and empty hooks" $ \store -> do+        absent <- opsApplication defaultOpsHttpConfig Ops.emptyAppHooks store+        refused <- call absent methodGet "/projection/consumers?category=order" ""+        assertError refused status400 "invalid_request"+        Right registry <- pure (Consumers.durableConsumerRegistry [])+        app <- opsApplication defaultOpsHttpConfig Ops.emptyAppHooks {Ops.durableConsumers = Just registry} store+        empty <- call app methodGet "/projection/consumers?category=order" ""+        simpleStatus empty `shouldBe` status200+        field "items" (body empty) `shouldBe` Just (Aeson.toJSON ([] :: [Value]))+        forM_ [("category=", "invalid_category"), ("category=%20order", "invalid_category"), ("category=order&limit=501", "invalid_page_size"), ("category=order&after=bad", "invalid_cursor")] $ \(query, code) -> do+          invalid <- call app methodGet ("/projection/consumers?" <> query) ""+          assertError invalid status400 code+        oversizedCursor <- call app methodGet ("/projection/consumers?category=order&after=" <> Text.replicate 4097 "a") ""+        simpleStatus oversizedCursor `shouldBe` status400+        zero <- call app methodGet "/projection/consumers?category=order&limit=0" ""+        simpleStatus zero `shouldBe` status400+      it "pages durable bindings through HTTP and rejects cross-category or stale continuations" $ \store -> do+        let description name = Consumers.DurableConsumerDescription Consumers.DurableSubscriber name ("sub-" <> name) Consumers.AllCategories+        Right registry <- pure (Consumers.durableConsumerRegistry (map description ["a", "b", "c"]))+        let mounted = Ops.emptyAppHooks {Ops.durableConsumers = Just registry}+        app <- opsApplication defaultOpsHttpConfig mounted store+        first <- call app methodGet "/projection/consumers?category=order&limit=1" ""+        simpleStatus first `shouldBe` status200+        Just (String cursor) <- pure (field "next_cursor" (body first))+        second <- call app methodGet ("/projection/consumers?category=order&limit=1&after=" <> cursor) ""+        simpleStatus second `shouldBe` status200+        body second `shouldNotBe` body first+        wrong <- call app methodGet ("/projection/consumers?category=invoice&after=" <> cursor) ""+        assertError wrong status400 "invalid_cursor"+        Right changed <- pure (Consumers.durableConsumerRegistry [description "changed"])+        changedApp <- opsApplication defaultOpsHttpConfig mounted {Ops.durableConsumers = Just changed} store+        stale <- call changedApp methodGet ("/projection/consumers?category=order&after=" <> cursor) ""+        assertError stale status400 "invalid_cursor"+      it "orders combined sources and paginates across the source boundary" $ \store -> do+        Right registry <-+          pure+            ( Consumers.durableConsumerRegistry+                [ Consumers.DurableConsumerDescription Consumers.ProcessManagerConsumer "fulfillment" "orders" (Consumers.ExactCategory "order"),+                  Consumers.DurableConsumerDescription Consumers.DurableSubscriber "audit" "auditor" Consumers.AllCategories+                ]+            )+        let mounted = Ops.emptyAppHooks {Ops.durableConsumers = Just registry, Ops.projectionCatalog = Just orderCatalogOperations}+        app <- opsApplication defaultOpsHttpConfig mounted store+        first <- call app methodGet "/projection/consumers?category=order&limit=2" ""+        simpleStatus first `shouldBe` status200+        Just (Array items) <- pure (field "items" (body first))+        map (field "kind") (foldr (:) [] items) `shouldBe` [Just (String "durable_subscriber"), Just (String "process_manager")]+        Just (String cursor) <- pure (field "next_cursor" (body first))+        last <- call app methodGet ("/projection/consumers?category=order&limit=2&after=" <> cursor) ""+        Just (Array rest) <- pure (field "items" (body last))+        map (field "kind") (foldr (:) [] rest) `shouldBe` [Just (String "projection_group")]+        field "next_cursor" (body last) `shouldBe` Nothing+      it "bounds an oversized single item without truncating bindings and keeps schema refusal" $ \store -> do+        let description = Consumers.DurableConsumerDescription Consumers.DurableSubscriber "large" (Text.replicate 1100000 "s") (Consumers.ExactCategory "order")+        Right registry <- pure (Consumers.durableConsumerRegistry [description])+        app <- opsApplication defaultOpsHttpConfig Ops.emptyAppHooks {Ops.durableConsumers = Just registry} store+        oversized <- call app methodGet "/projection/consumers?category=order&limit=1" ""+        assertError oversized status413 "inspection_result_too_large"+        Right () <- Pool.use store.pool (Session.script "ALTER TABLE keiro.keiro_timers ADD COLUMN consumers_drift text")+        drift <- call app methodGet "/projection/consumers?category=order" ""+        assertError drift status409 "schema_drift"     describe "mounted process-manager discovery" $ do       it "serves Jitsurei metadata without journal instances and agrees with the parsed CLI" $ \store -> do         let mounted = Ops.emptyAppHooks {Ops.processManagers = Just jitsureiProcessManagerInspectors}@@ -200,6 +342,38 @@         Right () <- Pool.use store.pool (Session.script "ALTER TABLE keiro.keiro_timers ADD COLUMN pm_types_drift text")         response <- call app methodGet "/pm/types" ""         assertError response status409 "schema_drift"+    describe "workflow counts routes" $ do+      it "agrees with the owner read, translates exact names, and returns zeroes" $ \store -> do+        seedInspection store+        Right () <- Pool.use store.pool (Session.script "INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status) VALUES ('special','支払_%''\\ λ','failed')")+        app <- opsApplication defaultOpsHttpConfig hooks store+        forM_ [Nothing, Just "approval", Just "missing", Just "支払_%'\\ λ"] $ \name -> do+          expected <- runStoreIO store (Inspection.countWorkflowInstancesByStatus (Inspection.WorkflowCountsQuery name)) >>= either (fail . show) pure+          let query = maybe "" (\n -> "?workflow_name=" <> Text.decodeUtf8 (urlEncode True (Text.encodeUtf8 n))) name+          response <- call app methodGet ("/wf/inspect/counts" <> query) ""+          simpleStatus response `shouldBe` status200+          field "counts" (body response) `shouldBe` Just (Aeson.toJSON expected.counts)+          field "observed_at" (body response) `shouldSatisfy` maybe False (\case String _ -> True; _ -> False)+        rejected <- call app methodGet "/wf/inspect/counts?limit=1" ""+        assertError rejected status400 "invalid_request"+        Right () <- Pool.use store.pool (Session.script "ALTER TABLE keiro.keiro_timers ADD COLUMN counts_drift text")+        drift <- call app methodGet "/wf/inspect/counts" ""+        assertError drift status409 "schema_drift"+      it "shares admission and returns its permit after a deadline" $ \store -> do+        entered <- newEmptyMVar+        let config = defaultOpsHttpConfig {inspectionLimits = defaultInspectionLimits {maxConcurrentReads = 1, readDeadlineSeconds = 1}}+            operations = OpsHttpOperations (putMVar entered () >> threadDelay 3000000 >> pure (Right [])) (\_ _ -> pure (Succeeded emptyResult))+        budget <- newInspectionBudget config.inspectionLimits+        app <- opsApplicationWithOperations config budget Ops.emptyAppHooks store operations+        withAsync (call app methodGet "/wf/inspect/counts" "") $ \job -> do+          takeMVar entered+          refused <- call app methodGet "/wf/inspect/counts" ""+          assertError refused status503 "inspection_busy"+          expired <- wait job+          assertError expired status504 "inspection_timeout"+        healthy <- opsApplicationWithBudget config budget Ops.emptyAppHooks store+        recovered <- call healthy methodGet "/wf/inspect/counts" ""+        simpleStatus recovered `shouldBe` status200     describe "workflow page routes" $ do       it "rejects an unrepresentable outgoing cursor instead of returning an unusable token" $ \store -> do         result <- Pool.use store.pool (Session.script "INSERT INTO keiro.keiro_workflows(workflow_id,workflow_name,status) VALUES ('a',repeat('界',2000),'running'),('b',repeat('界',2000),'running')")@@ -600,6 +774,30 @@                 Just (Right bytes) | Char8.null bytes -> pure ()                 _ -> expectationFailure ("stopped runner retained an idle connection on iteration " <> show iteration)   around (withFreshDatabase fixture) $ do+    describe "category consumers database boundary" $ do+      it "rejects invalid input before borrowing and uses only ordinary schema verification" $ \url ->+        withStore (defaultConnectionSettings url) $ \store -> do+          calls <- newIORef (0 :: Int)+          let observe = \case Observation.ConnectionObservation _ Observation.InUseConnectionStatus -> atomicModifyIORef' calls (\n -> (n + 1, ())); _ -> pure ()+              settings = (defaultConnectionSettings url) {observationHandler = Just observe}+              mounted = Ops.emptyAppHooks {Ops.projectionCatalog = Just orderCatalogOperations, Ops.aggregates = Just Map.empty}+          withOpsInspectionPool defaultInspectionLimits settings $ \pool -> do+            app <- opsApplication defaultOpsHttpConfig mounted store {pool}+            before <- readIORef calls+            invalid <- call app methodGet "/projection/consumers?category=order&after=bad" ""+            assertError invalid status400 "invalid_cursor"+            readIORef calls `shouldReturn` before+            aggregate <- call app methodGet "/aggregate/types" ""+            simpleStatus aggregate `shouldBe` status200+            afterAggregate <- readIORef calls+            consumers <- call app methodGet "/projection/consumers?category=order" ""+            simpleStatus consumers `shouldBe` status200+            afterConsumers <- readIORef calls+            afterConsumers - afterAggregate `shouldBe` afterAggregate - before+            afterAggregate - before `shouldBe` 1+            repeated <- call app methodGet "/projection/consumers?category=order" ""+            body repeated `shouldBe` body consumers+            readIORef calls `shouldReturn` (afterConsumers + 1)     describe "process-manager discovery database boundary" $ do       it "performs only aggregate-types verifier work and rejects parsing before borrowing" $ \url ->         withStore (defaultConnectionSettings url) $ \store -> do@@ -642,9 +840,25 @@             -- One schema-verification session and one instance-read session.             readIORef calls `shouldReturn` 2     describe "schema handshake" $ do+      it "cancels a blocked count statement and recovers the pool and admission permit" $ \url ->+        withStore (defaultConnectionSettings url) $ \store -> do+          let limits = defaultInspectionLimits {statementTimeoutMillis = 100, maxConcurrentReads = 1}+              config = defaultOpsHttpConfig {inspectionLimits = limits}+          withOpsInspectionPool limits (defaultConnectionSettings url) $ \pool -> do+            app <- opsApplication config hooks store {pool}+            bracket (Connection.acquire (Settings.connectionString url) >>= either (fail . show) pure) Connection.release $ \control -> do+              Right () <- Connection.use control (Session.script "BEGIN; LOCK TABLE keiro.keiro_workflows IN ACCESS EXCLUSIVE MODE")+              expired <- call app methodGet "/wf/inspect/counts" ""+              simpleStatus expired `shouldSatisfy` (\status -> status /= status200)+              field "counts" (body expired) `shouldBe` Nothing+              Right () <- Connection.use control (Session.script "ROLLBACK")+              recovered <- call app methodGet "/wf/inspect/counts" ""+              simpleStatus recovered `shouldBe` status200+       it "bounds SQL on every inspection connection and releases pool capacity on cancellation" $ \url -> do         let limits = defaultInspectionLimits {statementTimeoutMillis = 100, maxConcurrentReads = 1}         withOpsInspectionPool limits (defaultConnectionSettings url) $ \pool -> do+          Right () <- Pool.use pool (Session.script "DO $$ BEGIN IF current_setting('max_parallel_workers_per_gather') <> '0' THEN RAISE EXCEPTION 'inspection parallel workers are enabled'; END IF; END $$")           slow <- Pool.use pool (Session.script "SELECT pg_sleep(1)")           slow `shouldSatisfy` either (const True) (const False)           result <- timeout 1000000 (Pool.use pool (Session.script "SELECT 1"))@@ -725,7 +939,9 @@  cliReads :: [(Text, [String])] cliReads =-  [ ("/wf/list", ["wf", "list"]),+  [ ("/wf/inspect/counts", ["wf", "inspect", "counts"]),+    ("/wf/inspect/counts?workflow_name=approval", ["wf", "inspect", "counts", "--workflow-name", "approval"]),+    ("/wf/list", ["wf", "list"]),     ("/timer/pending/list?limit=1", ["timer", "pending", "list", "--limit", "1"]),     ("/shard/list", ["shard", "list"]),     ("/timer/stuck/list?min_age=0s", ["timer", "stuck", "list", "--min-age", "0s"]),
test/Test/WebSocketSpec.hs view
@@ -16,7 +16,7 @@ import Hasql.Session qualified as Session import Keiro.Ops (emptyAppHooks) import Keiro.Ops.Http hiding (subscribe)-import Keiro.Ops.Projection (projectionGroupStatusJson)+import Keiro.Ops.Projection (projectionGroupStatusRowJson) import Keiro.ReadModel (registerReadModel) import Keiro.ReadModel.Rebuild import Keiro.Test.Postgres (Fixture, withFreshStore)@@ -97,7 +97,7 @@         projection <- receive connection         field "feed" projection `shouldBe` Just (String "projection_groups")         Right page <- stored store (listProjectionGroupStatusesPage (ProjectionStatusPageRequest 100 Nothing))-        field "changed" projection `shouldBe` Just (Aeson.toJSON (map projectionGroupStatusJson page.items))+        field "changed" projection `shouldBe` Just (Aeson.toJSON (map projectionGroupStatusRowJson page.items))         void (stored store (appendJournalEntry (WorkflowName "head") (WorkflowId "one") (StepRecorded "head" Null now)))         headOnly <- receive connection         field "feed" headOnly `shouldBe` Just (String "projection_groups")