{-# LANGUAGE OverloadedStrings #-}

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

import Arbiter.Core.CronSchedule qualified as CS
import Arbiter.Core.Enum (showText)
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, getSchema, withDbTransaction)
import Arbiter.Core.Operations qualified as Ops
import Control.Monad (unless, void, when)
import Control.Monad.IO.Class (MonadIO)
import Data.Either (fromRight)
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 (NominalDiffTime, UTCTime, addUTCTime, diffUTCTime, getCurrentTime)
import System.Cron (CronSchedule, parseCronSchedule)
import UnliftIO
  ( MonadUnliftIO
  , SomeException
  , TVar
  , atomically
  , checkSTM
  , liftIO
  , orElse
  , readTVar
  , registerDelay
  , tryAny
  , writeTVar
  )

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

-- | Upsert the default expression, overlap, timezone and initial enabled flag of
-- each 'CronJob' into the @cron_schedules@ table. Preserves user overrides. An
-- existing row keeps its enabled state.
initCronSchedules
  :: (MonadArbiter m)
  => LogConfig
  -> Text
  -- ^ Queue name
  -> [CronJob payload]
  -> m ()
initCronSchedules :: forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
logCfg Text
queueName [CronJob payload]
jobs = do
  schemaName <- m Text
forall (m :: * -> *). MonadArbiter m => m Text
getSchema
  for_ jobs $ \CronJob payload
cron ->
    Text
-> Text -> Text -> Text -> Text -> Maybe Text -> Bool -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text
-> Text -> Text -> Text -> Text -> Maybe Text -> Bool -> 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)
      (CronJob payload -> Bool
forall payload. CronJob payload -> Bool
initiallyEnabled CronJob payload
cron)
  tryLog logCfg Info $
    "Cron schedules initialized: " <> showText (length jobs) <> " schedule(s) upserted"

-- | Scheduler entry point. Exits when the worker state becomes 'ShuttingDown'.
runCronScheduler
  :: (QueueOperation m payload)
  => LogConfig
  -> TVar WorkerState
  -> TVar Bool
  -- ^ Set by the run-now listener when a schedule this pool owns is requested.
  -> SchemaName
  -> Text
  -- ^ Queue name (recorded on each schedule row).
  -> [CronJob payload]
  -> m ()
runCronScheduler :: forall (m :: * -> *) payload.
QueueOperation m payload =>
LogConfig
-> TVar WorkerState
-> TVar Bool
-> Text
-> Text
-> [CronJob payload]
-> m ()
runCronScheduler LogConfig
logCfg TVar WorkerState
stateVar TVar Bool
runNowVar Text
schemaName Text
queueName [CronJob payload]
jobs = do
  LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
logCfg Text
queueName [CronJob payload]
jobs
  cronLog <- LogConfig -> m CronLog
forall (m :: * -> *). MonadIO m => LogConfig -> m CronLog
newCronLog LogConfig
logCfg
  startupNow <- liftIO getCurrentTime
  shuttingDown <- atomically (isShuttingDown stateVar)
  unless shuttingDown $ catchUp cronLog startupNow
  logCron cronLog Info $ "Cron scheduler started with " <> showText (length jobs) <> " schedule(s)"
  loop cronLog
  where
    catchUp :: CronLog -> UTCTime -> m ()
catchUp CronLog
cronLog UTCTime
now = do
      CronLog -> Text -> [CronJob payload] -> UTCTime -> m ()
forall payload (m :: * -> *).
QueueOperation m payload =>
CronLog -> Text -> [CronJob payload] -> UTCTime -> m ()
processRunRequests CronLog
cronLog Text
schemaName [CronJob payload]
jobs UTCTime
now
      CronLog -> Text -> Text -> [CronJob payload] -> UTCTime -> m ()
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
    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 <- 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
          IO UTCTime -> m UTCTime
forall a. IO a -> m a
forall (m :: * -> *) a. MonadIO m => IO a -> m a
liftIO IO UTCTime
getCurrentTime m UTCTime -> (UTCTime -> m ()) -> m ()
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= CronLog -> UTCTime -> m ()
catchUp CronLog
cronLog
          CronLog -> m ()
