diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,17 @@ ## [Unreleased] +## 0.18.0.0 — 2026-09-20 + + +### New Features + +- Re-export `Keiro.Codec.Base16Bytes`, `Keiro.Codec.CalendarDay`, and + `Keiro.Codec.TextSet` from `keiro-core`. Generated codecs import these + modules directly whenever a spec declares a base16, calendar-day, or text-set + mapping, so a consumer of a generated service continues to need only a single + direct `keiro` dependency. + ## 0.17.0.0 — 2026-09-17 ### Breaking Changes 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.17.0.0 +version: 0.18.0.0 synopsis: Event sourcing framework and workflow engine description: A library that composes kiroku, keiki, and shibuya into an @@ -131,10 +131,13 @@ reexported-modules: keiro-core:Keiro.Codec, + keiro-core:Keiro.Codec.Base16Bytes, + keiro-core:Keiro.Codec.CalendarDay, keiro-core:Keiro.Codec.IdDomain, keiro-core:Keiro.Codec.Nominal, keiro-core:Keiro.Codec.Structural, keiro-core:Keiro.Codec.Structural.Generic, + keiro-core:Keiro.Codec.TextSet, keiro-core:Keiro.EventStream, keiro-core:Keiro.EventStream.Validate, keiro-core:Keiro.Integration.Event, @@ -164,7 +167,7 @@ hs-opentelemetry-semantic-conventions >=1.40 && <2, keiki >=0.9 && <0.10, keiki-codec-json >=0.9 && <0.10, - keiro-core ^>=0.17.0.0, + keiro-core ^>=0.18.0.0, kiroku-store >=0.8 && <0.9, lens >=5.2 && <5.4, mmzk-typeid >=0.7 && <0.8, @@ -196,11 +199,14 @@ Paths_keiro PreCanonicalRecoverySpec PreimageSpec + ProcessManagerReplayCompatibilitySpec ProjectionReplaySpec ReactionExample ReadModelSpec + ReplayCompatibilitySpec VersionedRebuildSpec VersionedTargetPostgresSpec + WorkflowReplayCompatibilitySpec ghc-options: -threaded @@ -227,7 +233,7 @@ keiki, keiki-codec-json, keiro, - keiro-test-support ^>=0.17.0.0, + keiro-test-support ^>=0.18.0.0, kiroku-store >=0.8 && <0.9, process >=1.6 && <1.7, shibuya-core ^>=0.9.0.0, @@ -269,8 +275,8 @@ hs-opentelemetry-sdk >=1.0 && <1.1, keiki >=0.9 && <0.10, keiro, - keiro-core ^>=0.17.0.0, - keiro-test-support ^>=0.17.0.0, + 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, streamly-core >=0.3 && <0.4, diff --git a/test/Main.hs b/test/Main.hs --- a/test/Main.hs +++ b/test/Main.hs @@ -240,6 +240,7 @@ withFreshStores2, withMigratedSuite, ) +import Keiro.Test.ReplayCompatibility qualified as ReplayEvidence import Keiro.Timer import Keiro.Timer qualified as Timer import Keiro.Wake @@ -409,9 +410,11 @@ import Paths_keiro qualified as Package import PreCanonicalRecoverySpec qualified import PreimageSpec qualified +import ProcessManagerReplayCompatibilitySpec qualified import ProjectionReplaySpec qualified import ReactionExample qualified import ReadModelSpec qualified +import ReplayCompatibilitySpec qualified import Shibuya.Adapter (Adapter (..)) import Shibuya.Core.Ack (AckDecision (..), DeadLetterReason (..), HaltReason (..), RetryDelay (..), deadLetterCodeText, deadLetterReasonCode, deadLetterReasonDetail, renderDeadLetterReason) import Shibuya.Core.AckHandle (AckHandle (..)) @@ -424,6 +427,7 @@ import Test.Hspec import VersionedRebuildSpec qualified import VersionedTargetPostgresSpec qualified +import WorkflowReplayCompatibilitySpec qualified import "hasql-transaction" Hasql.Transaction qualified as Tx main :: IO () @@ -437,8 +441,11 @@ VersionedTargetPostgresSpec.spec fixture VersionedRebuildSpec.spec fixture PreCanonicalRecoverySpec.spec fixture + ProcessManagerReplayCompatibilitySpec.spec ProjectionReplaySpec.spec fixture ReadModelSpec.spec + ReplayCompatibilitySpec.spec fixture + WorkflowReplayCompatibilitySpec.spec fixture describe "catalog-fenced inline projections" $ around (withFreshResourceStore fixture) $ do it "rolls back the event append and target write while its group rebuilds" $ \(_storeHandle, StoreRunner runStore) -> do @@ -4572,6 +4579,83 @@ Right events <- runner $ Store.readStreamForward (StreamName "reaction-strict-target:order") (StreamVersion 0) 10 traverse (decodeRecorded feasibilityGateCodec) (Vector.toList events) `shouldBe` Right [GateOpened, GateAccepted 7] + + it "compares cross-build partial recovery traces and locates changed command meaning" $ \(_storeHandle, StoreRunner runner) -> do + let sourceEvent = recordedFromEventId (EventId sampleUuid) (CounterAdded 1) + runScenario correlation targetName timerId candidateAmount = do + let target = stream targetName :: Stream FeasibilityGateCommand + committedTimer = counterTimerRequest & #timerId .~ timerId + plan acceptedAmount = + Reaction.AdvanceReaction + (Add 1) + [ Reaction.FollowSchedule Reaction.Once committedTimer, + Reaction.FollowDispatch (PMCommand target (TryAccept acceptedAmount)), + Reaction.FollowDispatch (PMCommand target OpenGate) + ] + [] + Right (Right first) <- + runner $ + Reaction.runReactiveProcessManagerOnce + defaultRunCommandOptions + strictTargetReactionManager + sourceEvent + (correlation, plan 7) + first ^. #commandResults `shouldSatisfy` \case + [PMCommandFailed _ CommandRejected, PMCommandAppended {}] -> True + _ -> False + first ^. #timerEffects `shouldBe` Reaction.ReactionTimerEffects 1 1 0 + Right (Right recovered) <- + runner $ + Reaction.runReactiveProcessManagerOnce + defaultRunCommandOptions + strictTargetReactionManager + sourceEvent + (correlation, plan candidateAmount) + recovered ^. #managerResult `shouldSatisfy` \case + Reaction.ReactionDuplicate {} -> True + _ -> False + recovered ^. #commandResults `shouldSatisfy` \case + [PMCommandAppended {}, PMCommandDuplicate {}] -> True + _ -> False + recovered ^. #timerEffects `shouldBe` Reaction.ReactionTimerEffects 0 0 0 + Right sagaEvents <- + runner $ + Store.readStreamForward + (StreamName ("reaction-strict-saga:" <> correlation)) + (StreamVersion 0) + 10 + Right targetEvents <- + runner $ + Store.readStreamForward (StreamName targetName) (StreamVersion 0) 10 + Right decodedSaga <- pure (traverse (decodeRecorded counterCodec) (Vector.toList sagaEvents)) + Right decodedTarget <- pure (traverse (decodeRecorded feasibilityGateCodec) (Vector.toList targetEvents)) + Right timer <- runner (lookupTimer timerId) + let timerTrace = case timer of + Just row -> Text.pack (show (row ^. #status, row ^. #payload)) + Nothing -> "missing" + pure + ReplayEvidence.Observation + { durableState = + Map.fromList + [ ("saga-events", Aeson.toJSON (map (Text.pack . show) decodedSaga)), + ("target-events", Aeson.toJSON (map (Text.pack . show) decodedTarget)), + ("timer-state", Aeson.toJSON timerTrace) + ], + continuations = [], + durableIdentities = + Map.fromList + [ ("manager", "strict-target-reaction"), + ("source-event", Text.pack (show (sourceEvent ^. #eventId))), + ("target-occurrences", "logical-target:0,logical-target:1") + ], + freshAllocations = [] + } + baseline <- runScenario "cross-build-baseline" "cross-build-target-baseline" (TimerId (UUID.fromWords 0 0 0 201)) 7 + refactored <- runScenario "cross-build-refactored" "cross-build-target-refactored" (TimerId (UUID.fromWords 0 0 0 202)) 7 + changed <- runScenario "cross-build-changed" "cross-build-target-changed" (TimerId (UUID.fromWords 0 0 0 203)) 8 + ReplayEvidence.compareObservation "process/partial-recovery" baseline refactored `shouldBe` [] + ReplayEvidence.compareObservation "process/partial-recovery" baseline changed + `shouldBe` [ReplayEvidence.ObservationMismatch "process/partial-recovery"] it "reconciles a concurrent target loser that rehydrates to a silent result" $ \(_storeHandle, StoreRunner runner) -> do arrivals <- newMVar (0 :: Int) diff --git a/test/ProcessManagerReplayCompatibilitySpec.hs b/test/ProcessManagerReplayCompatibilitySpec.hs new file mode 100644 --- /dev/null +++ b/test/ProcessManagerReplayCompatibilitySpec.hs @@ -0,0 +1,84 @@ +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE OverloadedRecordDot #-} +{-# LANGUAGE OverloadedStrings #-} + +module ProcessManagerReplayCompatibilitySpec + ( spec, + ) +where + +import Data.Aeson qualified as Aeson +import Data.Foldable (traverse_) +import Data.Map.Strict qualified as Map +import Data.Text (Text) +import Keiro.Test.ReplayCompatibility +import Test.Hspec + +spec :: Spec +spec = describe "process-manager replay observation contract" do + it "accepts an unchanged target-keyed continuation trace" do + compareObservation processCase baselineObservation baselineObservation `shouldBe` [] + + it "rejects a migration from positional to target-keyed command identity" do + mismatch (changeIdentity "identity-family" "positional") + + it "rejects manager, correlation, source interpretation, timer, and witness drift independently" do + traverse_ + mismatch + [ changeIdentity "manager" "billing-v2", + changeIdentity "correlation" "account-42-v2", + changeState "source-meaning" (Aeson.String "amount=13"), + changeState "timer-decoder" (Aeson.String "narrow-v2"), + changeState "accepted-witness" (Aeson.String "undecodable") + ] + + it "rejects same-target reorder and payload changes" do + mismatch (baselineObservation {continuations = reverse baselineObservation.continuations}) + mismatch + ( baselineObservation + { continuations = + Aeson.object ["target" Aeson..= ("account-42" :: Text), "occurrence" Aeson..= (0 :: Int), "amount" Aeson..= (99 :: Int)] + : drop 1 baselineObservation.continuations + } + ) + where + mismatch candidate = + compareObservation processCase baselineObservation candidate + `shouldBe` [ObservationMismatch processCase] + +processCase :: Text +processCase = "process/billing/partial-recovery" + +baselineObservation :: Observation +baselineObservation = + Observation + { durableState = + Map.fromList + [ ("accepted-witness", Aeson.String "source-event-9"), + ("source-meaning", Aeson.String "amount=12"), + ("saga-state", Aeson.String "accepted"), + ("timer-decoder", Aeson.String "timer-v1"), + ("timer-payload", Aeson.object ["attempt" Aeson..= (1 :: Int)]), + ("timer-status", Aeson.String "scheduled") + ], + continuations = + [ Aeson.object ["target" Aeson..= ("account-42" :: Text), "occurrence" Aeson..= (0 :: Int), "amount" Aeson..= (12 :: Int)], + Aeson.object ["target" Aeson..= ("account-42" :: Text), "occurrence" Aeson..= (1 :: Int), "amount" Aeson..= (7 :: Int)] + ], + durableIdentities = + Map.fromList + [ ("manager", "billing"), + ("correlation", "account-42"), + ("source-event", "source-event-9"), + ("identity-family", "target-keyed") + ], + freshAllocations = [] + } + +changeIdentity :: Text -> Text -> Observation +changeIdentity key value = + baselineObservation {durableIdentities = Map.insert key value baselineObservation.durableIdentities} + +changeState :: Text -> Aeson.Value -> Observation +changeState key value = + baselineObservation {durableState = Map.insert key value baselineObservation.durableState} diff --git a/test/ReplayCompatibilitySpec.hs b/test/ReplayCompatibilitySpec.hs new file mode 100644 --- /dev/null +++ b/test/ReplayCompatibilitySpec.hs @@ -0,0 +1,170 @@ +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE DuplicateRecordFields #-} +{-# LANGUAGE OverloadedRecordDot #-} +{-# LANGUAGE OverloadedStrings #-} + +module ReplayCompatibilitySpec + ( spec, + ) +where + +import Data.Aeson (Value, object, withObject, (.:)) +import Data.Aeson qualified as Aeson +import Data.Aeson.Types (parseEither) +import Data.List.NonEmpty (NonEmpty (..)) +import Data.Map.Strict qualified as Map +import Data.Text (Text) +import Data.Text qualified as Text +import Data.Vector qualified as Vector +import Keiro.Codec +import Keiro.Test.Postgres (Fixture, withFreshStore) +import Keiro.Test.ReplayCompatibility +import Kiroku.Store qualified as Store +import Kiroku.Store.Types (ExpectedVersion (..), StreamName (..), StreamVersion (..)) +import Test.Hspec + +spec :: Fixture -> Spec +spec fixture = do + describe "serialized aggregate replay compatibility" $ around (withFreshStore fixture) do + it "crosses the PostgreSQL event envelope before candidate decoding" $ \store -> do + encoded <- shouldBeRight (encodeForAppend baselineCodec (WordHead 7)) + Right _ <- Store.runStoreIO store (Store.appendToStream (StreamName "replay-compatibility-aggregate") NoStream [encoded]) + Right stored <- Store.runStoreIO store (Store.readStreamForward (StreamName "replay-compatibility-aggregate") (StreamVersion 0) 10) + case Vector.toList stored of + [recorded] -> do + decodeRecorded baselineCodec recorded `shouldBe` Right (WordHead 7) + decodeRecorded refactoredCodec recorded `shouldBe` Right (WordHead 7) + events -> expectationFailure ("expected one stored event, got " <> show (length events)) + + describe "complete transition-prefix observations" do + it "keeps a source-module refactor equivalent at every complete prefix" do + let transitionWords = [[WordHead 2, WordTail 3], [WordHead 5, WordTail 7]] + traverse observeCompletePrefix (completePrefixes transitionWords) + `shouldBe` traverse observeRefactoredPrefix (completePrefixes transitionWords) + + it "rejects a truncated multi-event transition word" do + observeCompletePrefix [WordHead 2] `shouldBe` Left "transition word ended after its head" + + it "reports changed writes, retired inverse edges, and lost head data" do + let complete = [WordHead 2, WordTail 3] + observeChangedWrite complete `shouldNotBe` observeCompletePrefix complete + observeRetiredInverse complete `shouldBe` Left "WordHead inverse is unavailable" + observeCompletePrefix [WordTail 3] `shouldBe` Left "transition word lost its head" + + it "turns a semantic prefix drift into a compatibility failure" do + let baselineObservation = observationFor 5 + candidateObservation = observationFor 6 + oldReport = reportFor BaselineCapture baselineObservation + newReport = reportFor CandidateCapture candidateObservation + validateCompatibility prefixInventory oldReport newReport + `shouldContain` [ObservationMismatch "aggregate/word/prefix-1"] + +data WordEvent = WordHead Int | WordTail Int + deriving stock (Eq, Show) + +baselineCodec :: Codec WordEvent +baselineCodec = + Codec + { eventTypes = EventType "WordHead" :| [EventType "WordTail"], + eventType = \case WordHead {} -> EventType "WordHead"; WordTail {} -> EventType "WordTail", + schemaVersion = 1, + encode = encodeWord, + decode = decodeWord, + upcasters = [] + } + +-- A Haskell-only module/selector refactor retains the stable wire tags and keys. +refactoredCodec :: Codec WordEvent +refactoredCodec = baselineCodec {encode = refactoredEncode, decode = refactoredDecode} + where + refactoredEncode = encodeWord + refactoredDecode = decodeWord + +encodeWord :: WordEvent -> Value +encodeWord = \case + WordHead amount -> object ["amount" Aeson..= amount] + WordTail amount -> object ["amount" Aeson..= amount] + +decodeWord :: EventType -> Value -> Either Text WordEvent +decodeWord (EventType tag) value = case tag of + "WordHead" -> WordHead <$> parseAmount value + "WordTail" -> WordTail <$> parseAmount value + _ -> Left ("unknown word event: " <> tag) + +parseAmount :: Value -> Either Text Int +parseAmount value = case parseEither (withObject "WordEvent" (.: "amount")) value of + Left problem -> Left (Text.pack problem) + Right amount -> Right amount + +completePrefixes :: [[event]] -> [[event]] +completePrefixes = drop 1 . scanl (<>) [] + +observeCompletePrefix :: [WordEvent] -> Either Text Int +observeCompletePrefix = go 0 + where + go total [] = Right total + go _ [WordHead _] = Left "transition word ended after its head" + go _ (WordTail _ : _) = Left "transition word lost its head" + go _ (WordHead _ : WordHead _ : _) = Left "transition word contains consecutive heads" + go total (WordHead headAmount : WordTail tailAmount : rest) = go (total + headAmount + tailAmount) rest + +observeRefactoredPrefix :: [WordEvent] -> Either Text Int +observeRefactoredPrefix events = observeCompletePrefix events + +observeChangedWrite :: [WordEvent] -> Either Text Int +observeChangedWrite = fmap (+ 1) . observeCompletePrefix + +observeRetiredInverse :: [WordEvent] -> Either Text Int +observeRetiredInverse events + | any isHead events = Left "WordHead inverse is unavailable" + | otherwise = observeCompletePrefix events + where + isHead WordHead {} = True + isHead WordTail {} = False + +prefixCase :: RequiredCase +prefixCase = RequiredCase "aggregate/word/prefix-1" SemanticEquivalence prefixSurface + +prefixSurface :: PersistedSurface +prefixSurface = PersistedSurface "aggregate-stream" "word" "WordHead+WordTail:v1" + +prefixBuildPair :: BuildPair +prefixBuildPair = + BuildPair + { baseline = BuildIdentity "baseline" "language-v5" "runtime-v1" "plan-a", + candidate = BuildIdentity "candidate" "language-v5" "runtime-v1" "plan-b" + } + +prefixInventory :: EvidenceInventory +prefixInventory = + EvidenceInventory + inventoryVersionV1 + "prefix-inventory" + prefixBuildPair + [ InventoryContribution source applicability cases + | source <- inventorySourcesV1, + let applicable = source == BaselinePersistedSurfaces, + let applicability = if applicable then Applicable else NotApplicable "fixture does not exercise this source", + let cases = if applicable then [prefixCase] else [] + ] + +observationFor :: Int -> Observation +observationFor total = Observation (Map.fromList [("total", Aeson.toJSON total)]) [] Map.empty [] + +reportFor :: CaptureRole -> Observation -> CaptureReport +reportFor role observation = + CaptureReport + { reportVersion = reportVersionV1, + role, + buildPair = prefixBuildPair, + inventoryId = "prefix-inventory", + corpusHash = "prefix-corpus", + observationContractVersion = "word-observation/v1", + highWaterMarks = [HighWaterMark "word-1" 2], + selectedSurfaces = [prefixSurface], + determinismInputs = DeterminismInputs "fixed-clock" "fixed-randomness" "responses" "failures", + results = [CaseResult prefixCase Passed (Just observation)] + } + +shouldBeRight :: (Show error) => Either error value -> IO value +shouldBeRight = either (\problem -> expectationFailure (show problem) >> fail "unreachable") pure diff --git a/test/WorkflowReplayCompatibilitySpec.hs b/test/WorkflowReplayCompatibilitySpec.hs new file mode 100644 --- /dev/null +++ b/test/WorkflowReplayCompatibilitySpec.hs @@ -0,0 +1,197 @@ +{-# LANGUAGE BlockArguments #-} +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE OverloadedRecordDot #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE TypeApplications #-} +{-# LANGUAGE TypeOperators #-} + +module WorkflowReplayCompatibilitySpec + ( spec, + ) +where + +import Control.Exception (Exception, throwIO, try) +import Data.Aeson qualified as Aeson +import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef) +import Data.Set qualified as Set +import Data.Time (getCurrentTime) +import Effectful (Eff, IOE, (:>)) +import Effectful.Error.Static (Error) +import Keiro.Prelude (liftIO) +import Keiro.Test.Postgres (Fixture, withFreshStore) +import Keiro.Workflow +import Kiroku.Store qualified as Store +import Test.Hspec + +spec :: Fixture -> Spec +spec fixture = describe "cross-version workflow replay compatibility" $ around (withFreshStore fixture) do + it "continues a captured prefix under a source-only refactor without repeating its effect" $ \store -> do + baselineEffects <- newIORef (0 :: Int) + candidateEffects <- newIORef (0 :: Int) + let name = WorkflowName "compatibleRefactor" + wid = WorkflowId "instance-1" + captureCrashedPrefix store name wid (stableBaseline baselineEffects) + candidate <- Store.runStoreIO store $ runWorkflow name wid (stableCandidate candidateEffects) + candidate `shouldBe` Right (Completed 1) + readIORef baselineEffects `shouldReturn` 1 + readIORef candidateEffects `shouldReturn` 0 + + it "fails closed when candidate code cannot decode a captured result" $ \store -> do + baselineEffects <- newIORef (0 :: Int) + candidateEffects <- newIORef (0 :: Int) + let name = WorkflowName "decoderBreak" + wid = WorkflowId "instance-1" + captureCrashedPrefix store name wid (stableBaseline baselineEffects) + candidate <- + try (Store.runStoreIO store $ runWorkflow name wid (incompatibleCandidate candidateEffects)) :: + IO (Either WorkflowError (Either Store.StoreError (WorkflowOutcome Bool))) + candidate `shouldSatisfy` \case + Left (WorkflowStepDecodeError "stable-result" _) -> True + _ -> False + readIORef candidateEffects `shouldReturn` 0 + + it "exposes a decoder that succeeds with changed meaning at the first stable key" $ \store -> do + baselineEffects <- newIORef (0 :: Int) + candidateEffects <- newIORef (0 :: Int) + let name = WorkflowName "semanticDrift" + wid = WorkflowId "instance-1" + captureCrashedPrefix store name wid (semanticBaseline baselineEffects) + candidate <- Store.runStoreIO store $ runWorkflow name wid (semanticCandidate candidateEffects) + candidate `shouldBe` Right (Completed (CandidateAmount 6)) + CandidateAmount 6 `shouldNotBe` CandidateAmount 5 + readIORef candidateEffects `shouldReturn` 0 + + it "treats a renamed step as new work and therefore rejects it as a pure refactor" $ \store -> do + baselineEffects <- newIORef (0 :: Int) + candidateEffects <- newIORef (0 :: Int) + let name = WorkflowName "renamedStep" + wid = WorkflowId "instance-1" + captureCrashedPrefix store name wid (stableBaseline baselineEffects) + candidate <- Store.runStoreIO store $ runWorkflow name wid (renamedCandidate candidateEffects) + candidate `shouldBe` Right (Completed 1) + readIORef candidateEffects `shouldReturn` 1 + + it "keeps an in-flight instance on its recorded patch branch" $ \store -> do + baselineEffects <- newIORef (0 :: Int) + candidateEffects <- newIORef (0 :: Int) + let name = WorkflowName "patchedContinuation" + wid = WorkflowId "instance-1" + options = defaultWorkflowRunOptions {activePatches = Set.singleton compatibilityPatch} + baseline <- Store.runStoreIO store $ runWorkflow name wid (prePatchWorkflow baselineEffects) + baseline `shouldBe` Right Suspended + candidate <- Store.runStoreIO store $ runWorkflowWith options name wid (postPatchWorkflow candidateEffects) + candidate `shouldBe` Right (Completed "old-branch") + readIORef baselineEffects `shouldReturn` 1 + readIORef candidateEffects `shouldReturn` 0 + + it "reuses an indexed await result without rearming under candidate code" $ \store -> do + baselineArms <- newIORef (0 :: Int) + candidateArms <- newIORef (0 :: Int) + let name = WorkflowName "awaitContinuation" + wid = WorkflowId "instance-1" + baseline <- Store.runStoreIO store $ runWorkflow name wid (awaitingWorkflow baselineArms) + baseline `shouldBe` Right Suspended + now <- getCurrentTime + Right () <- + Store.runStoreIO store $ + appendJournalEntry name wid (StepRecorded "stable-await" (Aeson.toJSON (7 :: Int)) now) + candidate <- Store.runStoreIO store $ runWorkflow name wid (awaitingWorkflow candidateArms) + candidate `shouldBe` Right (Completed 7) + readIORef baselineArms `shouldReturn` 1 + readIORef candidateArms `shouldReturn` 0 + + it "continues from a baseline-carried seed after continueAsNew" $ \store -> do + let name = WorkflowName "rotatedContinuation" + wid = WorkflowId "instance-1" + baseline <- Store.runStoreIO store $ runWorkflow name wid baselineRotation + baseline `shouldBe` Right ContinuedAsNew + candidate <- Store.runStoreIO store $ runWorkflow name wid candidateRotation + candidate `shouldBe` Right (Completed 41) + +data CapturedPrefix = CapturedPrefix + deriving stock (Show) + +instance Exception CapturedPrefix + +captureCrashedPrefix :: + Store.KirokuStore -> + WorkflowName -> + WorkflowId -> + Eff '[Workflow, Store.Store, Error Store.StoreError, IOE] value -> + IO () +captureCrashedPrefix store name wid body = do + captured <- try @CapturedPrefix (Store.runStoreIO store $ runWorkflow name wid body) + case captured of + Left CapturedPrefix -> pure () + Right _ -> expectationFailure "baseline workflow completed instead of leaving a captured prefix" + +incrementAndRead :: IORef Int -> IO Int +incrementAndRead ref = atomicModifyIORef' ref (\value -> (value + 1, value + 1)) + +stableBaseline :: (Workflow :> es, IOE :> es) => IORef Int -> Eff es Int +stableBaseline effects = do + result <- step (StepName "stable-result") (liftIO (incrementAndRead effects)) + liftIO (failWithCapturedPrefix result) + +stableCandidate :: (Workflow :> es, IOE :> es) => IORef Int -> Eff es Int +stableCandidate effects = step (StepName "stable-result") (liftIO (incrementAndRead effects)) + +incompatibleCandidate :: (Workflow :> es, IOE :> es) => IORef Int -> Eff es Bool +incompatibleCandidate effects = step (StepName "stable-result") (liftIO (incrementAndRead effects) >> pure True) + +renamedCandidate :: (Workflow :> es, IOE :> es) => IORef Int -> Eff es Int +renamedCandidate effects = step (StepName "renamed-result") (liftIO (incrementAndRead effects)) + +newtype BaselineAmount = BaselineAmount Int + deriving stock (Eq, Show) + +instance Aeson.ToJSON BaselineAmount where + toJSON (BaselineAmount amount) = Aeson.toJSON amount + +instance Aeson.FromJSON BaselineAmount where + parseJSON value = BaselineAmount <$> Aeson.parseJSON value + +newtype CandidateAmount = CandidateAmount Int + deriving stock (Eq, Show) + +instance Aeson.ToJSON CandidateAmount where + toJSON (CandidateAmount amount) = Aeson.toJSON amount + +instance Aeson.FromJSON CandidateAmount where + parseJSON value = CandidateAmount . (+ 1) <$> Aeson.parseJSON value + +semanticBaseline :: (Workflow :> es, IOE :> es) => IORef Int -> Eff es BaselineAmount +semanticBaseline effects = do + result <- step (StepName "stable-result") (BaselineAmount <$> liftIO (incrementAndRead effects >> pure 5)) + liftIO (failWithCapturedPrefix result) + +semanticCandidate :: (Workflow :> es, IOE :> es) => IORef Int -> Eff es CandidateAmount +semanticCandidate effects = + step (StepName "stable-result") (CandidateAmount <$> liftIO (incrementAndRead effects >> pure 5)) + +failWithCapturedPrefix :: value -> IO value +failWithCapturedPrefix _ = throwIO CapturedPrefix + +compatibilityPatch :: PatchId +compatibilityPatch = PatchId "compatibility-v2" + +prePatchWorkflow :: (Workflow :> es, IOE :> es) => IORef Int -> Eff es String +prePatchWorkflow effects = do + _ <- step (StepName "stable-result") (liftIO (incrementAndRead effects)) + (_ :: ()) <- awaitStep (StepName "never-resolved") (pure ()) + pure "baseline" + +postPatchWorkflow :: (Workflow :> es, IOE :> es) => IORef Int -> Eff es String +postPatchWorkflow effects = do + _ <- step (StepName "stable-result") (liftIO (incrementAndRead effects)) + useNew <- patch compatibilityPatch + if useNew then pure "new-branch" else pure "old-branch" + +awaitingWorkflow :: (Workflow :> es, IOE :> es) => IORef Int -> Eff es Int +awaitingWorkflow arms = awaitStep (StepName "stable-await") (liftIO (incrementAndRead arms) >> pure ()) + +baselineRotation :: (Workflow :> es) => Eff es Int +baselineRotation = continueAsNew (41 :: Int) + +candidateRotation :: (Workflow :> es) => Eff es Int +candidateRotation = restoreSeed (0 :: Int)