{-# LANGUAGE OverloadedStrings #-}

-- | Cron schedule reconciliation and execution.
module Arbiter.Worker.Cron.Scheduler
  ( CronLog
  , newCronLog
  , runCronScheduler
  , initCronSchedules
  , processCronCatchUp
  , processRunRequests
  , enumerateCatchUpTicks
  , makeDedupKey
  , computeDelayMicros
  ) where

import Arbiter.Core.CronSchedule qualified as CS
import Arbiter.Core.Exceptions (displayEx)
import Arbiter.Core.HighLevel (QueueOperation)
import Arbiter.Core.HighLevel qualified as HL
import Arbiter.Core.Job.Schema (SchemaName)
import Arbiter.Core.Job.Types (DedupKey (IgnoreDuplicate), setDedupKey)
import Arbiter.Core.MonadArbiter (MonadArbiter, withDbTransaction)
import Arbiter.Core.Operations qualified as Ops
import Control.Concurrent.STM (retry)
import Control.Monad (unless, void, when)
import Control.Monad.IO.Class (MonadIO)
import Data.Foldable (for_, traverse_)
import Data.Maybe (fromMaybe, isNothing)
import Data.Set qualified as Set
import Data.Text (Text)
import Data.Text qualified as T
import Data.Time (UTCTime, addUTCTime, diffUTCTime, getCurrentTime)
import System.Cron (CronSchedule, parseCronSchedule)
import UnliftIO
  ( MonadUnliftIO
  , SomeException
  , TVar
  , atomically
  , liftIO
  , readTVar
  , readTVarIO
  , registerDelay
  , tryAny
  , writeTVar
  )

import Arbiter.Worker.Cron.Types
import Arbiter.Worker.Logger (FailureGates, LogConfig, LogLevel (..), newFailureGates, tryLog, tryReportedOn)
import Arbiter.Worker.WorkerState (WorkerState (..))

-- | Upsert default expression and overlap for each 'CronJob' into the
-- @cron_schedules@ table. Preserves any user overrides and enabled state.
initCronSchedules
  :: (MonadArbiter m)
  => SchemaName -> Text -> [CronJob payload] -> LogConfig -> m ()
initCronSchedules :: forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> [CronJob payload] -> LogConfig -> m ()
initCronSchedules Text
schemaName Text
queueName [CronJob payload]
jobs LogConfig
logCfg = do
  [CronJob payload] -> (CronJob payload -> m Int64) -> m ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
t a -> (a -> f b) -> f ()
for_ [CronJob payload]
jobs ((CronJob payload -> m Int64) -> m ())
-> (CronJob payload -> m Int64) -> m ()
forall a b. (a -> b) -> a -> b
$ \CronJob payload
cron ->
    Text -> Text -> Text -> Text -> Text -> Maybe Text -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Text -> Text -> Text -> Maybe Text -> m Int64
Ops.upsertCronDefault
      Text
schemaName
      (CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron)
      Text
queueName
      (CronJob payload -> Text
forall payload. CronJob payload -> Text
cronExpression CronJob payload
cron)
      (OverlapPolicy -> Text
overlapPolicyToText (CronJob payload -> OverlapPolicy
forall payload. CronJob payload -> OverlapPolicy
overlap CronJob payload
cron))
      (CronJob payload -> Maybe Text
forall payload. CronJob payload -> Maybe Text
timezone CronJob payload
cron)
  IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> (Text -> IO ()) -> Text -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. LogConfig -> LogLevel -> Text -> IO ()
forall (m :: * -> *).
MonadUnliftIO m =>
LogConfig -> LogLevel -> Text -> m ()
tryLog LogConfig
logCfg LogLevel
Info (Text -> m ()) -> Text -> m ()
forall a b. (a -> b) -> a -> b
$
    Text
"Cron schedules initialized: " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> String -> Text
T.pack (Int -> String
forall a. Show a => a -> String
show ([CronJob payload] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length [CronJob payload]
jobs)) Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" schedule(s) upserted"

-- | Scheduler entry point. Exits when the worker state becomes 'ShuttingDown'.
runCronScheduler
  :: (QueueOperation m payload)
  => TVar WorkerState
  -> TVar Bool
  -- ^ Set by the run-now listener when a schedule this pool owns is requested.
  -> LogConfig
  -> SchemaName
  -> Text
  -- ^ Queue name (recorded on each schedule row).
  -> [CronJob payload]
  -> m ()
runCronScheduler :: forall (m :: * -> *) payload.
QueueOperation m payload =>
TVar WorkerState
-> TVar Bool
-> LogConfig
-> Text
-> Text
-> [CronJob payload]
-> m ()
runCronScheduler TVar WorkerState
stateVar TVar Bool
runNowVar LogConfig
logCfg Text
schemaName Text
queueName [CronJob payload]
jobs = do
  Text -> Text -> [CronJob payload] -> LogConfig -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
Text -> Text -> [CronJob payload] -> LogConfig -> m ()
initCronSchedules Text
schemaName Text
queueName [CronJob payload]
jobs LogConfig
logCfg
  cronLog <- LogConfig -> m CronLog
forall (m :: * -> *). MonadIO m => LogConfig -> m CronLog
newCronLog LogConfig
logCfg
  startupNow <- liftIO getCurrentTime
  shuttingDown <- isShuttingDown stateVar
  unless shuttingDown $ do
    processRunRequests cronLog schemaName jobs startupNow
    processCronCatchUp cronLog schemaName queueName jobs startupNow
  logCron cronLog Info $ "Cron scheduler started with " <> T.pack (show (length jobs)) <> " schedule(s)"
  loop cronLog
  where
    loop :: CronLog -> m ()
loop CronLog
cronLog = do
      now <- IO UTCTime -> m UTCTime
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO IO UTCTime
getCurrentTime
      timerVar <- liftIO $ registerDelay (computeDelayMicros now)
      serve cronLog timerVar
    serve :: CronLog -> TVar Bool -> m ()
serve CronLog
cronLog TVar Bool
timerVar = do
      wake <- TVar WorkerState -> TVar Bool -> TVar Bool -> m WakeReason
forall (m :: * -> *).
MonadIO m =>
TVar WorkerState -> TVar Bool -> TVar Bool -> m WakeReason
waitForWake TVar WorkerState
stateVar TVar Bool
runNowVar TVar Bool
timerVar
      case wake of
        WakeReason
WakeShutdown -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
        WakeReason
WakeMinute -> do
          now <- IO UTCTime -> m UTCTime
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO IO UTCTime
getCurrentTime
          processRunRequests cronLog schemaName jobs now
          processCronCatchUp cronLog schemaName queueName jobs now
          loop cronLog
        WakeReason
WakeRunNow -> do
          now <- IO UTCTime -> m UTCTime
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO IO UTCTime
getCurrentTime
          processRunRequests cronLog schemaName jobs now
          serve cronLog timerVar

-- | Scheduler catch-up step. Each cron runs in its own transaction.
-- Backfill schedules hold a per-(schema, queue, name) advisory lock.
processCronCatchUp
  :: (QueueOperation m payload)
  => CronLog
  -> Text
  -> Text
  -- ^ Queue name
  -> [CronJob payload]
  -> UTCTime
  -- ^ Current wall-clock time
  -> m ()
processCronCatchUp :: forall (m :: * -> *) payload.
QueueOperation m payload =>
CronLog -> Text -> Text -> [CronJob payload] -> UTCTime -> m ()
processCronCatchUp CronLog
cronLog Text
schemaName Text
queueName [CronJob payload]
jobs UTCTime
now = do
  let currentTick :: UTCTime
currentTick = UTCTime -> UTCTime
truncateToMinute UTCTime
now
  (CronJob payload -> m ()) -> [CronJob payload] -> m ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (UTCTime -> CronJob payload -> m ()
processOneCron UTCTime
currentTick) [CronJob payload]
jobs
  where
    processOneCron :: UTCTime -> CronJob payload -> m ()
processOneCron UTCTime
currentTick CronJob payload
cron = do
      outcome <- CronLog
-> Text -> m TickOutcome -> m (Either SomeException TickOutcome)
forall (m :: * -> *) a.
MonadUnliftIO m =>
CronLog -> Text -> m a -> m (Either SomeException a)
tryCron CronLog
cronLog (Text
"Cron '" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"' tick") (m TickOutcome -> m (Either SomeException TickOutcome))
-> (m TickOutcome -> m TickOutcome)
-> m TickOutcome
-> m (Either SomeException TickOutcome)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. m TickOutcome -> m TickOutcome
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m TickOutcome -> m (Either SomeException TickOutcome))
-> m TickOutcome -> m (Either SomeException TickOutcome)
forall a b. (a -> b) -> a -> b
$ do
        haveLeader <- case CronJob payload -> BackfillPolicy
