keiro-0.12.0.0: src/Keiro/Command.hs
-- | The command side of the framework: hydrate an aggregate, transduce, append.
--
-- Running a command against an 'EventStream' follows one pipeline:
--
-- 1. /Hydrate/ — replay the stream's stored events (optionally fast-forwarding
-- from a snapshot) through the keiki transducer to recover the current
-- @(state, registers)@ and stream version.
-- 2. /Transduce/ — step the transducer with the command. A rejected transition
-- yields 'CommandRejected', while multiple matching transitions yield
-- 'CommandAmbiguous'; a transition that emits no events yields a no-op
-- 'CommandResult'.
-- 3. /Append/ — encode the emitted events with the stream's 'Codec' and append
-- them at the expected version. An optimistic-concurrency conflict is
-- retried up to 'retryLimit' times by rehydrating and replaying; exhausting
-- the budget yields 'RetryExhausted'.
--
-- Three runners expose this pipeline at increasing levels of integration:
--
-- * 'runCommand' — append only.
-- * 'runCommandWithSql' — run an extra @afterAppend@ action in the /same/
-- transaction as the append (e.g. update an inline read model).
-- * 'runCommandWithSqlEvents' — same, but the callback also receives the
-- emitted events paired with their 'RecordedEvent's. This is the primitive
-- the projection, process-manager, and router layers build on.
--
-- The transactional runners apply Kiroku's configured @enrichEvent@ hook before
-- event preparation, exactly like 'runCommand'. They therefore require a
-- 'KirokuStoreResource' in the effect stack; install it with @withKirokuStore@
-- and interpret 'Store' with @runStoreResource@.
--
-- Per-stream hydration honors Kiroku's stream-truncation marker. A retained
-- snapshot must cover every hidden event (snapshot version at least marker minus
-- one); otherwise hydration fails with 'HydrationGapDetected'. A marker above
-- the stream head can instead appear as an empty stream and repeated append
-- conflicts end in 'ConflictFixpoint'. Keiro never truncates streams itself.
-- Kiroku's @$all@ and category/subscription reads are unaffected by per-stream
-- truncation: the marker hides events from stream reads rather than deleting
-- them from the global log.
--
-- Every successful append is replayed immediately from its pre-command state so
-- an unreplayable batch is witnessed at the moment it poisons the stream. The
-- post-commit witness is counted and attached to the command span without
-- changing the successful result. The same replay fold feeds transparent
-- snapshot writes when the stream's 'Keiro.EventStream.SnapshotPolicy' fires;
-- post-commit snapshot failures are likewise swallowed and counted. Every runner
-- accepts a tracer for optional OpenTelemetry spans.
--
-- The additive hydration primitives are also consumed by "Keiro.ReplayAudit".
-- The audit deliberately calls the seeded and full variants separately so the
-- public command-serving fallback cannot hide a stale or unreplayable seed.
module Keiro.Command
( -- * Results and errors
CommandResult (..),
DomainDecision (..),
DomainCommandOutcome (..),
SilentCommandContext (..),
SilentDomainDecision (..),
DomainCommandHandler (..),
CommandError (..),
HydrationReplayReason (..),
commandErrorClass,
-- * Options
RunCommandOptions (..),
defaultRunCommandOptions,
-- * Running commands
runCommand,
runDomainCommand,
forgetDomainDecision,
runCommandWithSql,
runCommandWithSqlEvents,
SqlTransactionDecision (..),
SqlCommandOutcome (..),
runCommandWithSqlEventsControlled,
runDomainCommandWithSql,
runDomainCommandWithSqlEvents,
DomainSqlCommandOutcome (..),
runDomainCommandWithSqlEventsControlled,
-- * Hydration primitives (replay audit)
Hydrated (..),
hydrate,
hydrateFull,
hydrateSeeded,
)
where
import Control.Concurrent (threadDelay)
import Control.Exception (displayException)
import Data.Aeson qualified as Aeson
import Data.ByteString.Lazy.Char8 qualified as LazyByteString
import Data.Functor (($>))
import Data.Int (Int32)
import Data.List.NonEmpty qualified as NonEmpty
import Data.Text qualified as Text
import Data.Void (Void)
import Effectful (Eff, IOE, (:>))
import Effectful.Concurrent (runConcurrent)
import Effectful.Concurrent.Async qualified as Async
import Effectful.Error.Static (Error, tryError)
import Effectful.Exception (trySync)
import GHC.Clock (getMonotonicTimeNSec)
import GHC.Stack (HasCallStack)
import Keiki.Core (BoolAlg, RegFile)
import Keiki.Core qualified as Keiki
import Keiro.Codec (Codec, CodecError, decodeRecorded, encodeForAppendWithMetadata)
import Keiro.Command.Domain (SilentCommandContext (..), SilentDomainDecision (..))
import Keiro.EventStream (EventStream, StateCodec, Terminality (..))
import Keiro.EventStream.Validate (ValidatedEventStream, unvalidated)
import Keiro.Prelude
import Keiro.ReplayDigest (canonicalJsonBytes, replayDigest)
import Keiro.Snapshot
( SnapshotLookup (..),
SnapshotMissReason (..),
SnapshotSeed,
encodeSnapshotStrict,
lookupSnapshotSeed,
writeSnapshotEncoded,
)
import Keiro.Snapshot.Policy (shouldSnapshotSpan)
import Keiro.Stream (Stream)
import Keiro.Telemetry
( CommandDecisionClass (..),
KeiroMetrics,
commandDecisionClassText,
keiro_command_decision,
keiro_events_appended,
keiro_replay_divergence,
keiro_retry_attempt,
recordCommandConflicts,
recordCommandDecision,
recordCommandDuplicates,
recordCommandRetries,
recordSnapshotApplyDivergence,
recordSnapshotDecodeFailures,
recordSnapshotEncodeFailures,
recordSnapshotReadHits,
recordSnapshotReadMisses,
recordSnapshotSeedDivergence,
recordSnapshotWriteFailures,
withCommandSpan,
)
import Kiroku.Store.Append (appendToStream)
import Kiroku.Store.Effect (Store)
import Kiroku.Store.Effect.Resource (KirokuStoreResource, getKirokuStore)
import Kiroku.Store.Error (StoreError (..))
import Kiroku.Store.Read (readStreamForwardStream)
import Kiroku.Store.Transaction
( PreparedEvent,
appendConflictToStoreError,
appendToStreamTx,
enrichEventsIO,
prepareEventsIO,
runTransaction,
)
import Kiroku.Store.Types
( AppendResult,
EventData,
EventId (..),
ExpectedVersion (..),
GlobalPosition (..),
RecordedEvent (..),
StreamName (..),
StreamVersion (..),
)
import OpenTelemetry.Attributes.Key (unkey)
import OpenTelemetry.SemanticConventions (db_system_name, error_type)
import OpenTelemetry.Trace.Core (Span, SpanStatus (..), Tracer, addAttribute, setStatus)
import Streamly.Data.Fold qualified as Fold
import Streamly.Data.Stream qualified as Streamly
import System.IO (stderr)
import System.Random.Stateful (globalStdGen, uniformRM)
import "hasql-transaction" Hasql.Transaction qualified as Tx
import Prelude qualified
-- | The outcome of a successfully handled command.
--
-- Reports the target 'Stream', the stream version after the command, the global
-- log position only when this command appended and the store assigned a real
-- one, and how many events were appended. A no-op reports @0@ events and
-- @Nothing@ for its global position because per-stream reads cannot recover a
-- true global position.
data CommandResult target = CommandResult
{ target :: !(Stream target),
streamVersion :: !StreamVersion,
-- | 'Just' only when this command appended; 'Nothing' for a no-op.
globalPosition :: !(Maybe GlobalPosition),
eventsAppended :: !Int
}
deriving stock (Generic, Eq, Show)
-- | The application-level decision made by one selected live edge.
-- Infrastructure failures and commands for which no edge was selected remain
-- 'CommandError's outside this value.
data DomainDecision co rejection noOp
= -- | The exact non-empty event batch that was encoded and appended.
DomainAccepted !(NonEmpty co)
| -- | An explicitly selected silent edge classified as a rejection.
DomainRejected !rejection
| -- | An explicitly selected silent edge classified as a successful no-op.
DomainNoOp !noOp
deriving stock (Generic, Eq, Show)
-- | A typed domain decision paired with the ordinary persistence metadata.
data DomainCommandOutcome target co rejection noOp = DomainCommandOutcome
{ decision :: !(DomainDecision co rejection noOp),
result :: !(CommandResult target)
}
deriving stock (Generic, Eq, Show)
-- | A validated stream plus pure application policy for selected silent edges.
-- The classifier does not select an edge and is invoked only after Keiki has
-- selected exactly one live edge whose output word is empty.
data DomainCommandHandler phi rs s ci co rejection noOp = DomainCommandHandler
{ eventStream :: !(ValidatedEventStream phi rs s ci co),
classifySilent :: !(SilentCommandContext rs s ci -> SilentDomainDecision rejection noOp)
}
deriving stock (Generic)
-- Internal adapter for the historical command API. Every selected silent edge
-- is a successful no-op; the rejection type is uninhabited because this
-- classifier never constructs 'SilentRejected'.
silentNoOpHandler ::
ValidatedEventStream phi rs s ci co ->
DomainCommandHandler phi rs s ci co Void ()
silentNoOpHandler eventStream =
DomainCommandHandler
{ eventStream,
classifySilent = \_ -> SilentNoOp ()
}
-- | Erase the typed domain decision while retaining historical command
-- persistence metadata. This is a collapse of successful matched decisions;
-- unmatched commands remain an outer 'Left' and never reach this adapter.
forgetDomainDecision :: DomainCommandOutcome target co rejection noOp -> CommandResult target
forgetDomainDecision DomainCommandOutcome {result} = result
-- | Why a command did not complete.
data CommandError
= -- | A stored event could not be decoded while rehydrating the aggregate.
HydrationDecodeFailed !CodecError
| -- | Replay of the stored events through the transducer stalled. The
-- version identifies the failing stored event; for
-- 'HydrationTruncatedChain' it identifies the last stored event, after
-- which the expected multi-event chain remained incomplete.
HydrationReplayFailed !StreamVersion !HydrationReplayReason
| -- | Hydration observed a non-contiguous stream version. Carries the
-- expected version followed by the observed version. The store writes
-- contiguous versions, so this indicates that stream truncation hid
-- events not covered by the hydration seed. Restore visibility with
-- @clearStreamTruncateBefore@ or provide a covering snapshot before
-- retrying the command.
HydrationGapDetected !StreamVersion !StreamVersion
| -- | No transducer edge matched the command in the hydrated state.
CommandRejected
| -- | Two or more transducer edges matched the command in the hydrated
-- state. This is a deterministic aggregate-definition bug rather than a
-- business rejection; the list contains the zero-based matched edge
-- indices in declaration order.
CommandAmbiguous ![Int]
| -- | An emitted event could not be encoded for append.
EncodeFailed !CodecError
| -- | The underlying store rejected the append.
StoreFailed !StoreError
| -- | Optimistic-concurrency retries were exhausted (carries the total
-- attempts made and the last store error).
RetryExhausted !Int !StoreError
| -- | Retrying after a 'StreamAlreadyExists' conflict re-observed the same
-- stream version: the store says the stream exists but reading it shows no
-- progress. The typical cause is a soft-deleted stream, where reads return
-- nothing but appends still collide. Carries the observed version and the
-- conflict.
ConflictFixpoint !StreamVersion !StoreError
deriving stock (Generic, Eq, Show)
-- | Whether an in-transaction command callback accepts the append or asks the
-- runner to roll the whole transaction back while retaining a typed outcome.
data SqlTransactionDecision a
= CommitSqlTransaction !a
| RollbackSqlTransaction !a
deriving stock (Generic, Eq, Show)
-- | Result of a controlled transactional command. A rolled-back outcome has no
-- 'CommandResult' because neither its event append nor its SQL effects exist.
data SqlCommandOutcome target a
= SqlCommandNoOp !(CommandResult target)
| SqlCommandCommitted !(CommandResult target) !a
| SqlCommandRolledBack !a
deriving stock (Generic, Eq, Show)
-- | Result of a controlled transactional domain command. Selected silent
-- decisions perform no transaction callback. A rolled-back accepted append has
-- no 'DomainCommandOutcome' because neither its append nor its SQL effects
-- exist.
data DomainSqlCommandOutcome target co rejection noOp a
= DomainSqlCommandSilent !(DomainCommandOutcome target co rejection noOp)
| DomainSqlCommandCommitted !(DomainCommandOutcome target co rejection noOp) !a
| DomainSqlCommandRolledBack !a
deriving stock (Generic, Eq, Show)
forgetDomainSqlOutcome ::
DomainSqlCommandOutcome target co rejection noOp a ->
SqlCommandOutcome target a
forgetDomainSqlOutcome = \case
DomainSqlCommandSilent outcome -> SqlCommandNoOp (forgetDomainDecision outcome)
DomainSqlCommandCommitted outcome userValue ->
SqlCommandCommitted (forgetDomainDecision outcome) userValue
DomainSqlCommandRolledBack userValue -> SqlCommandRolledBack userValue
-- | Why replay of stored events stalled, projected from keiki's structured
-- failure types onto a monomorphic vocabulary suitable for 'CommandError'.
data HydrationReplayReason
= -- | No edge's first output template could have produced the event.
HydrationNoInvertingEdge
| -- | More than one edge could have produced the event.
HydrationAmbiguousInversion
| -- | An event did not match the next expected event in a chain.
HydrationQueueMismatch
| -- | The stream ended in the middle of a multi-event chain.
HydrationTruncatedChain
deriving stock (Generic, Eq, Show)
-- | Knobs controlling a single command invocation.
--
-- * 'retryLimit' — how many times to rehydrate-and-replay after an
-- optimistic-concurrency conflict before giving up with 'RetryExhausted'.
-- * 'pageSize' — batch size when reading the stream during hydration.
-- * 'eventIds' — caller-supplied ids assigned to the emitted events in order;
-- the basis for deterministic, idempotent appends (see 'Keiro.Router' and
-- 'Keiro.ProcessManager').
-- * 'beforeAppend' — an observation/test hook run immediately before each
-- append attempt, primarily for injecting concurrent writes. It is not an
-- application transaction callback: it may run for an accepted decision that
-- later conflicts and is discarded.
-- * 'retryBackoffMicros' — base delay before the k-th OCC retry, capped at
-- 100 ms and jittered. Set to 0 to disable backoff.
-- * 'metrics' — optional metrics handle for command and snapshot counters.
-- * 'verifyReplayOnAppend' — replay every just-appended batch from the
-- pre-command state. Divergence is a post-commit advisory: it is counted and
-- attached to the command span, but the already-successful command still
-- succeeds. Snapshot-enabled streams always run the fold because snapshots
-- consume its result.
-- * 'seedVerifySampleRate' — verify one in N snapshot seeds against a full
-- replay through the seed version. The replay runs asynchronously and never
-- blocks or fails the command. This detects hand-written fold changes that
-- leave the snapshot discriminator unchanged; @0@ disables the witness.
data RunCommandOptions = RunCommandOptions
{ retryLimit :: !Int,
pageSize :: !Int32,
eventIds :: ![EventId],
beforeAppend :: !(IO ()),
retryBackoffMicros :: !Int,
metrics :: !(Maybe KeiroMetrics),
verifyReplayOnAppend :: !Bool,
seedVerifySampleRate :: !Int,
-- | Optional OpenTelemetry tracer. When 'Just', the command runner
-- opens an 'Internal'-kind span around each invocation, named after
-- the resolved stream identifier and decorated with the messaging /
-- error semantic-conventions attributes audited in
-- 'docs/research/opentelemetry-semconv-audit.md'. When 'Nothing',
-- the runner emits no spans.
tracer :: !(Maybe Tracer),
-- | Optional JSON merged into every event's metadata for this command
-- invocation. Carries ambient context such as actor type, agent id,
-- and session id. The codec always adds a @schemaVersion@ key; the
-- keys here are merged on top (see 'Keiro.Codec.metadataFor'). When
-- 'Nothing', events carry only the schema-version marker, exactly as
-- before this field existed.
metadata :: !(Maybe Value)
}
deriving stock (Generic)
-- | Sensible defaults: 3 retries, 256-event read pages, no caller-assigned
-- event ids, a no-op pre-append hook, 5ms retry backoff, no metrics, post-append
-- replay verification enabled, one sampled snapshot-seed verification per 1000
-- snapshot hits, no tracer, and no extra metadata.
defaultRunCommandOptions :: RunCommandOptions
defaultRunCommandOptions =
RunCommandOptions
{ retryLimit = 3,
pageSize = 256,
eventIds = [],
beforeAppend = pure (),
retryBackoffMicros = 5000,
metrics = Nothing,
verifyReplayOnAppend = True,
seedVerifySampleRate = 1000,
tracer = Nothing,
metadata = Nothing
}
data Hydrated rs s = Hydrated
{ state :: !s,
registers :: !(RegFile rs),
streamVersion :: !StreamVersion
}
deriving stock (Generic)
data DomainCommandPlan target rs s co rejection noOp
= DomainCommandSilent !(SilentDomainDecision rejection noOp) !(CommandResult target)
| DomainCommandAppend !(Hydrated rs s) !(NonEmpty co) ![EventData]
deriving stock (Generic)
hydrate ::
forall phi rs s ci co es.
(HasCallStack, IOE :> es, Store :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>
RunCommandOptions ->
EventStream phi rs s ci co ->
Stream (EventStream phi rs s ci co) ->
Eff es (Either CommandError (Hydrated rs s))
hydrate options eventStream targetStream =
snapshotSeed >>= \case
Nothing -> hydrateFull options eventStream targetStream
Just seed -> do
replayed <-
hydrateSeeded
options
eventStream
targetStream
(seed ^. #state)
(seed ^. #registers)
(seed ^. #streamVersion)
case replayed of
Left _ -> hydrateFull options eventStream targetStream
Right hydrated -> do
for_ (eventStream ^. #stateCodec) $ \codec ->
scheduleSeedVerification options eventStream targetStream codec seed
pure (Right hydrated)
where
snapshotSeed =
case eventStream ^. #stateCodec of
Nothing -> pure Nothing
Just codec -> do
lookupSnapshotSeed ((eventStream ^. #resolveStreamName) targetStream) codec >>= \case
SnapshotHit seed -> do
recordSnapshotReadHits (options ^. #metrics) 1
pure (Just seed)
SnapshotUnavailable reason -> do
recordSnapshotReadMisses (options ^. #metrics) 1
case reason of
SnapshotDecodeFailed _ -> recordSnapshotDecodeFailures (options ^. #metrics) 1
_ -> pure ()
pure Nothing
hydrateFull ::
forall phi rs s ci co es.
(HasCallStack, Store :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>
RunCommandOptions ->
EventStream phi rs s ci co ->
Stream (EventStream phi rs s ci co) ->
Eff es (Either CommandError (Hydrated rs s))
hydrateFull options eventStream targetStream =
hydrateSeeded
options
eventStream
targetStream
(eventStream ^. #initialState)
(eventStream ^. #initialRegisters)
(StreamVersion 0)
-- | Replay a stored stream from an arbitrary snapshot or initial-state seed.
--
-- The store stream is grouped into bounded lists and each decoded prefix is
-- handed to keiki's 'Keiki.replayEvents'. Decoding stops at the first bad event in
-- a group, but the valid prefix is replayed first so an earlier replay failure
-- retains precedence over a later codec failure.
hydrateSeeded ::
forall phi rs s ci co es.
(HasCallStack, Store :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>
RunCommandOptions ->
EventStream phi rs s ci co ->
Stream (EventStream phi rs s ci co) ->
s ->
RegFile rs ->
StreamVersion ->
Eff es (Either CommandError (Hydrated rs s))
hydrateSeeded options eventStream targetStream seedState seedRegisters seedVersion = do
hydrateSeededThrough
Nothing
options
eventStream
targetStream
seedState
seedRegisters
seedVersion
hydrateSeededThrough ::
forall phi rs s ci co es.
(HasCallStack, Store :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>
Maybe StreamVersion ->
RunCommandOptions ->
EventStream phi rs s ci co ->
Stream (EventStream phi rs s ci co) ->
s ->
RegFile rs ->
StreamVersion ->
Eff es (Either CommandError (Hydrated rs s))
hydrateSeededThrough replayThrough options eventStream targetStream seedState seedRegisters seedVersion = do
replayed <-
Streamly.fold
(Fold.foldlM' replayPage (pure (Right initialReplay)))
recordedPages
pure (finishReplay replayed)
where
readPageSize = Prelude.max 1 (options ^. #pageSize)
groupSize = Prelude.fromIntegral readPageSize
recordedPages =
Streamly.foldMany
(Fold.take groupSize Fold.toList)
boundedRecorded
boundedRecorded =
case replayThrough of
Nothing -> readStreamForwardStream resolvedName seedVersion readPageSize
Just endVersion ->
Streamly.takeWhile
(\recorded -> recorded ^. #streamVersion <= endVersion)
(readStreamForwardStream resolvedName seedVersion readPageSize)
resolvedName = (eventStream ^. #resolveStreamName) targetStream
initialReplay = (Keiki.Settled seedState, seedRegisters, Nothing)
replayPage ::
Either CommandError (Keiki.InFlight s co, RegFile rs, Maybe RecordedEvent) ->
[RecordedEvent] ->
Eff es (Either CommandError (Keiki.InFlight s co, RegFile rs, Maybe RecordedEvent))
replayPage (Left err) _ = pure (Left err)
replayPage (Right (wrapper, registers, previousRecorded)) page =
pure $ case Keiki.replayEvents (eventStream ^. #transducer) (wrapper, registers) decodedEvents of
Left replayFailure ->
Left (hydrationReplayError previousRecorded decodedRecorded replayFailure)
Right (nextWrapper, nextRegisters) ->
case pendingInputFailure of
Just err -> Left err
Nothing ->
Right
( nextWrapper,
nextRegisters,
latestRecorded decodedRecorded previousRecorded
)
where
(decodedRecorded, decodedEvents, pendingInputFailure) = decodePrefix previousRecorded page
decodePrefix :: Maybe RecordedEvent -> [RecordedEvent] -> ([RecordedEvent], [co], Maybe CommandError)
decodePrefix previousRecorded = go [] [] startingVersion
where
startingVersion = maybe seedVersion (^. #streamVersion) previousRecorded
go recordedAcc eventAcc lastSeen = \case
[] -> (Prelude.reverse recordedAcc, Prelude.reverse eventAcc, Nothing)
recorded : rest ->
let observed = recorded ^. #streamVersion
expected = nextStreamVersion lastSeen
in if observed /= expected
then
( Prelude.reverse recordedAcc,
Prelude.reverse eventAcc,
Just (HydrationGapDetected expected observed)
)
else case decodeRecorded (eventStream ^. #eventCodec) recorded of
Left err ->
( Prelude.reverse recordedAcc,
Prelude.reverse eventAcc,
Just (HydrationDecodeFailed err)
)
Right event ->
go (recorded : recordedAcc) (event : eventAcc) observed rest
nextStreamVersion (StreamVersion version) = StreamVersion (version Prelude.+ 1)
hydrationReplayError ::
Maybe RecordedEvent ->
[RecordedEvent] ->
Keiki.ReplayFailure s co ->
CommandError
hydrationReplayError previousRecorded decodedRecorded replayFailure =
HydrationReplayFailed failureVersion (toHydrationReason (Keiki.replayFailureReason replayFailure))
where
failureVersion =
maybe seedVersion (^. #streamVersion) failureRecorded
failureRecorded =
case Keiki.replayFailureReason replayFailure of
Keiki.ReplayLogTruncated {} -> latestRecorded decodedRecorded previousRecorded
Keiki.ReplayEventFailed {} ->
case recordedAt (Keiki.replayFailedIndex replayFailure) decodedRecorded of
Just recorded -> Just recorded
Nothing -> latestRecorded decodedRecorded previousRecorded
finishReplay = \case
Left err -> Left err
Right (wrapper, finalRegisters, lastRecorded) ->
case wrapper of
Keiki.Settled finalState ->
Right
Hydrated
{ state = finalState,
registers = finalRegisters,
streamVersion = maybe seedVersion (^. #streamVersion) lastRecorded
}
Keiki.InFlight {} ->
Left
( HydrationReplayFailed
(maybe seedVersion (^. #streamVersion) lastRecorded)
HydrationTruncatedChain
)
toHydrationReason = \case
Keiki.ReplayEventFailed stepFailure -> case stepFailure of
Keiki.ReplayNoInvertingEdge {} -> HydrationNoInvertingEdge
Keiki.ReplayAmbiguousInversions {} -> HydrationAmbiguousInversion
Keiki.ReplayQueueMismatch {} -> HydrationQueueMismatch
Keiki.ReplayLogTruncated {} -> HydrationTruncatedChain
latestRecorded recorded fallback =
case lastMaybe recorded of
Just latest -> Just latest
Nothing -> fallback
lastMaybe = \case
[] -> Nothing
first : rest -> Just (Prelude.foldl (\_ current -> current) first rest)
recordedAt eventIndex recorded =
case Prelude.drop eventIndex recorded of
found : _ -> Just found
[] -> Nothing
scheduleSeedVerification ::
forall phi rs s ci co es.
(HasCallStack, IOE :> es, Store :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>
RunCommandOptions ->
EventStream phi rs s ci co ->
Stream (EventStream phi rs s ci co) ->
StateCodec (s, RegFile rs) ->
SnapshotSeed rs s ->
Eff es ()
scheduleSeedVerification options eventStream targetStream codec seed = do
void $ trySync $ do
sampled <-
case options ^. #seedVerifySampleRate of
rate | rate <= 0 -> pure False
1 -> pure True
rate -> liftIO ((== (1 :: Int)) <$> uniformRM (1, rate) globalStdGen)
when sampled
$ void
$ runConcurrent
$ Async.async
$ verifySnapshotSeed options eventStream targetStream codec seed
verifySnapshotSeed ::
forall phi rs s ci co es.
(HasCallStack, IOE :> es, Store :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>
RunCommandOptions ->
EventStream phi rs s ci co ->
Stream (EventStream phi rs s ci co) ->
StateCodec (s, RegFile rs) ->
SnapshotSeed rs s ->
Eff es ()
verifySnapshotSeed options eventStream targetStream codec seed = do
let seedVersion = seed ^. #streamVersion
streamName = (eventStream ^. #resolveStreamName) targetStream
full <-
hydrateSeededThrough
(Just seedVersion)
options
eventStream
targetStream
(eventStream ^. #initialState)
(eventStream ^. #initialRegisters)
(StreamVersion 0)
seededEncoded <-
liftIO
$ encodeSnapshotStrict
codec
(seed ^. #state, seed ^. #registers)
case (seededEncoded, full) of
(Left seedEncodeError, _) ->
reportSeedDivergence
options
streamName
seedVersion
("encode-failed:" <> Text.pack (displayException seedEncodeError))
(case full of Left replayError -> "replay-failed:" <> Text.pack (show replayError); Right _ -> "not-compared:seed-encode-failed")
(Right seededValue, Left replayError) ->
reportSeedDivergence
options
streamName
seedVersion
(replayDigest seededValue)
("replay-failed:" <> Text.pack (show replayError))
(Right seededValue, Right fullHydrated)
| fullHydrated ^. #streamVersion /= seedVersion ->
reportSeedDivergence
options
streamName
seedVersion
(replayDigest seededValue)
("version-mismatch:" <> Text.pack (show (fullHydrated ^. #streamVersion)))
| otherwise -> do
fullEncoded <-
liftIO
$ encodeSnapshotStrict
codec
(fullHydrated ^. #state, fullHydrated ^. #registers)
case fullEncoded of
Left fullEncodeError ->
reportSeedDivergence
options
streamName
seedVersion
(replayDigest seededValue)
("encode-failed:" <> Text.pack (displayException fullEncodeError))
Right fullValue ->
unless
(canonicalJsonBytes seededValue == canonicalJsonBytes fullValue)
( reportSeedDivergence
options
streamName
seedVersion
(replayDigest seededValue)
(replayDigest fullValue)
)
reportSeedDivergence ::
(IOE :> es) =>
RunCommandOptions ->
StreamName ->
StreamVersion ->
Text ->
Text ->
Eff es ()
reportSeedDivergence options (StreamName streamName) (StreamVersion seedVersion) seededDigest fullDigest = do
recordSnapshotSeedDivergence (options ^. #metrics) 1
liftIO
$ LazyByteString.hPutStrLn stderr
$ Aeson.encode
$ Aeson.object
[ "event" Aeson..= ("keiro.snapshot.seed.divergence" :: Text),
"level" Aeson..= ("error" :: Text),
"stream" Aeson..= streamName,
"seedVersion" Aeson..= seedVersion,
"seededDigest" Aeson..= seededDigest,
"fullDigest" Aeson..= fullDigest
]
-- | Hydrate the target stream, transduce the command, and append any emitted
-- events. Retries optimistic-concurrency conflicts up to 'retryLimit'. This
-- is the plain runner with no in-transaction side effects.
runCommand ::
forall phi rs s ci co es.
(HasCallStack, IOE :> es, Store :> es, Error StoreError :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>
RunCommandOptions ->
ValidatedEventStream phi rs s ci co ->
Stream (EventStream phi rs s ci co) ->
ci ->
Eff es (Either CommandError (CommandResult (EventStream phi rs s ci co)))
runCommand options validatedEventStream targetStream command =
withCommandSpan (options ^. #tracer) (resolvedStreamName eventStream targetStream) Nothing $ \mSpan -> do
(outcome, attemptNo) <-
domainCommandAttempts
options
(silentNoOpHandler validatedEventStream)
targetStream
command
mSpan
let result = fmap forgetDomainDecision outcome
recordCommandOutcome mSpan (^. #eventsAppended) attemptNo result
pure result
where
eventStream = unvalidated validatedEventStream
-- | Hydrate, select and evaluate one live edge, then return the exact typed
-- domain decision from the successful final optimistic-concurrency attempt.
-- Eventful decisions append the same non-empty batch carried by
-- 'DomainAccepted'. Selected silent edges are classified purely and perform no
-- append. No matching edge and every infrastructure failure remain
-- 'CommandError's.
runDomainCommand ::
forall phi rs s ci co rejection noOp es.
(HasCallStack, IOE :> es, Store :> es, Error StoreError :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>
RunCommandOptions ->
DomainCommandHandler phi rs s ci co rejection noOp ->
Stream (EventStream phi rs s ci co) ->
ci ->
Eff es (Either CommandError (DomainCommandOutcome (EventStream phi rs s ci co) co rejection noOp))
runDomainCommand options handler@DomainCommandHandler {eventStream = validatedEventStream} targetStream command =
withCommandSpan (options ^. #tracer) (resolvedStreamName (unvalidated validatedEventStream) targetStream) Nothing $ \mSpan -> do
(outcome, attemptNo) <- domainCommandAttempts options handler targetStream command mSpan
recordDomainCommandOutcome options mSpan attemptNo outcome
pure outcome
-- | The optimistic-concurrency attempt loop shared by the plain and
-- transactional domain command runners: hydrate, detect a conflict fixpoint,
-- prepare the plan, classify a silent decision, or hand an accepted batch to
-- the caller's append action. The append action receives the retry
-- continuation so 'retryOrFail' can re-enter the loop.
domainCommandAttemptLoop ::
forall phi rs s ci co rejection noOp outcome es.
(HasCallStack, IOE :> es, Store :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>
RunCommandOptions ->
DomainCommandHandler phi rs s ci co rejection noOp ->
Stream (EventStream phi rs s ci co) ->
ci ->
(DomainCommandOutcome (EventStream phi rs s ci co) co rejection noOp -> outcome) ->
( (Int -> Maybe (StoreError, StreamVersion) -> Eff es (Either CommandError outcome, Int)) ->
Int ->
Hydrated rs s ->
NonEmpty co ->
[EventData] ->
Eff es (Either CommandError outcome, Int)
) ->
Eff es (Either CommandError outcome, Int)
domainCommandAttemptLoop options handler@DomainCommandHandler {eventStream = validatedEventStream} targetStream command wrapSilent appendAction =
attempt 1 Nothing
where
eventStream' = unvalidated validatedEventStream
attempt attemptNo lastConflict = do
hydrated <- hydrate options eventStream' targetStream
either (\err -> pure (Left err, attemptNo)) (runPlan attemptNo lastConflict) hydrated
runPlan attemptNo lastConflict current =
case conflictFixpoint lastConflict (current ^. #streamVersion) of
Just err -> pure (Left err, attemptNo)
Nothing ->
case prepareDomainCommandPlan options handler eventStream' targetStream current command of
Left err -> pure (Left err, attemptNo)
Right (DomainCommandSilent silentDecision result) ->
pure
( Right
( wrapSilent
DomainCommandOutcome
{ decision = domainDecisionFromSilent silentDecision,
result
}
),
attemptNo
)
Right (DomainCommandAppend current' events encoded) ->
appendAction attempt attemptNo current' events encoded
{-# INLINE domainCommandAttemptLoop #-}
domainCommandAttempts ::
forall phi rs s ci co rejection noOp es.
(HasCallStack, IOE :> es, Store :> es, Error StoreError :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>
RunCommandOptions ->
DomainCommandHandler phi rs s ci co rejection noOp ->
Stream (EventStream phi rs s ci co) ->
ci ->
Maybe Span ->
Eff es (Either CommandError (DomainCommandOutcome (EventStream phi rs s ci co) co rejection noOp), Int)
domainCommandAttempts options handler@DomainCommandHandler {eventStream = validatedEventStream} targetStream command mSpan =
domainCommandAttemptLoop options handler targetStream command Prelude.id appendOnce
where
eventStream' = unvalidated validatedEventStream
appendOnce retry attemptNo current events encoded = do
liftIO (options ^. #beforeAppend)
appended <-
tryError @StoreError
$ appendToStream
((eventStream' ^. #resolveStreamName) targetStream)
(expectedVersion (current ^. #streamVersion))
encoded
case appended of
Right appendResult -> do
verifyAndSnapshot options mSpan eventStream' current (NonEmpty.toList events) appendResult
pure
( Right
DomainCommandOutcome
{ decision = DomainAccepted events,
result = appendedResult targetStream appendResult (Prelude.length encoded)
},
attemptNo
)
Left (_, storeError) ->
retryOrFail options retry attemptNo (current ^. #streamVersion) storeError
-- | Like 'runCommand', but run @afterAppend@ inside the /same/ transaction
-- as the append, so a read-model write commits atomically with the events.
-- The callback's result is returned as @Just@ on append (and 'Nothing' for a
-- no-op command that appended nothing).
--
-- Requires 'KirokuStoreResource' so the transactional append applies the
-- configured @enrichEvent@ hook. See 'runCommandWithSqlEvents' for the callback's
-- locking and latency implications.
runCommandWithSql ::
forall phi rs s ci co a es.
(HasCallStack, IOE :> es, Store :> es, Error StoreError :> es, KirokuStoreResource :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>
RunCommandOptions ->
ValidatedEventStream phi rs s ci co ->
Stream (EventStream phi rs s ci co) ->
ci ->
(AppendResult -> Tx.Transaction a) ->
Eff es (Either CommandError (CommandResult (EventStream phi rs s ci co), Maybe a))
runCommandWithSql options eventStream targetStream command afterAppend =
runCommandWithSqlEvents options eventStream targetStream command (\_ appendResult -> afterAppend appendResult)
-- | The most general runner: like 'runCommandWithSql', but the
-- in-transaction callback also receives every emitted event paired with the
-- 'RecordedEvent' the store persisted for it, in append order. Inline
-- projections, process managers, and routers are all built on this.
--
-- The runner requires 'KirokuStoreResource' and applies the configured
-- @enrichEvent@ hook before preparing the append. The callback therefore sees
-- the enriched metadata that was persisted.
--
-- The append updates Kiroku's global @$all@ stream and holds its PostgreSQL row
-- lock until this transaction commits. Every SQL operation in the callback
-- therefore extends the store-wide append serialization window. Keep the
-- callback small: precompute outside the transaction where possible, batch
-- writes, and minimize database round trips.
runCommandWithSqlEvents ::
forall phi rs s ci co a es.
(HasCallStack, IOE :> es, Store :> es, Error StoreError :> es, KirokuStoreResource :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>
RunCommandOptions ->
ValidatedEventStream phi rs s ci co ->
Stream (EventStream phi rs s ci co) ->
ci ->
([(co, RecordedEvent)] -> AppendResult -> Tx.Transaction a) ->
Eff es (Either CommandError (CommandResult (EventStream phi rs s ci co), Maybe a))
runCommandWithSqlEvents options validatedEventStream targetStream command afterAppend =
fmap (fmap collapse)
$ runCommandWithSqlEventsControlled
options
validatedEventStream
targetStream
command
(\pairs appendResult -> CommitSqlTransaction <$> afterAppend pairs appendResult)
where
collapse = \case
SqlCommandNoOp result -> (result, Nothing)
SqlCommandCommitted result userValue -> (result, Just userValue)
SqlCommandRolledBack _ ->
error "runCommandWithSqlEvents: an always-commit callback rolled back"
-- | Variant of 'runCommandWithSqlEvents' whose callback may condemn the whole
-- append transaction and still return a typed result. Rolled-back attempts do
-- not run replay verification or snapshot writes. Catalog-derived projection
-- fencing uses this boundary so discovering a rebuilding group after the append
-- SQL has taken its locks cannot leave the event or a partial projection write.
runCommandWithSqlEventsControlled ::
forall phi rs s ci co a es.
(HasCallStack, IOE :> es, Store :> es, Error StoreError :> es, KirokuStoreResource :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>
RunCommandOptions ->
ValidatedEventStream phi rs s ci co ->
Stream (EventStream phi rs s ci co) ->
ci ->
([(co, RecordedEvent)] -> AppendResult -> Tx.Transaction (SqlTransactionDecision a)) ->
Eff es (Either CommandError (SqlCommandOutcome (EventStream phi rs s ci co) a))
runCommandWithSqlEventsControlled options validatedEventStream targetStream command afterAppend =
withCommandSpan (options ^. #tracer) (resolvedStreamName eventStream targetStream) Nothing $ \mSpan -> do
(outcome, attemptNo) <-
domainSqlCommandAttempts
options
(silentNoOpHandler validatedEventStream)
targetStream
command
afterAppend
mSpan
let result = fmap forgetDomainSqlOutcome outcome
recordCommandOutcome mSpan eventCount attemptNo result
pure result
where
eventStream = unvalidated validatedEventStream
eventCount = \case
SqlCommandNoOp result -> result ^. #eventsAppended
SqlCommandCommitted result _ -> result ^. #eventsAppended
SqlCommandRolledBack _ -> 0
-- | Domain-aware counterpart to 'runCommandWithSql'. The callback runs only
-- for an accepted non-empty event batch and commits atomically with it. A typed
-- rejection or no-op returns 'Nothing' and opens no SQL transaction.
runDomainCommandWithSql ::
forall phi rs s ci co rejection noOp a es.
(HasCallStack, IOE :> es, Store :> es, Error StoreError :> es, KirokuStoreResource :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>
RunCommandOptions ->
DomainCommandHandler phi rs s ci co rejection noOp ->
Stream (EventStream phi rs s ci co) ->
ci ->
(AppendResult -> Tx.Transaction a) ->
Eff es (Either CommandError (DomainCommandOutcome (EventStream phi rs s ci co) co rejection noOp, Maybe a))
runDomainCommandWithSql options handler targetStream command afterAppend =
runDomainCommandWithSqlEvents options handler targetStream command (\_ appendResult -> afterAppend appendResult)
-- | Domain-aware counterpart to 'runCommandWithSqlEvents'. Accepted commands
-- pass the exact typed events paired with their reconstructed persisted events
-- to the callback in append order. Selected silent decisions never invoke it.
runDomainCommandWithSqlEvents ::
forall phi rs s ci co rejection noOp a es.
(HasCallStack, IOE :> es, Store :> es, Error StoreError :> es, KirokuStoreResource :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>
RunCommandOptions ->
DomainCommandHandler phi rs s ci co rejection noOp ->
Stream (EventStream phi rs s ci co) ->
ci ->
([(co, RecordedEvent)] -> AppendResult -> Tx.Transaction a) ->
Eff es (Either CommandError (DomainCommandOutcome (EventStream phi rs s ci co) co rejection noOp, Maybe a))
runDomainCommandWithSqlEvents options handler targetStream command afterAppend =
fmap (fmap collapse)
$ runDomainCommandWithSqlEventsControlled
options
handler
targetStream
command
(\pairs appendResult -> CommitSqlTransaction <$> afterAppend pairs appendResult)
where
collapse = \case
DomainSqlCommandSilent outcome -> (outcome, Nothing)
DomainSqlCommandCommitted outcome userValue -> (outcome, Just userValue)
DomainSqlCommandRolledBack _ ->
error "runDomainCommandWithSqlEvents: an always-commit callback rolled back"
-- | Controlled domain transaction variant used by catalog projection fences.
-- A rollback discards the accepted batch and callback effects and therefore
-- cannot fabricate a successful 'DomainCommandOutcome'.
runDomainCommandWithSqlEventsControlled ::
forall phi rs s ci co rejection noOp a es.
(HasCallStack, IOE :> es, Store :> es, Error StoreError :> es, KirokuStoreResource :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>
RunCommandOptions ->
DomainCommandHandler phi rs s ci co rejection noOp ->
Stream (EventStream phi rs s ci co) ->
ci ->
([(co, RecordedEvent)] -> AppendResult -> Tx.Transaction (SqlTransactionDecision a)) ->
Eff es (Either CommandError (DomainSqlCommandOutcome (EventStream phi rs s ci co) co rejection noOp a))
runDomainCommandWithSqlEventsControlled options handler@DomainCommandHandler {eventStream = validatedEventStream} targetStream command afterAppend =
withCommandSpan (options ^. #tracer) (resolvedStreamName (unvalidated validatedEventStream) targetStream) Nothing $ \mSpan -> do
(outcome, attemptNo) <- domainSqlCommandAttempts options handler targetStream command afterAppend mSpan
recordDomainSqlCommandOutcome options mSpan attemptNo outcome
pure outcome
domainSqlCommandAttempts ::
forall phi rs s ci co rejection noOp a es.
(HasCallStack, IOE :> es, Store :> es, Error StoreError :> es, KirokuStoreResource :> es, BoolAlg phi (RegFile rs, ci), Eq co) =>
RunCommandOptions ->
DomainCommandHandler phi rs s ci co rejection noOp ->
Stream (EventStream phi rs s ci co) ->
ci ->
([(co, RecordedEvent)] -> AppendResult -> Tx.Transaction (SqlTransactionDecision a)) ->
Maybe Span ->
Eff es (Either CommandError (DomainSqlCommandOutcome (EventStream phi rs s ci co) co rejection noOp a), Int)
domainSqlCommandAttempts options handler@DomainCommandHandler {eventStream = validatedEventStream} targetStream command afterAppend mSpan =
domainCommandAttemptLoop options handler targetStream command DomainSqlCommandSilent appendWithSqlOnce
where
eventStream' = unvalidated validatedEventStream
appendWithSqlOnce retry attemptNo current events encoded = do
liftIO (options ^. #beforeAppend)
store <- getKirokuStore
enriched <- liftIO (enrichEventsIO store encoded)
prepared <- prepareEventsIO enriched
now <- liftIO getCurrentTime
let streamName = (eventStream' ^. #resolveStreamName) targetStream
expected = expectedVersion (current ^. #streamVersion)
body = do
appended <- appendToStreamTx streamName expected prepared now
case appended of
Left conflict ->
Tx.condemn $> Left (appendConflictToStoreError conflict)
Right appendResult -> do
let typedEvents = NonEmpty.toList events
recordeds = reconstructRecorded appendResult now prepared
sqlDecision <- afterAppend (Prelude.zip typedEvents recordeds) appendResult
case sqlDecision of
CommitSqlTransaction userValue ->
pure (Right (appendResult, Right userValue))
RollbackSqlTransaction userValue -> do
Tx.condemn
pure (Right (appendResult, Left userValue))
outcome <- tryError @StoreError (runTransaction body)
case outcome of
Right (Right (appendResult, Right userValue)) -> do
verifyAndSnapshot options mSpan eventStream' current (NonEmpty.toList events) appendResult
pure
( Right
( DomainSqlCommandCommitted
DomainCommandOutcome
{ decision = DomainAccepted events,
result = appendedResult targetStream appendResult (Prelude.length encoded)
}
userValue
),
attemptNo
)
Right (Right (_, Left userValue)) ->
pure (Right (DomainSqlCommandRolledBack userValue), attemptNo)
Right (Left storeError) ->
retryOrFail options retry attemptNo (current ^. #streamVersion) storeError
Left (_, storeError) ->
retryOrFail options retry attemptNo (current ^. #streamVersion) storeError
prepareDomainCommandPlan ::
(BoolAlg phi (RegFile rs, ci)) =>
RunCommandOptions ->
DomainCommandHandler phi rs s ci co rejection noOp ->
EventStream phi rs s ci co ->
Stream (EventStream phi rs s ci co) ->
Hydrated rs s ->
ci ->
Either CommandError (DomainCommandPlan (EventStream phi rs s ci co) rs s co rejection noOp)
prepareDomainCommandPlan options DomainCommandHandler {classifySilent} eventStream targetStream current command =
case Keiki.stepDetailedEither (eventStream ^. #transducer) (current ^. #state, current ^. #registers) command of
Left failure -> Left (commandStepFailure failure)
Right success ->
case Keiki.stepSuccessOutputs success of
[] ->
Right
( DomainCommandSilent
( classifySilent
SilentCommandContext
{ state = current ^. #state,
registers = current ^. #registers,
command,
selectedEdge = Keiki.stepSuccessEdge success
}
)
(noOpResult targetStream current)
)
event : events ->
let batch = event :| events
in DomainCommandAppend current batch
. assignEventIds (options ^. #eventIds)
<$> encodeEvents (eventStream ^. #eventCodec) (options ^. #metadata) (NonEmpty.toList batch)
domainDecisionFromSilent :: SilentDomainDecision rejection noOp -> DomainDecision co rejection noOp
domainDecisionFromSilent = \case
SilentRejected reason -> DomainRejected reason
SilentNoOp explanation -> DomainNoOp explanation
commandStepFailure :: Keiki.StepFailure s -> CommandError
commandStepFailure = \case
Keiki.NoOutgoingEdges {} -> CommandRejected
Keiki.NoMatchingEdge {} -> CommandRejected
Keiki.AmbiguousEdges _ matches ->
CommandAmbiguous
[ Keiki.edgeIndex (Keiki.matchedEdge matched)
| matched <- matches
]
-- | Render the stream that the command targets as plain 'Text', for use
-- as a span name.
resolvedStreamName ::
EventStream phi rs s ci co ->
Stream (EventStream phi rs s ci co) ->
Text
resolvedStreamName eventStream targetStream =
case (eventStream ^. #resolveStreamName) targetStream of
StreamName n -> n
-- | Attach the command-span outcome attributes after the runner returns.
--
-- On success: 'db.system.name' and 'keiro.events.appended'.
-- On failure: 'error.type' (low-cardinality classifier) and span status
-- 'Error' (carrying the rendered 'CommandError' as the description).
--
-- Pure no-op when no span is active ('Nothing' tracer, etc).
recordCommandOutcome ::
(IOE :> es) =>
Maybe Span ->
(a -> Int) ->
Int ->
Either CommandError a ->
Eff es ()
recordCommandOutcome Nothing _ _ _ = pure ()
recordCommandOutcome (Just sp) eventsOf attemptNo result = do
addAttribute sp (unkey db_system_name) ("postgresql" :: Text)
addAttribute sp (unkey keiro_retry_attempt) (Prelude.fromIntegral attemptNo :: Int64)
case result of
Right v ->
addAttribute sp (unkey keiro_events_appended) (Prelude.fromIntegral (eventsOf v) :: Int64)
Left err -> do
addAttribute sp (unkey error_type) (commandErrorClass err)
setStatus sp (Error (Text.take 256 (Text.pack (show err))))
recordDomainCommandOutcome ::
(IOE :> es) =>
RunCommandOptions ->
Maybe Span ->
Int ->
Either CommandError (DomainCommandOutcome target co rejection noOp) ->
Eff es ()
recordDomainCommandOutcome options mSpan attemptNo outcome = do
recordCommandOutcome mSpan ((^. #eventsAppended) . forgetDomainDecision) attemptNo outcome
case outcome of
Left _ -> pure ()
Right DomainCommandOutcome {decision} ->
recordDomainDecision options mSpan decision
recordDomainSqlCommandOutcome ::
(IOE :> es) =>
RunCommandOptions ->
Maybe Span ->
Int ->
Either CommandError (DomainSqlCommandOutcome target co rejection noOp a) ->
Eff es ()
recordDomainSqlCommandOutcome options mSpan attemptNo outcome = do
recordCommandOutcome mSpan eventsAppended attemptNo outcome
case outcome of
Right (DomainSqlCommandSilent DomainCommandOutcome {decision}) ->
recordDomainDecision options mSpan decision
Right (DomainSqlCommandCommitted DomainCommandOutcome {decision} _) ->
recordDomainDecision options mSpan decision
Right (DomainSqlCommandRolledBack _) -> pure ()
Left _ -> pure ()
where
eventsAppended = \case
DomainSqlCommandSilent DomainCommandOutcome {result} -> result ^. #eventsAppended
DomainSqlCommandCommitted DomainCommandOutcome {result} _ -> result ^. #eventsAppended
DomainSqlCommandRolledBack _ -> 0
recordDomainDecision ::
(IOE :> es) =>
RunCommandOptions ->
Maybe Span ->
DomainDecision co rejection noOp ->
Eff es ()
recordDomainDecision options mSpan domainDecision = do
let decisionClass = domainDecisionClass domainDecision
for_ mSpan $ \sp ->
addAttribute sp (unkey keiro_command_decision) (commandDecisionClassText decisionClass)
recordCommandDecision (options ^. #metrics) decisionClass
domainDecisionClass :: DomainDecision co rejection noOp -> CommandDecisionClass
domainDecisionClass = \case
DomainAccepted _ -> DecisionAccepted
DomainRejected _ -> DecisionRejected
DomainNoOp _ -> DecisionNoOp
-- | Low-cardinality classifier for a 'CommandError'. Used as the
-- @error.type@ attribute value on the command span.
commandErrorClass :: CommandError -> Text
commandErrorClass = \case
HydrationDecodeFailed {} -> "hydration_decode_failed"
HydrationReplayFailed _ HydrationNoInvertingEdge -> "hydration_replay_no_inverting_edge"
HydrationReplayFailed _ HydrationAmbiguousInversion -> "hydration_replay_ambiguous_inversion"
HydrationReplayFailed _ HydrationQueueMismatch -> "hydration_replay_queue_mismatch"
HydrationReplayFailed _ HydrationTruncatedChain -> "hydration_replay_truncated_chain"
HydrationGapDetected {} -> "hydration_gap_detected"
CommandRejected -> "command_rejected"
CommandAmbiguous {} -> "command_ambiguous"
EncodeFailed {} -> "encode_failed"
StoreFailed {} -> "store_failed"
RetryExhausted {} -> "retry_exhausted"
ConflictFixpoint {} -> "conflict_fixpoint"
verifyAndSnapshot ::
forall phi rs s ci co es.
(BoolAlg phi (RegFile rs, ci), IOE :> es, Store :> es, Error StoreError :> es, Eq co) =>
RunCommandOptions ->
Maybe Span ->
EventStream phi rs s ci co ->
Hydrated rs s ->
[co] ->
AppendResult ->
Eff es ()
verifyAndSnapshot options mSpan eventStream current events appendResult
| Prelude.not (options ^. #verifyReplayOnAppend),
Nothing <- eventStream ^. #stateCodec =
pure ()
| otherwise =
case Keiki.applyEventsEither (eventStream ^. #transducer) (state current, registers current) events of
Left failure -> do
recordSnapshotApplyDivergence (options ^. #metrics) 1
for_ mSpan $ \sp ->
addAttribute
sp
(unkey keiro_replay_divergence)
(Text.take 256 (renderReplayFailure failure))
Right finalState ->
case eventStream ^. #stateCodec of
Nothing -> pure ()
Just codec -> do
let finalVersion = appendResult ^. #streamVersion
terminality =
if Keiki.isFinal (eventStream ^. #transducer) (Prelude.fst finalState)
then Terminal
else NotTerminal
when (shouldSnapshotSpan (eventStream ^. #snapshotPolicy) terminality finalState (current ^. #streamVersion) finalVersion)
$ do
encoded <- liftIO (encodeSnapshotStrict codec finalState)
case encoded of
Left _ -> recordSnapshotEncodeFailures (options ^. #metrics) 1
Right value -> do
outcome <- tryError @StoreError (writeSnapshotEncoded (appendResult ^. #streamId) finalVersion codec value)
case outcome of
Right () -> pure ()
Left _ -> recordSnapshotWriteFailures (options ^. #metrics) 1
renderReplayFailure :: Keiki.ReplayFailure s co -> Text
renderReplayFailure failure =
"event_index="
<> Text.pack (show (Keiki.replayFailedIndex failure))
<> ";reason="
<> case Keiki.replayFailureReason failure of
Keiki.ReplayEventFailed stepFailure -> case stepFailure of
Keiki.ReplayNoInvertingEdge {} -> "no_inverting_edge"
Keiki.ReplayAmbiguousInversions {} -> "ambiguous_inversions"
Keiki.ReplayQueueMismatch {} -> "queue_mismatch"
Keiki.ReplayLogTruncated {} -> "log_truncated"
retryOrFail ::
(IOE :> es) =>
RunCommandOptions ->
(Int -> Maybe (StoreError, StreamVersion) -> Eff es (Either CommandError a, Int)) ->
Int ->
StreamVersion ->
StoreError ->
Eff es (Either CommandError a, Int)
retryOrFail options retry attemptNo observedVersion storeError
| isRetryableConflict storeError,
attemptNo <= options ^. #retryLimit = do
recordCommandConflicts (options ^. #metrics) 1
backoffDelay options attemptNo
recordCommandRetries (options ^. #metrics) 1
retry (attemptNo Prelude.+ 1) (Just (storeError, observedVersion))
| isRetryableConflict storeError = do
recordCommandConflicts (options ^. #metrics) 1
pure (Left (RetryExhausted attemptNo storeError), attemptNo)
| otherwise = do
case storeError of
DuplicateEvent {} -> recordCommandDuplicates (options ^. #metrics) 1
_ -> pure ()
pure (Left (StoreFailed storeError), attemptNo)
backoffDelay :: (IOE :> es) => RunCommandOptions -> Int -> Eff es ()
backoffDelay options attemptNo
| base <= 0 = pure ()
| otherwise = do
nanos <- liftIO getMonotonicTimeNSec
let exponential = min 100000 (base Prelude.* (2 Prelude.^ (attemptNo Prelude.- 1 :: Int)))
jitter =
Prelude.fromIntegral (nanos `Prelude.mod` Prelude.fromIntegral exponential)
Prelude.- (exponential `Prelude.div` 2)
liftIO (threadDelay (max 0 (exponential Prelude.+ jitter)))
where
base = options ^. #retryBackoffMicros
conflictFixpoint :: Maybe (StoreError, StreamVersion) -> StreamVersion -> Maybe CommandError
conflictFixpoint (Just (previousError@StreamAlreadyExists {}, previousVersion)) currentVersion
| currentVersion == previousVersion = Just (ConflictFixpoint currentVersion previousError)
conflictFixpoint _ _ = Nothing
encodeEvents :: Codec co -> Maybe Value -> [co] -> Either CommandError [EventData]
encodeEvents codec md =
Prelude.mapM (mapLeft EncodeFailed . encodeForAppendWithMetadata codec md)
assignEventIds :: [EventId] -> [EventData] -> [EventData]
assignEventIds [] events = events
assignEventIds _ [] = []
assignEventIds (supplied : suppliedRest) (event : eventRest) =
(event & #eventId .~ Just supplied) : assignEventIds suppliedRest eventRest
expectedVersion :: StreamVersion -> ExpectedVersion
expectedVersion (StreamVersion 0) = NoStream
expectedVersion version = ExactVersion version
noOpResult ::
Stream target ->
Hydrated rs s ->
CommandResult target
noOpResult targetStream current =
CommandResult
{ target = targetStream,
streamVersion = current ^. #streamVersion,
globalPosition = Nothing,
eventsAppended = 0
}
appendedResult ::
Stream target ->
AppendResult ->
Int ->
CommandResult target
appendedResult targetStream appendResult count =
CommandResult
{ target = targetStream,
streamVersion = appendResult ^. #streamVersion,
globalPosition = Just (appendResult ^. #globalPosition),
eventsAppended = count
}
-- | Rebuild the per-event 'RecordedEvent' values for a just-appended batch.
--
-- The store assigns each event in a batch a contiguous stream version and
-- global position: event @i@ (1-based) gets @last - count + i@ for both
-- counters, where @last@ is the position the 'AppendResult' reports for the
-- final event and @count@ is the batch size. (The kiroku append SQL numbers
-- events with @WITH ORDINALITY@ and inserts @initial + idx@; see EP-27's
-- Surprises & Discoveries.) We therefore reconstruct each 'RecordedEvent'
-- exactly, rather than reading the batch back. The @createdAt@ is the same
-- timestamp 'prepareEventsIO'/'appendToStreamTx' used for the insert.
--
-- This is a source append (events are written to their own stream), so
-- @streamVersion == originalVersion@ and @originalStreamId@ is the appended
-- stream's id, per the 'RecordedEvent' contract.
reconstructRecorded :: AppendResult -> UTCTime -> [PreparedEvent] -> [RecordedEvent]
reconstructRecorded appendResult now prepared =
Prelude.zipWith mk [0 ..] prepared
where
count = Prelude.length prepared
StreamVersion lastSv = appendResult ^. #streamVersion
GlobalPosition lastGp = appendResult ^. #globalPosition
firstSv = lastSv Prelude.- Prelude.fromIntegral count Prelude.+ 1
firstGp = lastGp Prelude.- Prelude.fromIntegral count Prelude.+ 1
mk :: Int64 -> PreparedEvent -> RecordedEvent
mk i prepared' =
RecordedEvent
{ eventId = EventId (prepared' ^. #peEventId),
eventType = prepared' ^. #peEventType,
streamVersion = StreamVersion (firstSv Prelude.+ i),
globalPosition = GlobalPosition (firstGp Prelude.+ i),
originalStreamId = appendResult ^. #streamId,
originalVersion = StreamVersion (firstSv Prelude.+ i),
payload = prepared' ^. #pePayload,
metadata = prepared' ^. #peMetadata,
causationId = prepared' ^. #peCausationId,
correlationId = prepared' ^. #peCorrelationId,
createdAt = now
}
isRetryableConflict :: StoreError -> Bool
isRetryableConflict = \case
WrongExpectedVersion {} -> True
StreamAlreadyExists {} -> True
_ -> False
mapLeft :: (e -> e') -> Either e a -> Either e' a
mapLeft f = \case
Left err -> Left (f err)
Right value -> Right value