loop CronLog
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 tick commits independently. Backfill ticks
-- acquire a per-schedule advisory lock for their transaction.
--
-- A failed tick is logged and skipped once a later tick fires. The watermark
-- of a Backfill schedule stays behind a failed newest tick, and the next pass
-- retries it.
processCronCatchUp
  :: (QueueOperation m payload)
  => CronLog
  -> SchemaName
  -> 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 = (CronJob payload -> m ()) -> [CronJob payload] -> m ()
forall (t :: * -> *) (f :: * -> *) a b.
(Foldable t, Applicative f) =>
(a -> f b) -> t a -> f ()
traverse_ CronJob payload -> m ()
processOneCron [CronJob payload]
jobs
  where
    currentTick :: UTCTime
currentTick = UTCTime -> UTCTime
truncateToMinute UTCTime
now
    processOneCron :: CronJob payload -> m ()
processOneCron CronJob payload
cron = do
      m (Either SomeException ()) -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (m (Either SomeException ()) -> m ())
-> m (Either SomeException ()) -> m ()
forall a b. (a -> b) -> a -> b
$ CronLog -> Text -> m () -> m (Either SomeException ())
forall (m :: * -> *) a.
MonadUnliftIO m =>
CronLog -> Text -> m a -> m (Either SomeException a)
tryCron CronLog
cronLog (CronJob payload -> Text
forall payload. CronJob payload -> Text
cronLabel CronJob payload
cron Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" tick") (m () -> m (Either SomeException ()))
-> m () -> m (Either SomeException ())
forall a b. (a -> b) -> a -> b
$ do
        mRow <- Text -> Text -> m (Maybe CronScheduleRow)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe CronScheduleRow)
Ops.getCronScheduleByName Text
schemaName (CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron)
        completed <- processOne mRow cron
        when completed $ void $ Ops.touchCronChecked schemaName [name cron] currentTick
    processOne :: Maybe CronScheduleRow -> CronJob payload -> m Bool
processOne Maybe CronScheduleRow
mRow 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 -> Bool -> m Bool
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure Bool
True
      ParseError Text
expr String
err ->
        Bool
