diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,43 @@ ## [Unreleased] +## 0.19.0.0 — 2026-09-25 + +### Breaking Changes + +- Require `shibuya-core ^>=0.10.0.0` and `kiroku-store >=0.9.0.1 && <0.10`. + Shibuya 0.10 changes `ShutdownConfig`, `ConfigError`, `PolicyError`, and + `ProcessorState`, and a permanent framework-owned finalization failure now + throws `ProcessorFailure` rather than halting gracefully. Kiroku 0.9 + requires schema migration `0012` before any 0.9 process appends; see the + root changelog for the cutover. +- `RunCommandOptions` gains `seedVerifyInFlightLimit`. Code that constructs + the record directly must supply it; callers updating `defaultRunCommandOptions` + receive the process-wide default of one. +- `KeiroMetrics` gains `snapshotSeedSkipped` and + `snapshotSeedVerificationFailed` fields. Code that constructs the record + directly rather than through `newKeiroMetrics` must supply them. + +### New Features + +- Limit sampled snapshot-seed full replays to one in-flight task per process by + default. `RunCommandOptions.seedVerifyInFlightLimit` tunes the limit, with + zero restoring unbounded verification. The new + `keiro.snapshot.seed.skipped` and + `keiro.snapshot.seed.verification.failed` counters expose skipped samples and + unexpected task failures. + +### Other Changes + +- Add the `keiro-retention` test suite to `just verify`, with post-major heap + measurements for isolated store, adapter, process-manager, router, and + projection workloads (BUG-1). +- Add long and short command, hydration-only, and Kiroku tail-read retention + legs to distinguish command heap growth from snapshot hydration and store + reads (BUG-2). +- Count active Haskell threads in retention reports, excluding finished async + handles that can remain reachable during a measurement loop. + ## 0.18.0.0 — 2026-09-20 diff --git a/keiro.cabal b/keiro.cabal --- a/keiro.cabal +++ b/keiro.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: keiro -version: 0.18.0.0 +version: 0.19.0.0 synopsis: Event sourcing framework and workflow engine description: A library that composes kiroku, keiki, and shibuya into an @@ -167,13 +167,13 @@ hs-opentelemetry-semantic-conventions >=1.40 && <2, keiki >=0.9 && <0.10, keiki-codec-json >=0.9 && <0.10, - keiro-core ^>=0.18.0.0, - kiroku-store >=0.8 && <0.9, + keiro-core ^>=0.19.0.0, + kiroku-store >=0.9.0.1 && <0.10, lens >=5.2 && <5.4, mmzk-typeid >=0.7 && <0.8, random >=1.2.1 && <1.4, scientific >=0.3 && <0.4, - shibuya-core ^>=0.9.0.0, + shibuya-core ^>=0.10.0.0, stm >=2.5 && <2.6, streamly >=0.11 && <0.12, streamly-core >=0.3 && <0.4, @@ -233,10 +233,10 @@ keiki, keiki-codec-json, keiro, - keiro-test-support ^>=0.18.0.0, - kiroku-store >=0.8 && <0.9, + keiro-test-support ^>=0.19.0.0, + kiroku-store >=0.9.0.1 && <0.10, process >=1.6 && <1.7, - shibuya-core ^>=0.9.0.0, + shibuya-core ^>=0.10.0.0, stm >=2.5 && <2.6, streamly-core >=0.3 && <0.4, text >=2.1 && <2.2, @@ -245,6 +245,41 @@ uuid >=1.3 && <1.4, vector >=0.13 && <0.14, +test-suite keiro-retention + import: warnings, shared + type: exitcode-stdio-1.0 + hs-source-dirs: retention + main-is: Main.hs + other-modules: + Retention.Fixture + Retention.Legs + Retention.Measure + + ghc-options: + -threaded + -rtsopts + "-with-rtsopts=-N -T" + + build-depends: + aeson >=2.2 && <2.3, + base >=4.21 && <5, + effectful >=2.6 && <2.7, + effectful-core >=2.6 && <2.7, + hspec >=2.11, + keiki >=0.9 && <0.10, + keiki-codec-json >=0.9 && <0.10, + keiro, + keiro-core ^>=0.19.0.0, + keiro-test-support ^>=0.19.0.0, + kiroku-store >=0.9.0.1 && <0.10, + shibuya-core ^>=0.10.0.0, + shibuya-kiroku-adapter ^>=0.5.1.5, + stm >=2.5 && <2.6, + streamly-core >=0.3 && <0.4, + text >=2.1 && <2.2, + uuid >=1.3 && <1.4, + vector >=0.13 && <0.14, + benchmark keiro-bench import: warnings, shared type: exitcode-stdio-1.0 @@ -275,10 +310,10 @@ hs-opentelemetry-sdk >=1.0 && <1.1, keiki >=0.9 && <0.10, keiro, - keiro-core ^>=0.18.0.0, - keiro-test-support ^>=0.18.0.0, - kiroku-store >=0.8 && <0.9, - shibuya-core ^>=0.9.0.0, + keiro-core ^>=0.19.0.0, + keiro-test-support ^>=0.19.0.0, + kiroku-store >=0.9.0.1 && <0.10, + shibuya-core ^>=0.10.0.0, streamly-core >=0.3 && <0.4, tasty-bench >=0.4, text >=2.1 && <2.2, diff --git a/retention/Main.hs b/retention/Main.hs new file mode 100644 --- /dev/null +++ b/retention/Main.hs @@ -0,0 +1,45 @@ +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE OverloadedRecordDot #-} + +module Main (main) where + +import Control.Monad (forM_, unless) +import Data.Text qualified as Text +import Keiro.Test.Postgres (withFreshResourceStore, withMigratedSuite) +import Retention.Legs (Leg (..), allLegs) +import Retention.Measure +import System.Environment (lookupEnv) +import Test.Hspec + +main :: IO () +main = do + config <- gateConfigFromEnvironment + selected <- selectLegs + withMigratedSuite \fixture -> + hspec $ + describe "Retention legs" $ + forM_ selected \leg -> + it (Text.unpack leg.legName) $ + withFreshResourceStore fixture \handle -> do + (verdict, samples) <- leg.run config handle + unless (config.reportOnly || leg.legName == "command-long-history-verify-every") $ do + verdict `shouldBe` Bounded + whenBaseline leg.legName config samples + +selectLegs :: IO [Leg] +selectLegs = do + selected <- lookupEnv "KEIRO_RETENTION_LEGS" + case selected of + Nothing -> pure (filter ((/= "command-long-history-verify-every") . (.legName)) allLegs) + Just raw -> do + let names = fmap Text.strip (Text.splitOn "," (Text.pack raw)) + unknown = filter (`notElem` fmap (.legName) allLegs) names + unless (not (null names) && all (not . Text.null) names && null unknown) $ + fail ("KEIRO_RETENTION_LEGS contains an unknown or empty leg: " <> show unknown) + pure (filter (\leg -> leg.legName `elem` names) allLegs) + +whenBaseline :: Text.Text -> GateConfig -> [HeapSample] -> Expectation +whenBaseline name config samples = + if name == "baseline-no-store" + then let (_, _, growth) = judge config samples in growth `shouldSatisfy` (< 512 * 1024) + else pure () diff --git a/retention/Retention/Fixture.hs b/retention/Retention/Fixture.hs new file mode 100644 --- /dev/null +++ b/retention/Retention/Fixture.hs @@ -0,0 +1,452 @@ +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE LambdaCase #-} +{-# LANGUAGE OverloadedRecordDot #-} +{-# LANGUAGE TypeApplications #-} + +module Retention.Fixture + ( Signal (..), + TargetCommand (..), + TargetEvent (..), + TargetEventStream, + retentionTargetStream, + LedgerCommand (..), + LedgerEvent (..), + LedgerRegs, + LedgerState (..), + RetentionLedgerStream, + retentionLedger, + ledgerTarget, + seedLedger, + seedLedgerRaw, + retentionManager, + retentionRouter, + retentionTarget, + signalStreamName, + appendSignals, + decodeSignal, + ackAdapter, + sampleOnAck, + ) +where + +import Control.Concurrent.STM (atomically, tryPutTMVar) +import Control.Monad (forM_, unless, void, when) +import Data.Aeson (FromJSON, Result (..), ToJSON, Value (..), fromJSON, object, toJSON, (.=)) +import Data.IORef (atomicModifyIORef', newIORef) +import Data.List.NonEmpty (NonEmpty (..)) +import Data.Proxy (Proxy (..)) +import Data.Text (Text) +import Data.Text qualified as Text +import Data.UUID qualified as UUID +import Effectful (Eff, IOE, liftIO, (:>)) +import Effectful.Error.Static (Error) +import GHC.Generics (Generic) +import Keiki.Core + ( Edge (..), + HsPred, + InCtor, + IndexN, + RegFile (..), + SymTransducer (..), + Update (..), + WireCtor, + inpCtor, + matchInCtor, + oNil, + pack, + proj, + unavailableInCtor, + unavailableWireCtor, + (*:), + ) +import Keiki.Core qualified as Keiki +import Keiki.Shape (CanonicalStateShape) +import Keiro.Codec (Codec (..), encodeForAppend) +import Keiro.Command qualified as Command +import Keiro.EventStream (EventStream (..), SnapshotPolicy (..)) +import Keiro.EventStream.Validate (ValidatedEventStream, mkEventStreamOrThrow, unvalidated) +import Keiro.ProcessManager (PMCommand (..), ProcessManager (..), ProcessManagerAction (..)) +import Keiro.Router (Router (..)) +import Keiro.Snapshot.Codec (defaultStateCodec) +import Keiro.Snapshot.Schema (SnapshotRow (..), lookupSnapshotRow) +import Keiro.Stream (Stream, stream) +import Keiro.Stream qualified as Stream +import Keiro.Test.Postgres (StoreRunner (..)) +import Kiroku.Store qualified as Store +import Kiroku.Store.Effect (Store) +import Kiroku.Store.Effect.Resource (KirokuStoreResource) +import Kiroku.Store.Error (StoreError) +import Kiroku.Store.Subscription.Stream (AckItem (..), subscriptionAckStream) +import Kiroku.Store.Subscription.Types (SubscriptionConfig) +import Kiroku.Store.Subscription.Types qualified as Sub +import Kiroku.Store.Types (EventData (..), EventId (..), EventType (..), ExpectedVersion (..), GlobalPosition (..), RecordedEvent (..), StreamName (..), StreamVersion (..)) +import Numeric.Natural (Natural) +import Shibuya.Adapter (Adapter (..)) +import Shibuya.Core.Ack (AckDecision (..)) +import Shibuya.Core.Ack qualified as Ack +import Shibuya.Core.AckHandle (AckHandle (..)) +import Shibuya.Core.Ingested (Ingested (..)) +import Shibuya.Core.Types (Attempt (..), Cursor (..), Envelope (..), MessageId (..)) +import Streamly.Data.Stream qualified as Streamly + +data Signal = Signal + { signalId :: !Text, + account :: !Int + } + deriving stock (Eq, Show, Generic) + deriving anyclass (FromJSON, ToJSON) + +data TargetCommand = Credit !Int + deriving stock (Eq, Show) + +data TargetEvent = Credited !Int + deriving stock (Eq, Show) + +data TargetState = TargetReady + deriving stock (Bounded, Enum, Eq, Generic, Ord, Show) + deriving anyclass (FromJSON, ToJSON) + +instance CanonicalStateShape TargetState + +type TargetEventStream = EventStream (HsPred '[] TargetCommand) '[] TargetState TargetCommand TargetEvent + +data LedgerCommand = Deposit !Int + deriving stock (Eq, Show) + +data LedgerEvent = Deposited !Int + deriving stock (Eq, Show) + +data LedgerState = LedgerReady + deriving stock (Bounded, Enum, Eq, Generic, Ord, Show) + deriving anyclass (FromJSON, ToJSON) + +instance CanonicalStateShape LedgerState + +type LedgerRegs = '[ '("balance", Int)] + +type RetentionLedgerStream = EventStream (HsPred LedgerRegs LedgerCommand) LedgerRegs LedgerState LedgerCommand LedgerEvent + +retentionLedger :: ValidatedEventStream (HsPred LedgerRegs LedgerCommand) LedgerRegs LedgerState LedgerCommand LedgerEvent +retentionLedger = mkEventStreamOrThrow "retention-ledger" ledgerStreamDef + +ledgerTarget :: Stream RetentionLedgerStream +ledgerTarget = stream "retentionledger-account" + +ledgerStreamDef :: RetentionLedgerStream +ledgerStreamDef = + EventStream + { transducer = + SymTransducer + { edgesOut = \LedgerReady -> + [ Edge + { guard = matchInCtor depositCtor, + update = + USet + (#balance :: IndexN "balance" LedgerRegs Int) + (proj (#balance :: Keiki.Index LedgerRegs Int) Keiki..+ inpCtor depositCtor #amount), + output = [pack depositCtor depositedCtor (inpCtor depositCtor #amount *: oNil)], + target = LedgerReady, + mode = Keiki.Live + } + ], + initial = LedgerReady, + initialRegs = RCons (Proxy @"balance") 0 RNil, + isFinal = \_ -> False + }, + initialState = LedgerReady, + initialRegisters = RCons (Proxy @"balance") 0 RNil, + eventCodec = + Codec + { eventTypes = EventType "Deposited" :| [], + eventType = \_ -> EventType "Deposited", + schemaVersion = 1, + encode = \(Deposited amount) -> toJSON amount, + decode = \_ value -> case fromJSON value of + Success amount -> Right (Deposited amount) + Error message -> Left (Text.pack message), + upcasters = [] + }, + resolveStreamName = Stream.streamName, + snapshotPolicy = Every 100, + stateCodec = Just (defaultStateCodec @LedgerRegs @LedgerState 1) + } + +depositCtor :: InCtor LedgerCommand '[ '("amount", Int)] +depositCtor = + unavailableInCtor + "Deposit" + (\(Deposit amount) -> Just (RCons Proxy amount RNil)) + (\(RCons _ amount RNil) -> Deposit amount) + +depositedCtor :: WireCtor LedgerEvent (Int, ()) +depositedCtor = + unavailableWireCtor + "Deposited" + (\(Deposited amount) -> Just (amount, ())) + (\(amount, ()) -> Deposited amount) + +seedLedger :: StoreRunner -> Stream RetentionLedgerStream -> Int -> IO () +seedLedger runner@(StoreRunner runStore) target count = do + unless (count > 0 && count `mod` 100 == 0) $ + fail "seedLedger requires a positive multiple of 100 events" + seedLedgerRaw runner target (count - 1) + commandResult <- runStore (Command.runCommand Command.defaultRunCommandOptions {Command.seedVerifySampleRate = 0} retentionLedger target (Deposit 1)) + case commandResult of + Right (Right result) | result.eventsAppended == 1 && result.streamVersion == StreamVersion (fromIntegral count) -> pure () + other -> fail ("seedLedger command: " <> show other) + snapshot <- runStore do + maybeId <- Store.lookupStreamId (Stream.streamName target) + traverse lookupSnapshotRow maybeId + case snapshot of + Right (Just (Just row)) | row.streamVersion == StreamVersion (fromIntegral count) -> pure () + other -> fail ("seedLedger snapshot: " <> show other) + +seedLedgerRaw :: StoreRunner -> Stream RetentionLedgerStream -> Int -> IO () +seedLedgerRaw (StoreRunner runStore) target count = appendChunks count + where + eventStream = unvalidated retentionLedger + name = Stream.streamName target + seededEvent = either (error . show) Prelude.id (encodeForAppend eventStream.eventCodec (Deposited 1)) + appendChunks remaining + | remaining <= 0 = pure () + | otherwise = do + let chunk = min 500 remaining + result <- runStore (Store.appendToStream name AnyVersion (replicate chunk seededEvent)) + case result of + Left failure -> fail ("seedLedger append: " <> show failure) + Right _ -> appendChunks (remaining - chunk) + +retentionTargetStream :: ValidatedEventStream (HsPred '[] TargetCommand) '[] TargetState TargetCommand TargetEvent +retentionTargetStream = mkEventStreamOrThrow "retention-target" targetStreamDef + +snapshottedTargetStream :: ValidatedEventStream (HsPred '[] TargetCommand) '[] TargetState TargetCommand TargetEvent +snapshottedTargetStream = + mkEventStreamOrThrow + "retention-target-snapshotted" + targetStreamDef + { snapshotPolicy = Every 100, + stateCodec = Just (defaultStateCodec @'[] @TargetState 1) + } + +targetStreamDef :: TargetEventStream +targetStreamDef = + EventStream + { transducer = targetTransducer, + initialState = TargetReady, + initialRegisters = RNil, + eventCodec = + Codec + { eventTypes = EventType "Credited" :| [], + eventType = \_ -> EventType "Credited", + schemaVersion = 1, + encode = \(Credited amount) -> toJSON amount, + decode = \_ value -> case fromJSON value of + Success amount -> Right (Credited amount) + Error message -> Left (Text.pack message), + upcasters = [] + }, + resolveStreamName = Stream.streamName, + snapshotPolicy = Never, + stateCodec = Nothing + } + +targetTransducer :: SymTransducer (HsPred '[] TargetCommand) '[] TargetState TargetCommand TargetEvent +targetTransducer = + SymTransducer + { edgesOut = \TargetReady -> + [ Edge + { guard = matchInCtor creditCtor, + update = UKeep, + output = [pack creditCtor creditedCtor (inpCtor creditCtor #amount *: oNil)], + target = TargetReady, + mode = Keiki.Live + } + ], + initial = TargetReady, + initialRegs = RNil, + isFinal = \_ -> False + } + +creditCtor :: InCtor TargetCommand '[ '("amount", Int)] +creditCtor = + unavailableInCtor + "Credit" + (\(Credit amount) -> Just (RCons Proxy amount RNil)) + (\(RCons _ amount RNil) -> Credit amount) + +creditedCtor :: WireCtor TargetEvent (Int, ()) +creditedCtor = + unavailableWireCtor + "Credited" + (\(Credited amount) -> Just (amount, ())) + (\(amount, ()) -> Credited amount) + +data ManagerCommand = Seen + deriving stock (Eq, Show) + +data ManagerEvent = SeenRecorded + deriving stock (Eq, Show) + +data ManagerState = ManagerReady + deriving stock (Bounded, Enum, Eq, Ord, Show) + +type ManagerEventStream = EventStream (HsPred '[] ManagerCommand) '[] ManagerState ManagerCommand ManagerEvent + +managerEventStream :: ValidatedEventStream (HsPred '[] ManagerCommand) '[] ManagerState ManagerCommand ManagerEvent +managerEventStream = mkEventStreamOrThrow "retention-manager" managerStream + +managerStream :: ManagerEventStream +managerStream = + EventStream + { transducer = + SymTransducer + { edgesOut = \ManagerReady -> + [ Edge + { guard = matchInCtor seenCtor, + update = UKeep, + output = [pack seenCtor seenRecordedCtor oNil], + target = ManagerReady, + mode = Keiki.Live + } + ], + initial = ManagerReady, + initialRegs = RNil, + isFinal = \_ -> False + }, + initialState = ManagerReady, + initialRegisters = RNil, + eventCodec = + Codec + { eventTypes = EventType "SeenRecorded" :| [], + eventType = \_ -> EventType "SeenRecorded", + schemaVersion = 1, + encode = \_ -> Null, + decode = \_ _ -> Right SeenRecorded, + upcasters = [] + }, + resolveStreamName = Stream.streamName, + snapshotPolicy = Never, + stateCodec = Nothing + } + +seenCtor :: InCtor ManagerCommand '[] +seenCtor = unavailableInCtor "Seen" (\Seen -> Just RNil) (\RNil -> Seen) + +seenRecordedCtor :: WireCtor ManagerEvent () +seenRecordedCtor = unavailableWireCtor "SeenRecorded" (\SeenRecorded -> Just ()) (\() -> SeenRecorded) + +retentionTarget :: Int -> Stream TargetCommand +retentionTarget accountNumber = stream ("retentiontarget-" <> Text.pack (show (accountNumber `mod` 16))) + +retentionManager :: ProcessManager Signal (HsPred '[] ManagerCommand) '[] ManagerState ManagerCommand ManagerEvent (HsPred '[] TargetCommand) '[] TargetState TargetCommand TargetEvent +retentionManager = + ProcessManager + { name = "retention-manager", + correlate = signalId, + eventStream = managerEventStream, + streamFor = \identifier -> stream ("pm:retention-" <> identifier), + targetEventStream = retentionTargetStream, + targetProjections = const [], + handle = \signal -> + ProcessManagerAction + { command = Seen, + commands = [PMCommand (retentionTarget signal.account) (Credit 1)], + timers = [] + } + } + +retentionRouter :: Bool -> Router Signal (HsPred '[] TargetCommand) '[] TargetState TargetCommand TargetEvent '[Store, Error StoreError, KirokuStoreResource, IOE] +retentionRouter snapshotTargets = + Router + { name = "retention-router", + key = signalId, + resolve = \_ -> pure [PMCommand (retentionTarget accountNumber) (Credit 1) | accountNumber <- [0 .. 3]], + targetEventStream = if snapshotTargets then snapshottedTargetStream else retentionTargetStream, + targetProjections = const [] + } + +signalStreamName :: Int -> StreamName +signalStreamName index = StreamName ("retentionsource-" <> Text.pack (show index)) + +appendSignals :: StoreRunner -> Int -> IO () +appendSignals (StoreRunner runStore) count = + forM_ [1 .. count] \index -> do + let signal = Signal (Text.pack (show index)) (index `mod` 16) + event = + EventData + { eventId = Nothing, + eventType = EventType "RetentionSignal", + payload = toJSON signal, + metadata = Nothing, + causationId = Nothing, + correlationId = Nothing + } + result <- runStore (Store.appendToStream (signalStreamName index) NoStream [event]) + case result of + Left failure -> fail ("appendSignals: " <> show failure) + Right _ -> pure () + +decodeSignal :: RecordedEvent -> Maybe (RecordedEvent, Signal) +decodeSignal recorded = + case fromJSON recorded.payload of + Success signal -> Just (recorded, signal) + Error _ -> Nothing + +ackAdapter :: (IOE :> es) => Store.KirokuStore -> SubscriptionConfig -> Natural -> IO (Adapter es RecordedEvent, IO ()) +ackAdapter store config bufferSize = do + (source, cancel) <- subscriptionAckStream store config bufferSize + let ingested = fmap (toIngested cancel) (Streamly.morphInner liftIO source) + pure + ( Adapter {adapterName = "retention-ack-bridge", source = ingested, shutdown = liftIO cancel}, + cancel + ) + +toIngested :: (IOE :> es) => IO () -> AckItem -> Ingested es RecordedEvent +toIngested cancel (AckItem event attempt reply) = + Ingested + { envelope = + Envelope + { messageId = case event.eventId of EventId uuid -> MessageId (Text.pack (UUID.toString uuid)), + cursor = case event.globalPosition of GlobalPosition position -> Just (CursorInt (fromIntegral position)), + partition = Nothing, + enqueuedAt = Just event.createdAt, + traceContext = Nothing, + headers = Nothing, + attempt = Just (Attempt attempt), + attributes = mempty, + payload = event + }, + ack = AckHandle \case + AckHalt _ -> liftIO cancel + decision -> liftIO $ atomically $ void $ tryPutTMVar reply (toSubscriptionResult attempt decision), + lease = Nothing + } + +toSubscriptionResult :: Word -> AckDecision -> Sub.SubscriptionResult +toSubscriptionResult attempt = \case + AckOk -> Sub.Continue + AckRetry (Ack.RetryDelay delay) -> Sub.Retry (Sub.RetryDelay delay) + AckDeadLetter reason -> Sub.DeadLetter $ case reason of + Ack.PoisonPill detail -> Sub.DeadLetterPoison detail + Ack.InvalidPayload detail -> Sub.DeadLetterInvalid detail + Ack.MaxRetriesExceeded -> Sub.DeadLetterMaxAttempts (fromIntegral attempt) + other@(Ack.ApplicationFailure code detail) -> + Sub.DeadLetterOther + (Ack.renderDeadLetterReason other) + (object ["code" .= Ack.deadLetterCodeText code, "detail" .= detail]) + AckHalt _ -> Sub.Continue + +sampleOnAck :: (IOE :> es) => Int -> Int -> (Int -> IO ()) -> Adapter es msg -> Eff es (Adapter es msg) +sampleOnAck expected blockSize sample adapter = do + countRef <- liftIO (newIORef 0) + pure adapter {source = fmap (wrap countRef) adapter.source} + where + wrap countRef ingested = + ingested + { ack = AckHandle \decision -> do + ingested.ack.finalize decision + unless (decision == AckOk) $ liftIO $ fail ("retention worker ack was " <> show decision) + count <- liftIO $ atomicModifyIORef' countRef (\previous -> let next = previous + 1 in (next, next)) + when (count == expected) adapter.shutdown + when (count `mod` blockSize == 0) $ liftIO (sample count) + } diff --git a/retention/Retention/Legs.hs b/retention/Retention/Legs.hs new file mode 100644 --- /dev/null +++ b/retention/Retention/Legs.hs @@ -0,0 +1,375 @@ +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE OverloadedRecordDot #-} + +module Retention.Legs + ( Leg (..), + allLegs, + ) +where + +import Control.Concurrent (threadDelay) +import Control.Concurrent.STM (atomically, putTMVar) +import Control.Exception (evaluate, finally) +import Control.Monad (forM_, unless, void, when) +import Data.Aeson (toJSON) +import Data.IORef (atomicModifyIORef', newIORef, readIORef) +import Data.Text (Text) +import Data.Text qualified as Text +import Data.Text.IO qualified as Text.IO +import Data.UUID qualified as UUID +import Data.Vector qualified as Vector +import Effectful (Eff, IOE, liftIO, (:>)) +import Effectful.Exception qualified as Effectful +import Keiro.Command (defaultRunCommandOptions) +import Keiro.Command qualified as Command +import Keiro.EventStream.Validate (unvalidated) +import Keiro.ProcessManager (defaultWorkerOptions, runProcessManagerWorkerWith) +import Keiro.Projection (AsyncApplyOutcome (..), AsyncProjection (..), applyAsyncProjection) +import Keiro.ReadModel.Schema (registerReadModel) +import Keiro.Router (runRouterWorkerWith) +import Keiro.Snapshot.Schema (SnapshotRow (..), lookupSnapshotRow) +import Keiro.Stream qualified as KeiroStream +import Keiro.Test.Postgres (StoreRunner (..)) +import Kiroku.Store qualified as Store +import Kiroku.Store.Subscription qualified as Subscription +import Kiroku.Store.Subscription.Stream (AckItem (..), subscriptionAckStream) +import Kiroku.Store.Subscription.Types (SubscriptionConfig, SubscriptionName (..), SubscriptionResult (..), SubscriptionTarget (..), defaultSubscriptionConfig) +import Kiroku.Store.Transaction qualified as StoreTx +import Kiroku.Store.Types (CategoryName (..), EventData (..), EventId (..), EventType (..), ExpectedVersion (..), RecordedEvent, StreamName (..), StreamVersion (..)) +import Retention.Fixture (LedgerCommand (..), TargetCommand (..), TargetEventStream, ackAdapter, appendSignals, decodeSignal, ledgerTarget, retentionLedger, retentionManager, retentionRouter, retentionTargetStream, sampleOnAck, seedLedger, seedLedgerRaw) +import Retention.Measure (GateConfig (..), HeapSample (..), Verdict (..), measureLeg, measureLegWithSampler, sampleHeap) +import Shibuya.Adapter (Adapter (..)) +import Shibuya.Adapter.Kiroku (defaultKirokuAdapterConfig, kirokuAdapter) +import Shibuya.Core.Ack (AckDecision (..)) +import Shibuya.Core.AckHandle (AckHandle (..)) +import Shibuya.Core.Ingested (Ingested (..)) +import Streamly.Data.Stream (Stream) +import Streamly.Data.Stream qualified as Streamly +import System.Environment (lookupEnv) +import System.Timeout (timeout) +import Text.Read (readMaybe) + +data Leg = Leg + { legName :: !Text, + run :: GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) + } + +allLegs :: [Leg] +allLegs = + [ Leg "baseline-no-store" baselineNoStore, + Leg "kiroku-append-probe" kirokuAppendProbe, + Leg "kiroku-append-only" kirokuAppendOnly, + Leg "kiroku-append-tx" kirokuAppendTx, + Leg "kiroku-probe-only" kirokuProbeOnly, + Leg "kiroku-subscribe-ack" kirokuSubscribeAck, + Leg "hand-bridge-ack" handBridgeAck, + Leg "shibuya-adapter-ack" shibuyaAdapterAck, + Leg "pm-worker" pmWorker, + Leg "router-worker" routerWorker, + Leg "projection-apply" projectionApply, + Leg "command-long-history" commandLongHistory, + Leg "hydrate-only" hydrateOnly, + Leg "kiroku-read-tail" kirokuReadTail, + Leg "command-short-history" commandShortHistory, + Leg "command-long-history-verify-every" commandLongHistoryVerifyEvery + ] + +operationCount :: GateConfig -> Int +operationCount config = config.blocks * config.blockSize + +commandLongHistory :: GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) +commandLongHistory = commandLongHistoryWithRate "command-long-history" 0 + +commandLongHistoryVerifyEvery :: GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) +commandLongHistoryVerifyEvery = commandLongHistoryWithRate "command-long-history-verify-every" 1 + +commandLongHistoryWithRate :: Text -> Int -> GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) +commandLongHistoryWithRate name rate config (_, runner@(StoreRunner runStore)) = do + seedLedger runner ledgerTarget 10_000 + let options = defaultRunCommandOptions {Command.seedVerifySampleRate = rate} + measured <- measureLeg config name \index -> do + outcome <- runStore (Command.runCommand options retentionLedger ledgerTarget (Deposit 1)) + case outcome of + Right (Right result) + | result.eventsAppended == 1 + && result.streamVersion == StreamVersion (fromIntegral (10_000 + index)) -> + pure () + other -> fail (Text.unpack name <> ": command " <> show index <> ": " <> show other) + snapshot <- runStore do + maybeId <- Store.lookupStreamId (KeiroStream.streamName ledgerTarget) + traverse lookupSnapshotRow maybeId + let expectedSnapshot = StreamVersion (fromIntegral (((10_000 + operationCount config) `div` 100) * 100)) + case snapshot of + Right (Just (Just row)) | row.streamVersion == expectedSnapshot -> pure () + other -> fail (Text.unpack name <> ": final snapshot: " <> show other) + postReturn name (operationCount config) + pure measured + +hydrateOnly :: GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) +hydrateOnly config (_, runner@(StoreRunner runStore)) = do + seedLedger runner ledgerTarget 10_000 + let options = defaultRunCommandOptions {Command.seedVerifySampleRate = 0} + measured <- measureLeg config "hydrate-only" \_ -> do + outcome <- runStore (Command.hydrate options (unvalidated retentionLedger) ledgerTarget) + case outcome of + Right (Right hydrated) | hydrated.streamVersion == StreamVersion 10_000 -> pure () + Right (Right hydrated) -> fail ("hydrate-only: unexpected version " <> show hydrated.streamVersion) + Right (Left failure) -> fail ("hydrate-only: " <> show failure) + Left failure -> fail ("hydrate-only: " <> show failure) + postReturn "hydrate-only" (operationCount config) + pure measured + +kirokuReadTail :: GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) +kirokuReadTail config (_, runner@(StoreRunner runStore)) = do + seedLedgerRaw runner ledgerTarget 10_000 + let name = KeiroStream.streamName ledgerTarget + measured <- measureLeg config "kiroku-read-tail" \_ -> do + outcome <- runStore do + maybeId <- Store.lookupStreamId name + events <- Store.readStreamForward name (StreamVersion 9_900) 256 + pure (maybeId, events) + case outcome of + Right (Just _, events) | Vector.length events == 100 -> pure () + other -> fail ("kiroku-read-tail: " <> show (fmap (\(streamId, events) -> (streamId, Vector.length events)) other)) + postReturn "kiroku-read-tail" (operationCount config) + pure measured + +commandShortHistory :: GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) +commandShortHistory config (_, StoreRunner runStore) = do + let options = defaultRunCommandOptions {Command.seedVerifySampleRate = 0} + measured <- measureLeg config "command-short-history" \index -> do + let target = KeiroStream.stream ("retentiontarget-" <> Text.pack (show (index `mod` 16))) :: KeiroStream.Stream TargetEventStream + expected = StreamVersion (fromIntegral ((index - 1) `div` 16 + 1)) + outcome <- runStore (Command.runCommand options retentionTargetStream target (Credit 1)) + case outcome of + Right (Right result) | result.eventsAppended == 1 && result.streamVersion == expected -> pure () + other -> fail ("command-short-history: command " <> show index <> ": " <> show other) + postReturn "command-short-history" (operationCount config) + pure measured + +baselineNoStore :: GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) +baselineNoStore config _ = + measureLeg config "baseline-no-store" \index -> + void (evaluate (sum (replicate 100 (index `mod` 17)))) + +kirokuAppendProbe :: GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) +kirokuAppendProbe config (_, StoreRunner runStore) = do + measured <- measureLegWithSampler config "kiroku-append-probe" \sample -> do + outcome <- runStore $ forM_ [1 .. operationCount config] \index -> do + let (streamName, eventId, event) = plainEvent index + _ <- Store.appendToStream streamName AnyVersion [event] + exists <- Store.eventExistsInStream streamName eventId + unless exists $ liftIO $ fail ("kiroku-append-probe: missing event " <> show index) + when (index `mod` config.blockSize == 0) $ liftIO (sample index) + void (expectRight outcome) + postReturn "kiroku-append-probe" (operationCount config) + settleSeconds <- lookupEnv "KEIRO_RETENTION_SETTLE_SECONDS" + forM_ settleSeconds \rawSeconds -> case readMaybe rawSeconds of + Just seconds | seconds > 0 -> do + threadDelay (seconds * 1_000_000) + settled <- sampleHeap (operationCount config) + Text.IO.putStrLn ("settled leg=kiroku-append-probe after_seconds=" <> Text.pack (show seconds) <> " live_bytes=" <> Text.pack (show settled.liveBytes) <> " large_objects_bytes=" <> Text.pack (show settled.largeObjectBytes)) + _ -> fail "KEIRO_RETENTION_SETTLE_SECONDS must be a positive integer" + pure measured + +kirokuAppendOnly :: GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) +kirokuAppendOnly config (_, StoreRunner runStore) = do + requestedRate <- lookupEnv "KEIRO_RETENTION_APPEND_RATE" + delayMicros <- case requestedRate of + Nothing -> pure 0 + Just rawRate -> case readMaybe rawRate of + Just rate | rate > 0 -> pure (1_000_000 `div` rate) + _ -> fail "KEIRO_RETENTION_APPEND_RATE must be a positive integer" + measured <- measureLegWithSampler config "kiroku-append-only" \sample -> do + outcome <- runStore $ forM_ [1 .. operationCount config] \index -> do + let (streamName, _, event) = plainEvent index + _ <- Store.appendToStream streamName AnyVersion [event] + when (delayMicros > 0) $ liftIO (threadDelay delayMicros) + when (index `mod` config.blockSize == 0) $ liftIO (sample index) + void (expectRight outcome) + postReturn "kiroku-append-only" (operationCount config) + pure measured + +kirokuAppendTx :: GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) +kirokuAppendTx config (_, StoreRunner runStore) = do + measured <- measureLegWithSampler config "kiroku-append-tx" \sample -> do + outcome <- runStore $ forM_ [1 .. operationCount config] \index -> do + let (streamName, _, event) = plainEvent index + result <- StoreTx.runTransactionAppendingResource streamName AnyVersion [event] (const (pure ())) + case result of + Left failure -> liftIO $ fail ("kiroku-append-tx: " <> show failure) + Right () -> pure () + when (index `mod` config.blockSize == 0) $ liftIO (sample index) + void (expectRight outcome) + postReturn "kiroku-append-tx" (operationCount config) + pure measured + +kirokuProbeOnly :: GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) +kirokuProbeOnly config (_, StoreRunner runStore) = do + setup <- runStore $ forM_ [1 .. operationCount config] \index -> do + let (streamName, _, event) = plainEvent index + void (Store.appendToStream streamName AnyVersion [event]) + void (expectRight setup) + measured <- measureLegWithSampler config "kiroku-probe-only" \sample -> do + outcome <- runStore $ forM_ [1 .. operationCount config] \index -> do + let (streamName, eventId, _) = plainEvent index + exists <- Store.eventExistsInStream streamName eventId + unless exists $ liftIO $ fail ("kiroku-probe-only: missing event " <> show index) + when (index `mod` config.blockSize == 0) $ liftIO (sample index) + void (expectRight outcome) + postReturn "kiroku-probe-only" (operationCount config) + pure measured + +plainEvent :: Int -> (StreamName, EventId, EventData) +plainEvent index = + let streamName = StreamName ("retentionplain-" <> Text.pack (show (index `mod` 16))) + eventId = EventId (UUID.fromWords 0 0 0 (fromIntegral index)) + event = + EventData + { eventId = Just eventId, + eventType = EventType "RetentionPlain", + payload = toJSON index, + metadata = Nothing, + causationId = Nothing, + correlationId = Nothing + } + in (streamName, eventId, event) + +sourceTarget :: SubscriptionTarget +sourceTarget = Category (CategoryName "retentionsource") + +sourceConfig :: Text -> SubscriptionConfig +sourceConfig name = defaultSubscriptionConfig (SubscriptionName name) sourceTarget (\_ -> pure Continue) + +kirokuSubscribeAck :: GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) +kirokuSubscribeAck config (store, runner) = do + appendSignals runner (operationCount config) + measured <- measureLegWithSampler config "kiroku-subscribe-ack" \sample -> do + (source, cancel) <- subscriptionAckStream store (sourceConfig "retention-kiroku-ack") 256 + consumeAck (operationCount config) config.blockSize sample source `finally` cancel + postReturn "kiroku-subscribe-ack" (operationCount config) + pure measured + +consumeAck :: Int -> Int -> (Int -> IO ()) -> Stream IO AckItem -> IO () +consumeAck expected blockSize sample = go 1 + where + go index source = do + next <- timeout 30_000_000 (Streamly.uncons source) + case next of + Nothing -> fail ("kiroku-subscribe-ack: timed out after " <> show (index - 1) <> " events") + Just Nothing -> fail ("kiroku-subscribe-ack: ended after " <> show (index - 1) <> " events") + Just (Just (item, rest)) -> do + atomically (putTMVar item.ackReply Continue) + when (index `mod` blockSize == 0) (sample index) + when (index < expected) (go (index + 1) rest) + +handBridgeAck :: GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) +handBridgeAck config (store, runner@(StoreRunner runStore)) = do + appendSignals runner (operationCount config) + measured <- measureLegWithSampler config "hand-bridge-ack" \sample -> do + outcome <- runStore do + (adapter, _) <- liftIO $ ackAdapter store (sourceConfig "retention-hand-bridge-ack") 256 + consumeAdapter (operationCount config) config.blockSize sample adapter `Effectful.finally` adapter.shutdown + void (expectRight outcome) + postReturn "hand-bridge-ack" (operationCount config) + pure measured + +shibuyaAdapterAck :: GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) +shibuyaAdapterAck config (store, runner@(StoreRunner runStore)) = do + appendSignals runner (operationCount config) + measured <- measureLegWithSampler config "shibuya-adapter-ack" \sample -> do + outcome <- runStore do + adapter <- kirokuAdapter store (defaultKirokuAdapterConfig (SubscriptionName "retention-shibuya-ack") sourceTarget) + consumeAdapter (operationCount config) config.blockSize sample adapter `Effectful.finally` adapter.shutdown + void (expectRight outcome) + postReturn "shibuya-adapter-ack" (operationCount config) + pure measured + +consumeAdapter :: (IOE :> es) => Int -> Int -> (Int -> IO ()) -> Adapter es RecordedEvent -> Eff es () +consumeAdapter expected blockSize sample adapter = go 1 adapter.source + where + go index source = do + next <- Streamly.uncons source + case next of + Nothing -> liftIO $ fail ("shibuya-adapter-ack: ended after " <> show (index - 1) <> " events") + Just (item, rest) -> do + item.ack.finalize AckOk + when (index `mod` blockSize == 0) $ liftIO (sample index) + when (index < expected) (go (index + 1) rest) + +pmWorker :: GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) +pmWorker config (store, runner@(StoreRunner runStore)) = do + appendSignals runner (operationCount config) + measured <- measureLegWithSampler config "pm-worker" \sample -> do + outcome <- runStore do + adapter <- kirokuAdapter store (defaultKirokuAdapterConfig (SubscriptionName "retention-pm") sourceTarget) + instrumented <- sampleOnAck (operationCount config) config.blockSize sample adapter + runProcessManagerWorkerWith defaultWorkerOptions defaultRunCommandOptions retentionManager instrumented decodeSignal + `Effectful.finally` adapter.shutdown + void (expectRight outcome) + assertTargets runner (operationCount config) [0 .. 15] (\accountNumber -> length [index | index <- [1 .. operationCount config], index `mod` 16 == accountNumber]) + postReturn "pm-worker" (operationCount config) + pure measured + +routerWorker :: GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) +routerWorker config (store, runner@(StoreRunner runStore)) = do + let snapshotTargets = operationCount config > 1500 + Text.IO.putStrLn $ if snapshotTargets then "router-worker target snapshots: Every 100" else "router-worker target snapshots: Never" + appendSignals runner (operationCount config) + measured <- measureLegWithSampler config "router-worker" \sample -> do + outcome <- runStore do + adapter <- kirokuAdapter store (defaultKirokuAdapterConfig (SubscriptionName "retention-router") sourceTarget) + instrumented <- sampleOnAck (operationCount config) config.blockSize sample adapter + runRouterWorkerWith defaultWorkerOptions defaultRunCommandOptions (retentionRouter snapshotTargets) instrumented decodeSignal + `Effectful.finally` adapter.shutdown + void (expectRight outcome) + assertTargets runner (operationCount config) [0 .. 3] (const (operationCount config)) + postReturn "router-worker" (operationCount config) + pure measured + +assertTargets :: StoreRunner -> Int -> [Int] -> (Int -> Int) -> IO () +assertTargets (StoreRunner runStore) total accounts expectedFor = + forM_ accounts \accountNumber -> do + let name = StreamName ("retentiontarget-" <> Text.pack (show accountNumber)) + events <- expectRight =<< runStore (Store.readStreamForward name (StreamVersion 0) (fromIntegral (total + 1))) + unless (Vector.length events == expectedFor accountNumber) $ + fail ("target " <> show name <> ": expected " <> show (expectedFor accountNumber) <> " events, got " <> show (Vector.length events)) + +projectionApply :: GateConfig -> (Store.KirokuStore, StoreRunner) -> IO (Verdict, [HeapSample]) +projectionApply config (store, runner@(StoreRunner runStore)) = do + void (expectRight =<< runStore (registerReadModel "retention-activity" 1 "retention-activity-v1")) + appendSignals runner (operationCount config) + measured <- measureLegWithSampler config "projection-apply" \sample -> do + countRef <- newIORef 0 + let projection = + AsyncProjection + { name = "retention-activity", + readModelName = "retention-activity", + subscriptionName = "retention-projection", + applyRecorded = \_ -> pure (), + idempotencyKey = (.eventId) + } + handler event = do + outcome <- expectRight =<< runStore (Store.runTransaction (applyAsyncProjection projection event)) + unless (outcome == AsyncApplied) $ fail ("projection-apply: " <> show outcome) + count <- atomicModifyIORef' countRef (\previous -> let next = previous + 1 in (next, next)) + when (count `mod` config.blockSize == 0) (sample count) + pure if count == operationCount config then Stop else Continue + config' = defaultSubscriptionConfig (SubscriptionName "retention-projection") sourceTarget handler + completed <- timeout 120_000_000 (Subscription.withSubscription store config' Subscription.wait) + case completed of + Nothing -> fail "projection-apply: subscription timed out" + Just (Left failure) -> fail ("projection-apply: " <> show failure) + Just (Right ()) -> pure () + count <- readIORef countRef + unless (count == operationCount config) $ fail ("projection-apply: processed " <> show count <> " events") + postReturn "projection-apply" (operationCount config) + pure measured + +postReturn :: Text -> Int -> IO () +postReturn name count = do + sample <- sampleHeap count + Text.IO.putStrLn ("post-return leg=" <> name <> " live_bytes=" <> Text.pack (show sample.liveBytes) <> " large_objects_bytes=" <> Text.pack (show sample.largeObjectBytes)) + +expectRight :: (Show errorType) => Either errorType a -> IO a +expectRight = either (fail . show) pure diff --git a/retention/Retention/Measure.hs b/retention/Retention/Measure.hs new file mode 100644 --- /dev/null +++ b/retention/Retention/Measure.hs @@ -0,0 +1,164 @@ +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE OverloadedRecordDot #-} + +module Retention.Measure + ( HeapSample (..), + Verdict (..), + GateConfig (..), + gateConfigFromEnvironment, + sampleHeap, + judge, + measureLeg, + measureLegWithSampler, + renderTable, + ) +where + +import Control.Monad (forM_, unless) +import Data.IORef (modifyIORef', newIORef, readIORef) +import Data.List (intercalate) +import Data.Text (Text) +import Data.Text qualified as Text +import Data.Text.IO qualified as Text.IO +import Data.Word (Word64) +import GHC.Conc (ThreadStatus (..), listThreads, threadStatus) +import GHC.Stats (GCDetails (..), RTSStats (..), getRTSStats, getRTSStatsEnabled) +import Numeric (showFFloat) +import System.Environment (lookupEnv) +import System.Mem (performMajorGC) +import Text.Read (readMaybe) + +data HeapSample = HeapSample + { operations :: !Int, + liveBytes :: !Word64, + largeObjectBytes :: !Word64, + threads :: !Int + } + deriving stock (Eq, Show) + +data Verdict = Bounded | Retained + deriving stock (Eq, Show) + +data GateConfig = GateConfig + { blocks :: !Int, + blockSize :: !Int, + warmupBlocks :: !Int, + slopeFloorBytesPerOp :: !Double, + growthFloorBytes :: !Word64, + reportOnly :: !Bool + } + deriving stock (Eq, Show) + +gateConfigFromEnvironment :: IO GateConfig +gateConfigFromEnvironment = do + total <- positiveEnv "KEIRO_RETENTION_OPERATIONS" 1500 + requestedBlocks <- lookupEnv "KEIRO_RETENTION_BLOCKS" + blockCount <- case requestedBlocks of + Nothing -> + case filter (\candidate -> total `mod` candidate == 0) [6 .. 12] of + first : _ -> pure first + [] -> fail "KEIRO_RETENTION_OPERATIONS needs a divisor from 6 to 12; set KEIRO_RETENTION_BLOCKS explicitly" + Just _ -> positiveEnv "KEIRO_RETENTION_BLOCKS" 6 + unless (blockCount >= 6 && total `mod` blockCount == 0) $ + fail "retention probe requires at least 6 blocks and operations divisible by blocks" + report <- maybe False (const True) <$> lookupEnv "KEIRO_RETENTION_REPORT_ONLY" + pure + GateConfig + { blocks = blockCount, + blockSize = total `div` blockCount, + warmupBlocks = 1, + slopeFloorBytesPerOp = 512, + growthFloorBytes = 2 * 1024 * 1024, + reportOnly = report + } + +positiveEnv :: String -> Int -> IO Int +positiveEnv name defaultValue = do + raw <- lookupEnv name + case raw of + Nothing -> pure defaultValue + Just value -> case readMaybe value of + Just parsed | parsed > 0 -> pure parsed + _ -> fail (name <> " must be a positive integer") + +sampleHeap :: Int -> IO HeapSample +sampleHeap count = do + enabled <- getRTSStatsEnabled + unless enabled $ fail "keiro-retention requires RTS statistics; run with +RTS -T -RTS" + performMajorGC + stats <- getRTSStats + statuses <- listThreads >>= traverse threadStatus + let threadCount = length (filter isActive statuses) + pure + HeapSample + { operations = count, + liveBytes = gcdetails_live_bytes stats.gc, + largeObjectBytes = gcdetails_large_objects_bytes stats.gc, + threads = threadCount + } + where + -- listThreads also returns finished Async handles that can remain + -- reachable during a measurement loop. They are not running workers. + isActive ThreadFinished = False + isActive ThreadDied = False + isActive _ = True + +judge :: GateConfig -> [HeapSample] -> (Verdict, Double, Word64) +judge config allSamples = + let kept = drop config.warmupBlocks allSamples + xs = fmap (fromIntegral . (.operations)) kept :: [Double] + ys = fmap (fromIntegral . (.liveBytes)) kept :: [Double] + count = fromIntegral (length kept) + xMean = sum xs / count + yMean = sum ys / count + numerator = sum (zipWith (\x y -> (x - xMean) * (y - yMean)) xs ys) + denominator = sum (fmap (\x -> (x - xMean) ^ (2 :: Int)) xs) + slope = if denominator == 0 then 0 else numerator / denominator + growth = case kept of + [] -> 0 + first : _ -> max 0 (toInteger (liveBytes (last kept)) - toInteger first.liveBytes) + verdict = if slope > config.slopeFloorBytesPerOp && growth > toInteger config.growthFloorBytes then Retained else Bounded + in (verdict, slope, fromInteger growth) + +measureLeg :: GateConfig -> Text -> (Int -> IO ()) -> IO (Verdict, [HeapSample]) +measureLeg config name operation = + measureLegWithSampler config name \sample -> + forM_ [1 .. config.blocks * config.blockSize] \index -> do + operation index + whenBlockEnd index sample + where + whenBlockEnd index sample = + if index `mod` config.blockSize == 0 then sample index else pure () + +measureLegWithSampler :: GateConfig -> Text -> ((Int -> IO ()) -> IO ()) -> IO (Verdict, [HeapSample]) +measureLegWithSampler config name drive = do + samplesRef <- newIORef [] + drive \count -> do + sample <- sampleHeap count + modifyIORef' samplesRef (sample :) + samples <- reverse <$> readIORef samplesRef + let expected = fmap (* config.blockSize) [1 .. config.blocks] + unless (fmap (.operations) samples == expected) $ + fail ("retention leg " <> Text.unpack name <> " sampled at unexpected operation counts: " <> show (fmap (.operations) samples)) + let result@(verdict, _, _) = judge config samples + Text.IO.putStrLn (renderTable name config samples result) + pure (verdict, samples) + +renderTable :: Text -> GateConfig -> [HeapSample] -> (Verdict, Double, Word64) -> Text +renderTable name config samples (verdict, slope, growth) = + Text.unlines $ + [ "leg=" <> name <> " operations=" <> decimal (config.blocks * config.blockSize) <> " blocks=" <> decimal config.blocks <> " block=" <> decimal config.blockSize, + "block ops live_bytes large_objects_bytes threads" + ] + <> zipWith renderSample [1 :: Int ..] samples + <> [ "slope=" <> Text.pack (show (round slope :: Int)) <> " B/op growth=" <> Text.pack (showFFloat (Just 2) (fromIntegral growth / (1024 * 1024) :: Double) "") <> " MiB verdict=" <> case verdict of + Bounded -> "bounded" + Retained -> "retained" + ] + where + decimal = Text.pack . show + renderSample block sample = + Text.intercalate " " [decimal block, decimal sample.operations, grouped sample.liveBytes, grouped sample.largeObjectBytes, decimal sample.threads] + grouped number = Text.pack (reverse (intercalate "," (chunksOfThree (reverse (show number))))) + chunksOfThree [] = [] + chunksOfThree digits = let (chunk, rest) = splitAt 3 digits in chunk : chunksOfThree rest diff --git a/src/Keiro/Command.hs b/src/Keiro/Command.hs --- a/src/Keiro/Command.hs +++ b/src/Keiro/Command.hs @@ -89,7 +89,8 @@ where import Control.Concurrent (threadDelay) -import Control.Exception (displayException) +import Control.Concurrent.STM (TVar, atomically, newTVarIO, readTVar, writeTVar) +import Control.Exception (SomeException, displayException) import Data.Aeson qualified as Aeson #ifdef KEIRO_REACTION_HYDRATION_PROBE import Data.ByteString.Char8 qualified as ByteString.Char8 @@ -104,7 +105,7 @@ import Effectful.Concurrent (runConcurrent) import Effectful.Concurrent.Async qualified as Async import Effectful.Error.Static (Error, tryError) -import Effectful.Exception (trySync) +import Effectful.Exception (finally, mask_, onException, trySync) import GHC.Clock (getMonotonicTimeNSec) import GHC.Stack (HasCallStack) import Keiki.Core (BoolAlg, RegFile) @@ -143,6 +144,8 @@ recordSnapshotReadHits, recordSnapshotReadMisses, recordSnapshotSeedDivergence, + recordSnapshotSeedSkipped, + recordSnapshotSeedVerificationFailed, recordSnapshotWriteFailures, withCommandSpan, ) @@ -175,6 +178,7 @@ import Streamly.Data.Fold qualified as Fold import Streamly.Data.Stream qualified as Streamly import System.IO (stderr) +import System.IO.Unsafe (unsafePerformIO) import System.Random.Stateful (globalStdGen, uniformRM) import "hasql-transaction" Hasql.Transaction qualified as Tx import Prelude qualified @@ -350,6 +354,9 @@ -- 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. +-- * 'seedVerifyInFlightLimit' — maximum concurrent sampled verifications in +-- this process. A sampled verification is skipped when the limit is full; +-- @0@ removes the limit. data RunCommandOptions = RunCommandOptions { retryLimit :: !Int, pageSize :: !Int32, @@ -359,6 +366,7 @@ metrics :: !(Maybe KeiroMetrics), verifyReplayOnAppend :: !Bool, seedVerifySampleRate :: !Int, + seedVerifyInFlightLimit :: !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 / @@ -391,6 +399,7 @@ metrics = Nothing, verifyReplayOnAppend = True, seedVerifySampleRate = 1000, + seedVerifyInFlightLimit = 1, tracer = Nothing, metadata = Nothing } @@ -667,17 +676,67 @@ SnapshotSeed rs s -> Eff es () scheduleSeedVerification options eventStream targetStream codec seed = do - void $ trySync $ do + scheduled <- 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 + when sampled $ mask_ $ do + let limit = options ^. #seedVerifyInFlightLimit + acquired <- liftIO (reserveSeedVerification limit) + if acquired + then + ( void + $ runConcurrent + $ Async.async + $ ( do + result <- trySync (verifySnapshotSeed options eventStream targetStream codec seed) + case result of + Left failure -> reportSeedVerificationFailure options failure + Right () -> pure () + ) + `finally` releaseSeedVerificationEff + ) + `onException` releaseSeedVerificationEff + else recordSnapshotSeedSkipped (options ^. #metrics) 1 + case scheduled of + Left failure -> void $ trySync (reportSeedVerificationFailure options failure) + Right () -> pure () + +reportSeedVerificationFailure :: (IOE :> es) => RunCommandOptions -> SomeException -> Eff es () +reportSeedVerificationFailure options failure = do + recordSnapshotSeedVerificationFailed (options ^. #metrics) 1 + liftIO + $ LazyByteString.hPutStrLn stderr + $ Aeson.encode + $ Aeson.object + [ "event" Aeson..= ("keiro.snapshot.seed.verification.failed" :: Text), + "level" Aeson..= ("error" :: Text), + "reason" Aeson..= displayException failure + ] + +-- A process-wide reservation prevents commands on different streams from +-- accumulating detached full-replay tasks. Atomic reservation includes the +-- limit check so competing command threads cannot oversubscribe it. +{-# NOINLINE seedVerificationsInFlight #-} +seedVerificationsInFlight :: TVar Int +seedVerificationsInFlight = unsafePerformIO (newTVarIO 0) + +releaseSeedVerificationEff :: (IOE :> es) => Eff es () +releaseSeedVerificationEff = liftIO releaseSeedVerification + +reserveSeedVerification :: Int -> IO Bool +reserveSeedVerification limit = atomically $ do + count <- readTVar seedVerificationsInFlight + if limit <= 0 Prelude.|| count < limit + then writeTVar seedVerificationsInFlight (count Prelude.+ 1) $> True + else pure False + +releaseSeedVerification :: IO () +releaseSeedVerification = atomically $ do + count <- readTVar seedVerificationsInFlight + writeTVar seedVerificationsInFlight (count Prelude.- 1) verifySnapshotSeed :: forall phi rs s ci co es. diff --git a/src/Keiro/Telemetry.hs b/src/Keiro/Telemetry.hs --- a/src/Keiro/Telemetry.hs +++ b/src/Keiro/Telemetry.hs @@ -108,6 +108,8 @@ keiroSnapshotWriteFailuresName, keiroSnapshotApplyDivergenceName, keiroSnapshotSeedDivergenceName, + keiroSnapshotSeedSkippedName, + keiroSnapshotSeedVerificationFailedName, keiroDispatchFailedName, keiroDispatchDeadletteredName, keiroSubscriptionDeadletteredName, @@ -162,6 +164,8 @@ recordSnapshotWriteFailures, recordSnapshotApplyDivergence, recordSnapshotSeedDivergence, + recordSnapshotSeedSkipped, + recordSnapshotSeedVerificationFailed, recordDispatchFailed, recordDispatchDeadLettered, recordSubscriptionDeadLettered, @@ -691,6 +695,12 @@ keiroSnapshotSeedDivergenceName :: Text keiroSnapshotSeedDivergenceName = "keiro.snapshot.seed.divergence" +keiroSnapshotSeedSkippedName :: Text +keiroSnapshotSeedSkippedName = "keiro.snapshot.seed.skipped" + +keiroSnapshotSeedVerificationFailedName :: Text +keiroSnapshotSeedVerificationFailedName = "keiro.snapshot.seed.verification.failed" + keiroDispatchFailedName :: Text keiroDispatchFailedName = "keiro.dispatch.failed" @@ -782,6 +792,8 @@ snapshotWriteFailures :: Counter Int64, snapshotApplyDivergence :: Counter Int64, snapshotSeedDivergence :: Counter Int64, + snapshotSeedSkipped :: Counter Int64, + snapshotSeedVerificationFailed :: Counter Int64, dispatchFailed :: Counter Int64, dispatchDeadlettered :: Counter Int64, subscriptionDeadlettered :: Counter Int64, @@ -844,6 +856,8 @@ snapshotWriteFailures' <- counterI64 keiroSnapshotWriteFailuresName "{failure}" "Post-commit snapshot writes that failed and were swallowed." snapshotApplyDivergence' <- counterI64 keiroSnapshotApplyDivergenceName "{failure}" "Just-appended event batches that failed to replay from the pre-command state; the stream is poisoned and its next hydration will fail." snapshotSeedDivergence' <- counterI64 keiroSnapshotSeedDivergenceName "{failure}" "Sampled snapshot seeds whose encoded state disagreed with a full replay through the seed version." + snapshotSeedSkipped' <- counterI64 keiroSnapshotSeedSkippedName "{verification}" "Sampled snapshot-seed verifications skipped because the in-flight limit was full." + snapshotSeedVerificationFailed' <- counterI64 keiroSnapshotSeedVerificationFailedName "{failure}" "Sampled snapshot-seed verification tasks that raised an exception." dispatchFailed' <- counterI64 keiroDispatchFailedName "{command}" "Process-manager/router dispatch commands that failed." dispatchDeadlettered' <- counterI64 keiroDispatchDeadletteredName "{command}" "Rejected process-manager/router dispatch commands handled by dead-letter or skip policy." subscriptionDeadlettered' <- counterI64 keiroSubscriptionDeadletteredName "{event}" "Kiroku source events dead-lettered by an explicit disposition or retry exhaustion." @@ -898,6 +912,8 @@ snapshotWriteFailures = snapshotWriteFailures', snapshotApplyDivergence = snapshotApplyDivergence', snapshotSeedDivergence = snapshotSeedDivergence', + snapshotSeedSkipped = snapshotSeedSkipped', + snapshotSeedVerificationFailed = snapshotSeedVerificationFailed', dispatchFailed = dispatchFailed', dispatchDeadlettered = dispatchDeadlettered', subscriptionDeadlettered = subscriptionDeadlettered', @@ -1065,6 +1081,12 @@ recordSnapshotSeedDivergence :: (MonadIO m) => Maybe KeiroMetrics -> Int64 -> m () recordSnapshotSeedDivergence = recordCounter snapshotSeedDivergence + +recordSnapshotSeedSkipped :: (MonadIO m) => Maybe KeiroMetrics -> Int64 -> m () +recordSnapshotSeedSkipped = recordCounter snapshotSeedSkipped + +recordSnapshotSeedVerificationFailed :: (MonadIO m) => Maybe KeiroMetrics -> Int64 -> m () +recordSnapshotSeedVerificationFailed = recordCounter snapshotSeedVerificationFailed recordDispatchFailed :: (MonadIO m) => Maybe KeiroMetrics -> Int64 -> m () recordDispatchFailed = recordCounter dispatchFailed diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -42,7 +42,7 @@ import Effectful.Error.Static (Error, throwError) import Effectful.Exception qualified as EffException import ExternalReadSpec qualified -import GHC.Conc (ThreadStatus (..), threadStatus) +import GHC.Conc (ThreadStatus (..), listThreads, threadStatus) import GroupRebuildSpec qualified import Hasql.Decoders qualified as D import Hasql.Encoders qualified as E @@ -2816,6 +2816,136 @@ Tx.statement targetName snapshotVersionForStreamStmt snapshotVersion `shouldBe` Just (StreamVersion 2) + it "bounds sampled full replays to one process-wide task and counts skipped samples" $ \storeHandle -> do + (exporter, metricsRef) <- inMemoryMetricExporter + (provider, _env) <- + createMeterProvider emptyMaterializedResources defaultSdkMeterProviderOptions {metricExporter = Just exporter} + meter <- getMeter provider Telemetry.keiroInstrumentationLibrary + keiroMetrics <- Telemetry.newKeiroMetrics meter + let targetName = "snapshot-seed-verification-bounded" + target = stream targetName :: Stream SnapshotCounterEventStream + candidateStream = + mkEventStreamOrThrow + "snapshot-counter-fold-v2-bounded" + (foldV2WithoutFingerprintBumpEventStreamDef & #snapshotPolicy .~ Never) + options = + defaultRunCommandOptions + & #metrics + ?~ keiroMetrics + & #seedVerifySampleRate + .~ 1 + & #seedVerifyInFlightLimit + .~ 1 + seedLongSnapshotCounter storeHandle target 10000 + baselineThreads <- length <$> listThreads + threadCounts <- forM [1 .. 20 :: Int] $ \_ -> do + result <- Store.runStoreIO storeHandle $ runCommand options candidateStream target (Add 1) + case result of + Right (Right commandResult) -> commandResult ^. #eventsAppended `shouldBe` 1 + other -> expectationFailure ("sampled command failed: " <> show other) + length <$> listThreads + maximum threadCounts `shouldSatisfy` (<= baselineThreads + 2) + observed <- timeout 15_000_000 $ do + let awaitCounts = do + _ <- forceFlushMeterProvider provider Nothing + exported <- readIORef metricsRef + let points = reverse (flattenScalarPoints exported) + count name = case lookup name points of + Just (IntNumber n) -> n + _ -> 0 + skipped = count "keiro.snapshot.seed.skipped" + diverged = count "keiro.snapshot.seed.divergence" + if skipped + diverged == 20 + then pure (skipped, diverged) + else threadDelay 10_000 >> awaitCounts + awaitCounts + case observed of + Just (skipped, diverged) -> do + skipped `shouldSatisfy` (> 0) + diverged `shouldSatisfy` (> 0) + Nothing -> expectationFailure "sampled verifications did not finish or account for all twenty commands" + + it "allows unbounded sampled verification when the limit is zero" $ \storeHandle -> do + (exporter, metricsRef) <- inMemoryMetricExporter + (provider, _env) <- + createMeterProvider emptyMaterializedResources defaultSdkMeterProviderOptions {metricExporter = Just exporter} + meter <- getMeter provider Telemetry.keiroInstrumentationLibrary + keiroMetrics <- Telemetry.newKeiroMetrics meter + let target = stream "snapshot-seed-verification-unbounded" :: Stream SnapshotCounterEventStream + candidateStream = + mkEventStreamOrThrow + "snapshot-counter-fold-v2-unbounded" + (foldV2WithoutFingerprintBumpEventStreamDef & #snapshotPolicy .~ Never) + options = + defaultRunCommandOptions + & #metrics + ?~ keiroMetrics + & #seedVerifySampleRate + .~ 1 + & #seedVerifyInFlightLimit + .~ 0 + seedLongSnapshotCounter storeHandle target 1000 + forM_ [1 .. 4 :: Int] $ \_ -> do + Right (Right result) <- Store.runStoreIO storeHandle $ runCommand options candidateStream target (Add 1) + result ^. #eventsAppended `shouldBe` 1 + observed <- timeout 10_000_000 $ do + let awaitCounts = do + _ <- forceFlushMeterProvider provider Nothing + exported <- readIORef metricsRef + case lookup "keiro.snapshot.seed.divergence" (reverse (flattenScalarPoints exported)) of + Just (IntNumber 4) -> pure () + _ -> threadDelay 10_000 >> awaitCounts + awaitCounts + observed `shouldBe` Just () + _ <- forceFlushMeterProvider provider Nothing + exported <- readIORef metricsRef + lookup "keiro.snapshot.seed.skipped" (flattenScalarPoints exported) `shouldBe` Nothing + + it "releases a verification slot after an unexpected replay exception" $ \storeHandle -> do + (exporter, metricsRef) <- inMemoryMetricExporter + (provider, _env) <- + createMeterProvider emptyMaterializedResources defaultSdkMeterProviderOptions {metricExporter = Just exporter} + meter <- getMeter provider Telemetry.keiroInstrumentationLibrary + keiroMetrics <- Telemetry.newKeiroMetrics meter + let target = stream "snapshot-seed-verification-failed" :: Stream SnapshotCounterEventStream + candidateStream = + mkEventStreamOrThrow + "snapshot-counter-fold-v1-broken-replay" + ( foldV1SnapshotCounterEventStreamDef + & #eventCodec + .~ ( counterCodec + & #decode + .~ ( \eventType value -> + case parseCounterEvent eventType value of + Right (CounterAdded 2) -> error "intentional seed verification replay failure" + decoded -> decoded + ) + ) + & #snapshotPolicy + .~ Never + ) + options = + defaultRunCommandOptions + & #metrics + ?~ keiroMetrics + & #seedVerifySampleRate + .~ 1 + & #seedVerifyInFlightLimit + .~ 1 + Right (Right _) <- Store.runStoreIO storeHandle $ runCommand defaultRunCommandOptions foldV1SnapshotCounterEventStream target (Add 2) + Right (Right _) <- Store.runStoreIO storeHandle $ runCommand defaultRunCommandOptions foldV1SnapshotCounterEventStream target (Add 3) + forM_ [1 .. 2 :: Int] $ \expected -> do + Right (Right _) <- Store.runStoreIO storeHandle $ runCommand options candidateStream target (Add 4) + observed <- timeout 5_000_000 $ do + let awaitFailure = do + _ <- forceFlushMeterProvider provider Nothing + exported <- readIORef metricsRef + case lookup "keiro.snapshot.seed.verification.failed" (reverse (flattenScalarPoints exported)) of + Just (IntNumber count) | count == fromIntegral expected -> pure () + _ -> threadDelay 10_000 >> awaitFailure + awaitFailure + observed `shouldBe` Just () + it "disables sampled seed verification at rate zero" $ \storeHandle -> do (exporter, metricsRef) <- inMemoryMetricExporter (provider, _env) <- @@ -15657,6 +15787,25 @@ snapshotCounterEventStream :: ValidatedSnapshotCounterEventStream snapshotCounterEventStream = mkEventStreamOrThrow "snapshot-counter" snapshotCounterEventStreamDef + +-- Seed the same accepted fold and snapshot shape as a long-lived stream +-- without running thousands of commands in the verification-bound examples. +seedLongSnapshotCounter :: Store.KirokuStore -> Stream SnapshotCounterEventStream -> Int -> IO () +seedLongSnapshotCounter storeHandle target historyLength = do + encoded <- shouldBeRight (encodeForAppend counterCodec (CounterAdded 1)) + let targetName = Stream.streamName target + rawCount = historyLength - 1 + (fullChunks, remainder) = rawCount `divMod` 500 + forM_ [1 .. fullChunks] $ \_ -> do + Right _ <- Store.runStoreIO storeHandle $ Store.appendToStream targetName AnyVersion (replicate 500 encoded) + pure () + when (remainder > 0) $ do + Right _ <- Store.runStoreIO storeHandle $ Store.appendToStream targetName AnyVersion (replicate remainder encoded) + pure () + Right (Right result) <- + Store.runStoreIO storeHandle $ + runCommand defaultRunCommandOptions foldV1SnapshotCounterEventStream target (Add 1) + result ^. #streamVersion `shouldBe` StreamVersion (fromIntegral historyLength) snapshotCounterTransducer :: SymTransducer (HsPred SnapshotCounterRegs CounterCommand) SnapshotCounterRegs CounterState CounterCommand CounterEvent snapshotCounterTransducer =