forall payload. CronJob payload -> BackfillPolicy
backfill CronJob payload
cron of
          BackfillPolicy
NoBackfill -> Bool -> m Bool
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
True
          Backfill NominalDiffTime
_ -> Text -> Text -> Text -> m Bool
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Text -> m Bool
Ops.tryAcquireCronLeader Text
schemaName Text
queueName (CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron)
        if not haveLeader
          then pure NotLeader
          else do
            mRow <- Ops.getCronScheduleByName schemaName (name cron)
            processOne mRow currentTick cron
            void $ Ops.touchCronChecked schemaName currentTick [name cron]
            pure Ran
      case outcome of
        Left SomeException
_ -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
        Right TickOutcome
NotLeader ->
          CronLog -> LogLevel -> Text -> m ()
forall (m :: * -> *).
MonadIO m =>
CronLog -> LogLevel -> Text -> m ()
logCron CronLog
cronLog LogLevel
Debug (Text -> m ()) -> Text -> m ()
forall a b. (a -> b) -> a -> b
$ Text
"Cron '" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"' skipped, another pool holds the lock"
        Right TickOutcome
Ran -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
    processOne :: Maybe CronScheduleRow -> UTCTime -> CronJob payload -> m ()
processOne Maybe CronScheduleRow
mRow UTCTime
currentTick CronJob payload
cron = case CronJob payload -> Maybe CronScheduleRow -> Resolved
forall payload.
CronJob payload -> Maybe CronScheduleRow -> Resolved
resolveAndParse CronJob payload
cron Maybe CronScheduleRow
mRow of
      Resolved
