From 817e745ba4c901524980e75ca6db66404d2fd58f Mon Sep 17 00:00:00 2001 From: 0xgleb <39841057+0xgleb@users.noreply.github.com> Date: Wed, 12 Aug 2026 00:17:39 -0300 Subject: [PATCH] deliver Haskell dispatch verdicts --- bindings/haskell/event-sorcery.cabal | 14 + .../haskell/src/EventSorcery/Aggregate.hs | 23 +- .../src/EventSorcery/Dispatch/Worker.hs | 232 ++++++++++++ .../haskell/src/EventSorcery/Job/Worker.hs | 233 +------------ .../src/EventSorcery/Job/Worker/Definition.hs | 77 ++++ .../src/EventSorcery/Job/Worker/Internal.hs | 275 +++++++++++++++ bindings/haskell/src/EventSorcery/Store.hs | 7 + bindings/haskell/test/DispatchWorkerSpec.hs | 330 ++++++++++++++++++ bindings/haskell/test/DomainSpec.hs | 9 +- bindings/haskell/test/StoreSpec.hs | 7 +- 10 files changed, 975 insertions(+), 232 deletions(-) create mode 100644 bindings/haskell/src/EventSorcery/Dispatch/Worker.hs create mode 100644 bindings/haskell/src/EventSorcery/Job/Worker/Definition.hs create mode 100644 bindings/haskell/src/EventSorcery/Job/Worker/Internal.hs create mode 100644 bindings/haskell/test/DispatchWorkerSpec.hs diff --git a/bindings/haskell/event-sorcery.cabal b/bindings/haskell/event-sorcery.cabal index 5b3ef70..37ad303 100644 --- a/bindings/haskell/event-sorcery.cabal +++ b/bindings/haskell/event-sorcery.cabal @@ -54,6 +54,7 @@ library EventSorcery.Aggregate EventSorcery.Dispatch EventSorcery.Dispatch.TestSupport + EventSorcery.Dispatch.Worker EventSorcery.Engine EventSorcery.Engine.Internal.FFI EventSorcery.Job @@ -68,6 +69,8 @@ library EventSorcery.Engine.Acquisition EventSorcery.Engine.Internal EventSorcery.Job.Definition + EventSorcery.Job.Worker.Definition + EventSorcery.Job.Worker.Internal extra-libraries: event_sorcery_ffi build-depends: @@ -177,3 +180,14 @@ test-suite job-worker-spec , bytestring >=0.11 && <0.13 , event-sorcery , text >=2.0 && <2.2 + +test-suite dispatch-worker-spec + import: strict + type: exitcode-stdio-1.0 + hs-source-dirs: test + main-is: DispatchWorkerSpec.hs + build-depends: + , base >=4.18 && <5 + , bytestring >=0.11 && <0.13 + , event-sorcery + , text >=2.0 && <2.2 diff --git a/bindings/haskell/src/EventSorcery/Aggregate.hs b/bindings/haskell/src/EventSorcery/Aggregate.hs index dea2ccd..898a101 100644 --- a/bindings/haskell/src/EventSorcery/Aggregate.hs +++ b/bindings/haskell/src/EventSorcery/Aggregate.hs @@ -1,3 +1,5 @@ +{-# LANGUAGE FunctionalDependencies #-} + module EventSorcery.Aggregate ( DecodeCause (..), DispatchIntent, @@ -8,6 +10,7 @@ module EventSorcery.Aggregate ( Member, SchemaVersion (..), dispatchIntent, + dispatchIntentJob, dispatchJobId, ) where @@ -18,11 +21,10 @@ import Data.Proxy (Proxy) import Data.Text (Text) import Data.Type.Equality (type (~)) import Data.Word (Word16) -import EventSorcery.Dispatch (JobDispatch) +import EventSorcery.Dispatch (DispatchOutcome, JobDispatch) import EventSorcery.Job.Definition ( Job (..), JobId, - jobType, ) import Prelude (Bool (..), Either, Eq, Ord, Show) @@ -39,11 +41,12 @@ newtype DecodeCause = DecodeCause Text deriving stock (Eq, Show) -data DispatchIntent job = DispatchIntent JobId Text ByteString +data DispatchIntent job = DispatchIntent JobId job -class Dispatches entity job where +class Dispatches entity job | job -> entity where injectDispatchIntent :: DispatchIntent job -> Event entity + injectDispatchOutcome :: DispatchOutcome job -> Command entity type Member item items = Elem item items ~ 'True @@ -57,6 +60,7 @@ type family Elem (item :: Type) (items :: [Type]) :: Bool where data Effect entity where Events :: NonEmpty (Event entity) -> Effect entity + Unchanged :: Effect entity Dispatch :: (Job job, Member job (Jobs entity), Dispatches entity job) => JobDispatch job @@ -90,10 +94,13 @@ class EventSourced entity where -> Either (CommandError entity) (Effect entity) -dispatchIntent :: forall job. Job job => JobId -> job -> DispatchIntent job -dispatchIntent identifier job = - DispatchIntent identifier (jobType @job) (encodeJob job) +dispatchIntent :: JobId -> job -> DispatchIntent job +dispatchIntent = DispatchIntent dispatchJobId :: DispatchIntent job -> JobId -dispatchJobId (DispatchIntent identifier _ _) = identifier +dispatchJobId (DispatchIntent identifier _) = identifier + + +dispatchIntentJob :: DispatchIntent job -> job +dispatchIntentJob (DispatchIntent _ job) = job diff --git a/bindings/haskell/src/EventSorcery/Dispatch/Worker.hs b/bindings/haskell/src/EventSorcery/Dispatch/Worker.hs new file mode 100644 index 0000000..9f9c139 --- /dev/null +++ b/bindings/haskell/src/EventSorcery/Dispatch/Worker.hs @@ -0,0 +1,232 @@ +module EventSorcery.Dispatch.Worker ( + DeliveryPolicy, + DispatchWorker, + OriginDeliveryError (..), + OriginJob (..), + OriginPort, + deliveryPolicy, + dispatchWorker, + originPort, + runDispatchJobOnce, + storeOriginPort, +) where + +import Data.Word (Word32) +import EventSorcery.Aggregate ( + Dispatches (injectDispatchOutcome), + EventSourced, + ) +import EventSorcery.Dispatch (DispatchFailure (..), DispatchOutcome) +import EventSorcery.Dispatch.Internal qualified as Dispatch +import EventSorcery.Job ( + DeadReason (RetriesExhausted), + Job (JobError, JobOutput), + JobId, + JobInstant, + ) +import EventSorcery.Job.Execution ( + DurableJob (renderJobError), + JobAttempt (JobAttempt), + ) +import EventSorcery.Job.Worker ( + JobRunError, + JobRunResult, + JobWorker, + ) +import EventSorcery.Job.Worker.Internal ( + SettlementStrategy (..), + VerdictDelivery (..), + runJobOnceWith, + ) +import EventSorcery.Store ( + Store, + StoreError (..), + executeCommand, + ) +import EventSorcery.Stream (StreamKey) +import Prelude (Either (..), IO, pure) + + +class + ( DurableJob job + , EventSourced (Origin job) + , Dispatches (Origin job) job + ) => + OriginJob job + where + type Origin job + originKey :: job -> StreamKey (Origin job) + + +data OriginDeliveryError failure + = OriginUnavailable failure + | OriginRefused failure + + +newtype OriginPort job failure + = OriginPort + (job -> DispatchOutcome job -> IO (Either (OriginDeliveryError failure) ())) + + +data DeliveryPolicy failure + = DeliveryPolicy + (JobInstant -> failure -> JobInstant) + (JobInstant -> failure -> JobInstant) + (JobId -> failure -> IO ()) + + +data DispatchWorker job failure + = DispatchWorker + (JobWorker job) + (OriginPort job failure) + (DeliveryPolicy failure) + + +originPort + :: (job -> DispatchOutcome job -> IO (Either (OriginDeliveryError failure) ())) + -> OriginPort job failure +originPort = OriginPort + + +deliveryPolicy + :: (JobInstant -> failure -> JobInstant) + -> (JobInstant -> failure -> JobInstant) + -> (JobId -> failure -> IO ()) + -> DeliveryPolicy failure +deliveryPolicy = DeliveryPolicy + + +dispatchWorker + :: JobWorker job + -> OriginPort job failure + -> DeliveryPolicy failure + -> DispatchWorker job failure +dispatchWorker = DispatchWorker + + +storeOriginPort + :: OriginJob job + => Store (Origin job) + -> OriginPort job (StoreError (Origin job)) +storeOriginPort store = OriginPort \job outcome -> do + delivered <- + executeCommand + store + (originKey job) + (injectDispatchOutcome outcome) + + pure case delivered of + Left failure@(StoreCommandRejected _) -> Left (OriginRefused failure) + Left failure@(StoreDecisionRejected _) -> Left (OriginRefused failure) + Left failure -> Left (OriginUnavailable failure) + Right _ -> Right () + + +runDispatchJobOnce + :: forall job failure + . OriginJob job + => DispatchWorker job failure + -> JobId + -> JobInstant + -> IO (Either JobRunError (JobRunResult (JobOutput job) (JobError job))) +runDispatchJobOnce (DispatchWorker worker port policy) identifier now = + runJobOnceWith (dispatchSettlement port policy now) worker identifier now + + +dispatchSettlement + :: OriginJob job + => OriginPort job failure + -> DeliveryPolicy failure + -> JobInstant + -> SettlementStrategy job +dispatchSettlement port policy now = + SettlementStrategy + { beforeSuccess = \identifier job attempt output -> + deliverVerdict + policy + now + port + job + (Dispatch.confirmedOutcome identifier output (attemptCount attempt)) + , beforeRejection = \identifier job attempt failure -> + deliverVerdict + policy + now + port + job + ( Dispatch.failedOutcome + identifier + (Rejected failure) + (attemptCount attempt) + ) + , beforeExhaustion = \identifier job attempt failure -> do + delivered <- + deliver + port + job + ( Dispatch.failedOutcome + identifier + ( DeadLettered + RetriesExhausted + (renderJobError job failure) + ) + (attemptCount attempt) + ) + + case delivered of + Left (OriginUnavailable deliveryFailure) -> + report policy identifier deliveryFailure + Left (OriginRefused deliveryFailure) -> + report policy identifier deliveryFailure + Right () -> pure () + } + + +deliverVerdict + :: DeliveryPolicy failure + -> JobInstant + -> OriginPort job failure + -> job + -> DispatchOutcome job + -> IO VerdictDelivery +deliverVerdict policy now port job outcome = do + delivered <- deliver port job outcome + + pure case delivered of + Left (OriginUnavailable failure) -> + VerdictDeferred (retryUnavailable policy now failure) + Left (OriginRefused failure) -> + VerdictDeferred (retryRefused policy now failure) + Right () -> VerdictAccepted + + +deliver + :: OriginPort job failure + -> job + -> DispatchOutcome job + -> IO (Either (OriginDeliveryError failure) ()) +deliver (OriginPort send) = send + + +retryUnavailable + :: DeliveryPolicy failure + -> JobInstant + -> failure + -> JobInstant +retryUnavailable (DeliveryPolicy schedule _ _) = schedule + + +retryRefused + :: DeliveryPolicy failure + -> JobInstant + -> failure + -> JobInstant +retryRefused (DeliveryPolicy _ schedule _) = schedule + + +report :: DeliveryPolicy failure -> JobId -> failure -> IO () +report (DeliveryPolicy _ _ send) = send + + +attemptCount :: JobAttempt -> Word32 +attemptCount (JobAttempt attempt) = attempt diff --git a/bindings/haskell/src/EventSorcery/Job/Worker.hs b/bindings/haskell/src/EventSorcery/Job/Worker.hs index dfeaaae..fdec7b6 100644 --- a/bindings/haskell/src/EventSorcery/Job/Worker.hs +++ b/bindings/haskell/src/EventSorcery/Job/Worker.hs @@ -8,104 +8,21 @@ module EventSorcery.Job.Worker ( runJobOnce, ) where -import Data.Maybe (Maybe (..)) -import Data.Unrestricted.Linear (Ur (Ur)) -import Data.Word (Word32) -import EventSorcery.Engine (EngineError, Store) -import EventSorcery.Job ( - ClaimBudget, - ClaimedJob, - DeadReason (..), - Job (JobError, JobOutput, decodeJob), - JobClaimDetails (..), - JobClaimResult (..), - JobDecodeError (..), - JobId, - JobInstant, - JobSettlement (..), - JobSettlementToken, - LeaseDuration, - WorkerId, - acknowledgeJob, - claimJob, - deadLetterJob, - deferJob, - retryJob, - settlementToken, - ) -import EventSorcery.Job.Execution ( - DurableJob (JobInput, renderJobError), - JobAttempt (JobAttempt), - JobContext (JobContext), - JobFailure (..), - JobOutcome (..), - executeDurableJob, +import EventSorcery.Job (Job (JobError, JobOutput), JobId, JobInstant) +import EventSorcery.Job.Execution (DurableJob) +import EventSorcery.Job.Worker.Definition ( + AttemptLimit, + JobRunError (..), + JobRunResult (..), + JobWorker, + jobWorker, + mkAttemptLimit, ) -import Prelude ( - Bool, - Either (..), - Eq, - IO, - Show, - maxBound, - otherwise, - pure, - (+), - (==), - (>=), +import EventSorcery.Job.Worker.Internal ( + genericSettlement, + runJobOnceWith, ) - - -newtype AttemptLimit = AttemptLimit Word32 - deriving stock (Eq, Show) - - -data JobWorker job = JobWorker - { store :: Store - , workerId :: WorkerId - , leaseDuration :: LeaseDuration - , claimBudget :: ClaimBudget - , attemptLimit :: AttemptLimit - , retryAt :: JobAttempt -> JobInstant - , input :: JobInput job - } - - -data JobRunResult output failure - = JobSucceeded output - | JobDeferredUntil JobInstant - | JobRetryScheduled JobAttempt JobInstant failure - | JobRejected failure - | JobRetriesExhausted JobAttempt failure - | JobRunAbandoned - | JobRunContended - | JobRunSkipped - | JobRunFenced - deriving stock (Eq, Show) - - -data JobRunError - = JobRunEngineFailed EngineError - | JobRunDecodeFailed JobId JobDecodeError - deriving stock (Eq, Show) - - -mkAttemptLimit :: Word32 -> Maybe AttemptLimit -mkAttemptLimit value - | value == 0 = Nothing - | otherwise = Just (AttemptLimit value) - - -jobWorker - :: Store - -> WorkerId - -> LeaseDuration - -> ClaimBudget - -> AttemptLimit - -> (JobAttempt -> JobInstant) - -> JobInput job - -> JobWorker job -jobWorker = JobWorker +import Prelude (Either, IO) runJobOnce @@ -115,126 +32,4 @@ runJobOnce -> JobId -> JobInstant -> IO (Either JobRunError (JobRunResult (JobOutput job) (JobError job))) -runJobOnce runtime identifier now = do - claimed <- - claimJob - runtime.store - identifier - runtime.workerId - now - runtime.leaseDuration - runtime.claimBudget - releaseClaim - - case claimed of - Left failure -> pure (Left (JobRunEngineFailed failure)) - Right (JobClaimed (details, token)) -> - case decodeJob @job details.payload of - Left failure -> rejectUndecodable runtime identifier token failure - Right job -> do - let attempt = JobAttempt details.attempt - context = JobContext identifier attempt - executed <- executeDurableJob context details.route runtime.input job - persistExecution runtime job token attempt executed - Right JobAbandoned -> pure (Right JobRunAbandoned) - Right JobContended -> pure (Right JobRunContended) - Right JobSkipped -> pure (Right JobRunSkipped) - - -rejectUndecodable - :: JobWorker job - -> JobId - -> JobSettlementToken - -> JobDecodeError - -> IO (Either JobRunError (JobRunResult output failure)) -rejectUndecodable runtime identifier token failure@(JobDecodeError cause) = do - settled <- deadLetterJob runtime.store token Undecodable cause - - pure case settled of - Left engineError -> Left (JobRunEngineFailed engineError) - Right SettlementApplied -> Left (JobRunDecodeFailed identifier failure) - Right SettlementFenced -> Right JobRunFenced - - -persistExecution - :: DurableJob job - => JobWorker job - -> job - -> JobSettlementToken - -> JobAttempt - -> Either (JobFailure (JobError job)) (JobOutcome (JobOutput job)) - -> IO (Either JobRunError (JobRunResult (JobOutput job) (JobError job))) -persistExecution runtime job token attempt executed = case executed of - Right (JobDone output) -> do - settled <- acknowledgeJob runtime.store token - pure (settlementResult (JobSucceeded output) settled) - Right (JobDeferred runAt) -> do - settled <- deferJob runtime.store token runAt - pure (settlementResult (JobDeferredUntil runAt) settled) - Left (TerminalFailure failure) -> do - settled <- - deadLetterJob - runtime.store - token - Rejected - (renderJobError job failure) - pure (settlementResult (JobRejected failure) settled) - Left (TransientFailure failure) -> - persistTransientFailure runtime job token attempt failure - - -persistTransientFailure - :: DurableJob job - => JobWorker job - -> job - -> JobSettlementToken - -> JobAttempt - -> JobError job - -> IO (Either JobRunError (JobRunResult (JobOutput job) (JobError job))) -persistTransientFailure runtime job token attempt failure = do - let next = nextAttempt attempt - failureText = renderJobError job failure - - if retryBudgetExhausted runtime.attemptLimit next - then do - settled <- - deadLetterJob - runtime.store - token - RetriesExhausted - failureText - pure (settlementResult (JobRetriesExhausted next failure) settled) - else do - let runAt = runtime.retryAt next - settled <- retryJob runtime.store token runAt failureText - pure (settlementResult (JobRetryScheduled next runAt failure) settled) - - -settlementResult - :: JobRunResult output failure - -> Either EngineError JobSettlement - -> Either JobRunError (JobRunResult output failure) -settlementResult result settled = case settled of - Left engineError -> Left (JobRunEngineFailed engineError) - Right SettlementApplied -> Right result - Right SettlementFenced -> Right JobRunFenced - - -nextAttempt :: JobAttempt -> JobAttempt -nextAttempt attempt@(JobAttempt current) - | current == maxBound = attempt - | otherwise = JobAttempt (current + 1) - - -retryBudgetExhausted :: AttemptLimit -> JobAttempt -> Bool -retryBudgetExhausted (AttemptLimit limit) (JobAttempt attempt) = - attempt >= limit - - -releaseClaim - :: JobClaimDetails - -> ClaimedJob - %1 -> Ur (JobClaimDetails, JobSettlementToken) -releaseClaim details won = - case settlementToken won of - Ur token -> Ur (details, token) +runJobOnce = runJobOnceWith genericSettlement diff --git a/bindings/haskell/src/EventSorcery/Job/Worker/Definition.hs b/bindings/haskell/src/EventSorcery/Job/Worker/Definition.hs new file mode 100644 index 0000000..95fb542 --- /dev/null +++ b/bindings/haskell/src/EventSorcery/Job/Worker/Definition.hs @@ -0,0 +1,77 @@ +module EventSorcery.Job.Worker.Definition ( + AttemptLimit (AttemptLimit), + JobRunError (..), + JobRunResult (..), + JobWorker (..), + jobWorker, + mkAttemptLimit, +) where + +import Data.Maybe (Maybe (..)) +import Data.Word (Word32) +import EventSorcery.Engine (EngineError, Store) +import EventSorcery.Job ( + ClaimBudget, + JobDecodeError, + JobId, + JobInstant, + LeaseDuration, + WorkerId, + ) +import EventSorcery.Job.Execution ( + DurableJob (JobInput), + JobAttempt, + ) +import Prelude (Eq, Show, otherwise, (==)) + + +newtype AttemptLimit = AttemptLimit Word32 + deriving stock (Eq, Show) + + +data JobWorker job = JobWorker + { store :: Store + , workerId :: WorkerId + , leaseDuration :: LeaseDuration + , claimBudget :: ClaimBudget + , attemptLimit :: AttemptLimit + , retryAt :: JobAttempt -> JobInstant + , input :: JobInput job + } + + +data JobRunResult output failure + = JobSucceeded output + | JobDeferredUntil JobInstant + | JobRetryScheduled JobAttempt JobInstant failure + | JobRejected failure + | JobRetriesExhausted JobAttempt failure + | JobRunAbandoned + | JobRunContended + | JobRunSkipped + | JobRunFenced + deriving stock (Eq, Show) + + +data JobRunError + = JobRunEngineFailed EngineError + | JobRunDecodeFailed JobId JobDecodeError + deriving stock (Eq, Show) + + +mkAttemptLimit :: Word32 -> Maybe AttemptLimit +mkAttemptLimit value + | value == 0 = Nothing + | otherwise = Just (AttemptLimit value) + + +jobWorker + :: Store + -> WorkerId + -> LeaseDuration + -> ClaimBudget + -> AttemptLimit + -> (JobAttempt -> JobInstant) + -> JobInput job + -> JobWorker job +jobWorker = JobWorker diff --git a/bindings/haskell/src/EventSorcery/Job/Worker/Internal.hs b/bindings/haskell/src/EventSorcery/Job/Worker/Internal.hs new file mode 100644 index 0000000..aad9f8b --- /dev/null +++ b/bindings/haskell/src/EventSorcery/Job/Worker/Internal.hs @@ -0,0 +1,275 @@ +module EventSorcery.Job.Worker.Internal ( + SettlementStrategy (..), + VerdictDelivery (..), + genericSettlement, + runJobOnceWith, +) where + +import Data.Unrestricted.Linear (Ur (Ur)) +import EventSorcery.Engine (EngineError) +import EventSorcery.Job ( + ClaimedJob, + DeadReason (..), + Job (JobError, JobOutput, decodeJob), + JobClaimDetails (..), + JobClaimResult (..), + JobDecodeError (..), + JobId, + JobInstant, + JobSettlement (..), + JobSettlementToken, + acknowledgeJob, + claimJob, + deadLetterJob, + deferJob, + retryJob, + settlementToken, + ) +import EventSorcery.Job.Execution ( + DurableJob (renderJobError), + JobAttempt (JobAttempt), + JobContext (JobContext), + JobFailure (..), + JobOutcome (..), + executeDurableJob, + ) +import EventSorcery.Job.Worker.Definition ( + AttemptLimit (AttemptLimit), + JobRunError (..), + JobRunResult (..), + JobWorker (..), + ) +import Prelude ( + Bool, + Either (..), + IO, + maxBound, + otherwise, + pure, + (+), + (==), + (>=), + ) + + +data VerdictDelivery + = VerdictAccepted + | VerdictDeferred JobInstant + + +data SettlementStrategy job = SettlementStrategy + { beforeSuccess + :: JobId + -> job + -> JobAttempt + -> JobOutput job + -> IO VerdictDelivery + , beforeRejection + :: JobId + -> job + -> JobAttempt + -> JobError job + -> IO VerdictDelivery + , beforeExhaustion + :: JobId + -> job + -> JobAttempt + -> JobError job + -> IO () + } + + +genericSettlement :: SettlementStrategy job +genericSettlement = + SettlementStrategy + { beforeSuccess = \_ _ _ _ -> pure VerdictAccepted + , beforeRejection = \_ _ _ _ -> pure VerdictAccepted + , beforeExhaustion = \_ _ _ _ -> pure () + } + + +runJobOnceWith + :: forall job + . DurableJob job + => SettlementStrategy job + -> JobWorker job + -> JobId + -> JobInstant + -> IO (Either JobRunError (JobRunResult (JobOutput job) (JobError job))) +runJobOnceWith settlement runtime identifier now = do + claimed <- + claimJob + runtime.store + identifier + runtime.workerId + now + runtime.leaseDuration + runtime.claimBudget + releaseClaim + + case claimed of + Left failure -> pure (Left (JobRunEngineFailed failure)) + Right (JobClaimed (details, token)) -> + case decodeJob @job details.payload of + Left failure -> rejectUndecodable runtime identifier token failure + Right job -> do + let attempt = JobAttempt details.attempt + context = JobContext identifier attempt + executed <- executeDurableJob context details.route runtime.input job + persistExecution + settlement + runtime + identifier + job + token + attempt + executed + Right JobAbandoned -> pure (Right JobRunAbandoned) + Right JobContended -> pure (Right JobRunContended) + Right JobSkipped -> pure (Right JobRunSkipped) + + +rejectUndecodable + :: JobWorker job + -> JobId + -> JobSettlementToken + -> JobDecodeError + -> IO (Either JobRunError (JobRunResult output failure)) +rejectUndecodable runtime identifier token failure@(JobDecodeError cause) = do + settled <- deadLetterJob runtime.store token Undecodable cause + + pure case settled of + Left engineError -> Left (JobRunEngineFailed engineError) + Right SettlementApplied -> Left (JobRunDecodeFailed identifier failure) + Right SettlementFenced -> Right JobRunFenced + + +persistExecution + :: DurableJob job + => SettlementStrategy job + -> JobWorker job + -> JobId + -> job + -> JobSettlementToken + -> JobAttempt + -> Either (JobFailure (JobError job)) (JobOutcome (JobOutput job)) + -> IO (Either JobRunError (JobRunResult (JobOutput job) (JobError job))) +persistExecution settlement runtime identifier job token attempt executed = + case executed of + Right (JobDone output) -> do + delivery <- + settlement.beforeSuccess identifier job (nextAttempt attempt) output + persistSuccess runtime token output delivery + Right (JobDeferred runAt) -> do + settled <- deferJob runtime.store token runAt + pure (settlementResult (JobDeferredUntil runAt) settled) + Left (TerminalFailure failure) -> do + delivery <- + settlement.beforeRejection identifier job (nextAttempt attempt) failure + persistRejection runtime job token failure delivery + Left (TransientFailure failure) -> + persistTransientFailure + settlement + runtime + identifier + job + token + attempt + failure + + +persistSuccess + :: JobWorker job + -> JobSettlementToken + -> output + -> VerdictDelivery + -> IO (Either JobRunError (JobRunResult output failure)) +persistSuccess runtime token output delivery = case delivery of + VerdictAccepted -> do + settled <- acknowledgeJob runtime.store token + pure (settlementResult (JobSucceeded output) settled) + VerdictDeferred runAt -> do + settled <- deferJob runtime.store token runAt + pure (settlementResult (JobDeferredUntil runAt) settled) + + +persistRejection + :: DurableJob job + => JobWorker job + -> job + -> JobSettlementToken + -> JobError job + -> VerdictDelivery + -> IO (Either JobRunError (JobRunResult output (JobError job))) +persistRejection runtime job token failure delivery = case delivery of + VerdictAccepted -> do + settled <- + deadLetterJob + runtime.store + token + Rejected + (renderJobError job failure) + pure (settlementResult (JobRejected failure) settled) + VerdictDeferred runAt -> do + settled <- deferJob runtime.store token runAt + pure (settlementResult (JobDeferredUntil runAt) settled) + + +persistTransientFailure + :: DurableJob job + => SettlementStrategy job + -> JobWorker job + -> JobId + -> job + -> JobSettlementToken + -> JobAttempt + -> JobError job + -> IO (Either JobRunError (JobRunResult (JobOutput job) (JobError job))) +persistTransientFailure settlement runtime identifier job token attempt failure = do + let next = nextAttempt attempt + failureText = renderJobError job failure + + if retryBudgetExhausted runtime.attemptLimit next + then do + settlement.beforeExhaustion identifier job next failure + settled <- + deadLetterJob + runtime.store + token + RetriesExhausted + failureText + pure (settlementResult (JobRetriesExhausted next failure) settled) + else do + let runAt = runtime.retryAt next + settled <- retryJob runtime.store token runAt failureText + pure (settlementResult (JobRetryScheduled next runAt failure) settled) + + +settlementResult + :: JobRunResult output failure + -> Either EngineError JobSettlement + -> Either JobRunError (JobRunResult output failure) +settlementResult result settled = case settled of + Left engineError -> Left (JobRunEngineFailed engineError) + Right SettlementApplied -> Right result + Right SettlementFenced -> Right JobRunFenced + + +nextAttempt :: JobAttempt -> JobAttempt +nextAttempt attempt@(JobAttempt current) + | current == maxBound = attempt + | otherwise = JobAttempt (current + 1) + + +retryBudgetExhausted :: AttemptLimit -> JobAttempt -> Bool +retryBudgetExhausted (AttemptLimit limit) (JobAttempt attempt) = + attempt >= limit + + +releaseClaim + :: JobClaimDetails + -> ClaimedJob + %1 -> Ur (JobClaimDetails, JobSettlementToken) +releaseClaim details won = + case settlementToken won of + Ur token -> Ur (details, token) diff --git a/bindings/haskell/src/EventSorcery/Store.hs b/bindings/haskell/src/EventSorcery/Store.hs index b1ce238..633b0af 100644 --- a/bindings/haskell/src/EventSorcery/Store.hs +++ b/bindings/haskell/src/EventSorcery/Store.hs @@ -66,6 +66,7 @@ data StoreError entity | StoreCommandRejected (Aggregate.CommandError entity) | StoreDecisionRejected (Aggregate.ApplyError entity) | StoreSnapshotDecodeFailed Word64 Aggregate.DecodeCause + | StoreUneventfulInitialization deriving stock instance @@ -90,6 +91,7 @@ data PreparedCommand entity where data CommitPlan entity = PreparedEvents entity Word64 (NonEmpty ProposedEvent) + | PreparedUnchanged entity | PreparedDispatch entity Word64 (NonEmpty ProposedEvent) JobSeed @@ -230,6 +232,10 @@ prepareEffect _ expected current (Events events) = ( PreparedCommand (Ur (PreparedEvents next expected (encodeEvents events))) ) +prepareEffect _ _ (Just entity) Unchanged = + pure (Right (PreparedCommand (Ur (PreparedUnchanged entity)))) +prepareEffect _ _ Nothing Unchanged = + pure (Left StoreUneventfulInitialization) prepareEffect (Store _ nextJobId) expected current (Dispatch request) = do identifier <- nextJobId let job = dispatchJob request @@ -271,6 +277,7 @@ commitPrepared engine identity (PreparedCommand (Ur plan)) = committed <- commit engine identity expected events pure (next <$ first StoreEngineFailed committed) + PreparedUnchanged entity -> pure (Right entity) PreparedDispatch next expected events seed -> do committed <- commitWithJob engine identity expected events seed diff --git a/bindings/haskell/test/DispatchWorkerSpec.hs b/bindings/haskell/test/DispatchWorkerSpec.hs new file mode 100644 index 0000000..ff98155 --- /dev/null +++ b/bindings/haskell/test/DispatchWorkerSpec.hs @@ -0,0 +1,330 @@ +module Main (main) where + +import Data.ByteString qualified as ByteString +import Data.IORef (IORef, modifyIORef', newIORef, readIORef) +import Data.List.NonEmpty (NonEmpty ((:|))) +import Data.Text (Text) +import EventSorcery.Aggregate ( + DecodeCause (DecodeCause), + Dispatches (..), + Effect (..), + EventSourced (..), + EventVersion (EventVersion), + SchemaVersion (SchemaVersion), + dispatchIntentJob, + dispatchJobId, + ) +import EventSorcery.Dispatch ( + DispatchEvent (..), + DispatchOutcome, + DispatchRefused, + DispatchReplay, + DispatchedJob (..), + evolveDispatch, + guardDispatch, + settleDispatch, + settledOutput, + ) +import EventSorcery.Dispatch.TestSupport (confirmedOutcome) +import EventSorcery.Dispatch.Worker ( + DeliveryPolicy, + OriginJob (..), + deliveryPolicy, + dispatchWorker, + runDispatchJobOnce, + storeOriginPort, + ) +import EventSorcery.Engine ( + OpenOptions (OpenOptions), + Store, + closeStore, + openStore, + ) +import EventSorcery.Job ( + ClaimBudget (ClaimBudget), + Job (..), + JobDecodeError (JobDecodeError), + JobId, + JobInstant (JobInstant), + LeaseDuration (LeaseDuration), + WorkerId (WorkerId), + mkJobId, + ) +import EventSorcery.Job.Execution ( + DurableJob (..), + JobAttempt, + JobOutcome (JobDone), + Reconciliation (Reconciled), + ) +import EventSorcery.Job.Worker ( + AttemptLimit, + JobRunResult (..), + JobWorker, + jobWorker, + mkAttemptLimit, + ) +import EventSorcery.Store ( + executeCommand, + loadEntity, + mkStore, + ) +import EventSorcery.Stream (StreamKey, streamKey) +import Prelude ( + Bool (False, True), + Either (Left, Right), + Eq, + IO, + Maybe (Just, Nothing), + Show, + String, + error, + fmap, + otherwise, + pure, + (&&), + (<>), + (==), + ) + + +newtype Account = Account (DispatchedJob ChargeCard) + + +data AccountId = AccountId + + +data AccountCommand + = OpenAccount + | StartCharge + | SettleCharge (DispatchOutcome ChargeCard) + + +data AccountEvent + = AccountOpened + | ChargeChanged (DispatchEvent ChargeCard) + + +data AccountError + = AccountAlreadyOpen + | AccountNotOpen + | ChargeRefused DispatchRefused + deriving stock (Eq, Show) + + +data AccountReplayError + = AccountReplayFailed DispatchReplay + | AccountOpenedTwice + deriving stock (Eq, Show) + + +data ChargeCard = ChargeCard + deriving stock (Eq, Show) + + +instance Job ChargeCard where + type JobType ChargeCard = "charge-card" + type JobOutput ChargeCard = Text + type JobError ChargeCard = Text + + + encodeJob _ = ByteString.empty + decodeJob bytes + | bytes == ByteString.empty = Right ChargeCard + | otherwise = Left (JobDecodeError "invalid charge job") + + +instance DurableJob ChargeCard where + type JobInput ChargeCard = IORef [Text] + + + renderJobError _ failure = failure + + + submit _ calls _ = do + modifyIORef' calls (<> ["submit"]) + pure (Right (JobDone "charged")) + + + reconcile _ calls _ = do + modifyIORef' calls (<> ["reconcile"]) + pure (Right (Reconciled "charged")) + + +instance OriginJob ChargeCard where + type Origin ChargeCard = Account + originKey _ = accountKey + + +instance Dispatches Account ChargeCard where + injectDispatchIntent intent = + ChargeChanged + (Dispatched (dispatchJobId intent) (dispatchIntentJob intent)) + injectDispatchOutcome = SettleCharge + + +instance EventSourced Account where + type EntityId Account = AccountId + type Command Account = AccountCommand + type Event Account = AccountEvent + type CommandError Account = AccountError + type ApplyError Account = AccountReplayError + type Jobs Account = '[ChargeCard] + + + aggregateType _ = "account" + encodeEntityId AccountId = "account-1" + eventType AccountOpened = "account-opened" + eventType (ChargeChanged (Dispatched _ _)) = "charge-dispatched" + eventType (ChargeChanged (ConfirmedEvent _)) = "charge-confirmed" + eventType (ChargeChanged (FailedEvent _)) = "charge-failed" + eventVersion _ = EventVersion "1" + schemaVersion _ = SchemaVersion 1 + encodeEvent AccountOpened = ByteString.singleton 0 + encodeEvent (ChargeChanged (Dispatched _ _)) = ByteString.singleton 1 + encodeEvent (ChargeChanged (ConfirmedEvent _)) = ByteString.singleton 2 + encodeEvent (ChargeChanged (FailedEvent _)) = ByteString.singleton 3 + decodeEvent bytes = case ByteString.unpack bytes of + [0] -> Right AccountOpened + [1] -> Right (ChargeChanged (Dispatched jobIdentifier ChargeCard)) + [2] -> Right confirmedEvent + _ -> Left (DecodeCause "invalid account event") + encodeSnapshot _ = ByteString.empty + decodeSnapshot _ = Right (Account Idle) + originate AccountOpened = Right (Account Idle) + originate _ = Left AccountOpenedTwice + evolve _ AccountOpened = Left AccountOpenedTwice + evolve (Account state) (ChargeChanged event) = + case evolveDispatch state event of + Left failure -> Left (AccountReplayFailed failure) + Right next -> Right (Account next) + initialize OpenAccount = Right (Events (AccountOpened :| [])) + initialize _ = Left AccountNotOpen + transition _ OpenAccount = Left AccountAlreadyOpen + transition (Account state) StartCharge = + case guardDispatch state ChargeCard of + Left failure -> Left (ChargeRefused failure) + Right request -> Right (Dispatch request) + transition (Account state) (SettleCharge outcome) = + case settleDispatch state outcome of + Left failure -> Left (ChargeRefused failure) + Right [] -> Right Unchanged + Right (event : remaining) -> + Right (Events (ChargeChanged event :| fmap ChargeChanged remaining)) + + +main :: IO () +main = do + opened <- openStore (OpenOptions "sqlite::memory:" 5000 1 1) + + case opened of + Left _ -> error "failed to open the shared engine" + Right engine -> do + let originStore = mkStore engine (pure jobIdentifier) + calls <- newIORef [] + + openedAccount <- executeCommand originStore accountKey OpenAccount + dispatched <- executeCommand originStore accountKey StartCharge + result <- + runDispatchJobOnce + ( dispatchWorker + (runner engine calls) + (storeOriginPort originStore) + policy + ) + jobIdentifier + now + settled <- loadEntity originStore accountKey + recorded <- readIORef calls + + expect "account did not open" (fmap isIdle openedAccount == Right True) + expect + "charge was not dispatched" + (fmap isInFlight dispatched == Right True) + expect + "verdict was not delivered before the job acknowledged" + ( result == Right (JobSucceeded "charged") + && fmap (fmap settledOutputOf) settled == Right (Just "charged") + && recorded == ["submit"] + ) + + closed <- closeStore engine + expect "failed to close the shared engine" (closed == Right ()) + + +runner + :: Store + -> IORef [Text] + -> JobWorker ChargeCard +runner engine = + jobWorker + engine + (WorkerId "dispatch-worker") + (LeaseDuration 30_000) + (ClaimBudget 50) + attemptLimit + retrySchedule + + +policy :: DeliveryPolicy failure +policy = deliveryPolicy deliveryRetryAt deliveryRetryAt (\_ _ -> pure ()) + + +deliveryRetryAt :: JobInstant -> failure -> JobInstant +deliveryRetryAt _ _ = later + + +retrySchedule :: JobAttempt -> JobInstant +retrySchedule _ = later + + +attemptLimit :: AttemptLimit +attemptLimit = case mkAttemptLimit 3 of + Just limit -> limit + Nothing -> error "valid attempt limit was rejected" + + +confirmedEvent :: AccountEvent +confirmedEvent = + case settleDispatch + (InFlight jobIdentifier) + (confirmedOutcome jobIdentifier "charged" 1) of + Right [event] -> ChargeChanged event + _ -> error "test verdict did not settle" + + +isIdle :: Account -> Bool +isIdle (Account Idle) = True +isIdle _ = False + + +isInFlight :: Account -> Bool +isInFlight (Account (InFlight _)) = True +isInFlight _ = False + + +settledOutputOf :: Account -> Text +settledOutputOf (Account (Confirmed settled)) = settledOutput settled +settledOutputOf _ = error "account charge was not confirmed" + + +accountKey :: StreamKey Account +accountKey = streamKey @Account AccountId + + +jobIdentifier :: JobId +jobIdentifier = case mkJobId "01ARZ3NDEKTSV4RRFFQ69G5FCA" of + Just identifier -> identifier + Nothing -> error "valid test job identifier was rejected" + + +now :: JobInstant +now = JobInstant 1_000 + + +later :: JobInstant +later = JobInstant 90_000 + + +expect :: String -> Bool -> IO () +expect _ True = pure () +expect message False = error message diff --git a/bindings/haskell/test/DomainSpec.hs b/bindings/haskell/test/DomainSpec.hs index 10d2b13..6e941da 100644 --- a/bindings/haskell/test/DomainSpec.hs +++ b/bindings/haskell/test/DomainSpec.hs @@ -4,7 +4,7 @@ import Data.ByteString qualified as ByteString import Data.List.NonEmpty (NonEmpty ((:|))) import EventSorcery.Aggregate ( DecodeCause (DecodeCause), - Dispatches (injectDispatchIntent), + Dispatches (..), Effect (..), EventSourced (..), EventVersion (EventVersion), @@ -12,7 +12,7 @@ import EventSorcery.Aggregate ( dispatchIntent, dispatchJobId, ) -import EventSorcery.Dispatch (kickoff) +import EventSorcery.Dispatch (DispatchOutcome, kickoff) import EventSorcery.Job ( Job (..), JobId, @@ -162,6 +162,7 @@ data AccountId = AccountId data AccountCommand = OpenAccount | SendWelcome + | SettleWelcome (DispatchOutcome SendWelcomeEmail) data AccountEvent @@ -187,6 +188,7 @@ instance Job SendWelcomeEmail where instance Dispatches Account SendWelcomeEmail where injectDispatchIntent intent = WelcomeRequested (dispatchJobId intent) + injectDispatchOutcome = SettleWelcome instance EventSourced Account where @@ -216,9 +218,10 @@ instance EventSourced Account where evolve account AccountOpened = Right account evolve (Account count) (WelcomeRequested _) = Right (Account count) initialize OpenAccount = Right (Events (AccountOpened :| [])) - initialize SendWelcome = Left AccountError + initialize _ = Left AccountError transition _ OpenAccount = Left AccountError transition _ SendWelcome = Right (Dispatch (kickoff SendWelcomeEmail)) + transition _ (SettleWelcome _) = Left AccountError testJobId :: JobId diff --git a/bindings/haskell/test/StoreSpec.hs b/bindings/haskell/test/StoreSpec.hs index 2dbe0db..e720f1d 100644 --- a/bindings/haskell/test/StoreSpec.hs +++ b/bindings/haskell/test/StoreSpec.hs @@ -5,14 +5,14 @@ import Data.List.NonEmpty (NonEmpty ((:|))) import Data.Text.Encoding qualified as Text import EventSorcery.Aggregate ( DecodeCause (DecodeCause), - Dispatches (injectDispatchIntent), + Dispatches (..), Effect (..), EventSourced (..), EventVersion (EventVersion), SchemaVersion (SchemaVersion), dispatchJobId, ) -import EventSorcery.Dispatch (kickoff) +import EventSorcery.Dispatch (DispatchOutcome, kickoff) import EventSorcery.Engine ( OpenOptions (OpenOptions), closeStore, @@ -150,6 +150,7 @@ data AccountCommand | Deposit Int | EmitInvalidEvent | SendWelcome + | SettleWelcome (DispatchOutcome SendWelcomeEmail) data AccountEvent @@ -184,6 +185,7 @@ instance Job SendWelcomeEmail where instance Dispatches Account SendWelcomeEmail where injectDispatchIntent intent = WelcomeRequested (dispatchJobId intent) + injectDispatchOutcome = SettleWelcome instance EventSourced Account where @@ -229,6 +231,7 @@ instance EventSourced Account where transition _ (Deposit amount) = Right (Events (FundsDeposited amount :| [])) transition _ EmitInvalidEvent = Right (Events (AccountOpened 99 :| [])) transition _ SendWelcome = Right (Dispatch (kickoff SendWelcomeEmail)) + transition _ (SettleWelcome _) = Left AlreadyOpen decodeAmount