diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/app/Main.hs b/app/Main.hs --- a/app/Main.hs +++ b/app/Main.hs @@ -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 diff --git a/bench/InspectionClients.hs b/bench/InspectionClients.hs --- a/bench/InspectionClients.hs +++ b/bench/InspectionClients.hs @@ -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)) diff --git a/bench/InspectionInterference.hs b/bench/InspectionInterference.hs --- a/bench/InspectionInterference.hs +++ b/bench/InspectionInterference.hs @@ -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) diff --git a/keiro-ops-http.cabal b/keiro-ops-http.cabal --- a/keiro-ops-http.cabal +++ b/keiro-ops-http.cabal @@ -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, diff --git a/src/Keiro/Ops/Http.hs b/src/Keiro/Ops/Http.hs --- a/src/Keiro/Ops/Http.hs +++ b/src/Keiro/Ops/Http.hs @@ -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 diff --git a/src/Keiro/Ops/Http/Host.hs b/src/Keiro/Ops/Http/Host.hs new file mode 100644 --- /dev/null +++ b/src/Keiro/Ops/Http/Host.hs @@ -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 diff --git a/src/Keiro/Ops/Http/Pool.hs b/src/Keiro/Ops/Http/Pool.hs --- a/src/Keiro/Ops/Http/Pool.hs +++ b/src/Keiro/Ops/Http/Pool.hs @@ -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) diff --git a/src/Keiro/Ops/Http/Views.hs b/src/Keiro/Ops/Http/Views.hs --- a/src/Keiro/Ops/Http/Views.hs +++ b/src/Keiro/Ops/Http/Views.hs @@ -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]) ) ) diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -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"]), diff --git a/test/Test/WebSocketSpec.hs b/test/Test/WebSocketSpec.hs --- a/test/Test/WebSocketSpec.hs +++ b/test/Test/WebSocketSpec.hs @@ -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")