Disabled -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
      ParseError Text
expr String
err ->
        CronLog -> LogLevel -> Text -> m ()
forall (m :: * -> *).
MonadIO m =>
CronLog -> LogLevel -> Text -> m ()
logCron CronLog
cronLog LogLevel
Error (Text -> m ()) -> Text -> m ()
forall a b. (a -> b) -> a -> b
$
          Text
"Cron schedule '"
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"' has invalid effective expression '"
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
expr
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"': "
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> String -> Text
T.pack String
err
      InvalidTimezone Text
tzName ->
        CronLog -> LogLevel -> Text -> m ()
forall (m :: * -> *).
MonadIO m =>
CronLog -> LogLevel -> Text -> m ()
logCron CronLog
cronLog LogLevel
Error (Text -> m ()) -> Text -> m ()
forall a b. (a -> b) -> a -> b
$
          Text
"Cron schedule '"
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"' has unknown timezone '"
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
tzName
            Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"'"
      Effective OverlapPolicy
effectiveOv CronSchedule
sched Maybe Text
effectiveTz -> do
        let ticksInWindow :: [UTCTime]
ticksInWindow = BackfillPolicy -> Maybe UTCTime -> UTCTime -> [UTCTime]
enumerateCatchUpTicks (CronJob payload -> BackfillPolicy
forall payload. CronJob payload -> BackfillPolicy
backfill CronJob payload
cron) (Maybe CronScheduleRow
mRow Maybe CronScheduleRow
-> (CronScheduleRow -> Maybe UTCTime) -> Maybe UTCTime
forall a b. Maybe a -> (a -> Maybe b) -> Maybe b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= CronScheduleRow -> Maybe UTCTime
CS.lastCheckedAt) UTCTime
currentTick
            ticksToFire :: [UTCTime]
ticksToFire = CronSchedule
-> Maybe Text -> OverlapPolicy -> [UTCTime] -> [UTCTime]
pickTicksToFire CronSchedule
sched Maybe Text
effectiveTz OverlapPolicy
effectiveOv [UTCTime]
ticksInWindow
            replayCount :: Int
replayCount = [UTCTime] -> Int
forall a. [a] -> Int
forall (t :: * -> *) a. Foldable t => t a -> Int
length ((UTCTime -> Bool) -> [UTCTime] -> [UTCTime]
forall a. (a -> Bool) -> [a] -> [a]
filter (UTCTime -> UTCTime -> Bool
forall a. Eq a => a -> a -> Bool
/= UTCTime
currentTick) [UTCTime]
ticksToFire)
        Bool -> m () -> m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
when (Int
replayCount Int -> Int -> Bool
forall a. Ord a => a -> a -> Bool
> Int
0)
          (m () -> m ()) -> m () -> m ()
forall a b. (a -> b) -> a -> b
$ CronLog -> LogLevel -> Text -> m ()
forall (m :: * -> *).
MonadIO m =>
CronLog -> LogLevel -> Text -> m ()
logCron CronLog
cronLog LogLevel
Info
          (Text -> m ()) -> Text -> m ()
forall a b. (a -> b) -> a -> b
$ Text
"Replaying " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> String -> Text
T.pack (Int -> String
forall a. Show a => a -> String
show Int
replayCount) Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" missed tick(s) for '" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"'"
        [UTCTime] -> (UTCTime -> m ()) -> m ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
