feat: renew Haskell job leases

bindings/haskell/event-sorcery.cabal

@@ -74,6 +74,7 @@ library
 
   extra-libraries: event_sorcery_ffi
   build-depends:
+    , async                      >=2.2    && <2.3
     , base                       >=4.18   && <5
     , bytestring                 >=0.11   && <0.13
     , cborg                      >=0.2.10 && <0.3
@@ -176,6 +177,7 @@ test-suite job-worker-spec
   hs-source-dirs: test
   main-is:        JobWorkerSpec.hs
   build-depends:
+    , async          >=2.2  && <2.3
     , base           >=4.18 && <5
     , bytestring     >=0.11 && <0.13
     , event-sorcery

bindings/haskell/src/EventSorcery/Job/Worker.hs

@@ -3,8 +3,11 @@ module EventSorcery.Job.Worker (
   JobRunError (..),
   JobRunResult (..),
   JobWorker,
+  RenewalSchedule,
   jobWorker,
   mkAttemptLimit,
+  renewalSchedule,
+  renewingJobWorker,
   runJobOnce,
 ) where
 
@@ -15,8 +18,11 @@ import EventSorcery.Job.Worker.Definition (
   JobRunError (..),
   JobRunResult (..),
   JobWorker,
+  RenewalSchedule,
   jobWorker,
   mkAttemptLimit,
+  renewalSchedule,
+  renewingJobWorker,
  )
 import EventSorcery.Job.Worker.Internal (
   genericSettlement,

bindings/haskell/src/EventSorcery/Job/Worker/Definition.hs

@@ -3,8 +3,11 @@ module EventSorcery.Job.Worker.Definition (
   JobRunError (..),
   JobRunResult (..),
   JobWorker (..),
+  RenewalSchedule (..),
   jobWorker,
   mkAttemptLimit,
+  renewalSchedule,
+  renewingJobWorker,
 ) where
 
 import Data.Maybe (Maybe (..))
@@ -22,7 +25,7 @@ import EventSorcery.Job.Execution (
   DurableJob (JobInput),
   JobAttempt,
  )
-import Prelude (Eq, Show, otherwise, (==))
+import Prelude (Eq, IO, Show, otherwise, (==))
 
 
 newtype AttemptLimit = AttemptLimit Word32
@@ -37,6 +40,13 @@ data JobWorker job = JobWorker
   , attemptLimit :: AttemptLimit
   , retryAt :: JobAttempt -> JobInstant
   , input :: JobInput job
+  , leaseRenewal :: Maybe RenewalSchedule
+  }
+
+
+data RenewalSchedule = RenewalSchedule
+  { waitBeforeRenewal :: IO ()
+  , renewalDeadline :: IO JobInstant
   }
 
 
@@ -74,4 +84,22 @@ jobWorker
   -> (JobAttempt -> JobInstant)
   -> JobInput job
   -> JobWorker job
-jobWorker = JobWorker
+jobWorker store workerId leaseDuration claimBudget attemptLimit retryAt input =
+  JobWorker
+    { store
+    , workerId
+    , leaseDuration
+    , claimBudget
+    , attemptLimit
+    , retryAt
+    , input
+    , leaseRenewal = Nothing
+    }
+
+
+renewalSchedule :: IO () -> IO JobInstant -> RenewalSchedule
+renewalSchedule = RenewalSchedule
+
+
+renewingJobWorker :: JobWorker job -> RenewalSchedule -> JobWorker job
+renewingJobWorker worker schedule = worker {leaseRenewal = Just schedule}

bindings/haskell/src/EventSorcery/Job/Worker/Internal.hs

@@ -5,23 +5,34 @@ module EventSorcery.Job.Worker.Internal (
   runJobOnceWith,
 ) where
 
+import Control.Concurrent.Async (cancel, link, withAsync)
+import Control.Concurrent.MVar (
+  MVar,
+  newEmptyMVar,
+  putMVar,
+  tryReadMVar,
+ )
+import Data.Maybe (Maybe (..))
 import Data.Unrestricted.Linear (Ur (Ur))
 import EventSorcery.Engine (EngineError)
 import EventSorcery.Job (
   ClaimedJob,
   DeadReason (..),
   Job (JobError, JobOutput, decodeJob),
   JobClaimDetails (..),
+  JobClaimReference,
   JobClaimResult (..),
   JobDecodeError (..),
   JobId,
   JobInstant,
+  JobLeaseResult (..),
   JobSettlement (..),
   JobSettlementToken,
   acknowledgeJob,
   claimJob,
   deadLetterJob,
   deferJob,
+  renewJob,
   retryJob,
   settlementToken,
  )
@@ -38,6 +49,7 @@ import EventSorcery.Job.Worker.Definition (
   JobRunError (..),
   JobRunResult (..),
   JobWorker (..),
+  RenewalSchedule (..),
  )
 import Prelude (
   Bool,
@@ -47,6 +59,7 @@ import Prelude (
   otherwise,
   pure,
   (+),
+  (<$>),
   (==),
   (>=),
  )
@@ -115,15 +128,25 @@ runJobOnceWith settlement runtime identifier now = do
         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
+          executed <-
+            executeWithRenewal
+              runtime
+              details.reference
+              (executeDurableJob context details.route runtime.input job)
+
+          case executed of
+            ExecutionCompleted result ->
+              persistExecution
+                settlement
+                runtime
+                identifier
+                job
+                token
+                attempt
+                result
+            ExecutionLeaseLost -> pure (Right JobRunFenced)
+            ExecutionRenewalFailed failure ->
+              pure (Left (JobRunEngineFailed failure))
     Right JobAbandoned -> pure (Right JobRunAbandoned)
     Right JobContended -> pure (Right JobRunContended)
     Right JobSkipped -> pure (Right JobRunSkipped)
@@ -273,3 +296,57 @@ releaseClaim
 releaseClaim details won =
   case settlementToken won of
     Ur token -> Ur (details, token)
+
+
+data RenewedExecution result
+  = ExecutionCompleted result
+  | ExecutionLeaseLost
+  | ExecutionRenewalFailed EngineError
+
+
+data RenewalTerminal
+  = RenewalLost
+  | RenewalFailed EngineError
+
+
+executeWithRenewal
+  :: JobWorker job
+  -> JobClaimReference
+  -> IO result
+  -> IO (RenewedExecution result)
+executeWithRenewal runtime reference execute = case runtime.leaseRenewal of
+  Nothing -> do
+    ExecutionCompleted <$> execute
+  Just schedule -> do
+    terminal <- newEmptyMVar
+
+    withAsync
+      (renewUntilTerminal runtime schedule reference terminal)
+      \renewer -> do
+        link renewer
+        result <- execute
+        cancel renewer
+        renewal <- tryReadMVar terminal
+
+        pure case renewal of
+          Nothing -> ExecutionCompleted result
+          Just RenewalLost -> ExecutionLeaseLost
+          Just (RenewalFailed failure) -> ExecutionRenewalFailed failure
+
+
+renewUntilTerminal
+  :: JobWorker job
+  -> RenewalSchedule
+  -> JobClaimReference
+  -> MVar RenewalTerminal
+  -> IO ()
+renewUntilTerminal runtime schedule reference terminal = do
+  schedule.waitBeforeRenewal
+  newLease <- schedule.renewalDeadline
+  renewed <- renewJob runtime.store reference newLease
+
+  case renewed of
+    Left failure -> putMVar terminal (RenewalFailed failure)
+    Right LeaseLost -> putMVar terminal RenewalLost
+    Right LeaseHeld ->
+      renewUntilTerminal runtime schedule reference terminal