{-# LANGUAGE DeriveAnyClass #-}
{-# LANGUAGE OverloadedStrings #-}

-- | Configuration types for the arbiter worker pool.
module Arbiter.Worker.Config
  ( -- * Worker configuration
    WorkerConfig (..)
  , transactionalWorkerConfig
  , manualWorkerConfig
  , batchedWorkerConfig
  , withHooks
  , withMaintenance
  , HandlerMode (..)
  , handlerBatchSize
  , MaintenanceOp (..)
  , maintenanceOpName
  , ResultOf
  , WorkerConfigException (..)
  , validateWorkerConfig
  , defaultMaintenanceSparseInterval
  , defaultMaintenanceBucketIdle
  , defaultMaintenanceTimeout

    -- * Batch callbacks
  , BatchCallbacks (..)
  , hoistBatchCallbacks

    -- * Worker state
  , WorkerState (..)
  , WorkerRuntime
  , shutdownWorkerPool
  , getWorkerState
  , getListenerReady
  , readEffectiveState

    -- * Internal

    -- | Pool runtime access for the arbiter packages. Not covered by the PVP.
    -- Use 'writePause' to change the pause flag. A direct write to 'pauseVar'
    -- bypasses the epoch guard.
  , writePause
  , writePauseIfCurrent
  , workerStateVar
  , pauseVar
  , pauseEpoch
  , heartbeatSignal
  , pulseHeartbeat
  , listenerReadyVar
  ) where

import Arbiter.Core.Job.Types (JobRead, JobWrite, ObservabilityHooks, andThen, defaultObservabilityHooks)
import Arbiter.Core.MonadArbiter (JobHandler, MonadArbiter, ResultOf)
import Arbiter.Core.Operations (singleJobBatchSize)
import Control.Exception (Exception)
import Control.Monad (void, when)
import Control.Monad.IO.Class (MonadIO, liftIO)
import Data.Aeson (Value, (.=))
import Data.Bifunctor (first)
import Data.Foldable (toList)
import Data.Int (Int64)
import Data.List.NonEmpty (NonEmpty ((:|)))
import Data.Text (Text)
import Data.Text qualified as T
import Data.Time (NominalDiffTime)
import Data.UUID (UUID, toString)
import Data.UUID.V4 qualified as UUID
import Data.Word (Word64)
import Network.HostName (getHostName)
import System.Directory (getTemporaryDirectory)
import UnliftIO (MonadUnliftIO)
import UnliftIO.STM (TMVar, TVar, newEmptyTMVarIO, newTVarIO)
import UnliftIO.STM qualified as STM

import Arbiter.Worker.BackoffStrategy (BackoffStrategy, Jitter (..), exponentialBackoff)
import Arbiter.Worker.Cron.Types (CronJob)
import Arbiter.Worker.Logger (LogConfig, defaultLogConfig, withIdentity)
import Arbiter.Worker.WorkerState (WorkerState (..))

-- | Which reaper op a maintenance report came from.
data MaintenanceOp
  = -- | Refresh the group summary rows.
    RefreshGroups
  | -- | Delete worker rows whose heartbeat is stale.
    SweepStaleWorkers
  | -- | Move exhausted jobs to the DLQ.
    SweepExhaustedJobs
  | -- | Delete force-cancelled jobs whose lease has lapsed.
    SweepCancelledJobs
  | -- | Delete idle rate-limit buckets that are full.
    PruneRateLimitBuckets
  | -- | Rebuild the concurrency counts after a crash truncated them.
    ReconcileConcurrencyStale
  | -- | Recount the concurrency keys, then delete the unused ones.
    ReconcilePruneConcurrency
  | -- | Delete expired archived jobs.
    PurgeArchives
  | -- | Delete DLQ jobs older than their queue's DLQ retention.
    PurgeDLQ
  deriving stock (MaintenanceOp
MaintenanceOp -> MaintenanceOp -> Bounded MaintenanceOp
forall a. a -> a -> Bounded a
$cminBound :: MaintenanceOp
minBound :: MaintenanceOp
$cmaxBound :: MaintenanceOp
maxBound :: MaintenanceOp
Bounded, Int -> MaintenanceOp
MaintenanceOp -> Int
MaintenanceOp -> [MaintenanceOp]
MaintenanceOp -> MaintenanceOp
MaintenanceOp -> MaintenanceOp -> [MaintenanceOp]
MaintenanceOp -> MaintenanceOp -> MaintenanceOp -> [MaintenanceOp]
(MaintenanceOp -> MaintenanceOp)
-> (MaintenanceOp -> MaintenanceOp)
-> (Int -> MaintenanceOp)
-> (MaintenanceOp -> Int)
-> (MaintenanceOp -> [MaintenanceOp])
-> (MaintenanceOp -> MaintenanceOp -> [MaintenanceOp])
-> (MaintenanceOp -> MaintenanceOp -> [MaintenanceOp])
-> (MaintenanceOp
    -> MaintenanceOp -> MaintenanceOp -> [MaintenanceOp])
-> Enum MaintenanceOp
forall a.
(a -> a)
-> (a -> a)
-> (Int -> a)
-> (a -> Int)
-> (a -> [a])
-> (a -> a -> [a])
-> (a -> a -> [a])
-> (a -> a -> a -> [a])
-> Enum a
$csucc :: MaintenanceOp -> MaintenanceOp
succ :: MaintenanceOp -> MaintenanceOp
$cpred :: MaintenanceOp -> MaintenanceOp
pred :: MaintenanceOp -> MaintenanceOp
$ctoEnum :: Int -> MaintenanceOp
toEnum :: Int -> MaintenanceOp
$cfromEnum :: MaintenanceOp -> Int
fromEnum :: MaintenanceOp -> Int
$cenumFrom :: MaintenanceOp -> [MaintenanceOp]
enumFrom :: MaintenanceOp -> [MaintenanceOp]
$cenumFromThen :: MaintenanceOp -> MaintenanceOp -> [MaintenanceOp]
enumFromThen :: MaintenanceOp -> MaintenanceOp -> [MaintenanceOp]
$cenumFromTo :: MaintenanceOp -> MaintenanceOp -> [MaintenanceOp]
enumFromTo :: MaintenanceOp -> MaintenanceOp -> [MaintenanceOp]
$cenumFromThenTo :: MaintenanceOp -> MaintenanceOp -> MaintenanceOp -> [MaintenanceOp]
enumFromThenTo :: MaintenanceOp -> MaintenanceOp -> MaintenanceOp -> [MaintenanceOp]
Enum, MaintenanceOp -> MaintenanceOp -> Bool
(MaintenanceOp -> MaintenanceOp -> Bool)
-> (MaintenanceOp -> MaintenanceOp -> Bool) -> Eq MaintenanceOp
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: MaintenanceOp -> MaintenanceOp -> Bool
== :: MaintenanceOp -> MaintenanceOp -> Bool
$c/= :: MaintenanceOp -> MaintenanceOp -> Bool
/= :: MaintenanceOp -> MaintenanceOp -> Bool
Eq, Eq MaintenanceOp
Eq MaintenanceOp =>
(MaintenanceOp -> MaintenanceOp -> Ordering)
-> (MaintenanceOp -> MaintenanceOp -> Bool)
-> (MaintenanceOp -> MaintenanceOp -> Bool)
-> (MaintenanceOp -> MaintenanceOp -> Bool)
-> (MaintenanceOp -> MaintenanceOp -> Bool)
-> (MaintenanceOp -> MaintenanceOp -> MaintenanceOp)
-> (MaintenanceOp -> MaintenanceOp -> MaintenanceOp)
-> Ord MaintenanceOp
MaintenanceOp -> MaintenanceOp -> Bool
MaintenanceOp -> MaintenanceOp -> Ordering
MaintenanceOp -> MaintenanceOp -> MaintenanceOp
forall a.
Eq a =>
(a -> a -> Ordering)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> Bool)
-> (a -> a -> a)
-> (a -> a -> a)
-> Ord a
$ccompare :: MaintenanceOp -> MaintenanceOp -> Ordering
compare :: MaintenanceOp -> MaintenanceOp -> Ordering
$c< :: MaintenanceOp -> MaintenanceOp -> Bool
< :: MaintenanceOp -> MaintenanceOp -> Bool
$c<= :: MaintenanceOp -> MaintenanceOp -> Bool
<= :: MaintenanceOp -> MaintenanceOp -> Bool
$c> :: MaintenanceOp -> MaintenanceOp -> Bool
> :: MaintenanceOp -> MaintenanceOp -> Bool
$c>= :: MaintenanceOp -> MaintenanceOp -> Bool
>= :: MaintenanceOp -> MaintenanceOp -> Bool
$cmax :: MaintenanceOp -> MaintenanceOp -> MaintenanceOp
max :: MaintenanceOp -> MaintenanceOp -> MaintenanceOp
$cmin :: MaintenanceOp -> MaintenanceOp -> MaintenanceOp
min :: MaintenanceOp -> MaintenanceOp -> MaintenanceOp
Ord, Int -> MaintenanceOp -> ShowS
[MaintenanceOp] -> ShowS
MaintenanceOp -> String
(Int -> MaintenanceOp -> ShowS)
-> (MaintenanceOp -> String)
-> ([MaintenanceOp] -> ShowS)
-> Show MaintenanceOp
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> MaintenanceOp -> ShowS
showsPrec :: Int -> MaintenanceOp -> ShowS
$cshow :: MaintenanceOp -> String
show :: MaintenanceOp -> String
$cshowList :: [MaintenanceOp] -> ShowS
showList :: [MaintenanceOp] -> ShowS
Show)

-- | The op's stable name, used to coordinate replicas and to label its metrics.
maintenanceOpName :: MaintenanceOp -> Text
maintenanceOpName :: MaintenanceOp -> Text
maintenanceOpName MaintenanceOp
operation = case MaintenanceOp
operation of
  MaintenanceOp
RefreshGroups -> Text
"refresh-all-groups"
  MaintenanceOp
SweepStaleWorkers -> Text
"sweep-stale-workers"
  MaintenanceOp
SweepExhaustedJobs -> Text
"sweep-exhausted-jobs"
  MaintenanceOp
SweepCancelledJobs -> Text
"sweep-cancelled-jobs"
  MaintenanceOp
PruneRateLimitBuckets -> Text
"prune-rate-limit-buckets"
  MaintenanceOp
ReconcileConcurrencyStale -> Text
"reconcile-concurrency-stale"
  MaintenanceOp
ReconcilePruneConcurrency -> Text
"reconcile-prune-concurrency"
  MaintenanceOp
PurgeArchives -> Text
"purge-archives"
  MaintenanceOp
PurgeDLQ -> Text
"purge-dlq"

-- | Invalid worker configuration detected before a pool starts.
newtype WorkerConfigException = WorkerConfigException Text
  deriving stock (WorkerConfigException -> WorkerConfigException -> Bool
(WorkerConfigException -> WorkerConfigException -> Bool)
-> (WorkerConfigException -> WorkerConfigException -> Bool)
-> Eq WorkerConfigException
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: WorkerConfigException -> WorkerConfigException -> Bool
== :: WorkerConfigException -> WorkerConfigException -> Bool
$c/= :: WorkerConfigException -> WorkerConfigException -> Bool
/= :: WorkerConfigException -> WorkerConfigException -> Bool
Eq, Int -> WorkerConfigException -> ShowS
[WorkerConfigException] -> ShowS
WorkerConfigException -> String
(Int -> WorkerConfigException -> ShowS)
-> (WorkerConfigException -> String)
-> ([WorkerConfigException] -> ShowS)
-> Show WorkerConfigException
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> WorkerConfigException -> ShowS
showsPrec :: Int -> WorkerConfigException -> ShowS
$cshow :: WorkerConfigException -> String
show :: WorkerConfigException -> String
$cshowList :: [WorkerConfigException] -> ShowS
showList :: [WorkerConfigException] -> ShowS
Show)
  deriving anyclass (Show WorkerConfigException
Typeable WorkerConfigException
(Typeable WorkerConfigException, Show WorkerConfigException) =>
(WorkerConfigException -> SomeException)
-> (SomeException -> Maybe WorkerConfigException)
-> (WorkerConfigException -> String)
-> (WorkerConfigException -> Bool)
-> Exception WorkerConfigException
SomeException -> Maybe WorkerConfigException
WorkerConfigException -> Bool
WorkerConfigException -> String
WorkerConfigException -> SomeException
forall e.
(Typeable e, Show e) =>
(e -> SomeException)
-> (SomeException -> Maybe e)
-> (e -> String)
-> (e -> Bool)
-> Exception e
$ctoException :: WorkerConfigException -> SomeException
toException :: WorkerConfigException -> SomeException
$cfromException :: SomeException -> Maybe WorkerConfigException
fromException :: SomeException -> Maybe WorkerConfigException
$cdisplayException :: WorkerConfigException -> String
displayException :: WorkerConfigException -> String
$cbacktraceDesired :: WorkerConfigException -> Bool
backtraceDesired :: WorkerConfigException -> Bool
Exception)

-- | Mutable state owned by one worker pool. The config constructors allocate it.
data WorkerRuntime = WorkerRuntime
  { WorkerRuntime -> TVar WorkerState
runtimeStateVar :: TVar WorkerState
  , WorkerRuntime -> TVar Bool
runtimePauseVar :: TVar Bool
  , WorkerRuntime -> TVar Word64
runtimePauseEpoch :: TVar Word64
  , WorkerRuntime -> TMVar ()
runtimeHeartbeatSignal :: TMVar ()
  , WorkerRuntime -> TVar Bool
runtimeListenerReadyVar :: TVar Bool
  }

-- | Configuration for a worker pool.
data WorkerConfig m payload = WorkerConfig
  { forall (m :: * -> *) payload. WorkerConfig m payload -> Int
workerCount :: Int
  -- ^ Number of concurrent worker threads.
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> HandlerMode m payload
handlerMode :: HandlerMode m payload
  -- ^ Job handler and claiming strategy. Set by 'transactionalWorkerConfig',
  -- 'manualWorkerConfig' and 'batchedWorkerConfig'.
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
pollInterval :: NominalDiffTime
  -- ^ Longest wait in seconds between dispatcher polls. A NOTIFY or a freed
  -- worker wakes the dispatcher sooner. Default: @5@.
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
visibilityTimeout :: NominalDiffTime
  -- ^ The lease length: seconds a claim holds its job before another worker can claim it.
  -- Must be greater than 'jobHeartbeatInterval'. Default: @60@.
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
jobHeartbeatInterval :: NominalDiffTime
  -- ^ Interval in seconds for extending a job's lease during processing.
  -- Must be less than 'visibilityTimeout'. Default: @30@.
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> Maybe NominalDiffTime
maxJobDuration :: Maybe NominalDiffTime
  -- ^ Interrupt a batch whose claim hooks and handler run longer than this many seconds.
  -- Its unfinalized jobs fail for retry. 'Nothing' sets no bound. Default: @Nothing@.
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
workerHeartbeatInterval :: NominalDiffTime
  -- ^ Minimum gap in seconds between beats. A beat bumps the worker registry heartbeat,
  -- touches the optional liveness file, and reconciles pause state from the DB.
  -- Unless the pool is paused, a beat also waits for a dispatcher claim attempt or a
  -- lease extend. Keep it well below 'workerStaleThreshold'. Default: @10@.
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> BackoffStrategy
backoffStrategy :: BackoffStrategy
  -- ^ Retry backoff strategy. Default: @exponentialBackoff 2 1048576@.
  , forall (m :: * -> *) payload. WorkerConfig m payload -> Jitter
jitter :: Jitter
  -- ^ Jitter strategy for retry delays. Default: 'EqualJitter'.
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> ObservabilityHooks m payload
observabilityHooks :: ObservabilityHooks m payload
  -- ^ Callbacks for metrics or tracing.
  -- Default: 'Arbiter.Core.Job.Types.defaultObservabilityHooks'.
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> MaintenanceOp -> Int64 -> m ()
onMaintenance :: MaintenanceOp -> Int64 -> m ()
  -- ^ Called after a reaper op this pool won the gate for, with the rows it touched.
  -- Reaper work is schema-wide and carries no queue. Default: no-op.
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> WorkerRuntime
workerRuntime :: WorkerRuntime
  -- ^ Mutable lifecycle state allocated for this pool.
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> Maybe String
livenessFile :: Maybe FilePath
  -- ^ When set, each beat touches this file, for a file-based liveness probe.
  -- See 'workerHeartbeatInterval' for when a beat occurs.
  -- Default: @arbiter-worker-\<uuid\>@ in the system temporary directory, from the UUID
  -- the constructor generated. Overriding 'workerId' does not change it.
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> Maybe NominalDiffTime
gracefulShutdownTimeout :: Maybe NominalDiffTime
  -- ^ Seconds a graceful shutdown waits on in-flight jobs before force-exiting.
  -- 'Nothing' waits indefinitely. Default: @Just 30@.
  , forall (m :: * -> *) payload. WorkerConfig m payload -> LogConfig
logConfig :: LogConfig
  -- ^ Structured JSON log destination, level, and additional context.
  -- Default: @Info@ to stdout, with a @worker_id@ context pair from the UUID the
  -- constructor generated. Overriding 'workerId' does not change it.
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> [CronJob payload]
cronJobs :: [CronJob payload]
  -- ^ Cron schedules. A non-empty list gives the pool a scheduler thread, which reads
  -- the @cron_schedules@ table each tick for runtime overrides. Default: @[]@.
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
reaperInterval :: NominalDiffTime
  -- ^ Sleep in seconds between reaper passes, and the gap between runs of each ordinary
  -- reaper operation. Default: @300@ (5 minutes).
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
reaperSparseInterval :: NominalDiffTime
  -- ^ Gap in seconds between runs of 'PruneRateLimitBuckets' and
  -- 'ReconcilePruneConcurrency'. Default: 'defaultMaintenanceSparseInterval'.
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
reaperBucketIdle :: NominalDiffTime
  -- ^ Idle age in seconds at which the reaper prunes a full rate-limit bucket. Default: 'defaultMaintenanceBucketIdle'.
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
reaperTimeout :: NominalDiffTime
  -- ^ Abort any single reaper statement that runs longer than this many seconds. Default: 'defaultMaintenanceTimeout'.
  , forall (m :: * -> *) payload. WorkerConfig m payload -> UUID
workerId :: UUID
  -- ^ This worker's registry id. Default: a fresh random UUID.
  , forall (m :: * -> *) payload. WorkerConfig m payload -> Maybe Text
workerHost :: Maybe Text
  -- ^ Hostname recorded in the worker registry. Default: this machine's hostname.
  , forall (m :: * -> *) payload. WorkerConfig m payload -> Maybe Value
workerMetadata :: Maybe Value
  -- ^ Arbitrary JSONB metadata for the worker registry row (image tag,
  -- git SHA, deploy id, etc.). Default: @Nothing@.
  , forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
workerStaleThreshold :: NominalDiffTime
  -- ^ The reaper's stale-worker sweep removes this worker's row when its heartbeat
  -- is older than this many seconds. Keep it well above 'workerHeartbeatInterval',
  -- 'pollInterval' and 'jobHeartbeatInterval'. Default: @300@ (5 minutes).
  }

-- | Per-job finalizers passed to a batched handler. When the handler returns, untouched
-- jobs are reprocessed after the lease expires, and their attempt is used. When it throws,
-- untouched jobs fail. A 'Arbiter.Core.Exceptions.JobNackException' releases them with
-- their attempt restored. A 'Arbiter.Core.Exceptions.JobScopedFailure' that names jobs
-- in the batch fails only those jobs. A 'Arbiter.Core.Exceptions.JobGoneException' that
-- names jobs in the batch reports only those jobs as unavailable, or as cancelled
-- after a force-cancel. Both release the rest. The @With@ variants store a result for
-- the job's parent rollup or its archive entry.
--
-- Each callback commits in its own transaction. Inside
-- 'Arbiter.Core.MonadArbiter.withDbTransaction', it runs as a savepoint in the
-- outer transaction. The success hook fires at savepoint release. An outer
-- rollback reprocesses the job after its lease expires.
--
-- 'ack', 'ackWith' and 'spawn' abort the handler if another worker holds the
-- job. The abort nacks the siblings still unfinalized.
data BatchCallbacks m payload result = BatchCallbacks
  { forall (m :: * -> *) payload result.
BatchCallbacks m payload result -> JobRead payload -> m ()
ack :: JobRead payload -> m ()
  -- ^ Ack and fire 'Arbiter.Core.Job.Types.onJobSuccess', storing no result.
  -- Available on any queue. A job acked this way is absent from its parent
  -- rollup's child results and leaves its archive entry's result @NULL@.
  , forall (m :: * -> *) payload result.
BatchCallbacks m payload result
-> JobRead payload -> result -> m ()
ackWith :: JobRead payload -> result -> m ()
  -- ^ Ack, store the result for the parent rollup or the job's archive entry,
  -- fire 'Arbiter.Core.Job.Types.onJobSuccess'.
  , forall (m :: * -> *) payload result.
BatchCallbacks m payload result -> [JobRead payload] -> m ()
ackAll :: [JobRead payload] -> m ()
  -- ^ Bulk-'ack' in one parent-aware transaction, storing no results. Fires
  -- 'Arbiter.Core.Job.Types.onJobSuccess' per acked job. A job another worker
  -- holds is reported and skipped. The handler continues.
  , forall (m :: * -> *) payload result.
BatchCallbacks m payload result
-> [(JobRead payload, result)] -> m ()
ackAllWith :: [(JobRead payload, result)] -> m ()
  -- ^ 'ackAll' storing each job's result for its parent rollup or archive entry.
  , forall (m :: * -> *) payload result.
BatchCallbacks m payload result -> JobRead payload -> Text -> m ()
failRetry :: JobRead payload -> Text -> m ()
  -- ^ Retry with backoff, then DLQ at the job's @maxAttempts@. Fires
  -- 'Arbiter.Core.Job.Types.onJobFailure', then 'Arbiter.Core.Job.Types.onJobRetry' or
  -- 'Arbiter.Core.Job.Types.onJobFailedAndMovedToDLQ'.
  , forall (m :: * -> *) payload result.
BatchCallbacks m payload result
-> JobRead payload -> NominalDiffTime -> Text -> m ()
failRetryAfter :: JobRead payload -> NominalDiffTime -> Text -> m ()
  -- ^ 'failRetry' after a delay in seconds in place of the backoff.
  , forall (m :: * -> *) payload result.
BatchCallbacks m payload result -> JobRead payload -> Text -> m ()
failPermanent :: JobRead payload -> Text -> m ()
  -- ^ Straight to the DLQ. Fires 'Arbiter.Core.Job.Types.onJobFailure' and
  -- 'Arbiter.Core.Job.Types.onJobFailedAndMovedToDLQ'.
  , forall (m :: * -> *) payload result.
BatchCallbacks m payload result -> JobRead payload -> Text -> m ()
cancelBranch :: JobRead payload -> Text -> m ()
  -- ^ Cancel the job's branch: its parent with all descendants, or the job's own
  -- subtree when it has no parent. Fires 'Arbiter.Core.Job.Types.onJobCancelled'.
  , forall (m :: * -> *) payload result.
BatchCallbacks m payload result -> JobRead payload -> Text -> m ()
cancelTree :: JobRead payload -> Text -> m ()
  -- ^ Cancel the whole tree from the root down. Fires
  -- 'Arbiter.Core.Job.Types.onJobCancelled'.
  , forall (m :: * -> *) payload result.
BatchCallbacks m payload result -> JobRead payload -> m ()
nack :: JobRead payload -> m ()
  -- ^ Reprocess after the lease expires. Records no failure and consumes no attempt.
  -- Fires no hook unless the job is no longer this worker's.
  , forall (m :: * -> *) payload result.
BatchCallbacks m payload result
-> JobRead payload -> NominalDiffTime -> m ()
nackAfter :: JobRead payload -> NominalDiffTime -> m ()
  -- ^ 'nack' that reprocesses after a delay in seconds in place of the lease's end.
  , forall (m :: * -> *) payload result.
BatchCallbacks m payload result
-> JobRead payload -> NonEmpty (JobWrite payload) -> m ()
spawn :: JobRead payload -> NonEmpty (JobWrite payload) -> m ()
  -- ^ Insert children under this job and suspend it, in one transaction. Fires
  -- 'Arbiter.Core.Job.Types.onJobSuccess'.
  }

-- | Callbacks for @m@, called from @n@ through the given natural transformation.
-- It must be a monad morphism that keeps the worker's connection and schema. Each
-- callback then joins the handler's transaction.
hoistBatchCallbacks :: (forall a. m a -> n a) -> BatchCallbacks m payload result -> BatchCallbacks n payload result
hoistBatchCallbacks :: forall (m :: * -> *) (n :: * -> *) payload result.
(forall a. m a -> n a)
-> BatchCallbacks m payload result
-> BatchCallbacks n payload result
hoistBatchCallbacks forall a. m a -> n a
nat BatchCallbacks m payload result
callbacks =
  BatchCallbacks
    { ack :: JobRead payload -> n ()
ack = m () -> n ()
forall a. m a -> n a
nat (m () -> n ())
-> (JobRead payload -> m ()) -> JobRead payload -> n ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. BatchCallbacks m payload result -> JobRead payload -> m ()
forall (m :: * -> *) payload result.
BatchCallbacks m payload result -> JobRead payload -> m ()
ack BatchCallbacks m payload result
callbacks
    , ackWith :: JobRead payload -> result -> n ()
ackWith = \JobRead payload
job result
result -> m () -> n ()
forall a. m a -> n a
nat (BatchCallbacks m payload result
-> JobRead payload -> result -> m ()
forall (m :: * -> *) payload result.
BatchCallbacks m payload result
-> JobRead payload -> result -> m ()
ackWith BatchCallbacks m payload result
callbacks JobRead payload
job result
result)
    , ackAll :: [JobRead payload] -> n ()
ackAll = m () -> n ()
forall a. m a -> n a
nat (m () -> n ())
-> ([JobRead payload] -> m ()) -> [JobRead payload] -> n ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. BatchCallbacks m payload result -> [JobRead payload] -> m ()
forall (m :: * -> *) payload result.
BatchCallbacks m payload result -> [JobRead payload] -> m ()
ackAll BatchCallbacks m payload result
callbacks
    , ackAllWith :: [(JobRead payload, result)] -> n ()
ackAllWith = m () -> n ()
forall a. m a -> n a
nat (m () -> n ())
-> ([(JobRead payload, result)] -> m ())
-> [(JobRead payload, result)]
-> n ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. BatchCallbacks m payload result
-> [(JobRead payload, result)] -> m ()
forall (m :: * -> *) payload result.
BatchCallbacks m payload result
-> [(JobRead payload, result)] -> m ()
ackAllWith BatchCallbacks m payload result
callbacks
    , failRetry :: JobRead payload -> Text -> n ()
failRetry = \JobRead payload
job Text
msg -> m () -> n ()
forall a. m a -> n a
nat (BatchCallbacks m payload result -> JobRead payload -> Text -> m ()
forall (m :: * -> *) payload result.
BatchCallbacks m payload result -> JobRead payload -> Text -> m ()
failRetry BatchCallbacks m payload result
callbacks JobRead payload
job Text
msg)
    , failRetryAfter :: JobRead payload -> NominalDiffTime -> Text -> n ()
failRetryAfter = \JobRead payload
job NominalDiffTime
delay Text
msg -> m () -> n ()
forall a. m a -> n a
nat (BatchCallbacks m payload result
-> JobRead payload -> NominalDiffTime -> Text -> m ()
forall (m :: * -> *) payload result.
BatchCallbacks m payload result
-> JobRead payload -> NominalDiffTime -> Text -> m ()
failRetryAfter BatchCallbacks m payload result
callbacks JobRead payload
job NominalDiffTime
delay Text
msg)
    , failPermanent :: JobRead payload -> Text -> n ()
failPermanent = \JobRead payload
job Text
msg -> m () -> n ()
forall a. m a -> n a
nat (BatchCallbacks m payload result -> JobRead payload -> Text -> m ()
forall (m :: * -> *) payload result.
BatchCallbacks m payload result -> JobRead payload -> Text -> m ()
failPermanent BatchCallbacks m payload result
callbacks JobRead payload
job Text
msg)
    , cancelBranch :: JobRead payload -> Text -> n ()
cancelBranch = \JobRead payload
job Text
msg -> m () -> n ()
forall a. m a -> n a
nat (BatchCallbacks m payload result -> JobRead payload -> Text -> m ()
forall (m :: * -> *) payload result.
BatchCallbacks m payload result -> JobRead payload -> Text -> m ()
cancelBranch BatchCallbacks m payload result
callbacks JobRead payload
job Text
msg)
    , cancelTree :: JobRead payload -> Text -> n ()
cancelTree = \JobRead payload
job Text
msg -> m () -> n ()
forall a. m a -> n a
nat (BatchCallbacks m payload result -> JobRead payload -> Text -> m ()
forall (m :: * -> *) payload result.
BatchCallbacks m payload result -> JobRead payload -> Text -> m ()
cancelTree BatchCallbacks m payload result
callbacks JobRead payload
job Text
msg)
    , nack :: JobRead payload -> n ()
nack = m () -> n ()
forall a. m a -> n a
nat (m () -> n ())
-> (JobRead payload -> m ()) -> JobRead payload -> n ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. BatchCallbacks m payload result -> JobRead payload -> m ()
forall (m :: * -> *) payload result.
BatchCallbacks m payload result -> JobRead payload -> m ()
nack BatchCallbacks m payload result
callbacks
    , nackAfter :: JobRead payload -> NominalDiffTime -> n ()
nackAfter = \JobRead payload
job NominalDiffTime
delay -> m () -> n ()
forall a. m a -> n a
nat (BatchCallbacks m payload result
-> JobRead payload -> NominalDiffTime -> m ()
forall (m :: * -> *) payload result.
BatchCallbacks m payload result
-> JobRead payload -> NominalDiffTime -> m ()
nackAfter BatchCallbacks m payload result
callbacks JobRead payload
job NominalDiffTime
delay)
    , spawn :: JobRead payload -> NonEmpty (JobWrite payload) -> n ()
spawn = \JobRead payload
job NonEmpty (JobWrite payload)
children -> m () -> n ()
forall a. m a -> n a
nat (BatchCallbacks m payload result
-> JobRead payload -> NonEmpty (JobWrite payload) -> m ()
forall (m :: * -> *) payload result.
BatchCallbacks m payload result
-> JobRead payload -> NonEmpty (JobWrite payload) -> m ()
spawn BatchCallbacks m payload result
callbacks JobRead payload
job NonEmpty (JobWrite payload)
children)
    }

-- | Job claim and handler mode. Set by 'transactionalWorkerConfig',
-- 'manualWorkerConfig' and 'batchedWorkerConfig'.
data HandlerMode m payload
  = -- | Automatic single-job mode: claim one job per group and run the handler
    -- in a worker transaction, storing its result and acking atomically.
    SingleJobMode (JobHandler m payload (ResultOf m payload))
  | -- | Batched callback mode: claim batches of up to N jobs and hand each batch to
    -- the handler with a 'BatchCallbacks' record to finalize each job. No worker
    -- transaction. A batch is one group, or ungrouped jobs. Batch size 1 is the manual
    -- single-job case.
    BatchedJobsMode
      Int
      -- ^ N, the maximum jobs in a batch.
      (NonEmpty (JobRead payload) -> BatchCallbacks m payload (ResultOf m payload) -> m ())

-- | The maximum jobs in a batch. A batch is one group, or ungrouped jobs.
handlerBatchSize :: WorkerConfig m payload -> Int
handlerBatchSize :: forall (m :: * -> *) payload. WorkerConfig m payload -> Int
handlerBatchSize WorkerConfig m payload
config = case WorkerConfig m payload -> HandlerMode m payload
forall (m :: * -> *) payload.
WorkerConfig m payload -> HandlerMode m payload
handlerMode WorkerConfig m payload
config of
  SingleJobMode JobHandler m payload (ResultOf m payload)
_ -> Int
singleJobBatchSize
  BatchedJobsMode Int
batchSize NonEmpty (JobRead payload)
-> BatchCallbacks m payload (ResultOf m payload) -> m ()
_ -> Int
batchSize

-- | Validate all invariants required for safe worker execution. Return all
-- violations.
validateWorkerConfig :: WorkerConfig m payload -> Either Text ()
validateWorkerConfig :: forall (m :: * -> *) payload.
WorkerConfig m payload -> Either Text ()
validateWorkerConfig WorkerConfig m payload
config =
  let Validation Either (NonEmpty Text) ()
result = WorkerConfig m payload -> Validation (WorkerConfig m payload)
forall (m :: * -> *) payload.
WorkerConfig m payload -> Validation (WorkerConfig m payload)
fieldInvariants WorkerConfig m payload
config Validation (WorkerConfig m payload)
-> Validation () -> Validation ()
forall a b. Validation a -> Validation b -> Validation b
forall (f :: * -> *) a b. Applicative f => f a -> f b -> f b
*> WorkerConfig m payload -> Validation ()
forall (m :: * -> *) payload.
WorkerConfig m payload -> Validation ()
crossFieldInvariants WorkerConfig m payload
config
   in (NonEmpty Text -> Text)
-> Either (NonEmpty Text) () -> Either Text ()
forall a b c. (a -> b) -> Either a c -> Either b c
forall (p :: * -> * -> *) a b c.
Bifunctor p =>
(a -> b) -> p a c -> p b c
first (Text -> [Text] -> Text
T.intercalate Text
"; " ([Text] -> Text)
-> (NonEmpty Text -> [Text]) -> NonEmpty Text -> Text
forall b c a. (b -> c) -> (a -> b) -> a -> c
. NonEmpty Text -> [Text]
forall a. NonEmpty a -> [a]
forall (t :: * -> *) a. Foldable t => t a -> [a]
toList) Either (NonEmpty Text) ()
result

-- | Rebuilds the config with every field checked or waived. A field added
-- to 'WorkerConfig' is a type error here until it is classified.
fieldInvariants :: WorkerConfig m payload -> Validation (WorkerConfig m payload)
fieldInvariants :: forall (m :: * -> *) payload.
WorkerConfig m payload -> Validation (WorkerConfig m payload)
fieldInvariants WorkerConfig m payload
config =
  Int
-> HandlerMode m payload
-> NominalDiffTime
-> NominalDiffTime
-> NominalDiffTime
-> Maybe NominalDiffTime
-> NominalDiffTime
-> BackoffStrategy
-> Jitter
-> ObservabilityHooks m payload
-> (MaintenanceOp -> Int64 -> m ())
-> WorkerRuntime
-> Maybe String
-> Maybe NominalDiffTime
-> LogConfig
-> [CronJob payload]
-> NominalDiffTime
-> NominalDiffTime
-> NominalDiffTime
-> NominalDiffTime
-> UUID
-> Maybe Text
-> Maybe Value
-> NominalDiffTime
-> WorkerConfig m payload
forall (m :: * -> *) payload.
Int
-> HandlerMode m payload
-> NominalDiffTime
-> NominalDiffTime
-> NominalDiffTime
-> Maybe NominalDiffTime
-> NominalDiffTime
-> BackoffStrategy
-> Jitter
-> ObservabilityHooks m payload
-> (MaintenanceOp -> Int64 -> m ())
-> WorkerRuntime
-> Maybe String
-> Maybe NominalDiffTime
-> LogConfig
-> [CronJob payload]
-> NominalDiffTime
-> NominalDiffTime
-> NominalDiffTime
-> NominalDiffTime
-> UUID
-> Maybe Text
-> Maybe Value
-> NominalDiffTime
-> WorkerConfig m payload
WorkerConfig
    (Int
 -> HandlerMode m payload
 -> NominalDiffTime
 -> NominalDiffTime
 -> NominalDiffTime
 -> Maybe NominalDiffTime
 -> NominalDiffTime
 -> BackoffStrategy
 -> Jitter
 -> ObservabilityHooks m payload
 -> (MaintenanceOp -> Int64 -> m ())
 -> WorkerRuntime
 -> Maybe String
 -> Maybe NominalDiffTime
 -> LogConfig
 -> [CronJob payload]
 -> NominalDiffTime
 -> NominalDiffTime
 -> NominalDiffTime
 -> NominalDiffTime
 -> UUID
 -> Maybe Text
 -> Maybe Value
 -> NominalDiffTime
 -> WorkerConfig m payload)
-> Validation Int
-> Validation
     (HandlerMode m payload
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> Maybe NominalDiffTime
      -> NominalDiffTime
      -> BackoffStrategy
      -> Jitter
      -> ObservabilityHooks m payload
      -> (MaintenanceOp -> Int64 -> m ())
      -> WorkerRuntime
      -> Maybe String
      -> Maybe NominalDiffTime
      -> LogConfig
      -> [CronJob payload]
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Text -> Int -> Validation Int
forall a. (Num a, Ord a) => Text -> a -> Validation a
positive Text
"workerCount" (WorkerConfig m payload -> Int
forall (m :: * -> *) payload. WorkerConfig m payload -> Int
workerCount WorkerConfig m payload
config)
    Validation
  (HandlerMode m payload
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> Maybe NominalDiffTime
   -> NominalDiffTime
   -> BackoffStrategy
   -> Jitter
   -> ObservabilityHooks m payload
   -> (MaintenanceOp -> Int64 -> m ())
   -> WorkerRuntime
   -> Maybe String
   -> Maybe NominalDiffTime
   -> LogConfig
   -> [CronJob payload]
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation (HandlerMode m payload)
-> Validation
     (NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> Maybe NominalDiffTime
      -> NominalDiffTime
      -> BackoffStrategy
      -> Jitter
      -> ObservabilityHooks m payload
      -> (MaintenanceOp -> Int64 -> m ())
      -> WorkerRuntime
      -> Maybe String
      -> Maybe NominalDiffTime
      -> LogConfig
      -> [CronJob payload]
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> (WorkerConfig m payload -> HandlerMode m payload
forall (m :: * -> *) payload.
WorkerConfig m payload -> HandlerMode m payload
handlerMode WorkerConfig m payload
config HandlerMode m payload
-> Validation Int -> Validation (HandlerMode m payload)
forall a b. a -> Validation b -> Validation a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ Text -> Int -> Validation Int
forall a. (Num a, Ord a) => Text -> a -> Validation a
positive Text
"handler batch size" (WorkerConfig m payload -> Int
forall (m :: * -> *) payload. WorkerConfig m payload -> Int
handlerBatchSize WorkerConfig m payload
config))
    Validation
  (NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> Maybe NominalDiffTime
   -> NominalDiffTime
   -> BackoffStrategy
   -> Jitter
   -> ObservabilityHooks m payload
   -> (MaintenanceOp -> Int64 -> m ())
   -> WorkerRuntime
   -> Maybe String
   -> Maybe NominalDiffTime
   -> LogConfig
   -> [CronJob payload]
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation NominalDiffTime
-> Validation
     (NominalDiffTime
      -> NominalDiffTime
      -> Maybe NominalDiffTime
      -> NominalDiffTime
      -> BackoffStrategy
      -> Jitter
      -> ObservabilityHooks m payload
      -> (MaintenanceOp -> Int64 -> m ())
      -> WorkerRuntime
      -> Maybe String
      -> Maybe NominalDiffTime
      -> LogConfig
      -> [CronJob payload]
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> NominalDiffTime -> Validation NominalDiffTime
forall a. (Num a, Ord a) => Text -> a -> Validation a
positive Text
"pollInterval" (WorkerConfig m payload -> NominalDiffTime
forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
pollInterval WorkerConfig m payload
config)
    Validation
  (NominalDiffTime
   -> NominalDiffTime
   -> Maybe NominalDiffTime
   -> NominalDiffTime
   -> BackoffStrategy
   -> Jitter
   -> ObservabilityHooks m payload
   -> (MaintenanceOp -> Int64 -> m ())
   -> WorkerRuntime
   -> Maybe String
   -> Maybe NominalDiffTime
   -> LogConfig
   -> [CronJob payload]
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation NominalDiffTime
-> Validation
     (NominalDiffTime
      -> Maybe NominalDiffTime
      -> NominalDiffTime
      -> BackoffStrategy
      -> Jitter
      -> ObservabilityHooks m payload
      -> (MaintenanceOp -> Int64 -> m ())
      -> WorkerRuntime
      -> Maybe String
      -> Maybe NominalDiffTime
      -> LogConfig
      -> [CronJob payload]
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> NominalDiffTime -> Validation NominalDiffTime
forall a. (Num a, Ord a) => Text -> a -> Validation a
positive Text
"visibilityTimeout" (WorkerConfig m payload -> NominalDiffTime
forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
visibilityTimeout WorkerConfig m payload
config)
    Validation
  (NominalDiffTime
   -> Maybe NominalDiffTime
   -> NominalDiffTime
   -> BackoffStrategy
   -> Jitter
   -> ObservabilityHooks m payload
   -> (MaintenanceOp -> Int64 -> m ())
   -> WorkerRuntime
   -> Maybe String
   -> Maybe NominalDiffTime
   -> LogConfig
   -> [CronJob payload]
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation NominalDiffTime
-> Validation
     (Maybe NominalDiffTime
      -> NominalDiffTime
      -> BackoffStrategy
      -> Jitter
      -> ObservabilityHooks m payload
      -> (MaintenanceOp -> Int64 -> m ())
      -> WorkerRuntime
      -> Maybe String
      -> Maybe NominalDiffTime
      -> LogConfig
      -> [CronJob payload]
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> NominalDiffTime -> Validation NominalDiffTime
forall a. (Num a, Ord a) => Text -> a -> Validation a
positive Text
"jobHeartbeatInterval" (WorkerConfig m payload -> NominalDiffTime
forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
jobHeartbeatInterval WorkerConfig m payload
config)
    Validation
  (Maybe NominalDiffTime
   -> NominalDiffTime
   -> BackoffStrategy
   -> Jitter
   -> ObservabilityHooks m payload
   -> (MaintenanceOp -> Int64 -> m ())
   -> WorkerRuntime
   -> Maybe String
   -> Maybe NominalDiffTime
   -> LogConfig
   -> [CronJob payload]
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation (Maybe NominalDiffTime)
-> Validation
     (NominalDiffTime
      -> BackoffStrategy
      -> Jitter
      -> ObservabilityHooks m payload
      -> (MaintenanceOp -> Int64 -> m ())
      -> WorkerRuntime
      -> Maybe String
      -> Maybe NominalDiffTime
      -> LogConfig
      -> [CronJob payload]
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> (NominalDiffTime -> Validation NominalDiffTime)
-> Maybe NominalDiffTime -> Validation (Maybe NominalDiffTime)
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b) -> Maybe a -> f (Maybe b)
traverse (Text -> NominalDiffTime -> Validation NominalDiffTime
forall a. (Num a, Ord a) => Text -> a -> Validation a
positive Text
"maxJobDuration") (WorkerConfig m payload -> Maybe NominalDiffTime
forall (m :: * -> *) payload.
WorkerConfig m payload -> Maybe NominalDiffTime
maxJobDuration WorkerConfig m payload
config)
    Validation
  (NominalDiffTime
   -> BackoffStrategy
   -> Jitter
   -> ObservabilityHooks m payload
   -> (MaintenanceOp -> Int64 -> m ())
   -> WorkerRuntime
   -> Maybe String
   -> Maybe NominalDiffTime
   -> LogConfig
   -> [CronJob payload]
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation NominalDiffTime
-> Validation
     (BackoffStrategy
      -> Jitter
      -> ObservabilityHooks m payload
      -> (MaintenanceOp -> Int64 -> m ())
      -> WorkerRuntime
      -> Maybe String
      -> Maybe NominalDiffTime
      -> LogConfig
      -> [CronJob payload]
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> NominalDiffTime -> Validation NominalDiffTime
forall a. (Num a, Ord a) => Text -> a -> Validation a
positive Text
"workerHeartbeatInterval" (WorkerConfig m payload -> NominalDiffTime
forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
workerHeartbeatInterval WorkerConfig m payload
config)
    Validation
  (BackoffStrategy
   -> Jitter
   -> ObservabilityHooks m payload
   -> (MaintenanceOp -> Int64 -> m ())
   -> WorkerRuntime
   -> Maybe String
   -> Maybe NominalDiffTime
   -> LogConfig
   -> [CronJob payload]
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation BackoffStrategy
-> Validation
     (Jitter
      -> ObservabilityHooks m payload
      -> (MaintenanceOp -> Int64 -> m ())
      -> WorkerRuntime
      -> Maybe String
      -> Maybe NominalDiffTime
      -> LogConfig
      -> [CronJob payload]
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> BackoffStrategy -> Validation BackoffStrategy
forall a. a -> Validation a
waived (WorkerConfig m payload -> BackoffStrategy
forall (m :: * -> *) payload.
WorkerConfig m payload -> BackoffStrategy
backoffStrategy WorkerConfig m payload
config)
    Validation
  (Jitter
   -> ObservabilityHooks m payload
   -> (MaintenanceOp -> Int64 -> m ())
   -> WorkerRuntime
   -> Maybe String
   -> Maybe NominalDiffTime
   -> LogConfig
   -> [CronJob payload]
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation Jitter
-> Validation
     (ObservabilityHooks m payload
      -> (MaintenanceOp -> Int64 -> m ())
      -> WorkerRuntime
      -> Maybe String
      -> Maybe NominalDiffTime
      -> LogConfig
      -> [CronJob payload]
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Jitter -> Validation Jitter
forall a. a -> Validation a
waived (WorkerConfig m payload -> Jitter
forall (m :: * -> *) payload. WorkerConfig m payload -> Jitter
jitter WorkerConfig m payload
config)
    Validation
  (ObservabilityHooks m payload
   -> (MaintenanceOp -> Int64 -> m ())
   -> WorkerRuntime
   -> Maybe String
   -> Maybe NominalDiffTime
   -> LogConfig
   -> [CronJob payload]
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation (ObservabilityHooks m payload)
-> Validation
     ((MaintenanceOp -> Int64 -> m ())
      -> WorkerRuntime
      -> Maybe String
      -> Maybe NominalDiffTime
      -> LogConfig
      -> [CronJob payload]
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> ObservabilityHooks m payload
-> Validation (ObservabilityHooks m payload)
forall a. a -> Validation a
waived (WorkerConfig m payload -> ObservabilityHooks m payload
forall (m :: * -> *) payload.
WorkerConfig m payload -> ObservabilityHooks m payload
observabilityHooks WorkerConfig m payload
config)
    Validation
  ((MaintenanceOp -> Int64 -> m ())
   -> WorkerRuntime
   -> Maybe String
   -> Maybe NominalDiffTime
   -> LogConfig
   -> [CronJob payload]
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation (MaintenanceOp -> Int64 -> m ())
-> Validation
     (WorkerRuntime
      -> Maybe String
      -> Maybe NominalDiffTime
      -> LogConfig
      -> [CronJob payload]
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> (MaintenanceOp -> Int64 -> m ())
-> Validation (MaintenanceOp -> Int64 -> m ())
forall a. a -> Validation a
waived (WorkerConfig m payload -> MaintenanceOp -> Int64 -> m ()
forall (m :: * -> *) payload.
WorkerConfig m payload -> MaintenanceOp -> Int64 -> m ()
onMaintenance WorkerConfig m payload
config)
    Validation
  (WorkerRuntime
   -> Maybe String
   -> Maybe NominalDiffTime
   -> LogConfig
   -> [CronJob payload]
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation WorkerRuntime
-> Validation
     (Maybe String
      -> Maybe NominalDiffTime
      -> LogConfig
      -> [CronJob payload]
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> WorkerRuntime -> Validation WorkerRuntime
forall a. a -> Validation a
waived (WorkerConfig m payload -> WorkerRuntime
forall (m :: * -> *) payload.
WorkerConfig m payload -> WorkerRuntime
workerRuntime WorkerConfig m payload
config)
    Validation
  (Maybe String
   -> Maybe NominalDiffTime
   -> LogConfig
   -> [CronJob payload]
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation (Maybe String)
-> Validation
     (Maybe NominalDiffTime
      -> LogConfig
      -> [CronJob payload]
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Maybe String -> Validation (Maybe String)
forall a. a -> Validation a
waived (WorkerConfig m payload -> Maybe String
forall (m :: * -> *) payload.
WorkerConfig m payload -> Maybe String
livenessFile WorkerConfig m payload
config)
    Validation
  (Maybe NominalDiffTime
   -> LogConfig
   -> [CronJob payload]
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation (Maybe NominalDiffTime)
-> Validation
     (LogConfig
      -> [CronJob payload]
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> (NominalDiffTime -> Validation NominalDiffTime)
-> Maybe NominalDiffTime -> Validation (Maybe NominalDiffTime)
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b) -> Maybe a -> f (Maybe b)
traverse (Text -> NominalDiffTime -> Validation NominalDiffTime
forall a. (Num a, Ord a) => Text -> a -> Validation a
nonNegative Text
"gracefulShutdownTimeout") (WorkerConfig m payload -> Maybe NominalDiffTime
forall (m :: * -> *) payload.
WorkerConfig m payload -> Maybe NominalDiffTime
gracefulShutdownTimeout WorkerConfig m payload
config)
    Validation
  (LogConfig
   -> [CronJob payload]
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation LogConfig
-> Validation
     ([CronJob payload]
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> LogConfig -> Validation LogConfig
forall a. a -> Validation a
waived (WorkerConfig m payload -> LogConfig
forall (m :: * -> *) payload. WorkerConfig m payload -> LogConfig
logConfig WorkerConfig m payload
config)
    Validation
  ([CronJob payload]
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation [CronJob payload]
-> Validation
     (NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> [CronJob payload] -> Validation [CronJob payload]
forall a. a -> Validation a
waived (WorkerConfig m payload -> [CronJob payload]
forall (m :: * -> *) payload.
WorkerConfig m payload -> [CronJob payload]
cronJobs WorkerConfig m payload
config)
    Validation
  (NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation NominalDiffTime
-> Validation
     (NominalDiffTime
      -> NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> NominalDiffTime -> Validation NominalDiffTime
forall a. (Num a, Ord a) => Text -> a -> Validation a
positive Text
"reaperInterval" (WorkerConfig m payload -> NominalDiffTime
forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
reaperInterval WorkerConfig m payload
config)
    Validation
  (NominalDiffTime
   -> NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation NominalDiffTime
-> Validation
     (NominalDiffTime
      -> NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> NominalDiffTime -> Validation NominalDiffTime
forall a. (Num a, Ord a) => Text -> a -> Validation a
positive Text
"reaperSparseInterval" (WorkerConfig m payload -> NominalDiffTime
forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
reaperSparseInterval WorkerConfig m payload
config)
    Validation
  (NominalDiffTime
   -> NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation NominalDiffTime
-> Validation
     (NominalDiffTime
      -> UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> NominalDiffTime -> Validation NominalDiffTime
forall a. (Num a, Ord a) => Text -> a -> Validation a
positive Text
"reaperBucketIdle" (WorkerConfig m payload -> NominalDiffTime
forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
reaperBucketIdle WorkerConfig m payload
config)
    Validation
  (NominalDiffTime
   -> UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation NominalDiffTime
-> Validation
     (UUID
      -> Maybe Text
      -> Maybe Value
      -> NominalDiffTime
      -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> NominalDiffTime -> Validation NominalDiffTime
forall a. (Num a, Ord a) => Text -> a -> Validation a
positive Text
"reaperTimeout" (WorkerConfig m payload -> NominalDiffTime
forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
reaperTimeout WorkerConfig m payload
config)
    Validation
  (UUID
   -> Maybe Text
   -> Maybe Value
   -> NominalDiffTime
   -> WorkerConfig m payload)
-> Validation UUID
-> Validation
     (Maybe Text
      -> Maybe Value -> NominalDiffTime -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> UUID -> Validation UUID
forall a. a -> Validation a
waived (WorkerConfig m payload -> UUID
forall (m :: * -> *) payload. WorkerConfig m payload -> UUID
workerId WorkerConfig m payload
config)
    Validation
  (Maybe Text
   -> Maybe Value -> NominalDiffTime -> WorkerConfig m payload)
-> Validation (Maybe Text)
-> Validation
     (Maybe Value -> NominalDiffTime -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Maybe Text -> Validation (Maybe Text)
forall a. a -> Validation a
waived (WorkerConfig m payload -> Maybe Text
forall (m :: * -> *) payload. WorkerConfig m payload -> Maybe Text
workerHost WorkerConfig m payload
config)
    Validation
  (Maybe Value -> NominalDiffTime -> WorkerConfig m payload)
-> Validation (Maybe Value)
-> Validation (NominalDiffTime -> WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Maybe Value -> Validation (Maybe Value)
forall a. a -> Validation a
waived (WorkerConfig m payload -> Maybe Value
forall (m :: * -> *) payload. WorkerConfig m payload -> Maybe Value
workerMetadata WorkerConfig m payload
config)
    Validation (NominalDiffTime -> WorkerConfig m payload)
-> Validation NominalDiffTime
-> Validation (WorkerConfig m payload)
forall a b. Validation (a -> b) -> Validation a -> Validation b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Text -> NominalDiffTime -> Validation NominalDiffTime
forall a. (Num a, Ord a) => Text -> a -> Validation a
positive Text
"workerStaleThreshold" (WorkerConfig m payload -> NominalDiffTime
forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
workerStaleThreshold WorkerConfig m payload
config)

-- | Invariants spanning more than one field.
crossFieldInvariants :: WorkerConfig m payload -> Validation ()
crossFieldInvariants :: forall (m :: * -> *) payload.
WorkerConfig m payload -> Validation ()
crossFieldInvariants WorkerConfig m payload
config =
  Bool -> Text -> Validation ()
require
    (WorkerConfig m payload -> NominalDiffTime
forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
jobHeartbeatInterval WorkerConfig m payload
config NominalDiffTime -> NominalDiffTime -> Bool
forall a. Ord a => a -> a -> Bool
< WorkerConfig m payload -> NominalDiffTime
forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
visibilityTimeout WorkerConfig m payload
config)
    Text
"jobHeartbeatInterval must be less than visibilityTimeout"
    Validation () -> Validation () -> Validation ()
forall a b. Validation a -> Validation b -> Validation b
forall (f :: * -> *) a b. Applicative f => f a -> f b -> f b
*> Bool -> Text -> Validation ()
require
      (WorkerConfig m payload -> NominalDiffTime
forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
workerHeartbeatInterval WorkerConfig m payload
config NominalDiffTime -> NominalDiffTime -> Bool
forall a. Ord a => a -> a -> Bool
< WorkerConfig m payload -> NominalDiffTime
forall (m :: * -> *) payload.
WorkerConfig m payload -> NominalDiffTime
workerStaleThreshold WorkerConfig m payload
config)
      Text
"workerHeartbeatInterval must be less than workerStaleThreshold"

newtype Validation a = Validation (Either (NonEmpty Text) a)
  deriving newtype ((forall a b. (a -> b) -> Validation a -> Validation b)
-> (forall a b. a -> Validation b -> Validation a)
-> Functor Validation
forall a b. a -> Validation b -> Validation a
forall a b. (a -> b) -> Validation a -> Validation b
forall (f :: * -> *).
(forall a b. (a -> b) -> f a -> f b)
-> (forall a b. a -> f b -> f a) -> Functor f
$cfmap :: forall a b. (a -> b) -> Validation a -> Validation b
fmap :: forall a b. (a -> b) -> Validation a -> Validation b
$c<$ :: forall a b. a -> Validation b -> Validation a
<$ :: forall a b. a -> Validation b -> Validation a
Functor)

instance Applicative Validation where
  pure :: forall a. a -> Validation a
pure = Either (NonEmpty Text) a -> Validation a
forall a. Either (NonEmpty Text) a -> Validation a
Validation (Either (NonEmpty Text) a -> Validation a)
-> (a -> Either (NonEmpty Text) a) -> a -> Validation a
forall b c a. (b -> c) -> (a -> b) -> a -> c
. a -> Either (NonEmpty Text) a
forall a b. b -> Either a b
Right
  Validation (Left NonEmpty Text
left) <*> :: forall a b. Validation (a -> b) -> Validation a -> Validation b
<*> Validation (Left NonEmpty Text
right) = Either (NonEmpty Text) b -> Validation b
forall a. Either (NonEmpty Text) a -> Validation a
Validation (NonEmpty Text -> Either (NonEmpty Text) b
forall a b. a -> Either a b
Left (NonEmpty Text
left NonEmpty Text -> NonEmpty Text -> NonEmpty Text
forall a. Semigroup a => a -> a -> a
<> NonEmpty Text
right))
  Validation Either (NonEmpty Text) (a -> b)
apply <*> Validation Either (NonEmpty Text) a
value = Either (NonEmpty Text) b -> Validation b
forall a. Either (NonEmpty Text) a -> Validation a
Validation (Either (NonEmpty Text) (a -> b)
apply Either (NonEmpty Text) (a -> b)
-> Either (NonEmpty Text) a -> Either (NonEmpty Text) b
forall a b.
Either (NonEmpty Text) (a -> b)
-> Either (NonEmpty Text) a -> Either (NonEmpty Text) b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Either (NonEmpty Text) a
value)

-- | A field carrying no invariant of its own.
waived :: a -> Validation a
waived :: forall a. a -> Validation a
waived = a -> Validation a
forall a. a -> Validation a
forall (f :: * -> *) a. Applicative f => a -> f a
pure

-- | Reject a non-positive field, passing it through unchanged.
positive :: (Num a, Ord a) => Text -> a -> Validation a
positive :: forall a. (Num a, Ord a) => Text -> a -> Validation a
positive Text
label a
value = a
value a -> Validation () -> Validation a
forall a b. a -> Validation b -> Validation a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ Bool -> Text -> Validation ()
require (a
value a -> a -> Bool
forall a. Ord a => a -> a -> Bool
> a
0) (Text
label Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" must be greater than zero")

-- | Reject a negative field and return a valid field unchanged. Zero disables
-- the applicable wait.
nonNegative :: (Num a, Ord a) => Text -> a -> Validation a
nonNegative :: forall a. (Num a, Ord a) => Text -> a -> Validation a
nonNegative Text
label a
value = a
value a -> Validation () -> Validation a
forall a b. a -> Validation b -> Validation a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ Bool -> Text -> Validation ()
require (a
value a -> a -> Bool
forall a. Ord a => a -> a -> Bool
>= a
0) (Text
label Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" must not be negative")

require :: Bool -> Text -> Validation ()
require :: Bool -> Text -> Validation ()
require Bool
condition Text
message = Either (NonEmpty Text) () -> Validation ()
forall a. Either (NonEmpty Text) a -> Validation a
Validation (if Bool
condition then () -> Either (NonEmpty Text) ()
forall a b. b -> Either a b
Right () else NonEmpty Text -> Either (NonEmpty Text) ()
forall a b. a -> Either a b
Left (Text
message Text -> [Text] -> NonEmpty Text
forall a. a -> [a] -> NonEmpty a
:| []))

-- | Create a t'WorkerConfig' running one job per group in a worker transaction
-- held for the duration of the handler.
transactionalWorkerConfig
  :: (MonadArbiter n, MonadIO m)
  => Int
  -- ^ Worker count
  -> JobHandler n payload (ResultOf n payload)
  -- ^ Handler returning the result type @payload@'s registry entry declares
  -> m (WorkerConfig n payload)
transactionalWorkerConfig :: forall (n :: * -> *) (m :: * -> *) payload.
(MonadArbiter n, MonadIO m) =>
Int
-> JobHandler n payload (ResultOf n payload)
-> m (WorkerConfig n payload)
transactionalWorkerConfig Int
workerCnt JobHandler n payload (ResultOf n payload)
handler =
  Int -> HandlerMode n payload -> m (WorkerConfig n payload)
forall (n :: * -> *) (m :: * -> *) payload.
(Applicative n, MonadIO m) =>
Int -> HandlerMode n payload -> m (WorkerConfig n payload)
mkDefaultConfig Int
workerCnt (JobHandler n payload (ResultOf n payload) -> HandlerMode n payload
forall (m :: * -> *) payload.
JobHandler m payload (ResultOf m payload) -> HandlerMode m payload
SingleJobMode JobHandler n payload (ResultOf n payload)
handler)

-- | Create a t'WorkerConfig' for batched job processing, no worker transaction.
-- The handler receives the batch and a 'BatchCallbacks' record to finalize each
-- job. To store a result per job, ack with 'ackWith' or 'ackAllWith'.
batchedWorkerConfig
  :: (MonadArbiter n, MonadIO m)
  => Int
  -- ^ Worker count
  -> Int
  -- ^ Batch size: max jobs per batch. A batch is one group, or ungrouped jobs.
  -> (NonEmpty (JobRead payload) -> BatchCallbacks n payload (ResultOf n payload) -> n ())
  -- ^ Batch handler
  -> m (WorkerConfig n payload)
batchedWorkerConfig :: forall (n :: * -> *) (m :: * -> *) payload.
(MonadArbiter n, MonadIO m) =>
Int
-> Int
-> (NonEmpty (JobRead payload)
    -> BatchCallbacks n payload (ResultOf n payload) -> n ())
-> m (WorkerConfig n payload)
batchedWorkerConfig Int
workerCnt Int
batchSize NonEmpty (JobRead payload)
-> BatchCallbacks n payload (ResultOf n payload) -> n ()
handler =
  Int -> HandlerMode n payload -> m (WorkerConfig n payload)
forall (n :: * -> *) (m :: * -> *) payload.
(Applicative n, MonadIO m) =>
Int -> HandlerMode n payload -> m (WorkerConfig n payload)
mkDefaultConfig Int
workerCnt (Int
-> (NonEmpty (JobRead payload)
    -> BatchCallbacks n payload (ResultOf n payload) -> n ())
-> HandlerMode n payload
forall (m :: * -> *) payload.
Int
-> (NonEmpty (JobRead payload)
    -> BatchCallbacks m payload (ResultOf m payload) -> m ())
-> HandlerMode m payload
BatchedJobsMode Int
batchSize NonEmpty (JobRead payload)
-> BatchCallbacks n payload (ResultOf n payload) -> n ()
handler)

-- | Create a t'WorkerConfig' that passes one job per handler call, no worker transaction.
-- The handler finalizes the job through 'BatchCallbacks'.
manualWorkerConfig
  :: (MonadArbiter n, MonadIO m)
  => Int
  -- ^ Worker count
  -> (JobRead payload -> BatchCallbacks n payload (ResultOf n payload) -> n ())
  -- ^ Job handler
  -> m (WorkerConfig n payload)
manualWorkerConfig :: forall (n :: * -> *) (m :: * -> *) payload.
(MonadArbiter n, MonadIO m) =>
Int
-> (JobRead payload
    -> BatchCallbacks n payload (ResultOf n payload) -> n ())
-> m (WorkerConfig n payload)
manualWorkerConfig Int
workerCnt JobRead payload
-> BatchCallbacks n payload (ResultOf n payload) -> n ()
handler =
  Int
-> Int
-> (NonEmpty (JobRead payload)
    -> BatchCallbacks n payload (ResultOf n payload) -> n ())
-> m (WorkerConfig n payload)
forall (n :: * -> *) (m :: * -> *) payload.
(MonadArbiter n, MonadIO m) =>
Int
-> Int
-> (NonEmpty (JobRead payload)
    -> BatchCallbacks n payload (ResultOf n payload) -> n ())
-> m (WorkerConfig n payload)
batchedWorkerConfig Int
workerCnt Int
singleJobBatchSize (\(JobRead payload
job :| [JobRead payload]
_) -> JobRead payload
-> BatchCallbacks n payload (ResultOf n payload) -> n ()
handler JobRead payload
job)

-- | Transform a pool's observability hooks.
withHooks
  :: (ObservabilityHooks m payload -> ObservabilityHooks m payload)
  -> WorkerConfig m payload
  -> WorkerConfig m payload
withHooks :: forall (m :: * -> *) payload.
(ObservabilityHooks m payload -> ObservabilityHooks m payload)
-> WorkerConfig m payload -> WorkerConfig m payload
withHooks ObservabilityHooks m payload -> ObservabilityHooks m payload
rework WorkerConfig m payload
cfg = WorkerConfig m payload
cfg {observabilityHooks = rework (observabilityHooks cfg)}

-- | Run @report@ before the pool's own maintenance callback. Each runs even when the other fails.
withMaintenance
  :: (MonadUnliftIO m)
  => (MaintenanceOp -> Int64 -> m ())
  -> WorkerConfig m payload
  -> WorkerConfig m payload
withMaintenance :: forall (m :: * -> *) payload.
MonadUnliftIO m =>
(MaintenanceOp -> Int64 -> m ())
-> WorkerConfig m payload -> WorkerConfig m payload
withMaintenance MaintenanceOp -> Int64 -> m ()
report WorkerConfig m payload
cfg =
  WorkerConfig m payload
cfg {onMaintenance = \MaintenanceOp
operation Int64
count -> MaintenanceOp -> Int64 -> m ()
report MaintenanceOp
operation Int64
count m () -> m () -> m ()
forall (m :: * -> *). MonadUnliftIO m => m () -> m () -> m ()
`andThen` WorkerConfig m payload -> MaintenanceOp -> Int64 -> m ()
forall (m :: * -> *) payload.
WorkerConfig m payload -> MaintenanceOp -> Int64 -> m ()
onMaintenance WorkerConfig m payload
cfg MaintenanceOp
operation Int64
count}

-- | Gap between polls when no notification arrives.
defaultPollInterval :: NominalDiffTime
defaultPollInterval :: NominalDiffTime
defaultPollInterval = NominalDiffTime
5

-- | Time a claimed job stays invisible to other workers.
defaultVisibilityTimeout :: NominalDiffTime
defaultVisibilityTimeout :: NominalDiffTime
defaultVisibilityTimeout = NominalDiffTime
60

-- | Gap between job lease renewals.
defaultJobHeartbeatInterval :: NominalDiffTime
defaultJobHeartbeatInterval :: NominalDiffTime
defaultJobHeartbeatInterval = NominalDiffTime
30

-- | Gap between worker registry heartbeats.
defaultWorkerHeartbeatInterval :: NominalDiffTime
defaultWorkerHeartbeatInterval :: NominalDiffTime
defaultWorkerHeartbeatInterval = NominalDiffTime
10

-- | Base of the default exponential backoff.
defaultBackoffBase :: Double
defaultBackoffBase :: Double
defaultBackoffBase = Double
2.0

-- | Cap of the default exponential backoff.
defaultBackoffCap :: NominalDiffTime
defaultBackoffCap :: NominalDiffTime
defaultBackoffCap = NominalDiffTime
1_048_576

-- | Time in-flight jobs get to finish at shutdown.
defaultGracefulShutdownTimeout :: NominalDiffTime
defaultGracefulShutdownTimeout :: NominalDiffTime
defaultGracefulShutdownTimeout = NominalDiffTime
30

-- | Gap between reaper passes.
defaultReaperInterval :: NominalDiffTime
defaultReaperInterval :: NominalDiffTime
defaultReaperInterval = NominalDiffTime
300

-- | Age after which a silent worker counts as stale.
defaultWorkerStaleThreshold :: NominalDiffTime
defaultWorkerStaleThreshold :: NominalDiffTime
defaultWorkerStaleThreshold = NominalDiffTime
300

-- | Gap the bucket prune and the concurrency reconcile-and-prune keep.
defaultMaintenanceSparseInterval :: NominalDiffTime
defaultMaintenanceSparseInterval :: NominalDiffTime
defaultMaintenanceSparseInterval = NominalDiffTime
3600

-- | Age after which an idle rate-limit bucket is pruned.
defaultMaintenanceBucketIdle :: NominalDiffTime
defaultMaintenanceBucketIdle :: NominalDiffTime
defaultMaintenanceBucketIdle = NominalDiffTime
300

-- | Statement timeout for one maintenance operation.
defaultMaintenanceTimeout :: NominalDiffTime
defaultMaintenanceTimeout :: NominalDiffTime
defaultMaintenanceTimeout = NominalDiffTime
300

-- | Internal helper to create a config with the given handler mode.
mkDefaultConfig
  :: (Applicative n, MonadIO m)
  => Int
  -> HandlerMode n payload
  -> m (WorkerConfig n payload)
mkDefaultConfig :: forall (n :: * -> *) (m :: * -> *) payload.
(Applicative n, MonadIO m) =>
Int -> HandlerMode n payload -> m (WorkerConfig n payload)
mkDefaultConfig Int
workerCnt HandlerMode n payload
mode = do
  heartbeatTMVar <- m (TMVar ())
forall (m :: * -> *) a. MonadIO m => m (TMVar a)
newEmptyTMVarIO
  stateTVar <- newTVarIO Running
  pauseTVar <- newTVarIO False
  pauseEpochTVar <- newTVarIO 0
  listenerReadyTVar <- newTVarIO False
  uuid <- liftIO UUID.nextRandom
  tmpDir <- liftIO getTemporaryDirectory
  host <- liftIO getHostName
  let livenessPath = String
tmpDir String -> ShowS
forall a. Semigroup a => a -> a -> a
<> String
"/arbiter-worker-" String -> ShowS
forall a. Semigroup a => a -> a -> a
<> UUID -> String
toString UUID
uuid
  pure
    WorkerConfig
      { workerCount = workerCnt
      , handlerMode = mode
      , pollInterval = defaultPollInterval
      , visibilityTimeout = defaultVisibilityTimeout
      , jobHeartbeatInterval = defaultJobHeartbeatInterval
      , maxJobDuration = Nothing
      , workerHeartbeatInterval = defaultWorkerHeartbeatInterval
      , backoffStrategy = exponentialBackoff defaultBackoffBase defaultBackoffCap
      , jitter = EqualJitter
      , observabilityHooks = defaultObservabilityHooks
      , onMaintenance = \MaintenanceOp
_ Int64
_ -> () -> n ()
forall a. a -> n a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      , workerRuntime =
          WorkerRuntime
            { runtimeStateVar = stateTVar
            , runtimePauseVar = pauseTVar
            , runtimePauseEpoch = pauseEpochTVar
            , runtimeHeartbeatSignal = heartbeatTMVar
            , runtimeListenerReadyVar = listenerReadyTVar
            }
      , livenessFile = Just livenessPath
      , gracefulShutdownTimeout = Just defaultGracefulShutdownTimeout
      , logConfig = withIdentity ("worker_id" .= uuid) defaultLogConfig
      , cronJobs = []
      , reaperInterval = defaultReaperInterval
      , reaperSparseInterval = defaultMaintenanceSparseInterval
      , reaperBucketIdle = defaultMaintenanceBucketIdle
      , reaperTimeout = defaultMaintenanceTimeout
      , workerId = uuid
      , workerHost = Just (T.pack host)
      , workerMetadata = Nothing
      , workerStaleThreshold = defaultWorkerStaleThreshold
      }

-- | Run\/shutdown state for a pool.
workerStateVar :: WorkerConfig n payload -> TVar WorkerState
workerStateVar :: forall (n :: * -> *) payload.
WorkerConfig n payload -> TVar WorkerState
workerStateVar = WorkerRuntime -> TVar WorkerState
runtimeStateVar (WorkerRuntime -> TVar WorkerState)
-> (WorkerConfig n payload -> WorkerRuntime)
-> WorkerConfig n payload
-> TVar WorkerState
forall b c a. (b -> c) -> (a -> b) -> a -> c
. WorkerConfig n payload -> WorkerRuntime
forall (m :: * -> *) payload.
WorkerConfig m payload -> WorkerRuntime
workerRuntime

-- | Pause state for a pool.
pauseVar :: WorkerConfig n payload -> TVar Bool
pauseVar :: forall (n :: * -> *) payload. WorkerConfig n payload -> TVar Bool
pauseVar = WorkerRuntime -> TVar Bool
runtimePauseVar (WorkerRuntime -> TVar Bool)
-> (WorkerConfig n payload -> WorkerRuntime)
-> WorkerConfig n payload
-> TVar Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. WorkerConfig n payload -> WorkerRuntime
forall (m :: * -> *) payload.
WorkerConfig m payload -> WorkerRuntime
workerRuntime

-- | Version of the current pause state.
pauseEpoch :: WorkerConfig n payload -> TVar Word64
pauseEpoch :: forall (n :: * -> *) payload. WorkerConfig n payload -> TVar Word64
pauseEpoch = WorkerRuntime -> TVar Word64
runtimePauseEpoch (WorkerRuntime -> TVar Word64)
-> (WorkerConfig n payload -> WorkerRuntime)
-> WorkerConfig n payload
-> TVar Word64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. WorkerConfig n payload -> WorkerRuntime
forall (m :: * -> *) payload.
WorkerConfig m payload -> WorkerRuntime
workerRuntime

-- | Signal used to coordinate worker heartbeats.
heartbeatSignal :: WorkerConfig n payload -> TMVar ()
heartbeatSignal :: forall (n :: * -> *) payload. WorkerConfig n payload -> TMVar ()
heartbeatSignal = WorkerRuntime -> TMVar ()
runtimeHeartbeatSignal (WorkerRuntime -> TMVar ())
-> (WorkerConfig n payload -> WorkerRuntime)
-> WorkerConfig n payload
-> TMVar ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. WorkerConfig n payload -> WorkerRuntime
forall (m :: * -> *) payload.
WorkerConfig m payload -> WorkerRuntime
workerRuntime

-- | Pulse the heartbeat signal.
pulseHeartbeat :: WorkerConfig n payload -> STM.STM ()
pulseHeartbeat :: forall (n :: * -> *) payload. WorkerConfig n payload -> STM ()
pulseHeartbeat WorkerConfig n payload
config = STM Bool -> STM ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (TMVar () -> () -> STM Bool
forall a. TMVar a -> a -> STM Bool
STM.tryPutTMVar (WorkerConfig n payload -> TMVar ()
forall (n :: * -> *) payload. WorkerConfig n payload -> TMVar ()
heartbeatSignal WorkerConfig n payload
config) ())

-- | State of the pool's notification listener.
listenerReadyVar :: WorkerConfig n payload -> TVar Bool
listenerReadyVar :: forall (n :: * -> *) payload. WorkerConfig n payload -> TVar Bool
listenerReadyVar = WorkerRuntime -> TVar Bool
runtimeListenerReadyVar (WorkerRuntime -> TVar Bool)
-> (WorkerConfig n payload -> WorkerRuntime)
-> WorkerConfig n payload
-> TVar Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. WorkerConfig n payload -> WorkerRuntime
forall (m :: * -> *) payload.
WorkerConfig m payload -> WorkerRuntime
workerRuntime

-- | Initiate graceful shutdown of the worker pool.
--
-- Stops claiming new jobs. In-flight jobs get up to 'gracefulShutdownTimeout' to
-- finish. Then the pool exits.
shutdownWorkerPool :: (MonadIO m) => WorkerConfig n payload -> m ()
shutdownWorkerPool :: forall (m :: * -> *) (n :: * -> *) payload.
MonadIO m =>
WorkerConfig n payload -> m ()
shutdownWorkerPool WorkerConfig n payload
config = STM () -> m ()
forall (m :: * -> *) a. MonadIO m => STM a -> m a
STM.atomically (STM () -> m ()) -> STM () -> m ()
forall a b. (a -> b) -> a -> b
$ TVar WorkerState -> WorkerState -> STM ()
forall a. TVar a -> a -> STM ()
STM.writeTVar (WorkerConfig n payload -> TVar WorkerState
forall (n :: * -> *) payload.
WorkerConfig n payload -> TVar WorkerState
workerStateVar WorkerConfig n payload
config) WorkerState
ShuttingDown

-- | The pool's state, with a pause reported as paused.
getWorkerState :: (MonadIO m) => WorkerConfig n payload -> m WorkerState
getWorkerState :: forall (m :: * -> *) (n :: * -> *) payload.
MonadIO m =>
WorkerConfig n payload -> m WorkerState
getWorkerState WorkerConfig n payload
config = STM WorkerState -> m WorkerState
forall (m :: * -> *) a. MonadIO m => STM a -> m a
STM.atomically (STM WorkerState -> m WorkerState)
-> STM WorkerState -> m WorkerState
forall a b. (a -> b) -> a -> b
$ WorkerConfig n payload -> STM WorkerState
forall (n :: * -> *) payload.
WorkerConfig n payload -> STM WorkerState
readEffectiveState WorkerConfig n payload
config

-- | Whether this pool's LISTEN channels are subscribed (or there is no listener).
getListenerReady :: (MonadIO m) => WorkerConfig n payload -> m Bool
getListenerReady :: forall (m :: * -> *) (n :: * -> *) payload.
MonadIO m =>
WorkerConfig n payload -> m Bool
getListenerReady WorkerConfig n payload
config = TVar Bool -> m Bool
forall (m :: * -> *) a. MonadIO m => TVar a -> m a
STM.readTVarIO (WorkerConfig n payload -> TVar Bool
forall (n :: * -> *) payload. WorkerConfig n payload -> TVar Bool
listenerReadyVar WorkerConfig n payload
config)

-- | Set the pause flag, superseding any reading a caller has in flight.
writePause :: WorkerConfig n payload -> Bool -> STM.STM ()
writePause :: forall (n :: * -> *) payload.
WorkerConfig n payload -> Bool -> STM ()
writePause WorkerConfig n payload
config Bool
paused = do
  TVar Bool -> Bool -> STM ()
forall a. TVar a -> a -> STM ()
STM.writeTVar (WorkerConfig n payload -> TVar Bool
forall (n :: * -> *) payload. WorkerConfig n payload -> TVar Bool
pauseVar WorkerConfig n payload
config) Bool
paused
  TVar Word64 -> (Word64 -> Word64) -> STM ()
forall a. TVar a -> (a -> a) -> STM ()
STM.modifyTVar' (WorkerConfig n payload -> TVar Word64
forall (n :: * -> *) payload. WorkerConfig n payload -> TVar Word64
pauseEpoch WorkerConfig n payload
config) (Word64 -> Word64 -> Word64
forall a. Num a => a -> a -> a
+ Word64
1)

-- | Apply a pause reading taken at @epoch@. A newer write wins.
writePauseIfCurrent :: WorkerConfig n payload -> Word64 -> Bool -> STM.STM ()
writePauseIfCurrent :: forall (n :: * -> *) payload.
WorkerConfig n payload -> Word64 -> Bool -> STM ()
writePauseIfCurrent WorkerConfig n payload
config Word64
epoch Bool
paused = do
  current <- TVar Word64 -> STM Word64
forall a. TVar a -> STM a
STM.readTVar (WorkerConfig n payload -> TVar Word64
forall (n :: * -> *) payload. WorkerConfig n payload -> TVar Word64
pauseEpoch WorkerConfig n payload
config)
  when (current == epoch) $ writePause config paused

-- | 'getWorkerState' inside 'UnliftIO.STM.STM'.
readEffectiveState :: WorkerConfig n payload -> STM.STM WorkerState
readEffectiveState :: forall (n :: * -> *) payload.
WorkerConfig n payload -> STM WorkerState
readEffectiveState WorkerConfig n payload
config = do
  state <- TVar WorkerState -> STM WorkerState
forall a. TVar a -> STM a
STM.readTVar (WorkerConfig n payload -> TVar WorkerState
forall (n :: * -> *) payload.
WorkerConfig n payload -> TVar WorkerState
workerStateVar WorkerConfig n payload
config)
  case state of
    WorkerState
ShuttingDown -> WorkerState -> STM WorkerState
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure WorkerState
ShuttingDown
    WorkerState
_ -> do
      paused <- TVar Bool -> STM Bool
forall a. TVar a -> STM a
STM.readTVar (WorkerConfig n payload -> TVar Bool
forall (n :: * -> *) payload. WorkerConfig n payload -> TVar Bool
pauseVar WorkerConfig n payload
config)
      pure $ if paused then Paused else state