t a -> (a -> f b) -> f ()
for_ [UTCTime]
ticksToFire ((UTCTime -> m ()) -> m ()) -> (UTCTime -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \UTCTime
tick ->
          CronLog
-> Text
-> CronJob payload
-> OverlapPolicy
-> Maybe Text
-> TickKind
-> UTCTime
-> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
CronLog
-> Text
-> CronJob payload
-> OverlapPolicy
-> Maybe Text
-> TickKind
-> UTCTime
-> m ()
tryInsertCronJob CronLog
cronLog Text
schemaName CronJob payload
cron OverlapPolicy
effectiveOv Maybe Text
effectiveTz (UTCTime -> UTCTime -> TickKind
tickKindFor UTCTime
currentTick UTCTime
tick) UTCTime
tick

data TickOutcome = NotLeader | Ran

-- | 'Live' for @currentTick@ and 'Replay' for any other tick.
tickKindFor :: UTCTime -> UTCTime -> TickKind
tickKindFor :: UTCTime -> UTCTime -> TickKind
tickKindFor UTCTime
currentTick UTCTime
tick = if UTCTime
tick UTCTime -> UTCTime -> Bool
forall a. Eq a => a -> a -> Bool
== UTCTime
currentTick then TickKind
Live else TickKind
Replay

-- | 'SkipOverlap' keeps the oldest match. 'AllowOverlap' keeps all.
-- Match evaluation uses the supplied timezone ('Nothing' = UTC).
pickTicksToFire :: CronSchedule -> Maybe Text -> OverlapPolicy -> [UTCTime] -> [UTCTime]
pickTicksToFire :: CronSchedule
-> Maybe Text -> OverlapPolicy -> [UTCTime] -> [UTCTime]
pickTicksToFire CronSchedule
sched Maybe Text
zone OverlapPolicy
overlapPolicy [UTCTime]
ticks =
  let matching :: [UTCTime]
matching = (UTCTime -> Bool) -> [UTCTime] -> [UTCTime]
forall a. (a -> Bool) -> [a] -> [a]
filter (Maybe Text -> CronSchedule -> UTCTime -> Bool
matchesInTimezone Maybe Text
zone CronSchedule
sched) [UTCTime]
ticks
   in case OverlapPolicy
overlapPolicy of
        OverlapPolicy
SkipOverlap -> Int -> [UTCTime] -> [UTCTime]
forall a. Int -> [a] -> [a]
take Int
1 [UTCTime]
matching
        OverlapPolicy
AllowOverlap -> [UTCTime]
matching

-- | Minutes to evaluate for a 'processCronCatchUp' call. Returns @[]@ when
-- the watermark is at or past @currentTick@.
enumerateCatchUpTicks :: BackfillPolicy -> Maybe UTCTime -> UTCTime -> [UTCTime]
enumerateCatchUpTicks :: BackfillPolicy -> Maybe UTCTime -> UTCTime -> [UTCTime]
enumerateCatchUpTicks BackfillPolicy
_ (Just UTCTime
lastChecked) UTCTime
currentTick
  | UTCTime -> UTCTime
truncateToMinute UTCTime
lastChecked UTCTime -> UTCTime -> Bool
forall a. Ord a => a -> a -> Bool
>= UTCTime
currentTick = []
enumerateCatchUpTicks BackfillPolicy
NoBackfill Maybe UTCTime
_ UTCTime
currentTick = [UTCTime
currentTick]
enumerateCatchUpTicks BackfillPolicy
_ Maybe UTCTime
Nothing UTCTime
currentTick = [UTCTime
currentTick]
enumerateCatchUpTicks (Backfill NominalDiffTime
window) (Just UTCTime
lastChecked) UTCTime
currentTick =
  let truncatedLast :: UTCTime
truncatedLast = UTCTime -> UTCTime
truncateToMinute UTCTime
lastChecked
      -- Truncate the window floor to a minute boundary.
      windowFloor :: UTCTime
windowFloor = UTCTime -> UTCTime
truncateToMinute (NominalDiffTime -> UTCTime -> UTCTime
addUTCTime (NominalDiffTime -> NominalDiffTime
forall a. Num a => a -> a
negate NominalDiffTime
window) UTCTime
currentTick)
      startMinute :: UTCTime
startMinute = UTCTime -> UTCTime -> UTCTime
forall a. Ord a => a -> a -> a
max (NominalDiffTime -> UTCTime -> UTCTime
addUTCTime NominalDiffTime
60 UTCTime
truncatedLast) UTCTime
windowFloor
   in UTCTime -> UTCTime -> [UTCTime]
enumMinutes UTCTime
startMinute UTCTime
currentTick

-- | Result of resolving a schedule's effective config (code defaults +
-- runtime DB overrides).
data Resolved
  = Disabled
  | ParseError Text String
  | InvalidTimezone Text
  | Effective OverlapPolicy CronSchedule (Maybe Text)

-- | The row's overlap override, falling back to the schedule's code default.
effectiveOverlapFor :: CronJob payload -> CS.CronScheduleRow -> OverlapPolicy
effectiveOverlapFor :: forall payload. CronJob payload -> CronScheduleRow -> OverlapPolicy
effectiveOverlapFor CronJob payload
cron CronScheduleRow
row = OverlapPolicy -> Maybe OverlapPolicy -> OverlapPolicy
forall a. a -> Maybe a -> a
fromMaybe (CronJob payload -> OverlapPolicy
forall payload. CronJob payload -> OverlapPolicy
overlap CronJob payload
cron) (Text -> Maybe OverlapPolicy
overlapPolicyFromText (CronScheduleRow -> Text
CS.effectiveOverlap CronScheduleRow
row))

resolveAndParse :: CronJob payload -> Maybe CS.CronScheduleRow -> Resolved
resolveAndParse :: forall payload.
CronJob payload -> Maybe CronScheduleRow -> Resolved
resolveAndParse CronJob payload
cron Maybe CronScheduleRow
mRow =
  let (Text
expr, OverlapPolicy
overlapPolicy, Maybe Text
zone, Bool
isEnabled) = case Maybe CronScheduleRow
mRow of
        Maybe CronScheduleRow