True Bool -> m () -> m Bool
forall a b. a -> m b -> m a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ CronLog -> LogLevel -> Text -> m ()
forall (m :: * -> *).
MonadIO m =>
CronLog -> LogLevel -> Text -> m ()
logCron CronLog
cronLog LogLevel
Error (CronJob payload -> Text
forall payload. CronJob payload -> Text
cronLabel 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 ->
        Bool
True Bool -> m () -> m Bool
forall a b. a -> m b -> m a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ CronLog -> LogLevel -> Text -> m ()
forall (m :: * -> *).
MonadIO m =>
CronLog -> LogLevel -> Text -> m ()
logCron CronLog
cronLog LogLevel
Error (CronJob payload -> Text
forall payload. CronJob payload -> Text
cronLabel 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 ticksToFire :: [UTCTime]
ticksToFire =
              (UTCTime -> Bool) -> [UTCTime] -> [UTCTime]
forall a. (a -> Bool) -> [a] -> [a]
filter (Maybe Text -> CronSchedule -> UTCTime -> Bool
matchesInTimezone Maybe Text
effectiveTz CronSchedule
sched) ([UTCTime] -> [UTCTime]) -> [UTCTime] -> [UTCTime]
forall a b. (a -> b) -> a -> b
$
                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
            fire :: UTCTime -> m TickOutcome
fire = CronLog
-> Text
-> Text
-> CronJob payload
-> OverlapPolicy
-> UTCTime
-> UTCTime
-> m TickOutcome
forall (m :: * -> *) payload.
QueueOperation m payload =>
CronLog
-> Text
-> Text
-> CronJob payload
-> OverlapPolicy
-> UTCTime
-> UTCTime
-> m TickOutcome
fireCronTick CronLog
cronLog Text
schemaName Text
queueName CronJob payload
cron OverlapPolicy
effectiveOv UTCTime
currentTick
        (replayed, lastOutcome) <- (TickOutcome -> Bool)
-> (UTCTime -> m TickOutcome)
-> UTCTime
-> [UTCTime]
-> m (Int, TickOutcome)
forall (m :: * -> *).
Monad m =>
(TickOutcome -> Bool)
-> (UTCTime -> m TickOutcome)
-> UTCTime
-> [UTCTime]
-> m (Int, TickOutcome)
fireTicks (OverlapPolicy -> TickOutcome -> Bool
stopsAt OverlapPolicy
effectiveOv) UTCTime -> m TickOutcome
fire UTCTime
currentTick [UTCTime]
ticksToFire
        when (replayed > 0)
          $ logCron cronLog Info
          $ cronLabel cron <> " replayed " <> showText replayed <> " missed tick(s)"
        pure (advancesWatermark (backfill cron) lastOutcome)

-- | Fire one catch-up tick in its own transaction and log how it ended.
fireCronTick
  :: (QueueOperation m payload)
  => CronLog -> SchemaName -> Text -> CronJob payload -> OverlapPolicy -> UTCTime -> UTCTime -> m TickOutcome
fireCronTick :: forall (m :: * -> *) payload.
QueueOperation m payload =>
CronLog
-> Text
-> Text
-> CronJob payload
-> OverlapPolicy
-> UTCTime
-> UTCTime
-> m TickOutcome
fireCronTick CronLog
cronLog Text
schemaName Text
queueName CronJob payload
cron OverlapPolicy
effectiveOv UTCTime
currentTick UTCTime
tick = do
  result <- CronLog
-> Text -> m TickOutcome -> m (Either SomeException TickOutcome)
forall (m :: * -> *) a.
MonadUnliftIO m =>
CronLog -> Text -> m a -> m (Either SomeException a)
tryCron CronLog
cronLog (CronJob payload -> Text
forall payload. CronJob payload -> Text
cronLabel CronJob payload
cron Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" insert") (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 haveLeader
      then TickHandled <$> insertCronJob schemaName cron effectiveOv (tickKindFor currentTick tick) tick
      else pure TickNoLeader
  let outcome = TickOutcome -> Either SomeException TickOutcome -> TickOutcome
forall b a. b -> Either a b -> b
fromRight TickOutcome
TickFailed Either SomeException TickOutcome
result
  case outcome of
    TickHandled Bool
_ -> 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
$ CronJob payload -> Text
forall payload. CronJob payload -> Text
cronLabel CronJob payload
cron Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" processed at " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> UTCTime -> Text
formatMinute UTCTime
tick
    TickOutcome
TickNoLeader -> 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
$ CronJob payload -> Text
forall payload. CronJob payload -> Text
cronLabel CronJob payload
cron Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" skipped, another pool holds the lock"
    TickOutcome
TickFailed -> () -> m ()
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure ()
  pure outcome

-- | Fire @ticks@ in order until @stop@ holds. Returns the replayed count and the last outcome.
fireTicks
  :: (Monad m) => (TickOutcome -> Bool) -> (UTCTime -> m TickOutcome) -> UTCTime -> [UTCTime] -> m (Int, TickOutcome)
fireTicks :: forall (m :: * -> *).
Monad m =>
(TickOutcome -> Bool)
-> (UTCTime -> m TickOutcome)
-> UTCTime
-> [UTCTime]
-> m (Int, TickOutcome)
fireTicks TickOutcome -> Bool
stop UTCTime -> m TickOutcome
fire UTCTime
currentTick = (Int, TickOutcome) -> [UTCTime] -> m (Int, TickOutcome)
go (Int
0, Bool -> TickOutcome
TickHandled Bool
False)
  where
    go :: (Int, TickOutcome) -> [UTCTime] -> m (Int, TickOutcome)
go (Int, TickOutcome)
acc [] = (Int, TickOutcome) -> m (Int, TickOutcome)
forall a. a -> m a
forall (f :: * -> *) a. Applicative f => a -> f a
pure (Int, TickOutcome)
acc
    go (Int
replayed, TickOutcome
_) (UTCTime
tick : [UTCTime]
ticks) = do
      outcome <- UTCTime -> m TickOutcome
fire UTCTime
tick
      let next = (Int
replayed Int -> Int -> Int
forall a. Num a => a -> a -> a
+ UTCTime -> TickOutcome -> Int
replayedBy UTCTime
tick TickOutcome
outcome, TickOutcome
outcome)
      if stop outcome then pure next else go next ticks
    replayedBy :: UTCTime -> TickOutcome -> Int
replayedBy UTCTime
tick TickOutcome
outcome = if UTCTime
tick UTCTime -> UTCTime -> Bool
forall a. Eq a => a -> a -> Bool
/= UTCTime
currentTick Bool -> Bool -> Bool
&& TickOutcome
outcome TickOutcome -> TickOutcome -> Bool
forall a. Eq a => a -> a -> Bool
== Bool -> TickOutcome
TickHandled Bool
True then Int
1 else Int
0

-- | Whether a tick outcome ends the catch-up pass under an overlap policy.
stopsAt :: OverlapPolicy -> TickOutcome -> Bool
stopsAt :: OverlapPolicy -> TickOutcome -> Bool
stopsAt OverlapPolicy
AllowOverlap = (TickOutcome -> TickOutcome -> Bool
forall a. Eq a => a -> a -> Bool
== TickOutcome
TickNoLeader)
stopsAt OverlapPolicy
SkipOverlap = (TickOutcome -> TickOutcome -> Bool
forall a. Eq a => a -> a -> Bool
/= TickOutcome
TickFailed)

-- | Whether the last tick outcome lets the pass advance the watermark to the current tick.
advancesWatermark :: BackfillPolicy -> TickOutcome -> Bool
advancesWatermark :: BackfillPolicy -> TickOutcome -> Bool
advancesWatermark BackfillPolicy
_ (TickHandled Bool
_) = Bool
True
advancesWatermark BackfillPolicy
NoBackfill TickOutcome
TickFailed = Bool
True
advancesWatermark BackfillPolicy
_ TickOutcome
_ = Bool
False

-- | How one tick's transaction ended. 'TickHandled' carries whether this pool
-- fired the gate.
data TickOutcome = TickHandled Bool | TickNoLeader | TickFailed
  deriving stock (TickOutcome -> TickOutcome -> Bool
(TickOutcome -> TickOutcome -> Bool)
-> (TickOutcome -> TickOutcome -> Bool) -> Eq TickOutcome
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: TickOutcome -> TickOutcome -> Bool
== :: TickOutcome -> TickOutcome -> Bool
$c/= :: TickOutcome -> TickOutcome -> Bool
/= :: TickOutcome -> TickOutcome -> Bool
Eq)

-- | '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

-- | 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
oneMinute 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
  | Bool -> Bool
not Bool
isEnabled = Resolved
Disabled
  | Bool
otherwise = (String -> Resolved)
-> (CronSchedule -> Resolved)
-> Either String CronSchedule
-> Resolved
forall a c b. (a -> c) -> (b -> c) -> Either a b -> c
either (Text -> String -> Resolved
ParseError Text
expr) CronSchedule -> Resolved
withSchedule (Text -> Either String CronSchedule
parseCronSchedule Text
expr)
  where
    (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, CronJob payload -> Bool
forall payload. CronJob payload -> Bool
initiallyEnabled CronJob payload
cron)
      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
        )
    withSchedule :: CronSchedule -> Resolved
withSchedule CronSchedule
sched
      | Just Text
tzName <- Maybe Text
zone, Maybe TZ -> Bool
forall a. Maybe a -> Bool
isNothing (Text -> Maybe TZ
resolveTZ Text
tzName) = Text -> Resolved
InvalidTimezone Text
tzName
      | Bool
otherwise = OverlapPolicy -> CronSchedule -> Maybe Text -> Resolved
Effective OverlapPolicy
overlapPolicy CronSchedule
sched Maybe Text
zone

-- | Fire the gate for @tick@, insert its job if the gate opened, and advance
-- the watermark to @tick@. Returns whether this call fired the gate.
insertCronJob
  :: (QueueOperation m payload)
  => SchemaName -> CronJob payload -> OverlapPolicy -> TickKind -> UTCTime -> m Bool
insertCronJob :: forall (m :: * -> *) payload.
QueueOperation m payload =>
Text
-> CronJob payload
-> OverlapPolicy
-> TickKind
-> UTCTime
-> m Bool
insertCronJob Text
schemaName CronJob payload
cron OverlapPolicy
effectiveOv TickKind
kind UTCTime
tick = 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 -> UTCTime -> Text
mkDedupKeyFromParts (CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron) OverlapPolicy
effectiveOv 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 [name cron] tick
  pure fired

data RunNowOutcome = Fired | Skipped | NotRequested

-- | Claim and fire every schedule with a pending run request. A 'SkipOverlap'
-- schedule reuses its constant dedup key, so its 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 -> SchemaName -> [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_ CronJob payload -> m ()
claimAndFire ((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)
    tick :: UTCTime
tick = UTCTime -> UTCTime
truncateToMinute UTCTime
now
    -- Every run-now failure is reported.
    claimAndFire :: CronJob payload -> m ()
claimAndFire 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 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
$ CronJob payload -> Text
forall payload. CronJob payload -> Text
cronLabel 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
$ CronJob payload -> Text
forall payload. CronJob payload -> Text
cronLabel 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
$ CronJob payload -> Text
forall payload. CronJob payload -> Text
cronLabel CronJob payload
cron Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" run-now skipped, a job is already active"
    fireClaimed :: CronJob payload -> CronScheduleRow -> m RunNowOutcome
fireClaimed 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 -> Text -> UTCTime -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> UTCTime -> m Int64
Ops.touchCronManualRun Text
schemaName (CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron) UTCTime
tick
          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

-- | A schedule's name as a log prefix.
cronLabel :: CronJob payload -> Text
cronLabel :: forall payload. CronJob payload -> Text
cronLabel CronJob payload
cron = 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
"'"

-- | 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. Report new failures and recovery. Suppress repeated
-- identical failures.
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)

-- | The dedup key for a schedule's tick. An 'AllowOverlap' key includes the UTC tick minute.
mkDedupKeyFromParts :: Text -> OverlapPolicy -> UTCTime -> Text
mkDedupKeyFromParts :: Text -> OverlapPolicy -> UTCTime -> Text
mkDedupKeyFromParts Text
jobName OverlapPolicy
overlapPolicy UTCTime
tick = case OverlapPolicy
overlapPolicy of
  OverlapPolicy
SkipOverlap -> Text -> Text
skipOverlapKey Text
jobName
  OverlapPolicy
AllowOverlap -> Text -> Text
skipOverlapKey Text
jobName Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
":" Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> UTCTime -> Text
formatMinute 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.
maxTickDelay :: NominalDiffTime
maxTickDelay :: NominalDiffTime
maxTickDelay = NominalDiffTime
120

-- | Delay in microseconds until the next minute boundary, clamped to
-- zero through 'maxTickDelay'.
computeDelayMicros :: UTCTime -> Int
computeDelayMicros :: UTCTime -> Int
computeDelayMicros UTCTime
now =
  let nextMinute :: UTCTime
nextMinute = UTCTime -> UTCTime
truncateToMinute (NominalDiffTime -> UTCTime -> UTCTime
addUTCTime NominalDiffTime
oneMinute 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 (NominalDiffTime -> Int
Ops.micros NominalDiffTime
maxTickDelay) (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 = STM WakeReason -> m 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
      reason <- (WakeReason
WakeMinute WakeReason -> STM () -> STM WakeReason
forall a b. a -> STM b -> STM a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ TVar Bool -> STM ()
awaitSet TVar Bool
timerVar) STM WakeReason -> STM WakeReason -> STM WakeReason
forall a. STM a -> STM a -> STM a
`orElse` (WakeReason
WakeRunNow WakeReason -> STM () -> STM WakeReason
forall a b. a -> STM b -> STM a
forall (f :: * -> *) a b. Functor f => a -> f b -> f a
<$ TVar Bool -> STM ()
awaitSet TVar Bool
runNowVar)
      writeTVar runNowVar False
      pure reason
  where
    awaitSet :: TVar Bool -> STM ()
awaitSet TVar Bool
var = TVar Bool -> STM Bool
forall a. TVar a -> STM a
readTVar TVar Bool
var STM Bool -> (Bool -> STM ()) -> STM ()
forall a b. STM a -> (a -> STM b) -> STM b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Bool -> STM ()
checkSTM