diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,57 @@ ## [Unreleased] +## 0.22.0.0 — 2026-10-27 + +### Breaking Changes + +- Require the Kiroku Store 0.12 cohort from + mori://shinzui/kiroku/packages/kiroku-store; apply migration 0016 from + mori://shinzui/kiroku/packages/kiroku-store-migrations with affected workers + stopped, then restart new binaries before using namespace targets. Update + exhaustive subscription target/event matches and low-level record constructors. + +- Direct `AppHooks` records must supply `aggregates` and `processManagers`; `emptyAppHooks` leaves them disabled. `OpsOutcome` adds `InspectionFailed`, requiring exhaustive consumers to handle structured codes. +- `wf show/steps/journal` return `InspectionFailed` for typed inspection errors. `pgmq dlq read` requires `--force` to reserve rows; an unforced invocation previews and exits 1. + +### New Features + +- Add embedded `pm types` with shared JSON and human metadata output, including + empty catalogs and aliases, without requiring a manager name or store reads. + Hosts may supply a bounded direct encoder through `runCommandWithReadEncoding`. + +- Add bounded projection-status pages and `projection status`, preserving the + frozen v1 relation and complete-list compatibility API. Page keys precede + correlated status reads; no writer index is added. HTTP feed protocol and + canonical view definitions are available as foundations for live feeds. + +- Add process-manager persisted-state inspection and bounded raw stream listing, + pending-timer pages with filter-bound cursors, and database-clock shard + ownership/listing. Reads add no worker queries, polling tasks or migrations. + Embedded `pm show/list` uses the additive `AppHooks.processManagers` field; + `emptyAppHooks` leaves this optional catalog off. + + +- An optional `AppHooks.aggregates` catalog mounts read-only `aggregate types|list|show|snapshot` commands. Snapshot show adds `observed_at` and `age_seconds`. + +- Bounded header presentation reuse through an alternative runner that checks the exact current row before returning prepared JSON bytes. + +- Bounded `wf inspect` commands and typed inspection errors; shared workflow rendering and paged compatibility reads. + + +- Export the authoritative command grammar, classifier, runner, and invocation fields for alternative front ends. + + +- `pm show/list` accepts reaction-state descriptors with explicitly unavailable provenance. Capture and `legacy-positional-v1` remain legacy positional features scheduled for retirement before 1.0. +- Workflow list/show JSON includes `status_class`; step rows include `recorded_at`. Show uses owner-table creation-key ordering for children and awakeables. Shard status includes database `observed_at`, `heartbeat_at`, `updated_at` and `lease_age_seconds`. + +### Bug Fixes + +- Treat `pgmq dlq read` as a mutation because it reserves messages for 30 seconds. It now previews by default and requires `--force` to read; the HTTP surface uses POST with explicit confirmation. + + +- Default workflow inspection consumes primary-key `wf-key2` pages; old default tokens fail closed and the obsolete exceptional-key refusal is removed. Filtered and creation-order reads retain host deadlines. + ## 0.21.0.0 — 2026-10-14 ### Bug Fixes diff --git a/keiro-ops.cabal b/keiro-ops.cabal --- a/keiro-ops.cabal +++ b/keiro-ops.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: keiro-ops -version: 0.21.0.0 +version: 0.22.0.0 synopsis: Operational command-line interface for Keiro deployments description: A standalone database-only console and embeddable command tree for inspecting @@ -49,12 +49,14 @@ hs-source-dirs: src exposed-modules: Keiro.Ops + Keiro.Ops.Aggregate Keiro.Ops.Embed Keiro.Ops.Env Keiro.Ops.Inbox Keiro.Ops.Outbox Keiro.Ops.Parse Keiro.Ops.Pgmq + Keiro.Ops.ProcessManager Keiro.Ops.Projection Keiro.Ops.Rebuild Keiro.Ops.Render @@ -64,6 +66,7 @@ Keiro.Ops.Stream Keiro.Ops.Timer Keiro.Ops.Workflow + Keiro.Ops.WorkflowHeaderCache build-depends: aeson >=2.2.2 && <2.3, @@ -73,10 +76,11 @@ effectful >=2.6 && <2.8, effectful-core >=2.6 && <2.7 || >=2.7.1.1 && <2.8, hasql >=1.10 && <1.11, - keiro ^>=0.21.0.0, - keiro-migrations ^>=0.21.0.0, - keiro-pgmq ^>=0.21.0.0, - kiroku-store ^>=0.11.0.0, + hasql-pool >=1.4 && <1.5, + keiro ^>=0.22.0.0, + keiro-migrations ^>=0.22.0.0, + keiro-pgmq ^>=0.22.0.0, + kiroku-store ^>=0.12.0.0, optparse-applicative >=0.18 && <0.20, text >=2.1 && <2.2, time >=1.12 && <1.15, @@ -106,6 +110,7 @@ -rtsopts -with-rtsopts=-N + other-modules: ReactionInspectionSpec build-tool-depends: keiro-ops:keiro-ops build-depends: aeson >=2.2 && <2.3, @@ -117,13 +122,14 @@ hasql >=1.10 && <1.11, hasql-transaction >=1.1 && <1.3, hspec >=2.11, - keiro ^>=0.21.0.0, + keiki >=0.9.1 && <0.10, + keiro ^>=0.22.0.0, keiro-ops, - keiro-pgmq ^>=0.21.0.0, - keiro-test-support ^>=0.21.0.0, - kiroku-store ^>=0.11.0.0, + keiro-pgmq ^>=0.22.0.0, + keiro-test-support ^>=0.22.0.0, + kiroku-store ^>=0.12.0.0, optparse-applicative >=0.18 && <0.20, - pgmq-migration >=0.6.1.1 && <0.7, + pgmq-migration >=0.7.0.0 && <0.8, process >=1.6 && <1.7, text >=2.1 && <2.2, time >=1.12 && <1.15, diff --git a/src/Keiro/Ops.hs b/src/Keiro/Ops.hs --- a/src/Keiro/Ops.hs +++ b/src/Keiro/Ops.hs @@ -2,26 +2,39 @@ ( main, mainWithHooks, AppHooks (..), + AggregateInspectors, + ProcessManagerInspectors, OpsAuditConfig (..), emptyAppHooks, - OpsInvocation, + OpsInvocation (..), + Command (..), opsCommandTree, + commandParser, + validateCommand, + isMutation, + runCommand, + runCommandWithHeaderCache, + runCommandWithReadEncoding, runOpsInvocation, ) where import Control.Exception (SomeException, displayException, fromException, try) +import Data.Aeson.Encoding (Encoding) +import Data.ByteString (ByteString) import Data.Foldable (traverse_) import Data.Maybe (isJust) import Data.Text qualified as Text import Data.Text.IO qualified as Text.IO import Hasql.Connection.Settings qualified as Settings import Keiro.Migrations.SchemaCheck (renderSchemaDrift, verifyExpectedSchema) +import Keiro.Ops.Aggregate qualified as Aggregate import Keiro.Ops.Embed import Keiro.Ops.Env import Keiro.Ops.Inbox qualified as Inbox import Keiro.Ops.Outbox qualified as Outbox import Keiro.Ops.Pgmq qualified as Pgmq +import Keiro.Ops.ProcessManager qualified as ProcessManager import Keiro.Ops.Projection qualified as Projection import Keiro.Ops.Rebuild qualified as Rebuild import Keiro.Ops.Render @@ -31,6 +44,7 @@ import Keiro.Ops.Stream qualified as Stream import Keiro.Ops.Timer qualified as Timer import Keiro.Ops.Workflow qualified as Workflow +import Keiro.Ops.WorkflowHeaderCache (WorkflowHeaderCache, runHeaderCached) import Kiroku.Store.Connection (defaultConnectionSettings, withStore) import Options.Applicative import System.Exit qualified as Exit @@ -60,8 +74,12 @@ opsCommand :: !Command } +-- | Programmatic front-end boundary. Alternative front ends must verify the +-- live schema, use JSON output mode, and respect 'isMutation'. data Command - = Workflow Workflow.Command + = ProcessManager ProcessManager.Command + | Aggregate Aggregate.Command + | Workflow Workflow.Command | Timer Timer.Command | Outbox Outbox.Command | Inbox Inbox.Command @@ -85,6 +103,7 @@ invocationParser :: AppHooks -> Parser OpsInvocation invocationParser hooks = OpsInvocation <$> globalOptionsParser <*> commandParser hooks +-- | The authoritative command grammar, including only mounted hooks. commandParser :: AppHooks -> Parser Command commandParser hooks = hsubparser @@ -123,8 +142,17 @@ (info (Stream <$> Stream.commandParser) (progDesc "Inspect and operate Kiroku streams")) <> replayAuditCommand <> rebuildCommand + <> aggregateCommand + <> processManagerCommand ) where + processManagerCommand = case hooks.processManagers of + Nothing -> mempty + Just _ -> command "pm" (info (ProcessManager <$> ProcessManager.commandParser) (progDesc "Inspect mounted process managers")) + aggregateCommand = + case hooks.aggregates of + Nothing -> mempty + Just _ -> command "aggregate" (info (Aggregate <$> Aggregate.commandParser) (progDesc "Inspect mounted aggregates")) replayAuditCommand = case hooks.replayAudit of Nothing -> mempty @@ -165,8 +193,20 @@ } runCommand hooks env opsCommand >>= finishOutcome env +-- | Validate supported command arguments before admission or schema work. +validateCommand :: Command -> Either (Text.Text, Text.Text) () +validateCommand (ProcessManager command) = ProcessManager.validateCommand command +validateCommand (Aggregate command) = Aggregate.validateCommand command +validateCommand (Workflow command) = Workflow.validateCommand command +validateCommand (Timer command) = Timer.validateCommand command +validateCommand (Projection command) = Projection.validateCommand command +validateCommand _ = Right () + +-- | Authoritative read/write classification for every front end. isMutation :: Command -> Bool isMutation = \case + ProcessManager command -> ProcessManager.isMutation command + Aggregate command -> Aggregate.isMutation command Workflow workflowCommand -> Workflow.isMutation workflowCommand Timer timerCommand -> Timer.isMutation timerCommand Outbox outboxCommand -> Outbox.isMutation outboxCommand @@ -179,8 +219,35 @@ ReplayAudit _ -> False Rebuild rebuildCommand -> Rebuild.isMutation rebuildCommand +-- | Reuse only pure header presentation after an exact fresh database check. +-- The optional bytes encode this outcome's successful read result. Other +-- commands use the ordinary authoritative runner without a cache lookup. +runCommandWithHeaderCache :: WorkflowHeaderCache -> AppHooks -> OpsEnv -> Command -> IO (OpsOutcome, Maybe ByteString) +runCommandWithHeaderCache cache _ env (Workflow (Workflow.Inspect (Workflow.InspectHeader ref))) = runHeaderCached cache env ref +runCommandWithHeaderCache _ hooks env command = do + outcome <- runCommand hooks env command + pure (outcome, Nothing) + +-- | Prepare supported direct encodings on demand using the host's bounded +-- encoder. Schema verification and admission remain the front end's job. +-- The callback must return the supplied encoding's exact bytes or refuse size. +-- Existing command/header behavior is delegated without changing its cache. +runCommandWithReadEncoding :: (Encoding -> IO (Maybe ByteString)) -> WorkflowHeaderCache -> AppHooks -> OpsEnv -> Command -> IO (OpsOutcome, Maybe ByteString) +runCommandWithReadEncoding encode cache hooks env command = case command of + ProcessManager ProcessManager.Types | Just catalog <- hooks.processManagers -> do + let (result, encoding) = ProcessManager.prepareTypes catalog + encoded <- encode encoding + pure $ case encoded of + Nothing -> (InspectionFailed "inspection_result_too_large" "encoded result exceeds the host limit; narrow the read", Nothing) + Just body -> (Succeeded result, Just body) + _ -> runCommandWithHeaderCache cache hooks env command + +-- | Execute a parsed command after schema verification. Front ends must use +-- JSON mode and enable force only after explicit mutation confirmation. runCommand :: AppHooks -> OpsEnv -> Command -> IO OpsOutcome runCommand hooks env = \case + ProcessManager command -> maybe (pure (Failed "process-manager inspection hook is not mounted")) (\catalog -> ProcessManager.runCommand catalog env command) hooks.processManagers + Aggregate command -> maybe (pure (Failed "aggregate inspection hook is not mounted")) (\catalog -> Aggregate.runCommand env catalog command) hooks.aggregates Workflow workflowCommand -> Workflow.runCommandWithResume hooks.workflowResume env workflowCommand Timer timerCommand -> Timer.runCommandWithFire hooks.timerFire env timerCommand Outbox outboxCommand -> Outbox.runCommand env outboxCommand @@ -210,6 +277,7 @@ Text.IO.hPutStrLn stderr ("preview only; re-run with --force: " <> reinvocation) pure (Exit.ExitFailure 1) Failed message -> operationalFailure message + InspectionFailed code message -> operationalFailure (code <> ": " <> message) operationalFailure :: Text.Text -> IO Exit.ExitCode operationalFailure message = do diff --git a/src/Keiro/Ops/Aggregate.hs b/src/Keiro/Ops/Aggregate.hs new file mode 100644 --- /dev/null +++ b/src/Keiro/Ops/Aggregate.hs @@ -0,0 +1,121 @@ +-- | Read-only adapters over the application's aggregate catalog. +module Keiro.Ops.Aggregate (Command (..), ShowOptions (..), ListOptions (..), commandParser, validateCommand, isMutation, runCommand) where + +import Data.Aeson (object, (.=)) +import Data.Aeson qualified as Aeson +import Data.Aeson.KeyMap qualified as KeyMap +import Data.Map.Strict qualified as Map +import Data.Text (Text) +import Data.Text qualified as Text +import Effectful (Eff, IOE) +import Effectful.Error.Static (Error) +import Keiro.Inspection.Aggregate qualified as Aggregate +import Keiro.Inspection.Cursor (InspectionCursor, decodeCursor) +import Keiro.Ops.Env (OpsEnv (..)) +import Keiro.Ops.Parse (cursorReader, positiveIntReader) +import Keiro.Ops.Render +import Kiroku.Store.Effect (Store, runStoreIO) +import Kiroku.Store.Error (StoreError) +import Kiroku.Store.Types (StreamName (..), StreamVersion (..)) +import Options.Applicative hiding (action, command, info, value) +import Options.Applicative qualified as Opt + +data ShowOptions = ShowOptions {name :: !Text, aggregateId :: !Text} + deriving stock (Eq, Show) + +data ListOptions = ListOptions {name :: !Text, limit :: !Int, after :: !(Maybe InspectionCursor)} + deriving stock (Eq, Show) + +data Command = Types | Show !ShowOptions | Snapshot !ShowOptions | List !ListOptions + deriving stock (Eq, Show) + +commandParser :: Parser Command +commandParser = + hsubparser + ( Opt.command "types" (Opt.info (pure Types) (progDesc "List mounted aggregate kinds")) + <> Opt.command "show" (Opt.info (Show <$> showOptions) (progDesc "Inspect folded aggregate state")) + <> Opt.command "snapshot" (Opt.info (Snapshot <$> showOptions) (progDesc "Inspect aggregate snapshot health")) + <> Opt.command "list" (Opt.info (List <$> listOptions) (progDesc "Read one aggregate page")) + ) + where + textOption key label = Text.pack <$> strOption (long key <> metavar label) + nameOption = textOption "name" "NAME" + showOptions = ShowOptions <$> nameOption <*> textOption "id" "ID" + listOptions = ListOptions <$> nameOption <*> option positiveIntReader (long "limit" <> Opt.value 50 <> metavar "N") <*> optional (option cursorReader (long "after" <> metavar "CURSOR")) + +isMutation :: Command -> Bool +isMutation = const False + +-- | Reject invalid input before a transport borrows a schema-check connection. +-- The owning page read additionally checks the catalog's exact category. +validateCommand :: Command -> Either (Text, Text) () +validateCommand = \case + Types -> Right () + Show options -> validateId options + Snapshot options -> validateId options + List options + | options.limit < 1 || options.limit > Aggregate.maxAggregatePageSize -> Left ("invalid_page_size", "page size must be 1..500") + | otherwise -> case options.after of + Nothing -> Right () + Just cursor -> case decodeCursor Aggregate.aggregateListCursorKind cursor of + Right (name, category, after :: Text) + | name == options.name && Text.isPrefixOf (category <> "-") after -> Right () + _ -> Left ("invalid_cursor", "invalid cursor or cursor belongs to another aggregate inspector") + where + validateId options + | Text.null (Text.strip options.aggregateId) = Left ("invalid_aggregate_id", "aggregate id must not be blank") + | otherwise = Right () + +runCommand :: OpsEnv -> Aggregate.AggregateInspectors -> Command -> IO OpsOutcome +runCommand env catalog command = case command of + Types -> + pure + ( Succeeded + ( OpsResult + ["name", "category", "snapshots_configured", "snapshot_policy"] + [[summary.name, summary.category, if summary.snapshotsConfigured then "true" else "false", summary.snapshotPolicy] | summary <- Aggregate.listAggregateTypes catalog] + (object ["items" .= map Aggregate.aggregateTypeJson (Aggregate.listAggregateTypes catalog)]) + ) + ) + Show options -> case Map.lookup options.name catalog of + Nothing -> pure (Failed ("unknown aggregate kind: " <> options.name)) + Just (Aggregate.SomeAggregateInspector inspector) -> + runAction env (Aggregate.inspectAggregate inspector options.aggregateId) $ \view -> + OpsResult + ["name", "aggregate_id", "stream_version", "state"] + [[view.name, view.aggregateId, versionText view.streamVersion, maybe "" id view.stateLabel]] + (Aggregate.aggregateViewJson view) + Snapshot options -> case Map.lookup options.name catalog of + Nothing -> pure (Failed ("unknown aggregate kind: " <> options.name)) + Just (Aggregate.SomeAggregateInspector inspector) -> + runAction env (Aggregate.snapshotStatusFor inspector options.aggregateId) $ \(StreamName streamName, StreamVersion version, status) -> + let value = case Aggregate.snapshotStatusJson status of + Aeson.Object fields -> Aeson.Object (fields <> KeyMap.fromList [("name", Aeson.toJSON options.name), ("aggregate_id", Aeson.toJSON options.aggregateId), ("stream_name", Aeson.toJSON streamName), ("stream_version", Aeson.toJSON version)]) + other -> other + in OpsResult ["name", "aggregate_id"] [[options.name, options.aggregateId]] value + List options -> case Map.lookup options.name catalog of + Nothing -> pure (Failed ("unknown aggregate kind: " <> options.name)) + Just (Aggregate.SomeAggregateInspector inspector) -> + runAction env (Aggregate.listAggregates inspector (Aggregate.AggregateListRequest options.limit options.after)) $ \page -> + OpsResult + ["aggregate_id", "stream_name", "stream_version"] + [[item.aggregateId, nameText item.streamName, versionText item.streamVersion] | item <- page.items] + (Aggregate.aggregateListPageJson page) + +runAction :: OpsEnv -> Eff '[Store, Error StoreError, IOE] (Either Aggregate.AggregateInspectionError a) -> (a -> OpsResult) -> IO OpsOutcome +runAction env action render = do + result <- runStoreIO env.store action + pure $ case result of + Left err -> Failed (Text.pack (show err)) + Right (Left err) -> case err of + Aggregate.InvalidPageSize _ -> InspectionFailed "invalid_page_size" (Aggregate.inspectionErrorText err) + Aggregate.InvalidCursor _ -> InspectionFailed "invalid_cursor" (Aggregate.inspectionErrorText err) + Aggregate.ResultTooLarge -> InspectionFailed "inspection_result_too_large" (Aggregate.inspectionErrorText err) + _ -> Failed (Aggregate.inspectionErrorText err) + Right (Right value) -> Succeeded (render value) + +versionText :: StreamVersion -> Text +versionText (StreamVersion version) = Text.pack (show version) + +nameText :: StreamName -> Text +nameText (StreamName name) = name diff --git a/src/Keiro/Ops/Embed.hs b/src/Keiro/Ops/Embed.hs --- a/src/Keiro/Ops/Embed.hs +++ b/src/Keiro/Ops/Embed.hs @@ -6,20 +6,26 @@ module Keiro.Ops.Embed ( AppHooks (..), OpsAuditConfig (..), + AggregateInspectors, + ProcessManagerInspectors, emptyAppHooks, ) where +import Keiro.Inspection.Aggregate (AggregateInspectors) import Keiro.Ops.ReplayAudit (OpsAuditConfig (..)) import Keiro.Ops.Timer (TimerFire) import Keiro.Ops.Workflow (ResumeHook) +import Keiro.ProcessManager.Inspect (ProcessManagerInspectors) import Keiro.Projection.Catalog.Operations (ProjectionCatalogOperations) data AppHooks = AppHooks { workflowResume :: !(Maybe ResumeHook), timerFire :: !(Maybe TimerFire), replayAudit :: !(Maybe OpsAuditConfig), - projectionCatalog :: !(Maybe ProjectionCatalogOperations) + projectionCatalog :: !(Maybe ProjectionCatalogOperations), + aggregates :: !(Maybe AggregateInspectors), + processManagers :: !(Maybe ProcessManagerInspectors) } emptyAppHooks :: AppHooks @@ -28,5 +34,7 @@ { workflowResume = Nothing, timerFire = Nothing, replayAudit = Nothing, - projectionCatalog = Nothing + projectionCatalog = Nothing, + aggregates = Nothing, + processManagers = Nothing } diff --git a/src/Keiro/Ops/Parse.hs b/src/Keiro/Ops/Parse.hs --- a/src/Keiro/Ops/Parse.hs +++ b/src/Keiro/Ops/Parse.hs @@ -5,13 +5,20 @@ positiveIntReader, nonNegativeReader, nonNegativeIntReader, + cursorReader, ) where import Data.Char (toLower) +import Data.Text qualified as Text import Data.Time (NominalDiffTime) -import Options.Applicative (ReadM, eitherReader) +import Keiro.Inspection.Cursor (InspectionCursor (..)) +import Options.Applicative (ReadM, eitherReader, str) import Text.Read qualified as Read + +-- | The owning read validates the opaque token's kind, owner, and payload. +cursorReader :: ReadM InspectionCursor +cursorReader = InspectionCursor . Text.pack <$> str durationReader :: ReadM NominalDiffTime durationReader = eitherReader parseDuration diff --git a/src/Keiro/Ops/Pgmq.hs b/src/Keiro/Ops/Pgmq.hs --- a/src/Keiro/Ops/Pgmq.hs +++ b/src/Keiro/Ops/Pgmq.hs @@ -46,7 +46,7 @@ dlqParser :: Parser DlqCommand dlqParser = hsubparser - ( command "read" (info readParser (progDesc "Read and decode visible DLQ entries")) + ( command "read" (info readParser (progDesc "Preview or reserve and decode visible DLQ entries for 30 seconds")) <> command "redrive" (info redriveParser (progDesc "Preview or move DLQ entries back to the main queue")) <> command "archive" (info archiveParser (progDesc "Preview or archive DLQ entries for retention")) <> command "purge" (info purgeParser (progDesc "Preview or permanently purge a DLQ")) @@ -75,14 +75,16 @@ isMutation :: Command -> Bool isMutation (Dlq dlqCommand) = case dlqCommand of - Read {} -> False + Read {} -> True Redrive {} -> True Archive {} -> True Purge {} -> True runCommand :: OpsEnv -> Command -> IO OpsOutcome runCommand env (Dlq dlqCommand) = case dlqCommand of - Read queue limit -> handlePgmq (runJobEff (pgmqRuntime env) (readDlq (rawJob queue) (fromIntegral limit))) (Succeeded . dlqListResult queue) + Read queue limit + | env.force -> handlePgmq (runJobEff (pgmqRuntime env) (readDlq (rawJob queue) (fromIntegral limit))) (Succeeded . dlqListResult queue) + | otherwise -> previewFromDepth env "read" queue (Just limit) ["pgmq", "dlq", "read", "--queue", queue, "--limit", showText limit] Redrive queue limit -> runRedrive env queue limit Archive queue entry limit -> runArchive env queue entry limit Purge queue -> runPurge env queue diff --git a/src/Keiro/Ops/ProcessManager.hs b/src/Keiro/Ops/ProcessManager.hs new file mode 100644 --- /dev/null +++ b/src/Keiro/Ops/ProcessManager.hs @@ -0,0 +1,136 @@ +-- | Thin read-only adapters over application-owned manager inspectors. +module Keiro.Ops.ProcessManager (Command (..), ShowOptions (..), ListOptions (..), commandParser, validateCommand, isMutation, runCommand, prepareTypes) where + +import Data.Aeson qualified as Aeson +import Data.Aeson.Encoding (Encoding) +import Data.Aeson.Encoding qualified as Encoding +import Data.Aeson.KeyMap qualified as KeyMap +import Data.Map.Lazy qualified as LazyMap +import Data.Map.Strict qualified as Map +import Data.Text (Text) +import Data.Text qualified as Text +import Keiro.Inspection.Cursor (InspectionCursor, decodeCursor) +import Keiro.Ops.Env (OpsEnv (..)) +import Keiro.Ops.Parse (cursorReader, positiveIntReader) +import Keiro.Ops.Render +import Keiro.ProcessManager.Inspect qualified as PM +import Kiroku.Store.Effect (runStoreIO) +import Options.Applicative hiding (command, info, value) +import Options.Applicative qualified as Opt + +data ShowOptions = ShowOptions {name :: !Text, correlationId :: !Text, timerLimit :: !Int} + deriving stock (Eq, Show) + +data ListOptions = ListOptions {name :: !Text, after :: !(Maybe InspectionCursor), limit :: !Int} + deriving stock (Eq, Show) + +data Command = Types | Show !ShowOptions | List !ListOptions deriving stock (Eq, Show) + +commandParser :: Parser Command +commandParser = + hsubparser + ( Opt.command "types" (Opt.info (pure Types) (progDesc "List mounted process managers")) + <> Opt.command + "show" + ( Opt.info + ( Show + <$> ( ShowOptions + <$> name + <*> textOption "correlation" "ID" + <*> option positiveIntReader (long "timer-limit" <> Opt.value 20 <> metavar "N") + ) + ) + (progDesc "Observe persisted manager state and intended dispatch witnesses") + ) + <> Opt.command + "list" + ( Opt.info + ( List + <$> ( ListOptions + <$> name + <*> optional (option cursorReader (long "after" <> metavar "CURSOR")) + <*> option positiveIntReader (long "limit" <> Opt.value 50 <> metavar "N") + ) + ) + (progDesc "Read one page of manager journal streams") + ) + ) + where + name = textOption "name" "NAME" + textOption key label = Text.pack <$> strOption (long key <> metavar label) + +isMutation :: Command -> Bool +isMutation = const False + +validateCommand :: Command -> Either (Text, Text) () +validateCommand Types = Right () +validateCommand (Show options) + | Text.null (Text.strip options.correlationId) = Left ("invalid_correlation_id", "correlation id must not be blank") + | options.timerLimit < 1 || options.timerLimit > 500 = Left ("invalid_page_size", "timer page size must be 1..500") + | otherwise = Right () +validateCommand (List options) + | options.limit < 1 || options.limit > 500 = Left ("invalid_page_size", "page size must be 1..500") + | otherwise = case options.after of + Nothing -> Right () + Just cursor -> case decodeCursor PM.instanceListCursorKind cursor of + Right (owner, category, after :: Text) | owner == options.name && Text.isPrefixOf (category <> "-") after -> Right () + _ -> Left ("invalid_cursor", "cursor belongs to another manager inspector or is malformed") + +runCommand :: PM.ProcessManagerInspectors -> OpsEnv -> Command -> IO OpsOutcome +runCommand catalog env command = case validateCommand command of + Left (code, detail) -> pure (InspectionFailed code detail) + Right () -> case command of + Types -> pure (Succeeded (fst (prepareTypes catalog))) + Show options -> case Map.lookup options.name catalog of + Nothing -> unknown options.name + Just (PM.SomeProcessManagerInspector inspector) -> do + result <- runStoreIO env.store (PM.inspectProcessManagerInstance inspector (PM.InstanceViewRequest options.timerLimit) options.correlationId) + pure $ case result of + Left err -> Failed (Text.pack (show err)) + Right (Left err) -> Failed (Text.pack (show err)) + Right (Right value) -> + Succeeded + ( OpsResult + ["name", "correlation_id", "state"] + (maybe [] (\view -> [[view.name, view.correlationId, maybe "" id view.stateLabel]]) value) + (maybe Aeson.Null PM.processManagerViewJson value) + ) + List options -> case Map.lookup options.name catalog of + Nothing -> unknown options.name + Just (PM.SomeProcessManagerInspector inspector) -> do + result <- runStoreIO env.store (PM.listProcessManagerInstances inspector (PM.InstanceListRequest options.limit options.after)) + pure $ case result of + Left err -> Failed (Text.pack (show err)) + Right (Left err) -> uncurry InspectionFailed (listError err) + Right (Right page) -> + Succeeded + ( OpsResult + ["correlation_id", "stream_version"] + [[item.correlationId, Text.pack (show item.streamVersion)] | item <- page.items] + (PM.instancePageJson page) + ) + where + unknown owner = pure (Failed ("unknown process manager: " <> owner)) + listError err = + ( case err of + PM.InvalidInstancePageSize _ -> "invalid_page_size" + PM.InvalidInstanceCursor _ -> "invalid_cursor" + PM.InstanceResultTooLarge -> "inspection_result_too_large", + Text.pack (show err) + ) + +-- | Canonical presentation and an equivalent direct encoding from one traversal. +-- The host must consume the encoding inside its own byte/deadline budget. +-- Neither representation is retained between requests. +prepareTypes :: PM.ProcessManagerInspectors -> (OpsResult, Encoding) +prepareTypes catalog = + let summaries = PM.listProcessManagerTypes catalog + -- Defer the canonical array until a caller actually consumes JSON. HTTP + -- can consume the direct encoding while retaining the same OpsResult API. + value = Aeson.Object (KeyMap.fromMap (LazyMap.singleton "items" (Aeson.toJSON summaries))) + in ( OpsResult + ["name", "category", "snapshots_configured", "snapshot_policy", "timers_configured"] + [[summary.name, summary.category, if summary.snapshotsConfigured then "true" else "false", summary.snapshotPolicy, maybe "unknown" (\configured -> if configured then "true" else "false") summary.timersConfigured] | summary <- summaries] + value, + Encoding.pairs (Encoding.pair "items" (Aeson.toEncoding summaries)) + ) diff --git a/src/Keiro/Ops/Projection.hs b/src/Keiro/Ops/Projection.hs --- a/src/Keiro/Ops/Projection.hs +++ b/src/Keiro/Ops/Projection.hs @@ -4,6 +4,9 @@ -- with ADR 28. module Keiro.Ops.Projection ( Command (..), + ProjectionStatusOptions (..), + projectionGroupStatusJson, + validateCommand, commandParser, isMutation, runCommand, @@ -18,10 +21,14 @@ import Data.Vector qualified as Vector import Effectful (Eff, IOE) import Effectful.Error.Static (Error) +import Keiro.Inspection.Cursor (InspectionCursor (..)) import Keiro.Ops.Env (OpsEnv (..), OutputMode (..)) +import Keiro.Ops.Parse (cursorReader, positiveIntReader) import Keiro.Ops.Render import Keiro.Projection (countAsyncProjectionDedupForBefore, pruneAsyncProjectionDedupForBefore) +import Keiro.Projection.Catalog (projectionRevisionIdText, rebuildGroupIdText) import Keiro.ReadModel (storeHeadPosition) +import Keiro.ReadModel.Rebuild import Kiroku.Store.Effect (Store, runStoreIO) import Kiroku.Store.Error (StoreError) import Kiroku.Store.Subscription @@ -31,18 +38,23 @@ subscriptionCheckpointInventory, ) import Kiroku.Store.Types (GlobalPosition (..)) -import Options.Applicative hiding (action, value) +import Options.Applicative hiding (action) import Prelude +data ProjectionStatusOptions = ProjectionStatusOptions {limit :: !Int, after :: !(Maybe InspectionCursor)} + deriving stock (Eq, Show) + data Command - = Position !Text + = Status !ProjectionStatusOptions + | Position !Text | PruneDedup !Text !UTCTime deriving stock (Eq, Show) commandParser :: Parser Command commandParser = hsubparser - ( command "position" (info positionParser (progDesc "Show one subscription's durable member checkpoints and floor")) + ( command "status" (info (Status <$> (ProjectionStatusOptions <$> option positiveIntReader (long "limit" <> value 100 <> metavar "N") <*> optional (option cursorReader (long "after" <> metavar "CURSOR")))) (progDesc "Observe one bounded page of projection-group status")) + <> command "position" (info positionParser (progDesc "Show one subscription's durable member checkpoints and floor")) <> command "prune-dedup" (info pruneParser (progDesc "Preview or prune one projection's old dedup rows")) ) where @@ -62,13 +74,49 @@ Right (parseTimeM True defaultTimeLocale "%Y-%m-%dT%H:%M:%S%QZ" raw) +statusInput :: ProjectionStatusOptions -> ProjectionStatusPageRequest +statusInput options = ProjectionStatusPageRequest options.limit options.after + +validateCommand :: Command -> Either (Text, Text) () +validateCommand (Status options) = either (Left . statusError) (const (Right ())) (validateProjectionStatusRequest (statusInput options)) +validateCommand _ = Right () + +statusError :: ProjectionStatusReadError -> (Text, Text) +statusError err = + ( case err of + InvalidProjectionStatusPageSize _ -> "invalid_page_size" + InvalidProjectionStatusCursor _ -> "invalid_cursor" + ProjectionStatusResultTooLarge -> "inspection_result_too_large", + Text.pack (show err) + ) + isMutation :: Command -> Bool isMutation = \case + Status {} -> False Position {} -> False PruneDedup {} -> True runCommand :: OpsEnv -> Command -> IO OpsOutcome runCommand env = \case + Status options -> runAction env action $ \case + Left err -> uncurry InspectionFailed (statusError err) + Right (page, head) -> + Succeeded + ( OpsResult + ["group_id", "lifecycle_phase", "reads_allowed", "serving_position_basis"] + [[rebuildGroupIdText row.groupId, row.lifecyclePhase, Text.pack (show row.readsAllowed), basisText row.servingPositionBasis] | row <- page.items] + ( object + ( ["observed_at" .= page.observedAt, "visible_store_head" .= positionInt head, "items" .= map projectionGroupStatusJson page.items] + <> maybe [] (\cursor -> ["next_cursor" .= cursor.cursorText]) page.next + ) + ) + ) + where + action = do + page <- listProjectionGroupStatusesPage (statusInput options) + case page of + Left err -> pure (Left err) + Right value -> Right . (value,) <$> storeHeadPosition Position subscription -> runPosition env subscription PruneDedup projection before | env.force -> @@ -79,6 +127,33 @@ PreviewRequired (pruneResult True projection before affected) (forceInvocation env ["projection", "prune-dedup", "--projection", projection, "--before", utcText before]) + +-- | The frozen v1 owner relation, using ADR-28 position vocabulary. +projectionGroupStatusJson :: ProjectionGroupStatusV1 -> Value +projectionGroupStatusJson row = + object + [ "group_id" .= rebuildGroupIdText row.groupId, + "lifecycle_phase" .= row.lifecyclePhase, + "reads_allowed" .= row.readsAllowed, + "writes_allowed" .= row.writesAllowed, + "serving_revision_id" .= fmap projectionRevisionIdText row.servingRevisionId, + "serving_epoch" .= row.servingEpoch, + "serving_position_basis" .= basisText row.servingPositionBasis, + "serving_applied_position" .= fmap positionInt row.servingAppliedPosition, + "active_run_id" .= fmap rebuildRunIdText row.activeRunId, + "candidate_revision_id" .= fmap projectionRevisionIdText row.candidateRevisionId, + "candidate_rebuild_position" .= fmap positionInt row.candidateRebuildPosition, + "candidate_rebuild_head" .= fmap positionInt row.candidateRebuildHead, + "query_models" .= row.queryModels, + "rebuild_started_at" .= row.rebuildStartedAt, + "last_promoted_at" .= row.lastPromotedAt, + "failed_at" .= row.failedAt, + "failure_code" .= row.failureCode, + "failure_detail" .= row.failureDetail + ] + +basisText :: ServingPositionBasis -> Text +basisText = \case ServingPositionAppend -> "append"; ServingPositionCheckpoint -> "checkpoint"; ServingPositionUnmanaged -> "unmanaged" runPosition :: OpsEnv -> Text -> IO OpsOutcome runPosition env subscription = diff --git a/src/Keiro/Ops/Render.hs b/src/Keiro/Ops/Render.hs --- a/src/Keiro/Ops/Render.hs +++ b/src/Keiro/Ops/Render.hs @@ -34,6 +34,7 @@ | SucceededWithExit !OpsResult !ExitCode | PreviewRequired !OpsResult !Text | Failed !Text + | InspectionFailed !Text !Text deriving stock (Eq, Show) emptyResult :: OpsResult diff --git a/src/Keiro/Ops/Shard.hs b/src/Keiro/Ops/Shard.hs --- a/src/Keiro/Ops/Shard.hs +++ b/src/Keiro/Ops/Shard.hs @@ -10,11 +10,12 @@ ) where -import Data.Aeson (object, (.=)) +import Data.Aeson (Value, object, (.=)) +import Data.Map.Strict qualified as Map import Data.Set qualified as Set import Data.Text (Text) import Data.Text qualified as Text -import Data.Time (UTCTime, getCurrentTime) +import Data.Time (UTCTime) import Data.UUID (UUID) import Data.UUID qualified as UUID import Effectful (Eff, IOE, (:>)) @@ -28,14 +29,16 @@ import Options.Applicative hiding (action, value) data Command - = Status !Text + = List + | Status !Text | Relinquish !Text !WorkerId deriving stock (Eq, Show) commandParser :: Parser Command commandParser = hsubparser - ( command "status" (info statusParser (progDesc "Show shard counts and ownership")) + ( command "list" (info (pure List) (progDesc "List sharded subscriptions and live ownership counts")) + <> command "status" (info statusParser (progDesc "Show shard counts and ownership")) <> command "relinquish" (info relinquishParser (progDesc "Preview or release every bucket owned by one worker")) ) where @@ -50,22 +53,16 @@ isMutation :: Command -> Bool isMutation = \case + List -> False Status {} -> False Relinquish {} -> True runCommand :: OpsEnv -> Command -> IO OpsOutcome runCommand env = \case - Status name -> do - now <- getCurrentTime - runAction env (statusAction name) (Succeeded . uncurry (statusResult now name)) + List -> runAction env listShardedSubscriptions (Succeeded . listResult) + Status name -> runAction env (shardOwnershipView (SubscriptionName name)) (Succeeded . statusResult) Relinquish name worker -> runRelinquish env name worker -statusAction :: (Store :> es) => Text -> Eff es ([(Int, Maybe WorkerId, Maybe UTCTime)], [(Int, Int)]) -statusAction name = do - ownership <- ownershipSnapshotFor (SubscriptionName name) - counts <- shardCountSnapshot (SubscriptionName name) - pure (ownership, counts) - runRelinquish :: OpsEnv -> Text -> WorkerId -> IO OpsOutcome runRelinquish env name worker | not env.force = @@ -94,38 +91,66 @@ result <- runStoreIO env.store action pure $ either (Failed . Text.pack . show) onSuccess result -statusResult :: UTCTime -> Text -> [(Int, Maybe WorkerId, Maybe UTCTime)] -> [(Int, Int)] -> OpsResult -statusResult now name ownership counts = +statusResult :: ShardOwnershipView -> OpsResult +statusResult view = OpsResult { headers = ["subscription", "bucket", "owner", "lease_expires_at", "lease_state", "configured_shards"], - rows = map rowText ownership, - jsonValue = object ["subscription" .= name, "shard_counts" .= map countJson counts, "ownership" .= map ownershipJson ownership] + rows = map rowText view.buckets, + jsonValue = + object + [ "subscription" .= name, + "observed_at" .= view.observedAt, + "shard_counts" .= map countJson counts, + "ownership" .= map ownershipJson view.buckets + ] } where + SubscriptionName name = view.subscriptionName + counts = Map.toAscList (Map.fromListWith (+) [(row.shardCount, 1 :: Int) | row <- view.buckets]) configured = Text.intercalate "," [showText shardCount <> " (" <> showText rowCount <> " rows)" | (shardCount, rowCount) <- counts] - rowText (bucket, owner, expiresAt) = + rowText row = [ name, - showText bucket, - maybe "unowned" workerText owner, - maybe "" timeText expiresAt, - leaseState now owner expiresAt, + showText row.bucket, + maybe "unowned" workerText row.owner, + maybe "" timeText row.leaseExpiresAt, + leaseStateText (leaseStateAt view.observedAt row), configured ] - countJson (shardCount, rowCount) = object ["shard_count" .= shardCount, "rows" .= rowCount] - ownershipJson (bucket, owner, expiresAt) = + ownershipJson row = object - [ "bucket" .= bucket, - "owner" .= fmap workerText owner, - "lease_expires_at" .= expiresAt, - "lease_state" .= leaseState now owner expiresAt + [ "bucket" .= row.bucket, + "owner" .= fmap workerText row.owner, + "lease_expires_at" .= row.leaseExpiresAt, + "lease_state" .= leaseStateText (leaseStateAt view.observedAt row), + "heartbeat_at" .= row.heartbeatAt, + "updated_at" .= row.updatedAt, + "lease_age_seconds" .= fmap (\age -> realToFrac age :: Double) (leaseAgeAt view.observedAt row) ] -leaseState :: UTCTime -> Maybe WorkerId -> Maybe UTCTime -> Text -leaseState _ Nothing _ = "unowned" -leaseState now (Just _) (Just expiry) - | expiry < now = "expired" - | otherwise = "live" -leaseState _ (Just _) Nothing = "invalid" +countJson :: (Int, Int) -> Value +countJson (shardCount, rowCount) = object ["shard_count" .= shardCount, "rows" .= rowCount] + +listResult :: ShardedSubscriptionList -> OpsResult +listResult view = + OpsResult + ["subscription", "buckets_total", "buckets_live", "live_owners"] + [[nameText row.subscriptionName, showText row.bucketsTotal, showText row.bucketsLive, showText row.liveOwners] | row <- view.subscriptions] + (object ["observed_at" .= view.observedAt, "items" .= map summaryJson view.subscriptions]) + where + nameText (SubscriptionName name) = name + summaryJson row = + object + [ "subscription" .= nameText row.subscriptionName, + "shard_counts" .= map countJson row.shardCounts, + "buckets_total" .= row.bucketsTotal, + "buckets_live" .= row.bucketsLive, + "buckets_expired" .= row.bucketsExpired, + "buckets_unowned" .= row.bucketsUnowned, + "live_owners" .= row.liveOwners + ] + +leaseStateText :: LeaseState -> Text +leaseStateText = \case LeaseLive -> "live"; LeaseExpired -> "expired"; LeaseUnowned -> "unowned"; LeaseInvalid -> "invalid" relinquishResult :: Bool -> Text -> WorkerId -> [Int] -> OpsResult relinquishResult preview name worker buckets = diff --git a/src/Keiro/Ops/Snapshot.hs b/src/Keiro/Ops/Snapshot.hs --- a/src/Keiro/Ops/Snapshot.hs +++ b/src/Keiro/Ops/Snapshot.hs @@ -15,10 +15,12 @@ import Data.Aeson (Value, object, (.=)) import Data.Aeson qualified as Aeson +import Data.Aeson.KeyMap qualified as KeyMap import Data.ByteString.Lazy qualified as LazyByteString import Data.Int (Int64) import Data.Text (Text) import Data.Text qualified as Text +import Data.Time (diffUTCTime) import Effectful (Eff, IOE, (:>)) import Effectful.Error.Static (Error) import Keiro.Ops.Env (OpsEnv (..), OutputMode (..)) @@ -92,7 +94,7 @@ runCommand :: OpsEnv -> Command -> IO OpsOutcome runCommand env = \case - Show name -> runAction env (lookupByName name) (Succeeded . snapshotResult name) + Show name -> runAction env (observationByName name) (Succeeded . snapshotObservedResult name) Delete name -> runDelete env name TruncationPreflight name before suppliedExpected -> runAction env (preflightFor name before suppliedExpected) (Succeeded . preflightResult) @@ -104,6 +106,11 @@ Nothing -> pure Nothing Just info -> lookupSnapshotRow info.id +observationByName :: (Store :> es) => Text -> Eff es (Maybe SnapshotObservation) +observationByName name = do + stream <- getStream (StreamName name) + traverse (observeSnapshotRow . (.id)) stream + runDelete :: OpsEnv -> Text -> IO OpsOutcome runDelete env name | not env.force = @@ -193,6 +200,21 @@ rows = maybe [] (pure . snapshotRowText name) row, jsonValue = maybe Aeson.Null (snapshotJson name) row } + +snapshotObservedResult :: Text -> Maybe SnapshotObservation -> OpsResult +snapshotObservedResult name observation = + case observation of + Just observed + | Just row <- observed.row -> + let age = realToFrac (diffUTCTime observed.observedAt row.updatedAt) :: Double + original = snapshotResult name (Just row) + fields = KeyMap.fromList [("observed_at", Aeson.toJSON observed.observedAt), ("age_seconds", Aeson.toJSON age)] + in original + { headers = original.headers <> ["observed_at", "age_seconds"], + rows = map (<> [showText observed.observedAt, showText age]) original.rows, + jsonValue = case original.jsonValue of Aeson.Object value -> Aeson.Object (value <> fields); value -> value + } + _ -> snapshotResult name Nothing snapshotRowText :: Text -> SnapshotRow -> [Text] snapshotRowText name row = diff --git a/src/Keiro/Ops/Timer.hs b/src/Keiro/Ops/Timer.hs --- a/src/Keiro/Ops/Timer.hs +++ b/src/Keiro/Ops/Timer.hs @@ -2,6 +2,9 @@ ( Command (..), DrainOptions (..), StuckListOptions (..), + PendingListOptions (..), + pendingTimerItemJson, + validateCommand, TimerFire, commandParser, commandParserWithDrain, @@ -20,8 +23,9 @@ import Data.UUID qualified as UUID import Effectful (Eff, IOE) import Effectful.Error.Static (Error) +import Keiro.Inspection.Cursor (InspectionCursor (..)) import Keiro.Ops.Env (OpsEnv (..), OutputMode (..)) -import Keiro.Ops.Parse (durationReader, nonNegativeIntReader, positiveIntReader) +import Keiro.Ops.Parse (cursorReader, durationReader, nonNegativeIntReader, positiveIntReader) import Keiro.Ops.Render import Keiro.Timer import Kiroku.Store.Effect (Store, runStoreIO) @@ -36,6 +40,10 @@ } deriving stock (Eq, Show) +data PendingListOptions = PendingListOptions + {manager :: !(Maybe Text), correlation :: !(Maybe Text), after :: !(Maybe InspectionCursor), limit :: !Int} + deriving stock (Eq, Show) + data DrainOptions = DrainOptions { limit :: !Int } @@ -45,6 +53,7 @@ data Command = StuckList !StuckListOptions + | PendingList !PendingListOptions | Requeue !TimerId | Cancel !TimerId | DeadLetter !TimerId !Text @@ -57,13 +66,14 @@ commandParserWithDrain :: Bool -> Parser Command commandParserWithDrain includeDrain = hsubparser - ( command - "stuck" - ( info - stuckCommandParser - (progDesc "Find timers left firing by an interrupted worker") - ) + ( command "pending" (info pendingCommandParser (progDesc "Read a bounded page of scheduled and firing timers")) <> command + "stuck" + ( info + stuckCommandParser + (progDesc "Find timers left firing by an interrupted worker") + ) + <> command "requeue" ( info (Requeue <$> timerIdArgument) @@ -97,6 +107,42 @@ ) | otherwise = mempty +pendingCommandParser :: Parser Command +pendingCommandParser = + hsubparser + ( command + "list" + ( info + ( PendingList + <$> ( PendingListOptions + <$> optional (Text.pack <$> strOption (long "manager" <> metavar "NAME")) + <*> optional (Text.pack <$> strOption (long "correlation" <> metavar "ID")) + <*> optional (option cursorReader (long "after" <> metavar "CURSOR")) + <*> option positiveIntReader (long "limit" <> metavar "N" <> Optparse.value 100) + ) + ) + (progDesc "Observe pending timers without claiming them") + ) + ) + +pendingInput :: PendingListOptions -> (PendingTimerFilter, PendingTimerPageRequest) +pendingInput options = (PendingTimerFilter options.manager options.correlation, PendingTimerPageRequest options.limit options.after) + +validateCommand :: Command -> Either (Text, Text) () +validateCommand (PendingList options) = + let (timerFilter, request) = pendingInput options + in either (Left . pendingError) (const (Right ())) (validatePendingTimerRequest timerFilter request) +validateCommand _ = Right () + +pendingError :: PendingTimerReadError -> (Text, Text) +pendingError err = + ( case err of + InvalidPendingTimerPageSize _ -> "invalid_page_size" + InvalidPendingTimerCursor _ -> "invalid_cursor" + PendingTimerResultTooLarge -> "inspection_result_too_large", + Text.pack (show err) + ) + stuckCommandParser :: Parser Command stuckCommandParser = hsubparser @@ -123,6 +169,7 @@ isMutation :: Command -> Bool isMutation = \case StuckList {} -> False + PendingList {} -> False Requeue {} -> True Cancel {} -> True DeadLetter {} -> True @@ -134,10 +181,30 @@ runCommandWithFire :: Maybe TimerFire -> OpsEnv -> Command -> IO OpsOutcome runCommandWithFire timerFire env = \case StuckList options -> runStuckList env options + PendingList options -> runPendingList env options Requeue timerId -> runMutation env RequeueOperation timerId Nothing Cancel timerId -> runMutation env CancelOperation timerId Nothing DeadLetter timerId reason -> runMutation env DeadLetterOperation timerId (Just reason) DrainOnce options -> runDrainOnce timerFire env options + +runPendingList :: OpsEnv -> PendingListOptions -> IO OpsOutcome +runPendingList env options = do + let (timerFilter, request) = pendingInput options + runAction env (findPendingTimers timerFilter request) $ \case + Left err -> uncurry InspectionFailed (pendingError err) + Right page -> + Succeeded + ( OpsResult + ["id", "manager", "correlation", "fire_at", "status", "attempts", "payload"] + (map (timerRow . (.timer)) page.timers) + ( object + ( ["observed_at" .= page.observedAt, "items" .= map (pendingTimerItemJson page.observedAt) page.timers] + <> maybe [] (\manager -> ["manager" .= manager]) options.manager + <> maybe [] (\correlation -> ["correlation_id" .= correlation]) options.correlation + <> maybe [] (\cursor -> ["next_cursor" .= cursor.cursorText]) page.next + ) + ) + ) runDrainOnce :: Maybe TimerFire -> OpsEnv -> DrainOptions -> IO OpsOutcome runDrainOnce Nothing _ _ = pure (Failed "timer fire hook is not mounted") diff --git a/src/Keiro/Ops/Workflow.hs b/src/Keiro/Ops/Workflow.hs --- a/src/Keiro/Ops/Workflow.hs +++ b/src/Keiro/Ops/Workflow.hs @@ -1,6 +1,7 @@ module Keiro.Ops.Workflow ( AwakeableCommand (..), Command (..), + InspectionCommand (..), GcOptions (..), InspectOptions (..), ListOptions (..), @@ -9,39 +10,37 @@ ResumeOptions (..), WorkflowRef (..), commandParser, + validateCommand, commandParserWithResume, isMutation, runCommand, runCommandWithResume, + headerOutcome, ) where import Data.Aeson (Value, object, (.=)) import Data.Aeson qualified as Aeson -import Data.Aeson.Types qualified as AesonTypes -import Data.Int (Int64) import Data.List.NonEmpty qualified as NonEmpty -import Data.Map.Strict (Map) import Data.Map.Strict qualified as Map -import Data.Maybe (catMaybes) import Data.Set qualified as Set import Data.Text (Text) import Data.Text qualified as Text import Data.Text.Encoding qualified as Text.Encoding -import Data.Time (NominalDiffTime, UTCTime, getCurrentTime) +import Data.Time (NominalDiffTime, UTCTime, defaultTimeLocale, getCurrentTime, parseTimeM) import Data.UUID (UUID) import Data.UUID qualified as UUID -import Data.Vector qualified as Vector import Effectful (Eff, IOE, (:>)) import Effectful.Error.Static (Error) -import Keiro.Codec (decodeRecorded) +import Keiro.Inspection.Cursor (InspectionCursor, maxCursorBytes) import Keiro.Ops.Env (OpsEnv (..), OutputMode (..)) -import Keiro.Ops.Parse (durationReader, nonNegativeReader, positiveIntReader) +import Keiro.Ops.Parse (cursorReader, durationReader, nonNegativeReader, positiveIntReader) import Keiro.Ops.Render import Keiro.Workflow.Awakeable (AwakeableId (..), cancelAwakeable, signalAwakeable) import Keiro.Workflow.Awakeable.Schema qualified as Awakeable import Keiro.Workflow.Child.Schema qualified as Child import Keiro.Workflow.Gc qualified as Gc +import Keiro.Workflow.Inspection qualified as Inspection import Keiro.Workflow.Instance qualified as Instance import Keiro.Workflow.Resume ( ResumeSummary (..), @@ -52,20 +51,10 @@ import Keiro.Workflow.Schema qualified as WorkflowSchema import Keiro.Workflow.Types ( WorkflowId (..), - WorkflowJournalEvent (..), WorkflowName (..), - awakeableAllocStepPrefix, - cancelledStepName, - completedStepName, - continuedAsNewStepName, - failedStepName, - workflowGenerationStreamName, - workflowJournalCodec, ) import Kiroku.Store.Effect (Store, runStoreIO) import Kiroku.Store.Error (StoreError) -import Kiroku.Store.Read qualified as StoreRead -import Kiroku.Store.Types import Options.Applicative hiding (action, value) import Options.Applicative qualified as Optparse @@ -119,6 +108,7 @@ data Command = List !ListOptions + | Inspect !InspectionCommand | Show !WorkflowRef | Steps !InspectOptions | Journal !InspectOptions @@ -130,6 +120,15 @@ | ResumeOnce !ResumeOptions deriving stock (Eq, Show) +data InspectionCommand + = InspectList !Inspection.WorkflowListQuery + | InspectHeader !WorkflowRef + | InspectJournal !WorkflowRef !Inspection.DetailPageQuery + | InspectSteps !WorkflowRef !Inspection.DetailPageQuery + | InspectChildren !WorkflowRef !Inspection.OwnerPageQuery + | InspectAwakeables !WorkflowRef !Inspection.OwnerPageQuery + deriving stock (Eq, Show) + commandParser :: Parser Command commandParser = commandParserWithResume False @@ -137,6 +136,7 @@ commandParserWithResume includeResume = hsubparser ( command "list" (info (List <$> listOptionsParser) (progDesc "List workflow instances with stable keyset paging")) + <> command "inspect" (info (Inspect <$> inspectionParser) (progDesc "Bounded workflow observation pages")) <> command "show" (info (Show <$> workflowRefParser) (progDesc "Show an instance, its children, and its awakeables")) <> command "steps" (info (Steps <$> inspectOptionsParser) (progDesc "Show the derived step index for one generation")) <> command "journal" (info (Journal <$> inspectOptionsParser) (progDesc "Decode one workflow journal generation in order")) @@ -180,6 +180,40 @@ <$> workflowRefParser <*> optional (option (nonNegativeReader "expected a non-negative generation") (long "generation" <> metavar "N" <> help "Journal generation; defaults to current")) +inspectionParser :: Parser InspectionCommand +inspectionParser = + hsubparser + ( command "list" (info (InspectList <$> listQuery) (progDesc "Page workflow instances")) + <> command "header" (info (InspectHeader <$> workflowRefParser) (progDesc "Read only the instance header")) + <> command "journal" (info (InspectJournal <$> workflowRefParser <*> detailQuery) (progDesc "Page one journal generation")) + <> command "steps" (info (InspectSteps <$> workflowRefParser <*> detailQuery) (progDesc "Page one generation's step index")) + <> command "children" (info (InspectChildren <$> workflowRefParser <*> ownerQuery) (progDesc "Page child links")) + <> command "awakeables" (info (InspectAwakeables <$> workflowRefParser <*> ownerQuery) (progDesc "Page allocated external promises")) + ) + where + limit = option positiveIntReader (long "limit" <> metavar "N" <> Optparse.value 100 <> help "Page size, at most 500") + after = optional (option cursorReader (long "after" <> metavar "CURSOR")) + detailQuery = + Inspection.DetailPageQuery + <$> optional (option (nonNegativeReader "expected a non-negative generation") (long "generation" <> metavar "N")) + <*> limit + <*> after + ownerQuery = Inspection.OwnerPageQuery <$> limit <*> after + listQuery = + Inspection.WorkflowListQuery + <$> (NonEmpty.nonEmpty <$> many (option statusClassReader (long "status-class" <> metavar "CLASS"))) + <*> optional (Text.pack <$> strOption (long "workflow-name" <> metavar "NAME")) + <*> optional (option utcReader (long "created-since" <> metavar "UTC")) + <*> optional (option utcReader (long "created-before" <> metavar "UTC")) + <*> optional (option utcReader (long "updated-since" <> metavar "UTC")) + <*> optional (option utcReader (long "updated-before" <> metavar "UTC")) + <*> option orderingReader (long "ordering" <> metavar "key|created_desc" <> Optparse.value Inspection.ByKey) + <*> limit + <*> after + statusClassReader = eitherReader $ \raw -> maybe (Left "expected running, retrying, sleeping, awaiting, completed, cancelled, or failed") Right (Inspection.statusClassFromText (Text.pack raw)) + orderingReader = eitherReader $ \case "key" -> Right Inspection.ByKey; "created_desc" -> Right Inspection.ByCreatedDesc; _ -> Left "expected key or created_desc" + utcReader = eitherReader $ \raw -> maybe (Left "expected a UTC timestamp such as 2026-10-20T01:02:03Z") Right (parseTimeM True defaultTimeLocale "%Y-%m-%dT%H:%M:%S%QZ" raw) + workflowRefParser :: Parser WorkflowRef workflowRefParser = WorkflowRef @@ -245,6 +279,7 @@ isMutation :: Command -> Bool isMutation = \case + Inspect {} -> False List {} -> False Show {} -> False Steps {} -> False @@ -263,6 +298,7 @@ runCommandWithResume :: Maybe ResumeHook -> OpsEnv -> Command -> IO OpsOutcome runCommandWithResume resumeHook env = \case + Inspect inspection -> runInspection env inspection List options -> runList env options Show ref -> runShow env ref Steps options -> runSteps env options @@ -274,6 +310,53 @@ GcRunOnce options -> runGc env options ResumeOnce options -> runResumeOnce resumeHook env options +runInspection :: OpsEnv -> InspectionCommand -> IO OpsOutcome +runInspection env = \case + InspectList query -> page (Inspection.listWorkflowInstancesPage query) + InspectHeader ref -> runAction env (Instance.lookupInstance (refName ref) (refId ref)) (headerOutcome ref) + InspectJournal ref query -> page (Inspection.readWorkflowJournalPage (refName ref) (refId ref) query) + InspectSteps ref query -> page (Inspection.loadStepIndexPage (refName ref) (refId ref) query) + InspectChildren ref query -> page (Inspection.listWorkflowChildrenPage (refName ref) (refId ref) query) + InspectAwakeables ref query -> page (Inspection.listWorkflowAwakeablesPage (refName ref) (refId ref) query) + where + page :: (Aeson.ToJSON a) => Eff '[Store, Error StoreError, IOE] (Either Inspection.InspectionError a) -> IO OpsOutcome + page action = runAction env action (either inspectionFailure (Succeeded . jsonResult . Aeson.toJSON)) + jsonResult value = OpsResult ["result"] [[jsonText value]] value + +-- | Canonical CLI and HTTP outcome for a freshly observed header. +headerOutcome :: WorkflowRef -> Maybe Instance.WorkflowInstanceRow -> OpsOutcome +headerOutcome ref = \case + Nothing -> Failed (workflowLabel ref <> " was not found") + Just row -> + let value = Inspection.instanceToJson row + in Succeeded (OpsResult ["result"] [[jsonText value]] value) + +inspectionFailure :: Inspection.InspectionError -> OpsOutcome +inspectionFailure err = uncurry InspectionFailed (inspectionError err) + +inspectionError :: Inspection.InspectionError -> (Text, Text) +inspectionError err = (code, message) + where + message = case err of + Inspection.CursorResultTooLarge -> "the continuation key exceeds the " <> Text.pack (show maxCursorBytes) <> "-byte cursor limit" + _ -> Text.pack (show err) + code = case err of + Inspection.InvalidPageSize _ -> "invalid_page_size" + Inspection.InvalidCursor _ -> "invalid_cursor" + Inspection.CursorGenerationMismatch _ _ -> "cursor_generation_mismatch" + Inspection.CursorResultTooLarge -> "inspection_result_too_large" + +-- | Validate the owning library's inspection grammar without any SQL. +validateCommand :: Command -> Either (Text, Text) () +validateCommand (Inspect command) = either (Left . inspectionError) Right $ case command of + InspectList query -> Inspection.validateWorkflowListQuery query + InspectHeader _ -> Right () + InspectJournal ref query -> Inspection.validateJournalQuery (refName ref) (refId ref) query + InspectSteps ref query -> Inspection.validateStepQuery (refName ref) (refId ref) query + InspectChildren ref query -> Inspection.validateChildQuery (refName ref) (refId ref) query + InspectAwakeables ref query -> Inspection.validateAwakeableQuery (refName ref) (refId ref) query +validateCommand _ = Right () + runResumeOnce :: Maybe ResumeHook -> OpsEnv -> ResumeOptions -> IO OpsOutcome runResumeOnce Nothing _ _ = pure (Failed "workflow resume hook is not mounted") @@ -360,41 +443,49 @@ options.limit runShow :: OpsEnv -> WorkflowRef -> IO OpsOutcome -runShow env ref = - runAction env action $ \case - Nothing -> Failed (workflowLabel ref <> " was not found") - Just details -> Succeeded (workflowDetailsResult details) +runShow env ref = runAction env action $ \case + Left err -> inspectionFailure err + Right Nothing -> Failed (workflowLabel ref <> " was not found") + Right (Just details) -> Succeeded (workflowDetailsResult details) where action = do instanceRow <- Instance.lookupInstance (refName ref) (refId ref) case instanceRow of - Nothing -> pure Nothing + Nothing -> pure (Right Nothing) Just row -> do - children <- Child.lookupChildrenOfParent ref.workflowId ref.workflowName - awakeables <- lookupWorkflowAwakeables ref - pure (Just (row, children, awakeables)) + children <- drainPages (\cursor -> Inspection.listWorkflowChildrenPage (refName ref) (refId ref) (Inspection.OwnerPageQuery 500 cursor)) (.items) (.nextCursor) + awakeables <- drainPages (\cursor -> Inspection.listWorkflowAwakeablesPage (refName ref) (refId ref) (Inspection.OwnerPageQuery 500 cursor)) (.items) (.nextCursor) + pure ((\cs as -> Just (row, cs, as)) <$> children <*> awakeables) runSteps :: OpsEnv -> InspectOptions -> IO OpsOutcome -runSteps env options = - runAction env action (Succeeded . uncurry stepsResult) +runSteps env options = runAction env action $ \(gen, rows) -> either inspectionFailure (Succeeded . stepsResult gen) rows where action = do - selectedGeneration <- resolveGeneration options - steps <- WorkflowSchema.loadStepIndex (refName options.target) (refId options.target) selectedGeneration - pure (selectedGeneration, steps) + gen <- resolveGeneration options + rows <- drainPages (\cursor -> Inspection.loadStepIndexPage (refName options.target) (refId options.target) (Inspection.DetailPageQuery (Just gen) 500 cursor)) (.items) (.nextCursor) + pure (gen, rows) runJournal :: OpsEnv -> InspectOptions -> IO OpsOutcome -runJournal env options = - runAction env action $ \(selectedGeneration, recorded) -> - case traverse decodeJournalView recorded of - Left err -> Failed err - Right views -> Succeeded (journalResult selectedGeneration views) +runJournal env options = runAction env action $ \(gen, rows) -> either inspectionFailure (Succeeded . journalResult gen) rows where action = do - selectedGeneration <- resolveGeneration options - recorded <- readJournalEvents (workflowGenerationStreamName (refName options.target) (refId options.target) selectedGeneration) - pure (selectedGeneration, recorded) + gen <- resolveGeneration options + rows <- drainPages (\cursor -> Inspection.readWorkflowJournalPage (refName options.target) (refId options.target) (Inspection.DetailPageQuery (Just gen) 500 cursor)) (.items) (.nextCursor) + pure (gen, rows) +-- Compatibility commands intentionally drain pages. Bounded inspect commands +-- above invoke one library page only, and the HTTP deadline still bounds these. +drainPages :: (Maybe InspectionCursor -> Eff es (Either Inspection.InspectionError page)) -> (page -> [a]) -> (page -> Maybe InspectionCursor) -> Eff es (Either Inspection.InspectionError [a]) +drainPages load items next = go Nothing [] + where + go cursor chunks = do + result <- load cursor + case result of + Left err -> pure (Left err) + Right page -> case next page of + Nothing -> pure (Right (concat (reverse (items page : chunks)))) + Just token -> go (Just token) (items page : chunks) + runAwakeable :: OpsEnv -> AwakeableCommand -> IO OpsOutcome runAwakeable env = \case AwakeableShow awakeableId -> @@ -514,76 +605,12 @@ pure options.generation -lookupWorkflowAwakeables :: (Store :> es) => WorkflowRef -> Eff es [Awakeable.AwakeableRow] -lookupWorkflowAwakeables ref = do - current <- WorkflowSchema.currentGeneration (refName ref) (refId ref) - stepIndexes <- traverse (WorkflowSchema.loadStepIndex (refName ref) (refId ref)) [0 .. current] - catMaybes <$> traverse Awakeable.lookupAwakeable (awakeableIds stepIndexes) - -awakeableIds :: [Map Text Value] -> [UUID] -awakeableIds indexes = - Set.toAscList . Set.fromList $ do - index <- indexes - (stepName, value) <- Map.toList index - if awakeableAllocStepPrefix `Text.isPrefixOf` stepName - then case Aeson.fromJSON value of - AesonTypes.Success (AwakeableId awakeableId) -> [awakeableId] - AesonTypes.Error _ -> [] - else [] - -readJournalEvents :: (Store :> es) => StreamName -> Eff es [RecordedEvent] -readJournalEvents streamName = go (StreamVersion 0) [] - where - pageSize = 256 - go cursor pages = do - page <- StoreRead.readStreamForward streamName cursor pageSize - if Vector.null page - then pure (concat (reverse pages)) - else - let nextCursor = (Vector.last page).streamVersion - in go nextCursor (Vector.toList page : pages) - -data JournalView = JournalView - { eventId :: !Text, - eventType :: !Text, - streamVersion :: !Int64, - globalPosition :: !Int64, - stepName :: !Text, - recordedAt :: !UTCTime, - payload :: !Value - } - -decodeJournalView :: RecordedEvent -> Either Text JournalView -decodeJournalView recorded = do - event <- firstShow (decodeRecorded workflowJournalCodec recorded) - let (stepName, recordedAt, payload) = case event of - StepRecorded name value timestamp -> (name, timestamp, value) - WorkflowCompleted timestamp -> (completedStepName, timestamp, Aeson.Null) - WorkflowCancelled timestamp -> (cancelledStepName, timestamp, Aeson.Null) - WorkflowFailed reason timestamp -> (failedStepName, timestamp, Aeson.toJSON reason) - WorkflowContinuedAsNew generation timestamp -> (continuedAsNewStepName, timestamp, Aeson.toJSON generation) - pure - JournalView - { eventId = case recorded.eventId of EventId value -> UUID.toText value, - eventType = case recorded.eventType of EventType value -> value, - streamVersion = case recorded.streamVersion of StreamVersion value -> value, - globalPosition = case recorded.globalPosition of GlobalPosition value -> value, - stepName, - recordedAt, - payload - } - -firstShow :: (Show err) => Either err value -> Either Text value -firstShow = \case - Left err -> Left ("workflow journal decode failed: " <> Text.pack (show err)) - Right value -> Right value - workflowListResult :: [Instance.WorkflowInstanceRow] -> OpsResult workflowListResult instances = OpsResult { headers = ["name", "id", "generation", "status", "attempts", "lease", "wake_after", "updated_at"], rows = map workflowListRow instances, - jsonValue = Aeson.toJSON (map workflowInstanceJson instances) + jsonValue = Aeson.toJSON (map Inspection.instanceToJson instances) } workflowListRow :: Instance.WorkflowInstanceRow -> [Text] @@ -598,24 +625,6 @@ timeText row.updatedAt ] -workflowInstanceJson :: Instance.WorkflowInstanceRow -> Value -workflowInstanceJson row = - object - [ "workflow_id" .= row.workflowId, - "workflow_name" .= row.workflowName, - "generation" .= row.generation, - "status" .= Instance.statusToText row.status, - "attempts" .= row.attempts, - "last_error" .= row.lastError, - "next_attempt_at" .= row.nextAttemptAt, - "wake_after" .= row.wakeAfter, - "leased_by" .= row.leasedBy, - "lease_expires_at" .= row.leaseExpiresAt, - "created_at" .= row.createdAt, - "updated_at" .= row.updatedAt, - "completed_at" .= row.completedAt - ] - workflowDetailsResult :: (Instance.WorkflowInstanceRow, [Child.ChildRow], [Awakeable.AwakeableRow]) -> OpsResult workflowDetailsResult (row, children, awakeables) = OpsResult @@ -634,41 +643,25 @@ ], jsonValue = object - [ "instance" .= workflowInstanceJson row, - "children" .= map childJson children, - "awakeables" .= map awakeableJson awakeables + [ "instance" .= Inspection.instanceToJson row, + "children" .= map Inspection.childToJson children, + "awakeables" .= map Inspection.awakeableToJson awakeables ] } -childJson :: Child.ChildRow -> Value -childJson row = - object - [ "child_id" .= row.childId, - "child_name" .= row.childName, - "parent_id" .= row.parentId, - "parent_name" .= row.parentName, - "await_step" .= row.awaitStep, - "status" .= Child.statusToText row.status, - "result" .= row.result, - "failure_reason" .= row.failureReason, - "created_at" .= row.createdAt, - "updated_at" .= row.updatedAt, - "completed_at" .= row.completedAt - ] - -stepsResult :: Int -> Map Text Value -> OpsResult +stepsResult :: Int -> [WorkflowSchema.WorkflowStepRow] -> OpsResult stepsResult generation steps = OpsResult { headers = ["step", "result"], - rows = [[name, truncateCell 120 (jsonText value)] | (name, value) <- Map.toAscList steps], + rows = [[row.stepName, truncateCell 120 (jsonText row.result)] | row <- steps], jsonValue = object [ "generation" .= generation, - "steps" .= [object ["step" .= name, "result" .= value] | (name, value) <- Map.toAscList steps] + "steps" .= map Inspection.stepToJson steps ] } -journalResult :: Int -> [JournalView] -> OpsResult +journalResult :: Int -> [Inspection.WorkflowJournalEntry] -> OpsResult journalResult generation views = OpsResult { headers = ["version", "event_type", "step", "recorded_at", "payload"], @@ -681,21 +674,9 @@ ] | view <- views ], - jsonValue = object ["generation" .= generation, "events" .= map journalViewJson views] + jsonValue = object ["generation" .= generation, "events" .= map Inspection.journalEntryToJson views] } -journalViewJson :: JournalView -> Value -journalViewJson view = - object - [ "event_id" .= view.eventId, - "event_type" .= view.eventType, - "stream_version" .= view.streamVersion, - "global_position" .= view.globalPosition, - "step_name" .= view.stepName, - "recorded_at" .= view.recordedAt, - "payload" .= view.payload - ] - awakeableResult :: Awakeable.AwakeableRow -> OpsResult awakeableResult row = OpsResult @@ -709,22 +690,9 @@ timeText row.updatedAt ] ], - jsonValue = awakeableJson row + jsonValue = Inspection.awakeableToJson row } -awakeableJson :: Awakeable.AwakeableRow -> Value -awakeableJson row = - object - [ "awakeable_id" .= UUID.toText row.awakeableId, - "owner_workflow_name" .= row.ownerWorkflowName, - "owner_workflow_id" .= row.ownerWorkflowId, - "status" .= Awakeable.statusToText row.status, - "payload" .= row.payload, - "created_at" .= row.createdAt, - "updated_at" .= row.updatedAt, - "completed_at" .= row.completedAt - ] - awakeablePreviewResult :: Text -> UUID -> Maybe Awakeable.AwakeableRow -> OpsResult awakeablePreviewResult operation awakeableId row = OpsResult @@ -735,7 +703,7 @@ [ "preview" .= True, "operation" .= operation, "disposition" .= disposition, - "awakeable" .= fmap awakeableJson row + "awakeable" .= fmap Inspection.awakeableToJson row ] } where @@ -756,7 +724,7 @@ [ "operation" .= operation, "outcome" .= outcome, "transitioned" .= transitioned, - "awakeable" .= fmap awakeableJson row + "awakeable" .= fmap Inspection.awakeableToJson row ] } where @@ -775,7 +743,7 @@ "operation" .= operation, "disposition" .= disposition, "target" .= object ["workflow_name" .= ref.workflowName, "workflow_id" .= ref.workflowId], - "instance" .= fmap workflowInstanceJson row + "instance" .= fmap Inspection.instanceToJson row ] } diff --git a/src/Keiro/Ops/WorkflowHeaderCache.hs b/src/Keiro/Ops/WorkflowHeaderCache.hs new file mode 100644 --- /dev/null +++ b/src/Keiro/Ops/WorkflowHeaderCache.hs @@ -0,0 +1,103 @@ +-- | Bounded reuse of pure header presentation. Every lookup still reads the +-- complete current header in PostgreSQL; callers must verify live schema first. +module Keiro.Ops.WorkflowHeaderCache + ( WorkflowHeaderCache, + newWorkflowHeaderCache, + workflowHeaderCacheUsage, + runHeaderCached, + ) +where + +import Data.Aeson (Value) +import Data.ByteString (ByteString) +import Data.ByteString qualified as ByteString +import Data.IORef +import Data.Map.Strict qualified as Map +import Data.Sequence qualified as Seq +import Data.Text (Text) +import Data.Text qualified as Text +import Data.Text.Encoding qualified as Text +import Hasql.Pool qualified as Pool +import Keiro.Ops.Env (OpsEnv (..)) +import Keiro.Ops.Render +import Keiro.Ops.Workflow (WorkflowRef (..), headerOutcome) +import Keiro.Workflow.Instance qualified as Instance +import Keiro.Workflow.Types (WorkflowId (..), WorkflowName (..)) +import Kiroku.Store.Connection (KirokuStore (..)) +import Kiroku.Store.Error (mapGenericUsageError) + +type View = (Text, Text) + +data Entry = Entry !Instance.WorkflowHeaderKey !OpsResult !ByteString !Integer + +data Entries = Entries !(Map.Map View Entry) !(Seq.Seq View) !Integer + +data WorkflowHeaderCache = WorkflowHeaderCache + { entries :: !(IORef Entries), + imageLimit :: !Int, + encode :: Value -> IO (Maybe ByteString) + } + +-- | At most 32 presentations, charged at most 1 MiB in total. Charge includes +-- input keys, exact row images, JSON values, rendered cells and encoded bodies. +-- A larger valid response can still be served without retaining its rendering. +-- The encoder must enforce the host's response cap and return exact JSON bytes. +newWorkflowHeaderCache :: Int -> (Value -> IO (Maybe ByteString)) -> IO WorkflowHeaderCache +newWorkflowHeaderCache responseLimit encode + | responseLimit <= 0 = ioError (userError "header response limit must be positive") + | otherwise = do + entries <- newIORef (Entries Map.empty Seq.empty 0) + -- PostgreSQL's image has field names and date arrays independent of the + -- public wire format. Allow its fixed overhead without an Int overflow. + let imageLimit = fromInteger (min (toInteger (maxBound :: Int)) (2 * toInteger responseLimit + 4096)) + pure WorkflowHeaderCache {entries, imageLimit, encode} + +-- | Charged retained bytes and entry count, for host diagnostics. +workflowHeaderCacheUsage :: WorkflowHeaderCache -> IO (Int, Integer) +workflowHeaderCacheUsage cache = do + Entries known _ bytes <- readIORef cache.entries + pure (Map.size known, bytes) + +runHeaderCached :: WorkflowHeaderCache -> OpsEnv -> WorkflowRef -> IO (OpsOutcome, Maybe ByteString) +runHeaderCached cache env ref = do + Entries known _ _ <- readIORef cache.entries + let view = (ref.workflowName, ref.workflowId) + previous = Map.lookup view known + key = case previous of Just (Entry image _ _ _) -> Just image; Nothing -> Nothing + forget = atomicModifyIORef' cache.entries (\entries -> (remove view entries, ())) + failure code message = forget >> pure (InspectionFailed code message, Nothing) + observed <- Pool.use env.store.pool (Instance.lookupInstanceConditionalSession cache.imageLimit key (WorkflowName ref.workflowName) (WorkflowId ref.workflowId)) + case observed of + Left err -> forget >> pure (Failed (Text.pack (show (mapGenericUsageError err))), Nothing) + Right Nothing -> forget >> pure (headerOutcome ref Nothing, Nothing) + Right (Just Instance.HeaderImageTooLarge) -> failure "inspection_result_too_large" "workflow header image exceeds the host limit" + Right (Just Instance.HeaderUnchanged) -> case previous of + Just (Entry _ result body _) -> pure (Succeeded result, Just body) + Nothing -> failure "inspection_header_inconsistent" "header comparison returned no initial image" + Right (Just (Instance.HeaderChanged image row)) -> case headerOutcome ref (Just row) of + Succeeded result -> do + encoded <- cache.encode result.jsonValue + case encoded of + Nothing -> failure "inspection_result_too_large" "encoded result exceeds the host limit; narrow the read" + Just body -> do + let charge = 4096 + toInteger (ByteString.length (Text.encodeUtf8 ref.workflowName)) + toInteger (ByteString.length (Text.encodeUtf8 ref.workflowId)) + 3 * toInteger (Instance.workflowHeaderKeyBytes image) + 6 * toInteger (ByteString.length body) + entry = Entry image result body charge + atomicModifyIORef' cache.entries (\entries -> (insert view entry entries, ())) + pure (Succeeded result, Just body) + outcome -> forget >> pure (outcome, Nothing) + +remove :: View -> Entries -> Entries +remove view (Entries known order bytes) = case Map.lookup view known of + Nothing -> Entries known order bytes + Just (Entry _ _ _ charge) -> Entries (Map.delete view known) (Seq.filter (/= view) order) (bytes - charge) + +insert :: View -> Entry -> Entries -> Entries +insert view entry@(Entry _ _ _ charge) entries = + let without = remove view entries + in if charge > 1024 * 1024 then without else fit without + where + fit (Entries known order bytes) + | Map.size known < 32 && bytes + charge <= 1024 * 1024 = Entries (Map.insert view entry known) (order Seq.|> view) (bytes + charge) + | otherwise = case Seq.viewl order of + oldest Seq.:< _ -> fit (remove oldest (Entries known order bytes)) + Seq.EmptyL -> Entries Map.empty Seq.empty 0 diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -29,6 +29,7 @@ import Hasql.Transaction qualified as Tx import Keiro.DeadLetter import Keiro.Inbox qualified as Inbox +import Keiro.Inspection.Cursor (InspectionCursor (..)) import Keiro.Integration.Event import Keiro.Ops (AppHooks (..)) import Keiro.Ops qualified as Ops @@ -74,6 +75,7 @@ import Kiroku.Store.Types import Options.Applicative qualified as Optparse import Pgmq.Migration qualified as PgmqMigration +import ReactionInspectionSpec qualified import System.Exit (ExitCode (..)) import System.Process (readProcessWithExitCode) import Test.Hspec @@ -89,7 +91,9 @@ { workflowResume = Just (Map.empty, defaultWorkflowResumeOptions), timerFire = Just (\_ -> pure Nothing), replayAudit = Just (OpsReplayAudit.OpsAuditConfig []), - projectionCatalog = Just emptyCatalogOperations + projectionCatalog = Just emptyCatalogOperations, + aggregates = Just Map.empty, + processManagers = Just Map.empty } emptyCatalogOperations :: CatalogOperations.ProjectionCatalogOperations @@ -111,8 +115,12 @@ spec :: Fixture -> Spec spec fixture = do + ReactionInspectionSpec.spec fixture describe "embedded command tree" do it "omits code-dependent commands from the standalone tree" do + isParseSuccess (parseOps Ops.emptyAppHooks ["pm", "show", "--name", "test", "--correlation", "one"]) `shouldBe` False + isParseSuccess (parseOps Ops.emptyAppHooks ["pm", "types"]) `shouldBe` False + isParseSuccess (parseOps Ops.emptyAppHooks ["aggregate", "types"]) `shouldBe` False isParseFailure (parseOps Ops.emptyAppHooks ["wf", "resume-once"]) `shouldBe` True isParseFailure (parseOps Ops.emptyAppHooks ["timer", "drain-once"]) `shouldBe` True isParseFailure (parseOps Ops.emptyAppHooks ["replay-audit", "--full"]) `shouldBe` True @@ -124,6 +132,14 @@ isParseFailure (parseOps Ops.emptyAppHooks ["rebuild", "reproject-stream", "ops-group", "ops-projection", "orders-1"]) `shouldBe` True it "mounts every code-dependent command from typed application hooks" do + isParseSuccess (parseOps embeddedHooks ["pm", "show", "--name", "test", "--correlation", "one"]) `shouldBe` True + isParseSuccess (parseOps embeddedHooks ["pm", "list", "--name", "test"]) `shouldBe` True + isParseSuccess (parseOps embeddedHooks ["pm", "types"]) `shouldBe` True + isParseSuccess (parseOps embeddedHooks ["pm", "types", "--name", "test"]) `shouldBe` False + isParseSuccess (parseOps embeddedHooks ["aggregate", "types"]) `shouldBe` True + isParseSuccess (parseOps embeddedHooks ["aggregate", "show", "--name", "counter", "--id", "one"]) `shouldBe` True + isParseSuccess (parseOps embeddedHooks ["aggregate", "snapshot", "--name", "counter", "--id", "one"]) `shouldBe` True + isParseSuccess (parseOps embeddedHooks ["aggregate", "list", "--name", "counter", "--limit", "2"]) `shouldBe` True isParseSuccess (parseOps embeddedHooks ["wf", "resume-once"]) `shouldBe` True isParseSuccess (parseOps embeddedHooks ["timer", "drain-once"]) `shouldBe` True isParseSuccess (parseOps embeddedHooks ["replay-audit", "--full"]) `shouldBe` True @@ -977,6 +993,34 @@ workflowStatus store ref `shouldReturn` Nothing describe "timer handlers" $ around (withFreshStore fixture) do + it "renders pending timer pages with reusable cursors and due time" $ \store -> do + now <- getCurrentTime + let request = timerRequest "018f5f43-8a70-7b9a-9a9b-59d391a76720" (addUTCTime (-60) now) + another = timerRequest "018f5f43-8a70-7b9a-9a9b-59d391a76721" (addUTCTime (-60) now) + options = OpsTimer.PendingListOptions Nothing Nothing Nothing 1 + expectStore store (runTransaction (Timer.scheduleTimerTx request >> Timer.scheduleTimerTx another)) + first <- OpsTimer.runCommand (opsEnv False store) (OpsTimer.PendingList options) + case first of + Succeeded result -> case result.jsonValue of + Aeson.Object fields -> do + Just (Aeson.String cursor) <- pure (KeyMap.lookup "next_cursor" fields) + Just (Aeson.Array items) <- pure (KeyMap.lookup "items" fields) + Vector.length items `shouldBe` 1 + case Vector.head items of + Aeson.Object item -> KeyMap.lookup "due_in_seconds" item `shouldSatisfy` (\case Just (Aeson.Number n) -> n < 0; _ -> False) + _ -> expectationFailure "expected timer item" + second <- OpsTimer.runCommand (opsEnv False store) (OpsTimer.PendingList (options {OpsTimer.after = Just (InspectionCursor cursor)})) + case second of + Succeeded rendered -> case rendered.jsonValue of + Aeson.Object page -> KeyMap.lookup "next_cursor" page `shouldBe` Nothing + _ -> expectationFailure "expected timer page" + _ -> expectationFailure (show second) + _ -> expectationFailure "expected timer page" + _ -> expectationFailure (show first) + filtered <- OpsTimer.runCommand (opsEnv False store) (OpsTimer.PendingList (options {OpsTimer.manager = Just "missing"})) + filtered `shouldSatisfy` isSucceeded + timerStatus store request.timerId `shouldReturn` Just Timer.Scheduled + it "previews and dispatches one bounded due-timer pass through the mounted hook" $ \store -> do now <- getCurrentTime let request = timerRequest "018f5f43-8a70-7b9a-9a9b-59d391a76722" (addUTCTime (-60) now) @@ -1253,6 +1297,22 @@ dlqDepth `shouldReturn` 0 describe "projection handlers" $ around (withFreshStore fixture) do + it "renders bounded projection status through the owning public relation" $ \store -> do + expectStore store (runTransaction (Tx.sql "CREATE SCHEMA app; CREATE TABLE app.ops_catalog (id bigint PRIMARY KEY)")) + catalog <- expectValidatedCatalog (opsCatalog "status-codec") + _ <- expectStore store (Rebuild.registerProjectionCatalog catalog) + Right page <- expectStore store (Rebuild.listProjectionGroupStatusesPage (Rebuild.ProjectionStatusPageRequest 1 Nothing)) + shown <- OpsProjection.runCommand (opsEnv False store) (OpsProjection.Status (OpsProjection.ProjectionStatusOptions 1 Nothing)) + case shown of + Succeeded result -> case result.jsonValue of + Aeson.Object fields -> do + KeyMap.lookup "items" fields `shouldBe` Just (Aeson.toJSON (map OpsProjection.projectionGroupStatusJson page.items)) + KeyMap.member "observed_at" fields `shouldBe` True + KeyMap.lookup "visible_store_head" fields `shouldBe` Just (Aeson.toJSON (0 :: Int)) + KeyMap.lookup "next_cursor" fields `shouldBe` Nothing + _ -> expectationFailure "expected projection status page" + _ -> expectationFailure (show shown) + it "prunes only the named dedup rows" $ \store -> do _ <- seedKirokuEvent store "projection-source" "018f5f43-8a70-7b9a-9a9b-59d391a76810" Nothing events <- expectStore store (readStreamForward (StreamName "projection-source") (StreamVersion 0) 1) @@ -1285,6 +1345,32 @@ status <- OpsShard.runCommand (opsEnv False store) (OpsShard.Status "ops-shards") resultArrayLengthFromObject "ownership" status `shouldBe` Just 2 + detailed <- expectStore store (Shard.shardOwnershipView subscription) + case status of + Succeeded result -> case result.jsonValue of + Aeson.Object fields -> do + KeyMap.member "observed_at" fields `shouldBe` True + Just (Aeson.Array ownership) <- pure (KeyMap.lookup "ownership" fields) + Vector.length ownership `shouldBe` length detailed.buckets + case Vector.head ownership of + Aeson.Object row -> do + KeyMap.member "heartbeat_at" row `shouldBe` True + KeyMap.member "lease_age_seconds" row `shouldBe` True + _ -> expectationFailure "expected ownership row" + _ -> expectationFailure "expected shard status" + _ -> expectationFailure (show status) + listed <- OpsShard.runCommand (opsEnv False store) OpsShard.List + case listed of + Succeeded result -> case result.jsonValue of + Aeson.Object fields -> do + Just (Aeson.Array items) <- pure (KeyMap.lookup "items" fields) + Vector.length items `shouldBe` 1 + case Vector.head items of + Aeson.Object row -> KeyMap.lookup "buckets_live" row `shouldBe` Just (Aeson.toJSON (2 :: Int)) + _ -> expectationFailure "expected shard summary" + _ -> expectationFailure "expected shard list" + _ -> expectationFailure (show listed) + preview <- OpsShard.runCommand (opsEnv False store) (OpsShard.Relinquish "ops-shards" worker) preview `shouldSatisfy` isPreview ownersBefore <- expectStore store (Shard.ownershipSnapshotFor subscription) @@ -1316,6 +1402,15 @@ regfileShapeHash = expected.regfileShapeHash, stateShapeHash = expected.stateShapeHash } + + shown <- OpsSnapshot.runCommand (opsEnv False store) (OpsSnapshot.Show "snapshot-ops") + case shown of + Succeeded result -> case result.jsonValue of + Aeson.Object fields -> do + KeyMap.lookup "observed_at" fields `shouldSatisfy` (\case Just (Aeson.String _) -> True; _ -> False) + KeyMap.lookup "age_seconds" fields `shouldSatisfy` (\case Just (Aeson.Number n) -> n >= 0; _ -> False) + _ -> expectationFailure "expected snapshot JSON object" + _ -> expectationFailure "expected snapshot show success" missing <- OpsSnapshot.runCommand (opsEnv False store) (OpsSnapshot.TruncationPreflight "no-snapshot" (StreamVersion 2) (Just expected)) jsonBool "passed" missing `shouldBe` Just False diff --git a/test/ReactionInspectionSpec.hs b/test/ReactionInspectionSpec.hs new file mode 100644 --- /dev/null +++ b/test/ReactionInspectionSpec.hs @@ -0,0 +1,170 @@ +module ReactionInspectionSpec (spec) where + +import Control.Exception (finally) +import Control.Monad (when) +import Data.Aeson (Value (..), object, (.=)) +import Data.Aeson qualified as Aeson +import Data.Aeson.Encoding qualified as Encoding +import Data.Aeson.Key qualified as Key +import Data.Aeson.KeyMap qualified as KeyMap +import Data.ByteString.Lazy qualified as Lazy +import Data.IORef (modifyIORef', newIORef, readIORef, writeIORef) +import Data.List.NonEmpty (NonEmpty (..)) +import Data.Map.Strict qualified as Map +import Data.Text (Text) +import Data.Text qualified as Text +import Data.Time (getCurrentTime) +import Data.UUID qualified as UUID +import Keiki.Core +import Keiro +import Keiro.Ops qualified as Ops +import Keiro.Ops.Env +import Keiro.Ops.ProcessManager qualified as PM +import Keiro.Ops.Render +import Keiro.Ops.WorkflowHeaderCache (newWorkflowHeaderCache, workflowHeaderCacheUsage) +import Keiro.ProcessManager.Inspect qualified as Inspect +import Keiro.ProcessManager.Reaction qualified as Reaction +import Keiro.Stream qualified as Stream +import Keiro.Test.Postgres +import Keiro.Timer qualified as Timer +import Kiroku.Store.Connection (ConnectionSettingsM (..)) +import Kiroku.Store.Effect (runStoreIO) +import Kiroku.Store.Read (getStream) +import Kiroku.Store.Types hiding (id) +import Options.Applicative qualified as Options +import Test.Hspec + +spec :: Fixture -> Spec +spec fixture = describe "reaction process-manager commands" $ + around (withFreshResourceStore fixture) $ do + it "lists metadata with an unusable store and fully renders both formats" $ \_ -> do + armed <- newIORef False + let observe _ = readIORef armed >>= \active -> when active (fail "pm types touched the store") + withFreshResourceStoreWith fixture (\settings -> settings {observationHandler = Just observe}) $ \(unusable, _) -> + flip finally (writeIORef armed False) $ do + writeIORef armed True + let unused = error "pm types evaluated an application callback" + manager = Reaction.ReactiveProcessManager "modern" id handler (\_ -> unused) definition (const []) (\_ -> unused) + descriptor = + (Inspect.reactiveProcessManagerInspector manager (Stream.categoryUnsafe "opsreaction") (\_ _ -> unused)) + { Inspect.correlationOf = \_ -> unused, + Inspect.stateLabel = \_ -> unused, + Inspect.enabledInputs = \_ -> unused + } + catalog = Map.fromList [("z-alias", Inspect.SomeProcessManagerInspector descriptor), ("a-alias", Inspect.SomeProcessManagerInspector descriptor)] + hooks = Ops.emptyAppHooks {Ops.processManagers = Just catalog} + env = OpsEnv unusable HumanTable False [] False + runStoreIO unusable (getStream (StreamName "sentinel")) `shouldThrow` anyIOException + PM.validateCommand PM.Types `shouldBe` Right () + PM.isMutation PM.Types `shouldBe` False + Succeeded result <- PM.runCommand catalog env PM.Types + result.rows `shouldBe` [["a-alias", "opsreaction", "false", "never", "unknown"], ["z-alias", "opsreaction", "false", "never", "unknown"]] + Text.length (renderHuman result) `shouldSatisfy` (> 0) + Text.length (jsonText result.jsonValue) `shouldSatisfy` (> 0) + case Options.execParserPure Options.defaultPrefs (Options.info (Ops.commandParser hooks) mempty) ["pm", "types"] of + Options.Success command -> Ops.runCommand hooks env command `shouldReturn` Succeeded result + _ -> expectationFailure "pm types did not parse" + Succeeded empty <- PM.runCommand Map.empty env PM.Types + empty.jsonValue `shouldBe` object ["items" .= ([] :: [Value])] + cache <- newWorkflowHeaderCache 1048576 (const (fail "metadata used the header cache encoder")) + calls <- newIORef (0 :: Int) + let encodeDirect encoding = do + modifyIORef' calls (+ 1) + pure (Just (Lazy.toStrict (Encoding.encodingToLazyByteString encoding))) + command = Ops.ProcessManager PM.Types + (Succeeded prepared, Just bytes) <- Ops.runCommandWithReadEncoding encodeDirect cache hooks env command + prepared `shouldBe` result + Aeson.eitherDecodeStrict bytes `shouldBe` Right result.jsonValue + _ <- Ops.runCommandWithReadEncoding encodeDirect cache hooks env command + readIORef calls `shouldReturn` 2 + workflowHeaderCacheUsage cache `shouldReturn` (0, 0) + (refused, body) <- Ops.runCommandWithReadEncoding (const (pure Nothing)) cache hooks env command + refused `shouldSatisfy` (\case InspectionFailed "inspection_result_too_large" _ -> True; _ -> False) + body `shouldBe` Nothing + it "mounts a handwritten reaction inspector through pm show and pm list" $ \(store, StoreRunner run) -> do + clock <- getCurrentTime + let source = RecordedEvent (EventId UUID.nil) (EventType "Input") (StreamVersion 1) (GlobalPosition 1) (StreamId 1) (StreamVersion 1) Null Nothing Nothing Nothing clock + timer = Timer.TimerRequest (Timer.TimerId UUID.nil) "modern" "one" clock (object []) + manager = + Reaction.ReactiveProcessManager + "modern" + id + handler + (\cid -> stream ("opsreaction-" <> cid)) + definition + (const []) + (const (Reaction.AdvanceReaction Go [] [Reaction.FollowSchedule Reaction.Once timer])) + descriptor = Inspect.reactiveProcessManagerInspector manager (Stream.categoryUnsafe "opsreaction") (\state _ -> String (Text.pack (show state))) + hooks = Ops.emptyAppHooks {Ops.processManagers = Just (Map.singleton "modern" (Inspect.SomeProcessManagerInspector descriptor))} + invoke args = case Options.execParserPure Options.defaultPrefs (Options.info (Ops.commandParser hooks) mempty) args of + Options.Success command -> + Ops.runCommand hooks (OpsEnv store Json False [] False) command >>= \case + Succeeded result -> pure result.jsonValue + other -> fail (show other) + _ -> fail "inspection command did not parse" + Right (Right _) <- run (Reaction.runReactiveProcessManagerOnce defaultRunCommandOptions manager source "one") + shown <- invoke ["pm", "show", "--name", "modern", "--correlation", "one"] + field "state" shown `shouldBe` Just (String "Finished") + field "state_version" shown `shouldBe` Just (Number 1) + case field "last_reaction" shown of + Just last -> do + field "provenance_status" last `shouldBe` Just (String "unavailable") + field "provenance" last `shouldBe` Just Null + Nothing -> expectationFailure "accepted saga event is missing" + case field "pending_timers" shown >>= field "items" of + Just (Array items) -> length items `shouldBe` 1 + _ -> expectationFailure "reaction timer is missing" + case field "waiting_for" shown of + Just (Array items) -> length items `shouldBe` 1 + _ -> expectationFailure "live input candidate is missing" + listed <- invoke ["pm", "list", "--name", "modern", "--limit", "1"] + case field "items" listed of + Just (Array rows) -> map (field "correlation_id") (foldr (:) [] rows) `shouldBe` [Just (String "one")] + _ -> expectationFailure "reaction journal is missing from the manager catalog" + +field :: Text -> Value -> Maybe Value +field key (Object fields) = KeyMap.lookup (Key.fromText key) fields +field _ _ = Nothing + +data State = Ready | Finished deriving stock (Eq, Show, Ord, Enum, Bounded) + +data Cmd = Go deriving stock (Eq, Show) + +data Event = Observed deriving stock (Eq, Show) + +definition :: ValidatedEventStream (HsPred '[] Cmd) '[] State Cmd Event +definition = + mkEventStreamOrThrow + "ops-reaction-inspection" + EventStream + { transducer = + SymTransducer + { edgesOut = const [Edge {guard = matchInCtor goCtor, update = UKeep, output = [pack goCtor eventCtor oNil], target = Finished, mode = Live}], + initial = Ready, + initialRegs = RNil, + isFinal = const False + }, + initialState = Ready, + initialRegisters = RNil, + eventCodec = + Codec + { eventTypes = EventType "Observed" :| [], + eventType = const (EventType "Observed"), + schemaVersion = 1, + encode = const Null, + decode = \kind _ -> if kind == EventType "Observed" then Right Observed else Left "unknown fixture event", + upcasters = [] + }, + resolveStreamName = Stream.streamName, + snapshotPolicy = Never, + stateCodec = Nothing + } + +handler :: DomainCommandHandler (HsPred '[] Cmd) '[] State Cmd Event Text Text +handler = DomainCommandHandler definition (const (SilentNoOp "unused")) + +goCtor :: InCtor Cmd '[] +goCtor = unavailableInCtor "Go" (const (Just RNil)) (\RNil -> Go) + +eventCtor :: WireCtor Event () +eventCtor = unavailableWireCtor "Observed" (const (Just ())) (const Observed)