Nothing -> (CronJob payload -> Text
forall payload. CronJob payload -> Text
cronExpression CronJob payload
cron, CronJob payload -> OverlapPolicy
forall payload. CronJob payload -> OverlapPolicy
overlap CronJob payload
cron, CronJob payload -> Maybe Text
forall payload. CronJob payload -> Maybe Text
timezone CronJob payload
cron, Bool
True)
        Just row :: CronScheduleRow
row@CS.CronScheduleRow {enabled :: CronScheduleRow -> Bool
CS.enabled = Bool
rowEnabled} ->
          ( CronScheduleRow -> Text
CS.effectiveExpression CronScheduleRow
row
          , CronJob payload -> CronScheduleRow -> OverlapPolicy
forall payload. CronJob payload -> CronScheduleRow -> OverlapPolicy
effectiveOverlapFor CronJob payload
cron CronScheduleRow
row
          , CronScheduleRow -> Maybe Text
CS.effectiveTimezone CronScheduleRow
row
          , Bool
rowEnabled
          )
   in if Bool
isEnabled
        then case Text -> Either String CronSchedule
parseCronSchedule Text
expr of
          Left String
err -> Text -> String -> Resolved
ParseError Text
expr String
err
          Right CronSchedule
sched -> case Maybe Text
zone of
            Just Text
tzName | Maybe TZ -> Bool
forall a. Maybe a -> Bool
isNothing (Text -> Maybe TZ
resolveTZ Text
tzName) -> Text -> Resolved
InvalidTimezone Text
tzName
            Maybe Text
_ -> OverlapPolicy -> CronSchedule -> Maybe Text -> Resolved
Effective OverlapPolicy
overlapPolicy CronSchedule
sched Maybe Text
zone
        else Resolved
Disabled

-- | Attempt to insert a single cron-tick job. Any failure is logged.
--
-- The insert and the per-tick @last_checked_at@ advance are atomic. If
-- either fails the other rolls back.
tryInsertCronJob
  :: (QueueOperation m payload)
  => CronLog -> Text -> CronJob payload -> OverlapPolicy -> Maybe Text -> TickKind -> UTCTime -> m ()
tryInsertCronJob :: forall (m :: * -> *) payload.
QueueOperation m payload =>
CronLog
-> Text
-> CronJob payload
-> OverlapPolicy
-> Maybe Text
-> TickKind
-> UTCTime
-> m ()
tryInsertCronJob CronLog
cronLog Text
schemaName CronJob payload
cron OverlapPolicy
effectiveOv Maybe Text
effectiveTz TickKind
kind UTCTime
tick = do
  result <- CronLog -> Text -> m () -> m (Either SomeException ())
forall (m :: * -> *) a.
MonadUnliftIO m =>
CronLog -> Text -> m a -> m (Either SomeException a)
tryCron CronLog
cronLog (Text
"Cron schedule '" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"' insert") (m () -> m (Either SomeException ()))
-> (m () -> m ()) -> m () -> m (Either SomeException ())
forall b c a. (b -> c) -> (a -> b) -> a -> c
. m () -> m ()
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m () -> m (Either SomeException ()))
-> m () -> m (Either SomeException ())
forall a b. (a -> b) -> a -> b
$ do
    -- Gate first. Another pool may have fired this minute.
    fired <- Text -> Text -> UTCTime -> m Bool
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> UTCTime -> m Bool
Ops.tryFireCronGate Text
schemaName (CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron) UTCTime
tick
    when fired $ do
      let key = Text -> OverlapPolicy -> Maybe Text -> UTCTime -> Text
makeDedupKeyFromParts (CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron) OverlapPolicy
effectiveOv Maybe Text
effectiveTz UTCTime
tick
          jobWrite = Maybe DedupKey -> JobWrite payload -> JobWrite payload
forall payload.
Maybe DedupKey -> JobWrite payload -> JobWrite payload
setDedupKey (DedupKey -> Maybe DedupKey
forall a. a -> Maybe a
Just (Text -> DedupKey
IgnoreDuplicate Text
key)) (JobWrite payload -> JobWrite payload)
-> JobWrite payload -> JobWrite payload
forall a b. (a -> b) -> a -> b
$ CronJob payload -> TickKind -> UTCTime -> JobWrite payload
forall payload.
CronJob payload -> TickKind -> UTCTime -> JobWrite payload
builder CronJob payload
cron TickKind
kind UTCTime
tick
      void $ HL.insertJob jobWrite
    void $ Ops.touchCronChecked schemaName tick [name cron]
  traverse_
    (const . logCron cronLog Debug $ "Cron schedule '" <> name cron <> "' processed at " <> formatMinute tick)
    result

data RunNowOutcome = Fired | Skipped | NotRequested

