keiro 0.18.0.0 → 0.19.0.0

raw patch · 9 files changed

+1358/−20 lines, 9 filesdep +shibuya-kiroku-adapterdep ~aesondep ~keiro-coredep ~keiro-test-supportPVP ok

version bump matches the API change (PVP)

Dependencies added: shibuya-kiroku-adapter

Dependency ranges changed: aeson, keiro-core, keiro-test-support, kiroku-store, shibuya-core

API changes (from Hackage documentation)

Files

• CHANGELOG.md +37/−0
• keiro.cabal +46/−11
• retention/Main.hs +45/−0
• retention/Retention/Fixture.hs +452/−0
• retention/Retention/Legs.hs +375/−0
• retention/Retention/Measure.hs +164/−0
• src/Keiro/Command.hs +67/−8
• src/Keiro/Telemetry.hs +22/−0
• test/Main.hs +150/−1

@@ -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  

@@ -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,

@@ -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 ()

@@ -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)+        }

@@ -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

@@ -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

@@ -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.

@@ -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

@@ -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 =