{-# LANGUAGE OverloadedStrings #-}

-- | The pool's side of a batch's lifecycle: the statements and hooks
-- "Arbiter.Worker.Batch" drives, unlifted to IO.
module Arbiter.Worker.Settlement
  ( PoolEffects
  , PoolMode
  , poolEffects
  , poolMode
  , batchLog
  ) where

import Arbiter.Core.Exceptions (throwJobGoneIds)
import Arbiter.Core.HighLevel (JobOperation)
import Arbiter.Core.HighLevel qualified as Arb
import Arbiter.Core.Job.Schema (SchemaName, TableName)
import Arbiter.Core.Job.Types qualified as Job
import Arbiter.Core.JobResult
import Arbiter.Core.MonadArbiter (MonadArbiter (..))
import Arbiter.Core.Operations qualified as Ops
import Arbiter.Core.Trace
  ( ConsumeShape (..)
  , ConsumeSpan
  , capturingContextIO
  , markSpanError
  , recordJobCancelled
  , recordJobFailure
  , resolveTracer
  , toConsumeShape
  , withConsumeSpan
  )
import Control.Monad (void, when)
import Control.Monad.IO.Class (liftIO)
import Data.Aeson (Value)
import Data.Bifunctor (second)
import Data.Foldable (traverse_)
import Data.Int (Int32)
import Data.List (partition)
import Data.List.NonEmpty (NonEmpty (..))
import Data.Maybe (fromMaybe)
import Data.Set qualified as Set
import Data.Text (Text)
import Data.Time (UTCTime)
import UnliftIO (MonadUnliftIO, UnliftIO (..), askUnliftIO)

import Arbiter.Worker.BackoffStrategy
import Arbiter.Worker.Batch
import Arbiter.Worker.Config
import Arbiter.Worker.Heartbeat.Guard (reclaimedReason)
import Arbiter.Worker.Logger
import Arbiter.Worker.Logger.Internal (jobHook, poolLog, withJobContext)
import Arbiter.Worker.Results (storeEncodedResult, storeEncodedResults, storeJobResult)

-- | Add a claimed batch to the pool log context.
batchLog :: WorkerConfig m payload -> NonEmpty (Job.JobRead payload) -> LogConfig
batchLog :: forall (m :: * -> *) payload.
WorkerConfig m payload -> NonEmpty (JobRead payload) -> LogConfig
batchLog WorkerConfig m payload
config = LogConfig -> NonEmpty (JobRead payload) -> LogConfig
forall payload.
LogConfig -> NonEmpty (JobRead payload) -> LogConfig
withJobContext (WorkerConfig m payload -> LogConfig
forall (m :: * -> *) payload. WorkerConfig m payload -> LogConfig
logConfig WorkerConfig m payload
config)

-- | Per-batch statements and hooks. Callback statements join the handler's
-- transaction. Other statements use the pool context. Built per pool and
-- bound to jobs per batch.
type PoolEffects m payload =
  NonEmpty (Job.JobRead payload)
  -> Effects IO (UnliftIO m) (Job.JobRead payload) (NonEmpty (Job.JobWrite payload)) Value

poolEffects
  :: forall payload m
   . (JobOperation m payload)
  => WorkerConfig m payload
  -> Ops.JobStatements
  -> ConsumeSpan
  -> m (PoolEffects m payload)
poolEffects :: forall payload (m :: * -> *).
JobOperation m payload =>
WorkerConfig m payload
-> JobStatements -> ConsumeSpan -> m (PoolEffects m payload)
poolEffects WorkerConfig m payload
config JobStatements
statements ConsumeSpan
consumeSpan = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  unlift@(UnliftIO run) <- askUnliftIO
  pure $ \jobs :: NonEmpty (JobRead payload)
jobs@(JobRead payload
firstJob :| [JobRead payload]
_) ->
    let queue :: Text
queue = JobRead payload -> Text
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> q
Job.queueName JobRead payload
firstJob
     in Effects
          { effectAmbient :: UnliftIO m
effectAmbient = UnliftIO m
unlift
          , effectSpan :: ((IO () -> IO ()) -> IO ()) -> IO ()
effectSpan = \(IO () -> IO ()) -> IO ()
body -> m () -> IO ()
forall a. m a -> IO a
run (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
              tracer <- m (Maybe Tracer)
forall (m :: * -> *). MonadIO m => m (Maybe Tracer)
resolveTracer
              withConsumeSpan tracer consumeSpan jobs (capturingContextIO >>= liftIO . body)
          , effectAck :: UnliftIO m -> JobRead payload -> Maybe Value -> IO ()
effectAck = \(UnliftIO forall a. m a -> IO a
runIn) JobRead payload
job Maybe Value
stored ->
              m () -> IO ()
forall a. m a -> IO a
runIn (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ m () -> m ()
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
                JobStatements -> JobRead payload -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
JobStatements -> JobRead payload -> m ()
ackOrGone JobStatements
statements JobRead payload
job
                JobRead payload -> Maybe Value -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
JobRead payload -> Maybe Value -> m ()
storeEncodedResult JobRead payload
job Maybe Value
stored
          , effectAckAll :: UnliftIO m
-> [(JobRead payload, Maybe Value)]
-> IO ([JobRead payload], [JobRead payload])
effectAckAll = \(UnliftIO forall a. m a -> IO a
runIn) [(JobRead payload, Maybe Value)]
pairs ->
              m ([JobRead payload], [JobRead payload])
-> IO ([JobRead payload], [JobRead payload])
forall a. m a -> IO a
runIn (m ([JobRead payload], [JobRead payload])
 -> IO ([JobRead payload], [JobRead payload]))
-> m ([JobRead payload], [JobRead payload])
-> IO ([JobRead payload], [JobRead payload])
forall a b. (a -> b) -> a -> b
$ m ([JobRead payload], [JobRead payload])
-> m ([JobRead payload], [JobRead payload])
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m ([JobRead payload], [JobRead payload])
 -> m ([JobRead payload], [JobRead payload]))
-> m ([JobRead payload], [JobRead payload])
-> m ([JobRead payload], [JobRead payload])
forall a b. (a -> b) -> a -> b
$ do
                let jobsToAck :: [JobRead payload]
jobsToAck = ((JobRead payload, Maybe Value) -> JobRead payload)
-> [(JobRead payload, Maybe Value)] -> [JobRead payload]
forall a b. (a -> b) -> [a] -> [b]
map (JobRead payload, Maybe Value) -> JobRead payload
forall a b. (a, b) -> a
fst [(JobRead payload, Maybe Value)]
pairs
                acked <- [Int64] -> Set Int64
forall a. Ord a => [a] -> Set a
Set.fromList ([Int64] -> Set Int64) -> m [Int64] -> m (Set Int64)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> AckStatements -> [JobRead payload] -> m [Int64]
forall (m :: * -> *) payload.
MonadArbiter m =>
AckStatements -> [JobRead payload] -> m [Int64]
Ops.ackJobsBatchWith (JobStatements -> AckStatements
Ops.statementsAck JobStatements
statements) [JobRead payload]
jobsToAck
                let ackedJob = (Int64 -> Set Int64 -> Bool
forall a. Ord a => a -> Set a -> Bool
`Set.member` Set Int64
acked) (Int64 -> Bool)
-> (JobRead payload -> Int64) -> JobRead payload -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
Job.primaryKey
                storeEncodedResults (filter (ackedJob . fst) pairs)
                pure (partition ackedJob jobsToAck)
          , effectFail :: UnliftIO m
-> Failure -> JobRead payload -> IO (Either Text Outcome)
effectFail = \(UnliftIO forall a. m a -> IO a
runIn) Failure
failure JobRead payload
job ->
              m (Either Text Outcome) -> IO (Either Text Outcome)
forall a. m a -> IO a
runIn (m (Either Text Outcome) -> IO (Either Text Outcome))
-> m (Either Text Outcome) -> IO (Either Text Outcome)
forall a b. (a -> b) -> a -> b
$ m (Either Text Outcome) -> m (Either Text Outcome)
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m (Either Text Outcome) -> m (Either Text Outcome))
-> m (Either Text Outcome) -> m (Either Text Outcome)
forall a b. (a -> b) -> a -> b
$ WorkerConfig m payload
-> TreeLocks
-> Failure
-> JobRead payload
-> m (Either Text Outcome)
forall payload (m :: * -> *).
MonadArbiter m =>
WorkerConfig m payload
-> TreeLocks
-> Failure
-> JobRead payload
-> m (Either Text Outcome)
handleJobFailure WorkerConfig m payload
config TreeLocks
Ops.TakeLocks Failure
failure JobRead payload
job
          , effectFailAll :: UnliftIO m
-> Failure
-> [JobRead payload]
-> [JobRead payload]
-> IO [(JobRead payload, Either Text Outcome)]
effectFailAll = \(UnliftIO forall a. m a -> IO a
runIn) Failure
failure [JobRead payload]
unhandled [JobRead payload]
unowned ->
              m [(JobRead payload, Either Text Outcome)]
-> IO [(JobRead payload, Either Text Outcome)]
forall a. m a -> IO a
runIn (m [(JobRead payload, Either Text Outcome)]
 -> IO [(JobRead payload, Either Text Outcome)])
-> m [(JobRead payload, Either Text Outcome)]
-> IO [(JobRead payload, Either Text Outcome)]
forall a b. (a -> b) -> a -> b
$ m [(JobRead payload, Either Text Outcome)]
-> m [(JobRead payload, Either Text Outcome)]
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m [(JobRead payload, Either Text Outcome)]
 -> m [(JobRead payload, Either Text Outcome)])
-> m [(JobRead payload, Either Text Outcome)]
-> m [(JobRead payload, Either Text Outcome)]
forall a b. (a -> b) -> a -> b
$ WorkerConfig m payload
-> Text
-> Text
-> Failure
-> [JobRead payload]
-> [JobRead payload]
-> m [(JobRead payload, Either Text Outcome)]
forall (m :: * -> *) payload.
MonadArbiter m =>
WorkerConfig m payload
-> Text
-> Text
-> Failure
-> [JobRead payload]
-> [JobRead payload]
-> m [(JobRead payload, Either Text Outcome)]
failAllLocked WorkerConfig m payload
config Text
schemaName Text
queue Failure
failure [JobRead payload]
unhandled [JobRead payload]
unowned
          , effectDeleteCancelled :: UnliftIO m -> [JobRead payload] -> IO (Set Int64)
effectDeleteCancelled = \(UnliftIO forall a. m a -> IO a
runIn) [JobRead payload]
gone ->
              m (Set Int64) -> IO (Set Int64)
forall a. m a -> IO a
runIn (m (Set Int64) -> IO (Set Int64))
-> (m [Int64] -> m (Set Int64)) -> m [Int64] -> IO (Set Int64)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ([Int64] -> Set Int64) -> m [Int64] -> m (Set Int64)
forall a b. (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap [Int64] -> Set Int64
forall a. Ord a => [a] -> Set a
Set.fromList
                (m [Int64] -> IO (Set Int64)) -> m [Int64] -> IO (Set Int64)
forall a b. (a -> b) -> a -> b
$ Text -> Text -> CancelledScope payload -> m [Int64]
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> CancelledScope payload -> m [Int64]
Ops.deleteCancelledJobs Text
schemaName Text
queue
                (CancelledScope payload -> m [Int64])
-> CancelledScope payload -> m [Int64]
forall a b. (a -> b) -> a -> b
$ UUID -> [JobRead payload] -> CancelledScope payload
forall payload. UUID -> [JobRead payload] -> CancelledScope payload
Ops.HeldBy (WorkerConfig m payload -> UUID
forall (m :: * -> *) payload. WorkerConfig m payload -> UUID
workerId WorkerConfig m payload
config) [JobRead payload]
gone
          , effectRelease :: UnliftIO m
-> Maybe NominalDiffTime -> [JobRead payload] -> IO (Set Int64)
effectRelease = \(UnliftIO forall a. m a -> IO a
runIn) Maybe NominalDiffTime
delay [JobRead payload]
released ->
              m (Set Int64) -> IO (Set Int64)
forall a. m a -> IO a
runIn ([Int64] -> Set Int64
forall a. Ord a => [a] -> Set a
Set.fromList ([Int64] -> Set Int64) -> m [Int64] -> m (Set Int64)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> ([JobRead payload] -> m [Int64])
-> (NominalDiffTime -> [JobRead payload] -> m [Int64])
-> Maybe NominalDiffTime
-> [JobRead payload]
-> m [Int64]
forall b a. b -> (a -> b) -> Maybe a -> b
maybe [JobRead payload] -> m [Int64]
forall payload (m :: * -> *).
MonadArbiter m =>
[JobRead payload] -> m [Int64]
Arb.nackJobsBatch NominalDiffTime -> [JobRead payload] -> m [Int64]
forall payload (m :: * -> *).
MonadArbiter m =>
NominalDiffTime -> [JobRead payload] -> m [Int64]
Arb.nackJobsBatchAfter Maybe NominalDiffTime
delay [JobRead payload]
released)
          , effectSpawn :: UnliftIO m
-> JobRead payload -> NonEmpty (JobWrite payload) -> IO ()
effectSpawn = \(UnliftIO forall a. m a -> IO a
runIn) JobRead payload
job NonEmpty (JobWrite payload)
kids ->
              m () -> IO ()
forall a. m a -> IO a
runIn (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ m () -> m ()
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
                m (NonEmpty (JobRead payload)) -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (m (NonEmpty (JobRead payload)) -> m ())
-> m (NonEmpty (JobRead payload)) -> m ()
forall a b. (a -> b) -> a -> b
$ JobRead payload
-> NonEmpty (JobWrite payload) -> m (NonEmpty (JobRead payload))
forall payload (m :: * -> *).
JobOperation m payload =>
JobRead payload
-> NonEmpty (JobWrite payload) -> m (NonEmpty (JobRead payload))
Arb.spawnChildren JobRead payload
job NonEmpty (JobWrite payload)
kids
                JobStatements -> JobRead payload -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
JobStatements -> JobRead payload -> m ()
ackOrGone JobStatements
statements JobRead payload
job
          , effectReport :: UnliftIO m -> Report (JobRead payload) -> IO ()
effectReport = \(UnliftIO forall a. m a -> IO a
runIn) -> m () -> IO ()
forall a. m a -> IO a
runIn (m () -> IO ())
-> (Report (JobRead payload) -> m ())
-> Report (JobRead payload)
-> IO ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. ConsumeShape -> Report (JobRead payload) -> m ()
report (NonEmpty (JobRead payload) -> ConsumeShape
forall payload. NonEmpty (JobRead payload) -> ConsumeShape
batchSpanShape NonEmpty (JobRead payload)
jobs)
          , effectLog :: LogLevel -> [JobRead payload] -> Text -> IO ()
effectLog = LogConfig -> LogLevel -> [JobRead payload] -> Text -> IO ()
forall payload.
LogConfig -> LogLevel -> [JobRead payload] -> Text -> IO ()
poolLog (WorkerConfig m payload -> LogConfig
forall (m :: * -> *) payload. WorkerConfig m payload -> LogConfig
logConfig WorkerConfig m payload
config)
          }
  where
    hooks :: ObservabilityHooks m payload
hooks = WorkerConfig m payload -> ObservabilityHooks m payload
forall (m :: * -> *) payload.
WorkerConfig m payload -> ObservabilityHooks m payload
observabilityHooks WorkerConfig m payload
config
    hook :: JobRead payload -> Text -> m () -> m ()
hook = LogConfig -> JobRead payload -> Text -> m () -> m ()
forall (m :: * -> *) payload.
MonadUnliftIO m =>
LogConfig -> JobRead payload -> Text -> m () -> m ()
jobHook (WorkerConfig m payload -> LogConfig
forall (m :: * -> *) payload. WorkerConfig m payload -> LogConfig
logConfig WorkerConfig m payload
config)
    report :: ConsumeShape -> Report (JobRead payload) -> m ()
report ConsumeShape
shape = \case
      Claimed JobRead payload
job UTCTime
startTime -> JobRead payload -> Text -> m () -> m ()
hook JobRead payload
job Text
"onJobClaimed" (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ ObservabilityHooks m payload
-> JobPayload payload => JobRead payload -> UTCTime -> m ()
forall (m :: * -> *) payload.
ObservabilityHooks m payload
-> JobPayload payload => JobRead payload -> UTCTime -> m ()
Job.onJobClaimed ObservabilityHooks m payload
hooks JobRead payload
job UTCTime
startTime
      Succeeded JobRead payload
job UTCTime
startTime UTCTime
endTime -> JobRead payload -> Text -> m () -> m ()
hook JobRead payload
job Text
"onJobSuccess" (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ ObservabilityHooks m payload
-> JobPayload payload =>
   JobRead payload -> UTCTime -> UTCTime -> m ()
forall (m :: * -> *) payload.
ObservabilityHooks m payload
-> JobPayload payload =>
   JobRead payload -> UTCTime -> UTCTime -> m ()
Job.onJobSuccess ObservabilityHooks m payload
hooks JobRead payload
job UTCTime
startTime UTCTime
endTime
      Failed JobRead payload
job Text
errorMsg UTCTime
startTime UTCTime
endTime Outcome
outcome -> do
        WorkerConfig m payload
-> ConsumeShape
-> JobRead payload
-> Text
-> UTCTime
-> UTCTime
-> m ()
forall (m :: * -> *) payload.
JobOperation m payload =>
WorkerConfig m payload
-> ConsumeShape
-> JobRead payload
-> Text
-> UTCTime
-> UTCTime
-> m ()
fireFailure WorkerConfig m payload
config ConsumeShape
shape JobRead payload
job Text
errorMsg UTCTime
startTime UTCTime
endTime
        case Outcome
outcome of
          Retrying NominalDiffTime
delay -> JobRead payload -> Text -> m () -> m ()
hook JobRead payload
job Text
"onJobRetry" (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ ObservabilityHooks m payload
-> JobPayload payload => JobRead payload -> NominalDiffTime -> m ()
forall (m :: * -> *) payload.
ObservabilityHooks m payload
-> JobPayload payload => JobRead payload -> NominalDiffTime -> m ()
Job.onJobRetry ObservabilityHooks m payload
hooks JobRead payload
job NominalDiffTime
delay
          Outcome
_ -> JobRead payload -> Text -> m () -> m ()
hook JobRead payload
job Text
"onJobFailedAndMovedToDLQ" (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ ObservabilityHooks m payload
-> JobPayload payload => JobRead payload -> Text -> m ()
forall (m :: * -> *) payload.
ObservabilityHooks m payload
-> JobPayload payload => JobRead payload -> Text -> m ()
Job.onJobFailedAndMovedToDLQ ObservabilityHooks m payload
hooks JobRead payload
job Text
errorMsg
      Cancelled JobRead payload
job Text
reason -> WorkerConfig m payload -> JobRead payload -> Text -> m ()
forall (m :: * -> *) payload.
JobOperation m payload =>
WorkerConfig m payload -> JobRead payload -> Text -> m ()
fireCancelled WorkerConfig m payload
config JobRead payload
job Text
reason
      Unavailable JobRead payload
job Text
reason -> JobRead payload -> Text -> m () -> m ()
hook JobRead payload
job Text
"onJobUnavailable" (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ ObservabilityHooks m payload
-> JobPayload payload => JobRead payload -> Text -> m ()
forall (m :: * -> *) payload.
ObservabilityHooks m payload
-> JobPayload payload => JobRead payload -> Text -> m ()
Job.onJobUnavailable ObservabilityHooks m payload
hooks JobRead payload
job Text
reason

-- | How a pool runs a batch.
type PoolMode m payload = Mode IO (UnliftIO m) (Job.JobRead payload) (NonEmpty (Job.JobWrite payload)) Value

-- | How the pool runs a batch's handler, unlifted to IO.
poolMode
  :: forall payload m
   . ( EncodeJobResult (ResultOf m payload)
     , MonadArbiter m
     )
  => WorkerConfig m payload
  -> Ops.JobStatements
  -> m (PoolMode m payload)
poolMode :: forall payload (m :: * -> *).
(EncodeJobResult (ResultOf m payload), MonadArbiter m) =>
WorkerConfig m payload -> JobStatements -> m (PoolMode m payload)
poolMode WorkerConfig m payload
config JobStatements
statements = do
  UnliftIO run <- m (UnliftIO m)
forall (m :: * -> *). MonadUnliftIO m => m (UnliftIO m)
askUnliftIO
  pure $ case handlerMode config of
    SingleJobMode JobHandler m payload (ResultOf m payload)
handler ->
      (JobRead payload -> IO ()) -> PoolMode m payload
forall (n :: * -> *) ctx job kids stored.
(job -> n ()) -> Mode n ctx job kids stored
SingleMode ((JobRead payload -> IO ()) -> PoolMode m payload)
-> (JobRead payload -> IO ()) -> PoolMode m payload
forall a b. (a -> b) -> a -> b
$ \JobRead payload
job ->
        m () -> IO ()
forall a. m a -> IO a
run (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ m () -> m ()
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
          handlerResult <- JobHandler m payload (ResultOf m payload)
-> JobRead payload -> m (ResultOf m payload)
forall payload.
JobHandler m payload (ResultOf m payload)
-> JobRead payload -> m (ResultOf m payload)
forall (m :: * -> *) payload.
MonadArbiter m =>
JobHandler m payload (ResultOf m payload)
-> JobRead payload -> m (ResultOf m payload)
runHandlerWithConnection JobHandler m payload (ResultOf m payload)
handler JobRead payload
job
          ackOrGone statements job
          storeJobResult job handlerResult
    BatchedJobsMode Int
_ NonEmpty (JobRead payload)
-> BatchCallbacks m payload (ResultOf m payload) -> m ()
handler ->
      (NonEmpty (JobRead payload)
 -> Callbacks
      IO
      (UnliftIO m)
      (JobRead payload)
      (NonEmpty (JobWrite payload))
      Value
 -> IO ())
-> PoolMode m payload
forall (n :: * -> *) ctx job kids stored.
(NonEmpty job -> Callbacks n ctx job kids stored -> n ())
-> Mode n ctx job kids stored
BatchedMode ((NonEmpty (JobRead payload)
  -> Callbacks
       IO
       (UnliftIO m)
       (JobRead payload)
       (NonEmpty (JobWrite payload))
       Value
  -> IO ())
 -> PoolMode m payload)
-> (NonEmpty (JobRead payload)
    -> Callbacks
         IO
         (UnliftIO m)
         (JobRead payload)
         (NonEmpty (JobWrite payload))
         Value
    -> IO ())
-> PoolMode m payload
forall a b. (a -> b) -> a -> b
$ \NonEmpty (JobRead payload)
jobs Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
callbacks -> m () -> IO ()
forall a. m a -> IO a
run (NonEmpty (JobRead payload)
-> BatchCallbacks m payload (ResultOf m payload) -> m ()
handler NonEmpty (JobRead payload)
jobs (Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
-> BatchCallbacks
     m payload (SpecResult (MatchIn payload '[] (RegistryOf m)))
forall result (m :: * -> *) payload.
(EncodeJobResult result, MonadUnliftIO m) =>
Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
-> BatchCallbacks m payload result
batchCallbacks Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
callbacks))

-- | The handler's callbacks over the batch's settle operations. Each captures the
-- handler's context at the call.
batchCallbacks
  :: (EncodeJobResult result, MonadUnliftIO m)
  => Callbacks IO (UnliftIO m) (Job.JobRead payload) (NonEmpty (Job.JobWrite payload)) Value
  -> BatchCallbacks m payload result
batchCallbacks :: forall result (m :: * -> *) payload.
(EncodeJobResult result, MonadUnliftIO m) =>
Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
-> BatchCallbacks m payload result
batchCallbacks Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
callbacks =
  BatchCallbacks
    { ack :: JobRead payload -> m ()
ack = Maybe Value -> JobRead payload -> m ()
ackAs Maybe Value
forall a. Maybe a
Nothing
    , ackWith :: JobRead payload -> result -> m ()
ackWith = \JobRead payload
job result
result -> Maybe Value -> JobRead payload -> m ()
ackAs (result -> Maybe Value
forall a. EncodeJobResult a => a -> Maybe Value
encodeJobResult result
result) JobRead payload
job
    , ackAll :: [JobRead payload] -> m ()
ackAll = \[JobRead payload]
jobs -> (UnliftIO m -> IO ()) -> m ()
forall {m :: * -> *} {b}.
MonadUnliftIO m =>
(UnliftIO m -> IO b) -> m b
here (\UnliftIO m
ctx -> Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
-> UnliftIO m -> [(JobRead payload, Maybe Value)] -> IO ()
forall (n :: * -> *) ctx job kids stored.
Callbacks n ctx job kids stored
-> ctx -> [(job, Maybe stored)] -> n ()
callbackAckAll Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
callbacks UnliftIO m
ctx ((JobRead payload -> (JobRead payload, Maybe Value))
-> [JobRead payload] -> [(JobRead payload, Maybe Value)]
forall a b. (a -> b) -> [a] -> [b]
map (,Maybe Value
forall a. Maybe a
Nothing) [JobRead payload]
jobs))
    , ackAllWith :: [(JobRead payload, result)] -> m ()
ackAllWith = \[(JobRead payload, result)]
pairs -> (UnliftIO m -> IO ()) -> m ()
forall {m :: * -> *} {b}.
MonadUnliftIO m =>
(UnliftIO m -> IO b) -> m b
here (\UnliftIO m
ctx -> Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
-> UnliftIO m -> [(JobRead payload, Maybe Value)] -> IO ()
forall (n :: * -> *) ctx job kids stored.
Callbacks n ctx job kids stored
-> ctx -> [(job, Maybe stored)] -> n ()
callbackAckAll Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
callbacks UnliftIO m
ctx (((JobRead payload, result) -> (JobRead payload, Maybe Value))
-> [(JobRead payload, result)] -> [(JobRead payload, Maybe Value)]
forall a b. (a -> b) -> [a] -> [b]
map ((result -> Maybe Value)
-> (JobRead payload, result) -> (JobRead payload, Maybe Value)
forall b c a. (b -> c) -> (a, b) -> (a, c)
forall (p :: * -> * -> *) b c a.
Bifunctor p =>
(b -> c) -> p a b -> p a c
second result -> Maybe Value
forall a. EncodeJobResult a => a -> Maybe Value
encodeJobResult) [(JobRead payload, result)]
pairs))
    , failRetry :: JobRead payload -> Text -> m ()
failRetry = FailureKind -> JobRead payload -> Text -> m ()
failAs FailureKind
RetryFailure
    , failRetryAfter :: JobRead payload -> NominalDiffTime -> Text -> m ()
failRetryAfter = \JobRead payload
job NominalDiffTime
delay -> FailureKind -> JobRead payload -> Text -> m ()
failAs (NominalDiffTime -> FailureKind
RetryAfterFailure NominalDiffTime
delay) JobRead payload
job
    , failPermanent :: JobRead payload -> Text -> m ()
failPermanent = FailureKind -> JobRead payload -> Text -> m ()
failAs FailureKind
PermanentFailure
    , cancelBranch :: JobRead payload -> Text -> m ()
cancelBranch = FailureKind -> JobRead payload -> Text -> m ()
failAs FailureKind
BranchCancelFailure
    , cancelTree :: JobRead payload -> Text -> m ()
cancelTree = FailureKind -> JobRead payload -> Text -> m ()
failAs FailureKind
TreeCancelFailure
    , nack :: JobRead payload -> m ()
nack = \JobRead payload
job -> (UnliftIO m -> IO ()) -> m ()
forall {m :: * -> *} {b}.
MonadUnliftIO m =>
(UnliftIO m -> IO b) -> m b
here (\UnliftIO m
ctx -> Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
-> UnliftIO m -> Maybe NominalDiffTime -> JobRead payload -> IO ()
forall (n :: * -> *) ctx job kids stored.
Callbacks n ctx job kids stored
-> ctx -> Maybe NominalDiffTime -> job -> n ()
callbackNack Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
callbacks UnliftIO m
ctx Maybe NominalDiffTime
forall a. Maybe a
Nothing JobRead payload
job)
    , nackAfter :: JobRead payload -> NominalDiffTime -> m ()
nackAfter = \JobRead payload
job NominalDiffTime
delay -> (UnliftIO m -> IO ()) -> m ()
forall {m :: * -> *} {b}.
MonadUnliftIO m =>
(UnliftIO m -> IO b) -> m b
here (\UnliftIO m
ctx -> Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
-> UnliftIO m -> Maybe NominalDiffTime -> JobRead payload -> IO ()
forall (n :: * -> *) ctx job kids stored.
Callbacks n ctx job kids stored
-> ctx -> Maybe NominalDiffTime -> job -> n ()
callbackNack Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
callbacks UnliftIO m
ctx (NominalDiffTime -> Maybe NominalDiffTime
forall a. a -> Maybe a
Just NominalDiffTime
delay) JobRead payload
job)
    , spawn :: JobRead payload -> NonEmpty (JobWrite payload) -> m ()
spawn = \JobRead payload
job NonEmpty (JobWrite payload)
children -> (UnliftIO m -> IO ()) -> m ()
forall {m :: * -> *} {b}.
MonadUnliftIO m =>
(UnliftIO m -> IO b) -> m b
here (\UnliftIO m
ctx -> Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
-> UnliftIO m
-> JobRead payload
-> NonEmpty (JobWrite payload)
-> IO ()
forall (n :: * -> *) ctx job kids stored.
Callbacks n ctx job kids stored -> ctx -> job -> kids -> n ()
callbackSpawn Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
callbacks UnliftIO m
ctx JobRead payload
job NonEmpty (JobWrite payload)
children)
    }
  where
    here :: (UnliftIO m -> IO b) -> m b
here UnliftIO m -> IO b
act = m (UnliftIO m)
forall (m :: * -> *). MonadUnliftIO m => m (UnliftIO m)
askUnliftIO m (UnliftIO m) -> (UnliftIO m -> m b) -> m b
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= IO b -> m b
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO b -> m b) -> (UnliftIO m -> IO b) -> UnliftIO m -> m b
forall b c a. (b -> c) -> (a -> b) -> a -> c
. UnliftIO m -> IO b
act
    ackAs :: Maybe Value -> JobRead payload -> m ()
ackAs Maybe Value
stored JobRead payload
job = (UnliftIO m -> IO ()) -> m ()
forall {m :: * -> *} {b}.
MonadUnliftIO m =>
(UnliftIO m -> IO b) -> m b
here (\UnliftIO m
ctx -> Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
-> UnliftIO m -> JobRead payload -> Maybe Value -> IO ()
forall (n :: * -> *) ctx job kids stored.
Callbacks n ctx job kids stored
-> ctx -> job -> Maybe stored -> n ()
callbackAck Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
callbacks UnliftIO m
ctx JobRead payload
job Maybe Value
stored)
    failAs :: FailureKind -> JobRead payload -> Text -> m ()
failAs FailureKind
kind JobRead payload
job Text
msg = (UnliftIO m -> IO ()) -> m ()
forall {m :: * -> *} {b}.
MonadUnliftIO m =>
(UnliftIO m -> IO b) -> m b
here (\UnliftIO m
ctx -> Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
-> UnliftIO m -> Failure -> JobRead payload -> IO ()
forall (n :: * -> *) ctx job kids stored.
Callbacks n ctx job kids stored -> ctx -> Failure -> job -> n ()
callbackFail Callbacks
  IO
  (UnliftIO m)
  (JobRead payload)
  (NonEmpty (JobWrite payload))
  Value
callbacks UnliftIO m
ctx (Text
msg, FailureKind
kind) JobRead payload
job)

-- | Ack a job inside the caller's transaction, throwing if another worker reclaimed it mid-flight.
ackOrGone :: (MonadArbiter m) => Ops.JobStatements -> Job.JobRead payload -> m ()
ackOrGone :: forall (m :: * -> *) payload.
MonadArbiter m =>
JobStatements -> JobRead payload -> m ()
ackOrGone JobStatements
statements JobRead payload
job = do
  rowsAffected <- AckStatements -> JobRead payload -> m Int64
forall (m :: * -> *) payload.
MonadArbiter m =>
AckStatements -> JobRead payload -> m Int64
Ops.ackJobWith (JobStatements -> AckStatements
Ops.statementsAck JobStatements
statements) JobRead payload
job
  when (rowsAffected == 0) $
    throwJobGoneIds reclaimedReason [Job.primaryKey job]

-- | The job's own attempt budget, or the default.
jobMaxAttempts :: Job.JobRead payload -> Int32
jobMaxAttempts :: forall payload. JobRead payload -> Int32
jobMaxAttempts JobRead payload
job = Int32 -> Maybe Int32 -> Int32
forall a. a -> Maybe a -> a
fromMaybe Int32
Job.defaultMaxAttempts (JobRead payload -> Maybe Int32
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> Maybe Int32
Job.maxAttempts JobRead payload
job)

-- | What the consumer span over this batch covers. A batch of one narrows it to
-- that job.
batchSpanShape :: NonEmpty (Job.JobRead payload) -> ConsumeShape
batchSpanShape :: forall payload. NonEmpty (JobRead payload) -> ConsumeShape
batchSpanShape = Int -> ConsumeShape
toConsumeShape (Int -> ConsumeShape)
-> (NonEmpty (JobRead payload) -> Int)
-> NonEmpty (JobRead payload)
-> ConsumeShape
forall b c a. (b -> c) -> (a -> b) -> a -> c
. NonEmpty (JobRead payload) -> Int
forall a. NonEmpty a -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length

-- | Report a failed job to its hooks and to the consumer span it ran under. Only a
-- single-job span takes the error status.
fireFailure
  :: (JobOperation m payload)
  => WorkerConfig m payload
  -> ConsumeShape
  -- ^ What the span this job ran under covers.
  -> Job.JobRead payload
  -> Text
  -> UTCTime
  -> UTCTime
  -> m ()
fireFailure :: forall (m :: * -> *) payload.
JobOperation m payload =>
WorkerConfig m payload
-> ConsumeShape
-> JobRead payload
-> Text
-> UTCTime
-> UTCTime
-> m ()
fireFailure WorkerConfig m payload
config ConsumeShape
shape JobRead payload
job Text
errorMsg UTCTime
startTime UTCTime
endTime = do
  JobRead payload -> Text -> m ()
forall (m :: * -> *) payload.
MonadIO m =>
JobRead payload -> Text -> m ()
recordJobFailure JobRead payload
job Text
errorMsg
  Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (ConsumeShape
shape ConsumeShape -> ConsumeShape -> Bool
forall a. Eq a => a -> a -> Bool
== ConsumeShape
PerJob) (Text -> m ()
forall (m :: * -> *). MonadIO m => Text -> m ()
markSpanError Text
errorMsg)
  LogConfig -> JobRead payload -> Text -> m () -> m ()
forall (m :: * -> *) payload.
MonadUnliftIO m =>
LogConfig -> JobRead payload -> Text -> m () -> m ()
jobHook (WorkerConfig m payload -> LogConfig
forall (m :: * -> *) payload. WorkerConfig m payload -> LogConfig
logConfig WorkerConfig m payload
config) JobRead payload
job Text
"onJobFailure" (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$
    ObservabilityHooks m payload
-> JobPayload payload =>
   JobRead payload -> Text -> UTCTime -> UTCTime -> m ()
forall (m :: * -> *) payload.
ObservabilityHooks m payload
-> JobPayload payload =>
   JobRead payload -> Text -> UTCTime -> UTCTime -> m ()
Job.onJobFailure (WorkerConfig m payload -> ObservabilityHooks m payload
forall (m :: * -> *) payload.
WorkerConfig m payload -> ObservabilityHooks m payload
observabilityHooks WorkerConfig m payload
config) JobRead payload
job Text
errorMsg UTCTime
startTime UTCTime
endTime

-- | Report a cancelled job.
fireCancelled
  :: (JobOperation m payload)
  => WorkerConfig m payload
  -> Job.JobRead payload
  -> Text
  -> m ()
fireCancelled :: forall (m :: * -> *) payload.
JobOperation m payload =>
WorkerConfig m payload -> JobRead payload -> Text -> m ()
fireCancelled WorkerConfig m payload
config JobRead payload
job Text
errorMsg = do
  JobRead payload -> Text -> m ()
forall (m :: * -> *) payload.
MonadIO m =>
JobRead payload -> Text -> m ()
recordJobCancelled JobRead payload
job Text
errorMsg
  LogConfig -> JobRead payload -> Text -> m () -> m ()
forall (m :: * -> *) payload.
MonadUnliftIO m =>
LogConfig -> JobRead payload -> Text -> m () -> m ()
jobHook (WorkerConfig m payload -> LogConfig
forall (m :: * -> *) payload. WorkerConfig m payload -> LogConfig
logConfig WorkerConfig m payload
config) JobRead payload
job Text
"onJobCancelled" (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$
    ObservabilityHooks m payload
-> JobPayload payload => JobRead payload -> Text -> m ()
forall (m :: * -> *) payload.
ObservabilityHooks m payload
-> JobPayload payload => JobRead payload -> Text -> m ()
Job.onJobCancelled (WorkerConfig m payload -> ObservabilityHooks m payload
forall (m :: * -> *) payload.
WorkerConfig m payload -> ObservabilityHooks m payload
observabilityHooks WorkerConfig m payload
config) JobRead payload
job Text
errorMsg

-- | Fail @unhandled@ and cancel @unowned@ after one pass locks every row they touch.
-- A cancel's pass covers the whole tree.
failAllLocked
  :: (MonadArbiter m)
  => WorkerConfig m payload
  -> SchemaName
  -> TableName
  -> Failure
  -> [Job.JobRead payload]
  -> [Job.JobRead payload]
  -> m [(Job.JobRead payload, Either Text Outcome)]
failAllLocked :: forall (m :: * -> *) payload.
MonadArbiter m =>
WorkerConfig m payload
-> Text
-> Text
-> Failure
-> [JobRead payload]
-> [JobRead payload]
-> m [(JobRead payload, Either Text Outcome)]
failAllLocked WorkerConfig m payload
config Text
schemaName Text
queue failure :: Failure
failure@(Text
_, FailureKind
kind) [JobRead payload]
unhandled [JobRead payload]
unowned = do
  m ()
lockParents
  held <- Text -> Text -> [Int64] -> m TreesLocked
lockTrees Text
schemaName Text
queue ((JobRead payload -> Int64) -> [JobRead payload] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
Job.primaryKey ([JobRead payload]
unhandled [JobRead payload] -> [JobRead payload] -> [JobRead payload]
forall a. Semigroup a => a -> a -> a
<> [JobRead payload]
unowned))
  outcomes <- traverse (\JobRead payload
job -> (JobRead payload
job,) (Either Text Outcome -> (JobRead payload, Either Text Outcome))
-> m (Either Text Outcome)
-> m (JobRead payload, Either Text Outcome)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> WorkerConfig m payload
-> TreeLocks
-> Failure
-> JobRead payload
-> m (Either Text Outcome)
forall payload (m :: * -> *).
MonadArbiter m =>
WorkerConfig m payload
-> TreeLocks
-> Failure
-> JobRead payload
-> m (Either Text Outcome)
handleJobFailure WorkerConfig m payload
config (TreesLocked -> TreeLocks
Ops.LocksHeld TreesLocked
held) Failure
failure JobRead payload
job) unhandled
  traverse_ (void . cancelJobFor kind) unowned
  pure outcomes
  where
    lockParents :: m ()
lockParents
      | FailureKind
kind FailureKind -> FailureKind -> Bool
forall a. Eq a => a -> a -> Bool
== FailureKind
BranchCancelFailure =
          Text -> Text -> [Int64] -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m ()
Ops.lockJobRootsAndParents Text
schemaName Text
queue ((JobRead payload -> Int64) -> [JobRead payload] -> [Int64]
forall a b. (a -> b) -> [a] -> [b]
map JobRead payload -> Int64
forall payload. JobRead payload -> Int64
branchRoot ([JobRead payload]
unhandled [JobRead payload] -> [JobRead payload] -> [JobRead payload]
forall a. Semigroup a => a -> a -> a
<> [JobRead payload]
unowned))
      | Bool
otherwise = Text -> Text -> [Maybe Int64] -> m ()
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Maybe Int64] -> m ()
Ops.lockJobParents Text
schemaName Text
queue ((JobRead payload -> Maybe Int64)
-> [JobRead payload] -> [Maybe Int64]
forall a b. (a -> b) -> [a] -> [b]
map JobRead payload -> Maybe Int64
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Maybe Int64
Job.parentId [JobRead payload]
unhandled)
    lockTrees :: Text -> Text -> [Int64] -> m TreesLocked
lockTrees
      | FailureKind -> Bool
cancelsTree FailureKind
kind = Text -> Text -> [Int64] -> m TreesLocked
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m TreesLocked
Ops.lockJobTreesFromRoot
      | Bool
otherwise = Text -> Text -> [Int64] -> m TreesLocked
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> [Int64] -> m TreesLocked
Ops.lockJobTrees

-- | The job's parent, or the job when it has none.
branchRoot :: Job.JobRead payload -> Job.JobId
branchRoot :: forall payload. JobRead payload -> Int64
branchRoot JobRead payload
job = Int64 -> Maybe Int64 -> Int64
forall a. a -> Maybe a -> a
fromMaybe (JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
Job.primaryKey JobRead payload
job) (JobRead payload -> Maybe Int64
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Maybe Int64
Job.parentId JobRead payload
job)

-- | Delete what a tree or branch cancel names, returning the ids deleted.
cancelJobFor
  :: (MonadArbiter m)
  => FailureKind
  -> Job.JobRead payload
  -> m [Job.JobId]
cancelJobFor :: forall (m :: * -> *) payload.
MonadArbiter m =>
FailureKind -> JobRead payload -> m [Int64]
cancelJobFor FailureKind
kind JobRead payload
job = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  case kind of
    FailureKind
BranchCancelFailure ->
      Text -> Text -> Int64 -> m [Int64]
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m [Int64]
Ops.cancelJobCascade Text
schemaName (JobRead payload -> Text
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> q
Job.queueName JobRead payload
job) (JobRead payload -> Int64
forall payload. JobRead payload -> Int64
branchRoot JobRead payload
job)
    FailureKind
_ -> Text -> Text -> Int64 -> m [Int64]
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Int64 -> m [Int64]
Ops.cancelJobTree Text
schemaName (JobRead payload -> Text
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> q
Job.queueName JobRead payload
job) (JobRead payload -> Int64
forall payload key q insertedAt adm.
Job payload key q insertedAt adm -> key
Job.primaryKey JobRead payload
job)

-- | Write a single job's failure (retry or move to DLQ), for the caller to report
-- once it commits. 'Left' why the write found no row.
handleJobFailure
  :: forall payload m
   . (MonadArbiter m)
  => WorkerConfig m payload
  -> Ops.TreeLocks
  -- ^ Whether the caller already holds the parent and tree locks.
  -> Failure
  -- ^ The handler exception, classified once for the whole batch.
  -> Job.JobRead payload
  -> m (Either Text Outcome)
handleJobFailure :: forall payload (m :: * -> *).
MonadArbiter m =>
WorkerConfig m payload
-> TreeLocks
-> Failure
-> JobRead payload
-> m (Either Text Outcome)
handleJobFailure WorkerConfig m payload
config TreeLocks
locks (Text
errorMsg, FailureKind
failureKind) JobRead payload
job
  -- A batch sibling's cancel takes out the whole tree. Zero rows still means gone.
  | FailureKind -> Bool
cancelsTree FailureKind
failureKind = Outcome -> Either Text Outcome
forall a b. b -> Either a b
Right (Outcome -> Either Text Outcome)
-> ([Int64] -> Outcome) -> [Int64] -> Either Text Outcome
forall b c a. (b -> c) -> (a -> b) -> a -> c
. [Int64] -> Outcome
TreeCancelled ([Int64] -> Either Text Outcome)
-> m [Int64] -> m (Either Text Outcome)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> FailureKind -> JobRead payload -> m [Int64]
forall (m :: * -> *) payload.
MonadArbiter m =>
FailureKind -> JobRead payload -> m [Int64]
cancelJobFor FailureKind
failureKind JobRead payload
job
  | FailureKind
failureKind FailureKind -> FailureKind -> Bool
forall a. Eq a => a -> a -> Bool
== FailureKind
PermanentFailure Bool -> Bool -> Bool
|| JobRead payload -> Int32
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Int32
Job.attempts JobRead payload
job Int32 -> Int32 -> Bool
forall a. Ord a => a -> a -> Bool
>= JobRead payload -> Int32
forall payload. JobRead payload -> Int32
jobMaxAttempts JobRead payload
job = do
      schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
      wrote "no longer available for the dead-letter queue" DeadLettered
        <$> Ops.moveToDLQ locks schemaName (Job.queueName job) errorMsg job
  | Bool
otherwise = do
      backoffSecs <- case FailureKind
failureKind of
        RetryAfterFailure NominalDiffTime
delay -> NominalDiffTime -> m NominalDiffTime
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure NominalDiffTime
delay
        FailureKind
_ -> IO NominalDiffTime -> m NominalDiffTime
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO NominalDiffTime -> m NominalDiffTime)
-> IO NominalDiffTime -> m NominalDiffTime
forall a b. (a -> b) -> a -> b
$ Jitter -> NominalDiffTime -> IO NominalDiffTime
applyJitter (WorkerConfig m payload -> Jitter
forall (m :: * -> *) payload. WorkerConfig m payload -> Jitter
jitter WorkerConfig m payload
config) (BackoffStrategy -> Int32 -> NominalDiffTime
calculateBackoff (WorkerConfig m payload -> BackoffStrategy
forall (m :: * -> *) payload.
WorkerConfig m payload -> BackoffStrategy
backoffStrategy WorkerConfig m payload
config) (JobRead payload -> Int32
forall payload q insertedAt adm.
Job payload Int64 q insertedAt adm -> Int32
Job.attempts JobRead payload
job))
      wrote "no longer available for retry" (Retrying backoffSecs)
        <$> Arb.updateJobForRetry backoffSecs errorMsg job
  where
    -- Nothing written means the job went elsewhere.
    wrote :: a -> b -> a -> Either a b
wrote a
reason b
outcome a
rowsAffected
      | a
rowsAffected a -> a -> Bool
forall a. Eq a => a -> a -> Bool
== a
0 = a -> Either a b
forall a b. a -> Either a b
Left a
reason
      | Bool
otherwise = b -> Either a b
forall a b. b -> Either a b
Right b
outcome