-- | Claim and fire every schedule with a pending run request. A 'SkipOverlap'
-- schedule reuses its constant dedup key. A manual run is skipped while one of
-- its jobs is active.
--
-- The claim and the insert are atomic. If either fails the other rolls back.
processRunRequests
  :: forall payload m
   . (QueueOperation m payload)
  => CronLog -> Text -> [CronJob payload] -> UTCTime -> m ()
processRunRequests :: forall payload (m :: * -> *).
QueueOperation m payload =>
CronLog -> Text -> [CronJob payload] -> UTCTime -> m ()
processRunRequests CronLog
cronLog Text
schemaName [CronJob payload]
jobs UTCTime
now = do
  scan <- CronLog -> Text -> m [Text] -> m (Either SomeException [Text])
forall (m :: * -> *) a.
MonadUnliftIO m =>
CronLog -> Text -> m a -> m (Either SomeException a)
tryCron CronLog
cronLog Text
"Cron run-request scan" (m [Text] -> m (Either SomeException [Text]))
-> m [Text] -> m (Either SomeException [Text])
forall a b. (a -> b) -> a -> b
$ Text -> [Text] -> m [Text]
forall (m :: * -> *). MonadArbiter m => Text -> [Text] -> m [Text]
Ops.pendingCronRuns Text
schemaName ((CronJob payload -> Text) -> [CronJob payload] -> [Text]
forall a b. (a -> b) -> [a] -> [b]
map CronJob payload -> Text
forall payload. CronJob payload -> Text
name [CronJob payload]
jobs)
  traverse_ (fireRequested . Set.fromList) scan
  where
    fireRequested :: Set Text -> m ()
fireRequested Set Text
requested =
      (CronJob payload -> m ()) -> [CronJob payload] -> m ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ (UTCTime -> CronJob payload -> m ()
claimAndFire (UTCTime -> UTCTime
truncateToMinute UTCTime
now)) ((CronJob payload -> Bool) -> [CronJob payload] -> [CronJob payload]
forall a. (a -> Bool) -> [a] -> [a]
filter (\CronJob payload
cron -> Text -> Set Text -> Bool
forall a. Ord a => a -> Set a -> Bool
Set.member (CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron) Set Text
requested) [CronJob payload]
jobs)
    -- Every run-now failure is reported.
    claimAndFire :: UTCTime -> CronJob payload -> m ()
claimAndFire UTCTime
tick CronJob payload
cron = do
      outcome <- m RunNowOutcome -> m (Either SomeException RunNowOutcome)
forall (m :: * -> *) a.
MonadUnliftIO m =>
m a -> m (Either SomeException a)
tryAny (m RunNowOutcome -> m (Either SomeException RunNowOutcome))
-> (m RunNowOutcome -> m RunNowOutcome)
-> m RunNowOutcome
-> m (Either SomeException RunNowOutcome)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. m RunNowOutcome -> m RunNowOutcome
forall a. m a -> m a
forall (m :: * -> *) a. MonadArbiter m => m a -> m a
withDbTransaction (m RunNowOutcome -> m (Either SomeException RunNowOutcome))
-> m RunNowOutcome -> m (Either SomeException RunNowOutcome)
forall a b. (a -> b) -> a -> b
$ do
        claimed <- Text -> Text -> m (Maybe CronScheduleRow)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe CronScheduleRow)
Ops.claimCronRun Text
schemaName (CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron)
        maybe (pure NotRequested) (fireClaimed tick cron) claimed
      case outcome of
        Left SomeException
exception ->
          CronLog -> LogLevel -> Text -> m ()
forall (m :: * -> *).
MonadIO m =>
CronLog -> LogLevel -> Text -> m ()
logCron CronLog
cronLog LogLevel
Error (Text -> m ()) -> Text -> m ()
forall a b. (a -> b) -> a -> b
$ Text
"Cron '" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"' run-now aborted: " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> SomeException -> Text
displayEx SomeException
exception
        Right RunNowOutcome
NotRequested -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
        Right RunNowOutcome
Fired ->
          CronLog -> LogLevel -> Text -> m ()
forall (m :: * -> *).
MonadIO m =>
CronLog -> LogLevel -> Text -> m ()
logCron CronLog
cronLog LogLevel
Info (Text -> m ()) -> Text -> m ()
forall a b. (a -> b) -> a -> b
$ Text
"Cron '" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"' run-now fired at " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> UTCTime -> Text
formatMinute UTCTime
tick
        Right RunNowOutcome
Skipped ->
          CronLog -> LogLevel -> Text -> m ()
forall (m :: * -> *).
MonadIO m =>
CronLog -> LogLevel -> Text -> m ()
logCron CronLog
cronLog LogLevel
Warning (Text -> m ()) -> Text -> m ()
forall a b. (a -> b) -> a -> b
$ Text
"Cron '" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
"' run-now skipped, a job is already active"
    fireClaimed :: UTCTime -> CronJob payload -> CronScheduleRow -> m RunNowOutcome
fireClaimed UTCTime
tick CronJob payload
cron CronScheduleRow
row = do
      let key :: Maybe DedupKey
key = case CronJob payload -> CronScheduleRow -> OverlapPolicy
forall payload. CronJob payload -> CronScheduleRow -> OverlapPolicy
effectiveOverlapFor CronJob payload
cron CronScheduleRow
row of
            OverlapPolicy
SkipOverlap -> DedupKey -> Maybe DedupKey
forall a. a -> Maybe a
Just (Text -> DedupKey
IgnoreDuplicate (Text -> Text
skipOverlapKey (CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron)))
            OverlapPolicy
AllowOverlap -> Maybe DedupKey
forall a. Maybe a
Nothing
          jobWrite :: JobWrite payload
jobWrite = Maybe DedupKey -> JobWrite payload -> JobWrite payload
forall payload.
Maybe DedupKey -> JobWrite payload -> JobWrite payload
setDedupKey Maybe DedupKey
key (JobWrite payload -> JobWrite payload)
-> JobWrite payload -> JobWrite payload
forall a b. (a -> b) -> a -> b
$ CronJob payload -> TickKind -> UTCTime -> JobWrite payload
forall payload.
CronJob payload -> TickKind -> UTCTime -> JobWrite payload
builder CronJob payload
cron TickKind
Live UTCTime
tick
      inserted <- JobWrite payload -> m (Maybe (JobRead payload))
forall payload (m :: * -> *).
QueueOperation m payload =>
JobWrite payload -> m (Maybe (JobRead payload))
HL.insertJob JobWrite payload
jobWrite
      case inserted of
        Just JobRead payload
_ -> do
          m Int64 -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (m Int64 -> m ()) -> m Int64 -> m ()
forall a b. (a -> b) -> a -> b
$ Text -> UTCTime -> Text -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> UTCTime -> Text -> m Int64
Ops.touchCronManualRun Text
schemaName UTCTime
tick (CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron)
          RunNowOutcome -> m RunNowOutcome
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure RunNowOutcome
Fired
        Maybe (JobRead payload)
Nothing -> RunNowOutcome -> m RunNowOutcome
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure RunNowOutcome
Skipped

-- | The scheduler's logger with the gates its repeating failures report through.
data CronLog = CronLog
  { CronLog -> LogConfig
cronLogConfig :: LogConfig
  , CronLog -> FailureGates
cronLogGates :: FailureGates
  }

-- | A 'CronLog' with no gate tripped, for one scheduler run.
newCronLog :: (MonadIO m) => LogConfig -> m CronLog
newCronLog :: forall (m :: * -> *). MonadIO m => LogConfig -> m CronLog
newCronLog LogConfig
logCfg = LogConfig -> FailureGates -> CronLog
CronLog LogConfig
logCfg (FailureGates -> CronLog) -> m FailureGates -> m CronLog
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> m FailureGates
forall (m :: * -> *). MonadIO m => m FailureGates
newFailureGates

-- | Log a cron message, swallowing logger failures.
logCron :: (MonadIO m) => CronLog -> LogLevel -> Text -> m ()
logCron :: forall (m :: * -> *).
MonadIO m =>
CronLog -> LogLevel -> Text -> m ()
logCron CronLog
cronLog LogLevel
level Text
msg = IO () -> m ()
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO () -> m ()) -> IO () -> m ()
forall a b. (a -> b) -> a -> b
$ LogConfig -> LogLevel -> Text -> IO ()
forall (m :: * -> *).
MonadUnliftIO m =>
LogConfig -> LogLevel -> Text -> m ()
tryLog (CronLog -> LogConfig
cronLogConfig CronLog
cronLog) LogLevel
level Text
msg

-- | Run a scheduler step, reporting only when its outcome changes. A schedule that
-- keeps failing the same way says so one time, then again when it recovers.
tryCron :: (MonadUnliftIO m) => CronLog -> Text -> m a -> m (Either SomeException a)
tryCron :: forall (m :: * -> *) a.
MonadUnliftIO m =>
CronLog -> Text -> m a -> m (Either SomeException a)
tryCron CronLog
cronLog = LogConfig
-> LogLevel
-> FailureGates
-> Text
-> m a
-> m (Either SomeException a)
forall (m :: * -> *) a.
MonadUnliftIO m =>
LogConfig
-> LogLevel
-> FailureGates
-> Text
-> m a
-> m (Either SomeException a)
tryReportedOn (CronLog -> LogConfig
cronLogConfig CronLog
cronLog) LogLevel
Error (CronLog -> FailureGates
cronLogGates CronLog
cronLog)

-- | Dedup key for a cron job, from its code-defined overlap and timezone.
makeDedupKey :: CronJob payload -> UTCTime -> Text
makeDedupKey :: forall payload. CronJob payload -> UTCTime -> Text
makeDedupKey CronJob payload
cron UTCTime
tick = Text -> OverlapPolicy -> Maybe Text -> UTCTime -> Text
makeDedupKeyFromParts (CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron) (CronJob payload -> OverlapPolicy
forall payload. CronJob payload -> OverlapPolicy
overlap CronJob payload
cron) (CronJob payload -> Maybe Text
forall payload. CronJob payload -> Maybe Text
timezone CronJob payload
cron) UTCTime
tick

-- | For 'AllowOverlap', the key includes the tick formatted in the schedule's
-- timezone. A DST fall-back minute fires one time.
makeDedupKeyFromParts :: Text -> OverlapPolicy -> Maybe Text -> UTCTime -> Text
makeDedupKeyFromParts :: Text -> OverlapPolicy -> Maybe Text -> UTCTime -> Text
makeDedupKeyFromParts Text
jobName OverlapPolicy
overlapPolicy Maybe Text
zone UTCTime
tick = case OverlapPolicy
overlapPolicy of
  OverlapPolicy
SkipOverlap -> Text -> Text
skipOverlapKey Text
jobName
  OverlapPolicy
AllowOverlap -> Text
"arbiter_cron:" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
jobName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
":" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Maybe Text -> UTCTime -> Text
formatMinuteInTimezone Maybe Text
zone UTCTime
tick

-- | The tick-independent key a 'SkipOverlap' schedule reuses. At most one of its
-- jobs is active.
skipOverlapKey :: Text -> Text
skipOverlapKey :: Text -> Text
skipOverlapKey Text
jobName = Text
"arbiter_cron:" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
jobName

-- | The longest a scheduler tick waits for the next minute boundary.
maxTickDelayMicros :: Int
maxTickDelayMicros :: Int
maxTickDelayMicros = Int
120_000_000

-- | Delay in microseconds until the next minute boundary, clamped to
-- @[0, 'maxTickDelayMicros']@.
computeDelayMicros :: UTCTime -> Int
computeDelayMicros :: UTCTime -> Int
computeDelayMicros UTCTime
now =
  let nextMinute :: UTCTime
nextMinute = UTCTime -> UTCTime
truncateToMinute (NominalDiffTime -> UTCTime -> UTCTime
addUTCTime NominalDiffTime
60 UTCTime
now)
   in Int -> Int -> Int
forall a. Ord a => a -> a -> a
max Int
0 (Int -> Int -> Int
forall a. Ord a => a -> a -> a
min Int
maxTickDelayMicros (NominalDiffTime -> Int
Ops.micros (UTCTime -> UTCTime -> NominalDiffTime
diffUTCTime UTCTime
nextMinute UTCTime
now)))

-- | Why the scheduler woke.
data WakeReason = WakeShutdown | WakeMinute | WakeRunNow

-- | Block until @timerVar@ elapses, a run-now signal arrives, or shutdown.
-- Shutdown wins, then an elapsed minute boundary, then a pending run-now.
waitForWake :: (MonadIO m) => TVar WorkerState -> TVar Bool -> TVar Bool -> m WakeReason
waitForWake :: forall (m :: * -> *).
MonadIO m =>
TVar WorkerState -> TVar Bool -> TVar Bool -> m WakeReason
waitForWake TVar WorkerState
stateVar TVar Bool
runNowVar TVar Bool
timerVar = IO WakeReason -> m WakeReason
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO (IO WakeReason -> m WakeReason)
-> (STM WakeReason -> IO WakeReason)
-> STM WakeReason
-> m WakeReason
forall b c a. (b -> c) -> (a -> b) -> a -> c
. STM WakeReason -> IO WakeReason
forall (m :: * -> *) a. MonadIO m => STM a -> m a
atomically (STM WakeReason -> m WakeReason) -> STM WakeReason -> m WakeReason
forall a b. (a -> b) -> a -> b
$ do
  state <- TVar WorkerState -> STM WorkerState
forall a. TVar a -> STM a
readTVar TVar WorkerState
stateVar
  case state of
    WorkerState
ShuttingDown -> WakeReason -> STM WakeReason
forall a. a -> STM a
forall (f :: * -> *) a. Applicative f => a -> f a
pure WakeReason
WakeShutdown
    WorkerState
_ -> do
      timedOut <- TVar Bool -> STM Bool
forall a. TVar a -> STM a
readTVar TVar Bool
timerVar
      reason <-
        if timedOut
          then pure WakeMinute
          else do
            requested <- readTVar runNowVar
            unless requested retry
            pure WakeRunNow
      writeTVar runNowVar False
      pure reason

-- | Snapshot of the current 'WorkerState' for use outside STM.
isShuttingDown :: (MonadIO m) => TVar WorkerState -> m Bool
isShuttingDown :: forall (m :: * -> *). MonadIO m => TVar WorkerState -> m Bool
isShuttingDown TVar WorkerState
stateVar = (WorkerState -> WorkerState -> Bool
forall a. Eq a => a -> a -> Bool
== WorkerState
ShuttingDown) (WorkerState -> Bool) -> m WorkerState -> m Bool
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> TVar WorkerState -> m WorkerState
forall (m :: * -> *) a. MonadIO m => TVar a -> m a
readTVarIO TVar WorkerState
stateVar