{-# LANGUAGE OverloadedStrings #-}
{-# LANGUAGE TypeFamilies #-}
{-# OPTIONS_GHC -Wno-incomplete-uni-patterns -Wno-x-partial #-}

-- | Cron scheduler tests, instantiated for each backend.
module Arbiter.Test.Worker.Cron (cronSpec) where

import Arbiter.Core.CronSchedule (CronScheduleUpdate (..))
import Arbiter.Core.CronSchedule qualified as CS
import Arbiter.Core.HighLevel (QueueOperation)
import Arbiter.Core.HighLevel qualified as HL
import Arbiter.Core.Job.Types (DedupKey (IgnoreDuplicate), JobRead, dedupKey, defaultJob, payload)
import Arbiter.Core.MonadArbiter (withDbTransaction)
import Arbiter.Core.Operations qualified as Ops
import Arbiter.Worker.Cron
  ( BackfillPolicy (..)
  , CronJob (..)
  , OverlapPolicy (..)
  , cronJob
  , cronJobInTimezone
  , formatMinute
  , initCronSchedules
  , newCronLog
  , processCronCatchUp
  , processRunRequests
  )
import Arbiter.Worker.Logger (silentLogConfig)
import Control.Monad (void)
import Control.Monad.IO.Class (liftIO)
import Data.Foldable (traverse_)
import Data.Maybe (isJust)
import Data.Text (Text)
import Data.Text qualified as T
import Data.Time (UTCTime, getCurrentTime)
import Database.PostgreSQL.Simple (Only (..))
import Database.PostgreSQL.Simple qualified as PG
import Test.Hspec (Spec, before, describe, expectationFailure, it, shouldBe, shouldNotBe, shouldSatisfy)
import UnliftIO (newEmptyMVar, putMVar, takeMVar)
import UnliftIO.Async (wait, withAsync)

import Arbiter.Test.Setup (mkTime, toQuery, withConn)
import Arbiter.Test.Worker.Backend (TestBackend (..))

-- | 'processCronCatchUp' under a fresh gate store.
catchUpAt
  :: (QueueOperation m payload)
  => Text
  -> Text
  -> [CronJob payload]
  -> UTCTime
  -> m ()
catchUpAt :: forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload]
jobs UTCTime
tick = do
  cronLog <- LogConfig -> m CronLog
forall (m :: * -> *). MonadIO m => LogConfig -> m CronLog
newCronLog LogConfig
silentLogConfig
  processCronCatchUp cronLog schema table jobs tick

-- | 'processRunRequests' under a fresh gate store.
runRequestsAt
  :: (QueueOperation m payload)
  => Text
  -> [CronJob payload]
  -> UTCTime
  -> m ()
runRequestsAt :: forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> [CronJob payload] -> UTCTime -> m ()
runRequestsAt Text
schema [CronJob payload]
jobs UTCTime
now = do
  cronLog <- LogConfig -> m CronLog
forall (m :: * -> *). MonadIO m => LogConfig -> m CronLog
newCronLog LogConfig
silentLogConfig
  processRunRequests cronLog schema jobs now

-- | An UPDATE on the schema's cron table by schedule name, with @?@ holes for the SET values.
cronUpdate :: Text -> Text -> PG.Query
cronUpdate :: Text -> Text -> Query
cronUpdate Text
schema Text
setClause = Text -> Query
toQuery (Text -> Query) -> Text -> Query
forall a b. (a -> b) -> a -> b
$ Text
"UPDATE " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text -> Text
CS.cronSchedulesTable Text
schema Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" SET " Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
setClause Text -> Text -> Text
forall a. Semigroup a => a -> a -> a
<> Text
" WHERE name = ?"

-- | Cron scheduler suite.
cronSpec
  :: forall payload m env
   . ( Eq payload
     , QueueOperation m payload
     , Show payload
     )
  => TestBackend payload m env
  -> Spec
cronSpec :: forall payload (m :: * -> *) env.
(Eq payload, QueueOperation m payload, Show payload) =>
TestBackend payload m env -> Spec
cronSpec TestBackend {Text
schema :: Text
schema :: forall payload (m :: * -> *) env. TestBackend payload m env -> Text
schema, Text
table :: Text
table :: forall payload (m :: * -> *) env. TestBackend payload m env -> Text
table, ByteString
connStr :: ByteString
connStr :: forall payload (m :: * -> *) env.
TestBackend payload m env -> ByteString
connStr, Text -> payload
mkSimple :: Text -> payload
mkSimple :: forall payload (m :: * -> *) env.
TestBackend payload m env -> Text -> payload
mkSimple, IO env
mkEnv :: IO env
mkEnv :: forall payload (m :: * -> *) env.
TestBackend payload m env -> IO env
mkEnv, forall a. env -> m a -> IO a
runM :: forall a. env -> m a -> IO a
runM :: forall payload (m :: * -> *) env.
TestBackend payload m env -> forall a. env -> m a -> IO a
runM} =
  IO env -> SpecWith env -> Spec
forall a. IO a -> SpecWith a -> Spec
before IO env
mkEnv (SpecWith env -> Spec) -> SpecWith env -> Spec
forall a b. (a -> b) -> a -> b
$ do
    String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"processCronCatchUp schedule matching" (SpecWith env -> SpecWith env) -> SpecWith env -> SpecWith env
forall a b. (a -> b) -> a -> b
$ do
      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"inserts a job when the schedule matches the tick time" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"every-min"
                Text
"* * * * *"
                OverlapPolicy
AllowOverlap
                (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"cron-fired"))
            tick :: UTCTime
tick = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
0 Int
0
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] UTCTime
tick

        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
10 NominalDiffTime
60 :: IO [JobRead payload]
        length jobs `shouldBe` 1
        payload (head jobs) `shouldBe` mkSimple "cron-fired"
        dedupKey (head jobs) `shouldBe` Just (IgnoreDuplicate "arbiter_cron:every-min:2025-06-15T12:00")

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"does not insert a job when the schedule does not match" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        -- "0 3 * * *" matches only at 03:00
        let Right CronJob payload
cron =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"nightly"
                Text
"0 3 * * *"
                OverlapPolicy
SkipOverlap
                (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"should-not-fire"))
            tick :: UTCTime
tick = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
0 Int
0 -- 12:00
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] UTCTime
tick

        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
10 NominalDiffTime
60 :: IO [JobRead payload]
        length jobs `shouldBe` 0

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"SkipOverlap: two ticks at different times produce only 1 job" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"every-min"
                Text
"* * * * *"
                OverlapPolicy
SkipOverlap
                (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"skip-test"))
            tick1 :: UTCTime
tick1 = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
0 Int
0
            tick2 :: UTCTime
tick2 = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
1 Int
0
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] UTCTime
tick1
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] UTCTime
tick2

        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
10 NominalDiffTime
60 :: IO [JobRead payload]
        -- Both ticks produce the same dedup key "arbiter_cron:every-min". One job results.
        length jobs `shouldBe` 1

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"AllowOverlap: two ticks at different times produce 2 jobs" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"every-min"
                Text
"* * * * *"
                OverlapPolicy
AllowOverlap
                (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"overlap-test"))
            tick1 :: UTCTime
tick1 = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
0 Int
0
            tick2 :: UTCTime
tick2 = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
1 Int
0
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] UTCTime
tick1
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] UTCTime
tick2

        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
10 NominalDiffTime
60 :: IO [JobRead payload]
        length jobs `shouldBe` 2

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"only matching schedules fire when multiple are provided" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cjAlways =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"always"
                Text
"* * * * *"
                OverlapPolicy
AllowOverlap
                (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"always-fires"))
            Right CronJob payload
cjNever =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"nightly"
                Text
"0 3 * * *"
                OverlapPolicy
AllowOverlap
                (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"should-not-fire"))
            tick :: UTCTime
tick = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
0 Int
0 -- 12:00
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cjAlways, CronJob payload
cjNever]
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cjAlways, CronJob payload
cjNever] UTCTime
tick

        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
10 NominalDiffTime
60 :: IO [JobRead payload]
        length jobs `shouldBe` 1
        payload (head jobs) `shouldBe` mkSimple "always-fires"

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"cronJobMake receives the tick time" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"time-check"
                Text
"* * * * *"
                OverlapPolicy
AllowOverlap
                (\TickKind
_ UTCTime
tickAt -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple (UTCTime -> Text
formatMinute UTCTime
tickAt)))
            tick :: UTCTime
tick = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
14 Int
30 Int
0
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] UTCTime
tick

        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
10 NominalDiffTime
60 :: IO [JobRead payload]
        length jobs `shouldBe` 1
        payload (head jobs) `shouldBe` mkSimple "2025-06-15T14:30"

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"advances a failed schedule and continues processing other schedules" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
good =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"good-1"
                Text
"* * * * *"
                OverlapPolicy
AllowOverlap
                (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"ok"))
            Right CronJob payload
bad =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"bad-1"
                Text
"* * * * *"
                OverlapPolicy
AllowOverlap
                (\TickKind
_ UTCTime
_ -> String -> JobWrite payload
forall a. String -> a
errorWithoutStackTrace String
"intentional builder failure")
            recovered :: CronJob payload
recovered = CronJob (ZonkAny 0)
forall {payload}. CronJob payload
bad {builder = \TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"recovered")}
            tick :: UTCTime
tick = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
0 Int
0
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
good, CronJob payload
forall {payload}. CronJob payload
bad]
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
good, CronJob payload
forall {payload}. CronJob payload
bad] UTCTime
tick

        rows <- env -> m [CronScheduleRow] -> IO [CronScheduleRow]
forall a. env -> m a -> IO a
runM env
env (m [CronScheduleRow] -> IO [CronScheduleRow])
-> m [CronScheduleRow] -> IO [CronScheduleRow]
forall a b. (a -> b) -> a -> b
$ Text -> Maybe Text -> m [CronScheduleRow]
forall (m :: * -> *).
MonadArbiter m =>
Text -> Maybe Text -> m [CronScheduleRow]
Ops.listCronSchedules Text
schema Maybe Text
forall a. Maybe a
Nothing
        let getRow Text
scheduleName = Text -> [(Text, CronScheduleRow)] -> Maybe CronScheduleRow
forall a b. Eq a => a -> [(a, b)] -> Maybe b
lookup Text
scheduleName [(CronScheduleRow -> Text
CS.name CronScheduleRow
row, CronScheduleRow
row) | CronScheduleRow
row <- [CronScheduleRow]
rows]
        case getRow "good-1" of
          Just CronScheduleRow
row -> CronScheduleRow -> Maybe UTCTime
CS.lastCheckedAt CronScheduleRow
row Maybe UTCTime -> Maybe UTCTime -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` UTCTime -> Maybe UTCTime
forall a. a -> Maybe a
Just UTCTime
tick
          Maybe CronScheduleRow
Nothing -> HasCallStack => String -> IO ()
String -> IO ()
expectationFailure String
"good-1 schedule missing"
        case getRow "bad-1" of
          Just CronScheduleRow
row -> CronScheduleRow -> Maybe UTCTime
CS.lastCheckedAt CronScheduleRow
row Maybe UTCTime -> Maybe UTCTime -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` UTCTime -> Maybe UTCTime
forall a. a -> Maybe a
Just UTCTime
tick
          Maybe CronScheduleRow
Nothing -> HasCallStack => String -> IO ()
String -> IO ()
expectationFailure String
"bad-1 schedule missing"

        -- The good cron's job was inserted.
        jobs <- runM env $ HL.claimNextVisibleJobs 10 60 :: IO [JobRead payload]
        map payload jobs `shouldBe` [mkSimple "ok"]

        runM env $ catchUpAt schema table [recovered] (mkTime 2025 6 15 12 1 0)
        recoveredJobs <- runM env $ HL.claimNextVisibleJobs 10 60 :: IO [JobRead payload]
        map payload recoveredJobs `shouldBe` [mkSimple "recovered"]

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"fires later backfill and live ticks after a tick keeps failing" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let failedTick :: UTCTime
failedTick = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
0 Int
0
            currentTick :: UTCTime
currentTick = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
2 Int
0
            Right CronJob payload
base =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"partial-backfill" Text
"* * * * *" OverlapPolicy
AllowOverlap ((TickKind -> UTCTime -> JobWrite payload)
 -> Either String (CronJob payload))
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall a b. (a -> b) -> a -> b
$ \TickKind
_ UTCTime
tick ->
                if UTCTime
tick UTCTime -> UTCTime -> Bool
forall a. Eq a => a -> a -> Bool
== UTCTime
failedTick
                  then String -> JobWrite payload
forall a. String -> a
errorWithoutStackTrace String
"persistent builder failure"
                  else payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple (UTCTime -> Text
formatMinute UTCTime
tick))
            cron :: CronJob payload
cron = CronJob payload
base {backfill = Backfill 600}
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          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.touchCronChecked Text
schema [CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron] (Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
11 Int
59 Int
0)
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] UTCTime
currentTick

        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
10 NominalDiffTime
60 :: IO [JobRead payload]
        map payload jobs `shouldBe` map (mkSimple . formatMinute) [mkTime 2025 6 15 12 1 0, currentTick]
        Just row <- runM env $ Ops.getCronScheduleByName schema (name cron)
        CS.lastCheckedAt row `shouldBe` Just currentTick

        runM env $ catchUpAt schema table [cron] (mkTime 2025 6 15 12 3 0)
        next <- runM env $ HL.claimNextVisibleJobs 10 60 :: IO [JobRead payload]
        map payload next `shouldBe` [mkSimple "2025-06-15T12:03"]

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"retries a failed newest backfill tick on the next pass" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let lastChecked :: UTCTime
lastChecked = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
11 Int
59 Int
0
            failedTick :: UTCTime
failedTick = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
1 Int
0
            nextTick :: UTCTime
nextTick = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
2 Int
0
            Right CronJob payload
base =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"retry-newest" Text
"* * * * *" OverlapPolicy
AllowOverlap ((TickKind -> UTCTime -> JobWrite payload)
 -> Either String (CronJob payload))
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall a b. (a -> b) -> a -> b
$ \TickKind
_ UTCTime
tick ->
                if UTCTime
tick UTCTime -> UTCTime -> Bool
forall a. Eq a => a -> a -> Bool
== UTCTime
failedTick
                  then String -> JobWrite payload
forall a. String -> a
errorWithoutStackTrace String
"transient builder failure"
                  else payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple (UTCTime -> Text
formatMinute UTCTime
tick))
            cron :: CronJob payload
cron = CronJob payload
base {backfill = Backfill 600}
            recovered :: CronJob payload
recovered = CronJob payload
cron {builder = \TickKind
_ UTCTime
tick -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple (UTCTime -> Text
formatMinute UTCTime
tick))}
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          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.touchCronChecked Text
schema [CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron] UTCTime
lastChecked
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] UTCTime
failedTick

        Just held <- env -> m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow)
forall a. env -> m a -> IO a
runM env
env (m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow))
-> m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow)
forall a b. (a -> b) -> a -> b
$ Text -> Text -> m (Maybe CronScheduleRow)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe CronScheduleRow)
Ops.getCronScheduleByName Text
schema (CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron)
        CS.lastCheckedAt held `shouldBe` Just (mkTime 2025 6 15 12 0 0)

        runM env $ catchUpAt schema table [recovered] nextTick
        jobs <- runM env $ HL.claimNextVisibleJobs 10 60 :: IO [JobRead payload]
        map payload jobs `shouldBe` map (mkSimple . formatMinute) [mkTime 2025 6 15 12 0 0, failedTick, nextTick]
        Just row <- runM env $ Ops.getCronScheduleByName schema (name cron)
        CS.lastCheckedAt row `shouldBe` Just nextTick

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"fires a later SkipOverlap tick after the oldest backfill tick fails" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let failedTick :: UTCTime
failedTick = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
0 Int
0
            currentTick :: UTCTime
currentTick = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
2 Int
0
            Right CronJob payload
base =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"skip-failed" Text
"* * * * *" OverlapPolicy
SkipOverlap ((TickKind -> UTCTime -> JobWrite payload)
 -> Either String (CronJob payload))
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall a b. (a -> b) -> a -> b
$ \TickKind
_ UTCTime
tick ->
                if UTCTime
tick UTCTime -> UTCTime -> Bool
forall a. Eq a => a -> a -> Bool
== UTCTime
failedTick
                  then String -> JobWrite payload
forall a. String -> a
errorWithoutStackTrace String
"persistent builder failure"
                  else payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple (UTCTime -> Text
formatMinute UTCTime
tick))
            cron :: CronJob payload
cron = CronJob payload
base {backfill = Backfill 600}
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          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.touchCronChecked Text
schema [CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron] (Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
11 Int
59 Int
0)
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] UTCTime
currentTick

        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
10 NominalDiffTime
60 :: IO [JobRead payload]
        map payload jobs `shouldBe` [mkSimple "2025-06-15T12:01"]
        Just row <- runM env $ Ops.getCronScheduleByName schema (name cron)
        CS.lastCheckedAt row `shouldBe` Just currentTick

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"retains the backfill watermark while another pool holds the leader lock" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let lastChecked :: UTCTime
lastChecked = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
11 Int
59 Int
0
            currentTick :: UTCTime
currentTick = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
1 Int
0
            Right CronJob payload
base = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"contended-backfill" Text
"* * * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
tick -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple (UTCTime -> Text
formatMinute UTCTime
tick)))
            cron :: CronJob payload
cron = CronJob payload
base {backfill = Backfill 600}
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          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.touchCronChecked Text
schema [CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron] UTCTime
lastChecked
        held <- IO (MVar Bool)
forall (m :: * -> *) a. MonadIO m => m (MVar a)
newEmptyMVar
        release <- newEmptyMVar
        let holdLeader = env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> (m () -> m ()) -> m () -> IO ()
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 () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
              got <- Text -> Text -> Text -> m Bool
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Text -> m Bool
Ops.tryAcquireCronLeader Text
schema Text
table (CronJob payload -> Text
forall payload. CronJob payload -> Text
name CronJob payload
cron)
              liftIO (putMVar held got)
              liftIO (takeMVar release)
        withAsync holdLeader $ \Async ()
holder -> do
          got <- MVar Bool -> IO Bool
forall (m :: * -> *) a. MonadIO m => MVar a -> m a
takeMVar MVar Bool
held
          got `shouldBe` True
          runM env $ catchUpAt schema table [cron] currentTick
          Just blocked <- runM env $ Ops.getCronScheduleByName schema (name cron)
          CS.lastCheckedAt blocked `shouldBe` Just lastChecked
          putMVar release ()
          wait holder

        runM env $ catchUpAt schema table [cron] currentTick
        jobs <- runM env $ HL.claimNextVisibleJobs 10 60 :: IO [JobRead payload]
        map payload jobs `shouldBe` map (mkSimple . formatMinute) [mkTime 2025 6 15 12 0 0, currentTick]

    String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"processRunRequests" (SpecWith env -> SpecWith env) -> SpecWith env -> SpecWith env
forall a b. (a -> b) -> a -> b
$ do
      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"fires a requested schedule the tick would not match" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"run-nightly"
                Text
"0 3 * * *"
                OverlapPolicy
AllowOverlap
                (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"manual"))
            now :: UTCTime
now = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
30 Int
0
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          _ <- Text -> Text -> m RunRequestOutcome
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m RunRequestOutcome
Ops.requestCronRun Text
schema Text
"run-nightly"
          runRequestsAt schema [cron] now

        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
10 NominalDiffTime
60 :: IO [JobRead payload]
        map payload jobs `shouldBe` [mkSimple "manual"]

        Just row <- runM env $ Ops.getCronScheduleByName schema "run-nightly"
        CS.runRequestedAt row `shouldBe` Nothing

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"keeps the request pending when the insert fails" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        -- The claim rolls back with the insert. A later pass fires it.
        let failing :: CronJob payload
            Right CronJob payload
failing =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"run-atomic"
                Text
"0 3 * * *"
                OverlapPolicy
AllowOverlap
                (\TickKind
_ UTCTime
_ -> String -> JobWrite payload
forall a. String -> a
errorWithoutStackTrace String
"intentional builder failure")
            Right CronJob payload
working = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"run-atomic" Text
"0 3 * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"manual"))
            now :: UTCTime
now = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
30 Int
0
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
failing]
          _ <- Text -> Text -> m RunRequestOutcome
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m RunRequestOutcome
Ops.requestCronRun Text
schema Text
"run-atomic"
          runRequestsAt schema [failing] now

        Just pendingRow <- env -> m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow)
forall a. env -> m a -> IO a
runM env
env (m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow))
-> m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow)
forall a b. (a -> b) -> a -> b
$ Text -> Text -> m (Maybe CronScheduleRow)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe CronScheduleRow)
Ops.getCronScheduleByName Text
schema Text
"run-atomic"
        CS.runRequestedAt pendingRow `shouldNotBe` Nothing

        runM env $ runRequestsAt schema [working] now
        jobs <- runM env $ HL.claimNextVisibleJobs 10 60 :: IO [JobRead payload]
        map payload jobs `shouldBe` [mkSimple "manual"]

        Just firedRow <- runM env $ Ops.getCronScheduleByName schema "run-atomic"
        CS.runRequestedAt firedRow `shouldBe` Nothing

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"records the manual run at the tick without advancing the gate" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"run-gate" Text
"0 3 * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"gate"))
            now :: UTCTime
now = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
30 Int
45
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          _ <- Text -> Text -> m RunRequestOutcome
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m RunRequestOutcome
Ops.requestCronRun Text
schema Text
"run-gate"
          runRequestsAt schema [cron] now

        Just row <- env -> m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow)
forall a. env -> m a -> IO a
runM env
env (m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow))
-> m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow)
forall a b. (a -> b) -> a -> b
$ Text -> Text -> m (Maybe CronScheduleRow)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe CronScheduleRow)
Ops.getCronScheduleByName Text
schema Text
"run-gate"
        CS.lastManualRunAt row `shouldBe` Just (mkTime 2025 6 15 12 30 0)
        CS.lastFiredAt row `shouldBe` Nothing

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"leaves last_manual_run_at alone when the run is skipped" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"run-skipmark" Text
"0 3 * * *" OverlapPolicy
SkipOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"skip"))
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          _ <- Text -> Text -> m RunRequestOutcome
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m RunRequestOutcome
Ops.requestCronRun Text
schema Text
"run-skipmark"
          runRequestsAt schema [cron] (mkTime 2025 6 15 12 30 0)
          _ <- Ops.requestCronRun schema "run-skipmark"
          runRequestsAt schema [cron] (mkTime 2025 6 15 12 31 0)

        Just row <- env -> m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow)
forall a. env -> m a -> IO a
runM env
env (m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow))
-> m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow)
forall a b. (a -> b) -> a -> b
$ Text -> Text -> m (Maybe CronScheduleRow)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe CronScheduleRow)
Ops.getCronScheduleByName Text
schema Text
"run-skipmark"
        CS.lastManualRunAt row `shouldBe` Just (mkTime 2025 6 15 12 30 0)

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"expires a request no pool claimed in time" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"run-expire" Text
"0 3 * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"expire"))
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          m RunRequestOutcome -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (m RunRequestOutcome -> m ()) -> m RunRequestOutcome -> m ()
forall a b. (a -> b) -> a -> b
$ Text -> Text -> m RunRequestOutcome
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m RunRequestOutcome
Ops.requestCronRun Text
schema Text
"run-expire"

        ByteString -> (Connection -> IO ()) -> IO ()
forall a. ByteString -> (Connection -> IO a) -> IO a
withConn ByteString
connStr ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn ->
          IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$
            Connection -> Query -> Only Text -> IO Int64
forall q. ToRow q => Connection -> Query -> q -> IO Int64
PG.execute
              Connection
conn
              (Text -> Text -> Query
cronUpdate Text
schema Text
"run_requested_at = NOW() - interval '10 minutes'")
              (Text -> Only Text
forall a. a -> Only a
Only (Text
"run-expire" :: Text))

        now <- IO UTCTime
getCurrentTime
        runM env $ runRequestsAt schema [cron] now
        jobs <- runM env $ HL.listJobs 100 0 :: IO [JobRead payload]
        map payload jobs `shouldBe` []

        Just row <- runM env $ Ops.getCronScheduleByName schema "run-expire"
        CS.runRequestedAt row `shouldBe` Nothing

        outcome <- runM env $ Ops.requestCronRun schema "run-expire"
        outcome `shouldBe` Ops.RunReqStamped

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"AllowOverlap: a manual run does not suppress a Backfill replay" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
base =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"run-backfill"
                Text
"0 * * * *"
                OverlapPolicy
AllowOverlap
                (\TickKind
_ UTCTime
tickAt -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple (UTCTime -> Text
formatMinute UTCTime
tickAt)))
            cron :: CronJob payload
cron = CronJob payload
base {backfill = Backfill 86400}
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          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.touchCronChecked Text
schema [Text
"run-backfill"] (Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
9 Int
0 Int
0)
          _ <- Text -> Text -> m RunRequestOutcome
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m RunRequestOutcome
Ops.requestCronRun Text
schema Text
"run-backfill"
          runRequestsAt schema [cron] (mkTime 2025 6 15 12 30 45)
          catchUpAt schema table [cron] (mkTime 2025 6 15 12 30 45)

        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
10 NominalDiffTime
60 :: IO [JobRead payload]
        let expected = (Text -> payload) -> [Text] -> [payload]
forall a b. (a -> b) -> [a] -> [b]
map Text -> payload
mkSimple [Text
"2025-06-15T10:00", Text
"2025-06-15T11:00", Text
"2025-06-15T12:00", Text
"2025-06-15T12:30"]
        length jobs `shouldBe` length expected
        map payload jobs `shouldSatisfy` all (`elem` expected)
        expected `shouldSatisfy` all (`elem` map payload jobs)

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"SkipOverlap: a manual run's active job dedups a Backfill replay" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
base =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"run-skipbf"
                Text
"0 * * * *"
                OverlapPolicy
SkipOverlap
                (\TickKind
_ UTCTime
tickAt -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple (UTCTime -> Text
formatMinute UTCTime
tickAt)))
            cron :: CronJob payload
cron = CronJob payload
base {backfill = Backfill 86400}
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          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.touchCronChecked Text
schema [Text
"run-skipbf"] (Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
9 Int
0 Int
0)
          _ <- Text -> Text -> m RunRequestOutcome
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m RunRequestOutcome
Ops.requestCronRun Text
schema Text
"run-skipbf"
          runRequestsAt schema [cron] (mkTime 2025 6 15 12 30 45)
          catchUpAt schema table [cron] (mkTime 2025 6 15 12 30 45)

        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
10 NominalDiffTime
60 :: IO [JobRead payload]
        map payload jobs `shouldBe` [mkSimple "2025-06-15T12:30"]

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"does nothing without a pending request" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"run-none" Text
"0 3 * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"nope"))
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> [CronJob payload] -> UTCTime -> m ()
runRequestsAt Text
schema [CronJob payload
cron] (Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
30 Int
0)

        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
10 NominalDiffTime
60 :: IO [JobRead payload]
        length jobs `shouldBe` 0

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"passes the builder a minute-truncated Live tick" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"run-tick"
                Text
"0 3 * * *"
                OverlapPolicy
AllowOverlap
                (\TickKind
_ UTCTime
tickAt -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple (String -> Text
T.pack (UTCTime -> String
forall a. Show a => a -> String
show UTCTime
tickAt))))
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          _ <- Text -> Text -> m RunRequestOutcome
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m RunRequestOutcome
Ops.requestCronRun Text
schema Text
"run-tick"
          runRequestsAt schema [cron] (mkTime 2025 6 15 12 30 45)

        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
10 NominalDiffTime
60 :: IO [JobRead payload]
        map payload jobs `shouldBe` [mkSimple (T.pack (show (mkTime 2025 6 15 12 30 0)))]

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"refuses a second request while one is still pending" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"run-coalesce" Text
"0 3 * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"once"))
        outcomes <- env
-> m (RunRequestOutcome, RunRequestOutcome)
-> IO (RunRequestOutcome, RunRequestOutcome)
forall a. env -> m a -> IO a
runM env
env (m (RunRequestOutcome, RunRequestOutcome)
 -> IO (RunRequestOutcome, RunRequestOutcome))
-> m (RunRequestOutcome, RunRequestOutcome)
-> IO (RunRequestOutcome, RunRequestOutcome)
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          first <- Text -> Text -> m RunRequestOutcome
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m RunRequestOutcome
Ops.requestCronRun Text
schema Text
"run-coalesce"
          second <- Ops.requestCronRun schema "run-coalesce"
          pure (first, second)
        outcomes `shouldBe` (Ops.RunReqStamped, Ops.RunReqPending)

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"accepts a fresh request once the pending one is claimed" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"run-again" Text
"0 3 * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"again"))
        outcome <- env -> m RunRequestOutcome -> IO RunRequestOutcome
forall a. env -> m a -> IO a
runM env
env (m RunRequestOutcome -> IO RunRequestOutcome)
-> m RunRequestOutcome -> IO RunRequestOutcome
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          _ <- Text -> Text -> m RunRequestOutcome
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m RunRequestOutcome
Ops.requestCronRun Text
schema Text
"run-again"
          _ <- Ops.claimCronRun schema "run-again"
          Ops.requestCronRun schema "run-again"
        outcome `shouldBe` Ops.RunReqStamped

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"claims a request exactly once across pools" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"run-once" Text
"0 3 * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"once"))
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          m RunRequestOutcome -> m ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (m RunRequestOutcome -> m ()) -> m RunRequestOutcome -> m ()
forall a b. (a -> b) -> a -> b
$ Text -> Text -> m RunRequestOutcome
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m RunRequestOutcome
Ops.requestCronRun Text
schema Text
"run-once"

        won <- env -> m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow)
forall a. env -> m a -> IO a
runM env
env (m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow))
-> m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow)
forall a b. (a -> b) -> a -> b
$ Text -> Text -> m (Maybe CronScheduleRow)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe CronScheduleRow)
Ops.claimCronRun Text
schema Text
"run-once"
        lost <- runM env $ Ops.claimCronRun schema "run-once"
        fmap CS.name won `shouldBe` Just "run-once"
        fmap CS.name lost `shouldBe` Nothing

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"SkipOverlap: a request is skipped while a job is already active" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"run-skip" Text
"0 3 * * *" OverlapPolicy
SkipOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"skip"))
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          _ <- Text -> Text -> m RunRequestOutcome
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m RunRequestOutcome
Ops.requestCronRun Text
schema Text
"run-skip"
          runRequestsAt schema [cron] (mkTime 2025 6 15 12 30 0)
          _ <- Ops.requestCronRun schema "run-skip"
          runRequestsAt schema [cron] (mkTime 2025 6 15 12 31 0)

        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
10 NominalDiffTime
60 :: IO [JobRead payload]
        length jobs `shouldBe` 1

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"disabling a schedule drops its pending request" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"run-off" Text
"0 3 * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"nope"))
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          _ <- Text -> Text -> m RunRequestOutcome
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m RunRequestOutcome
Ops.requestCronRun Text
schema Text
"run-off"
          _ <-
            Ops.updateCronSchedule
              schema
              "run-off"
              CronScheduleUpdate
                { overrideExpression = Nothing
                , overrideOverlap = Nothing
                , overrideTimezone = Nothing
                , enabled = Just False
                }
          runRequestsAt schema [cron] (mkTime 2025 6 15 12 30 0)

        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
10 NominalDiffTime
60 :: IO [JobRead payload]
        length jobs `shouldBe` 0
        Just row <- runM env $ Ops.getCronScheduleByName schema "run-off"
        CS.runRequestedAt row `shouldBe` Nothing

    String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"initCronSchedules" (SpecWith env -> SpecWith env) -> SpecWith env -> SpecWith env
forall a b. (a -> b) -> a -> b
$ do
      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"upserts rows" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cj1 = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"test-a" Text
"0 3 * * *" OverlapPolicy
SkipOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"a"))
            Right CronJob payload
cj2 = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"test-b" Text
"*/5 * * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"b"))
        -- Phase 1: Insert
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cj1, CronJob payload
cj2]
        rows <- env -> m [CronScheduleRow] -> IO [CronScheduleRow]
forall a. env -> m a -> IO a
runM env
env (m [CronScheduleRow] -> IO [CronScheduleRow])
-> m [CronScheduleRow] -> IO [CronScheduleRow]
forall a b. (a -> b) -> a -> b
$ Text -> Maybe Text -> m [CronScheduleRow]
forall (m :: * -> *).
MonadArbiter m =>
Text -> Maybe Text -> m [CronScheduleRow]
Ops.listCronSchedules Text
schema Maybe Text
forall a. Maybe a
Nothing
        length rows `shouldBe` 2
        map CS.name rows `shouldBe` ["test-a", "test-b"]
        map CS.defaultExpression rows `shouldBe` ["0 3 * * *", "*/5 * * * *"]

        -- Phase 2: Re-upsert with a modified expression. The row is updated in place.
        let Right cj1' = cronJob "test-a" "*/10 * * * *" SkipOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"a"))
        runM env $ initCronSchedules silentLogConfig table [cj1', cj2]
        rows2 <- runM env $ Ops.listCronSchedules schema Nothing
        length rows2 `shouldBe` 2
        map CS.defaultExpression rows2 `shouldBe` ["*/10 * * * *", "*/5 * * * *"]

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"skips disabled schedules" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"disabled-test" Text
"* * * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"should-skip"))
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          _ <-
            Text -> Text -> CronScheduleUpdate -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> CronScheduleUpdate -> m Int64
Ops.updateCronSchedule
              Text
schema
              Text
"disabled-test"
              CronScheduleUpdate
                { overrideExpression :: Maybe (Maybe Text)
overrideExpression = Maybe (Maybe Text)
forall a. Maybe a
Nothing
                , overrideOverlap :: Maybe (Maybe Text)
overrideOverlap = Maybe (Maybe Text)
forall a. Maybe a
Nothing
                , overrideTimezone :: Maybe (Maybe Text)
overrideTimezone = Maybe (Maybe Text)
forall a. Maybe a
Nothing
                , enabled :: Maybe Bool
enabled = Bool -> Maybe Bool
forall a. a -> Maybe a
Just Bool
False
                }
          catchUpAt schema table [cron] (mkTime 2025 6 15 12 0 0)

        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
10 NominalDiffTime
60 :: IO [JobRead payload]
        length jobs `shouldBe` 0

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"uses DB expression override over default" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        -- Create a schedule that fires every minute
        let Right CronJob payload
cron = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"override-test" Text
"* * * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"override"))
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          _ <-
            Text -> Text -> CronScheduleUpdate -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> CronScheduleUpdate -> m Int64
Ops.updateCronSchedule
              Text
schema
              Text
"override-test"
              CronScheduleUpdate
                { overrideExpression :: Maybe (Maybe Text)
overrideExpression = Maybe Text -> Maybe (Maybe Text)
forall a. a -> Maybe a
Just (Text -> Maybe Text
forall a. a -> Maybe a
Just Text
"0 3 * * *")
                , overrideOverlap :: Maybe (Maybe Text)
overrideOverlap = Maybe (Maybe Text)
forall a. Maybe a
Nothing
                , overrideTimezone :: Maybe (Maybe Text)
overrideTimezone = Maybe (Maybe Text)
forall a. Maybe a
Nothing
                , enabled :: Maybe Bool
enabled = Maybe Bool
forall a. Maybe a
Nothing
                }
          -- A tick at 12:00 does not fire under the 3am override.
          catchUpAt schema table [cron] (mkTime 2025 6 15 12 0 0)

        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
10 NominalDiffTime
60 :: IO [JobRead payload]
        length jobs `shouldBe` 0

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"updates last_fired_at on successful fire" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"fire-test" Text
"* * * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"fire"))
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] (Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
0 Int
0)

        mRow <- env -> m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow)
forall a. env -> m a -> IO a
runM env
env (m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow))
-> m (Maybe CronScheduleRow) -> IO (Maybe CronScheduleRow)
forall a b. (a -> b) -> a -> b
$ Text -> Text -> m (Maybe CronScheduleRow)
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> m (Maybe CronScheduleRow)
Ops.getCronScheduleByName Text
schema Text
"fire-test"
        case mRow of
          Maybe CronScheduleRow
Nothing -> HasCallStack => String -> IO ()
String -> IO ()
expectationFailure String
"Expected cron schedule row to exist"
          Just CronScheduleRow
row -> CronScheduleRow -> Maybe UTCTime
CS.lastFiredAt CronScheduleRow
row Maybe UTCTime -> (Maybe UTCTime -> Bool) -> IO ()
forall a. (HasCallStack, Show a) => a -> (a -> Bool) -> IO ()
`shouldSatisfy` Maybe UTCTime -> Bool
forall a. Maybe a -> Bool
isJust

    String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"processCronCatchUp backfill" (SpecWith env -> SpecWith env) -> SpecWith env -> SpecWith env
forall a b. (a -> b) -> a -> b
$ do
      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"fires every missed minute for schedules with Backfill policy" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        -- Simulate a scheduler wake-up after a 5-minute gap. last_checked_at
        -- is 5 minutes in the past. With Backfill 600 (10 min window) the
        -- catch-up fires a job for each missed minute.
        let Right CronJob payload
base =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"catchup-backfill"
                Text
"* * * * *"
                OverlapPolicy
AllowOverlap
                (\TickKind
_ UTCTime
tickAt -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple (UTCTime -> Text
formatMinute UTCTime
tickAt)))
            cron :: CronJob payload
cron = CronJob payload
base {backfill = Backfill 600}
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]

        ByteString -> (Connection -> IO ()) -> IO ()
forall a. ByteString -> (Connection -> IO a) -> IO a
withConn ByteString
connStr ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn ->
          IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$
            Connection -> Query -> Only Text -> IO Int64
forall q. ToRow q => Connection -> Query -> q -> IO Int64
PG.execute
              Connection
conn
              (Text -> Text -> Query
cronUpdate Text
schema Text
"last_checked_at = NOW() - interval '5 minutes'")
              (Text -> Only Text
forall a. a -> Only a
Only (Text
"catchup-backfill" :: Text))

        now <- IO UTCTime
getCurrentTime
        runM env $ catchUpAt schema table [cron] now

        jobs <- runM env $ HL.listJobs 100 0 :: IO [JobRead payload]
        -- Expect at least 5 missed minutes plus the current one.
        length jobs `shouldSatisfy` (>= 5)

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"does not replay missed minutes for NoBackfill schedules" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        -- NoBackfill replays no stale ticks. With a stale last_checked_at, only
        -- the current minute fires.
        let Right CronJob payload
cron =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"catchup-nobackfill"
                Text
"* * * * *"
                OverlapPolicy
AllowOverlap
                (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"no-replay"))
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]

        ByteString -> (Connection -> IO ()) -> IO ()
forall a. ByteString -> (Connection -> IO a) -> IO a
withConn ByteString
connStr ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn ->
          IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$
            Connection -> Query -> Only Text -> IO Int64
forall q. ToRow q => Connection -> Query -> q -> IO Int64
PG.execute
              Connection
conn
              (Text -> Text -> Query
cronUpdate Text
schema Text
"last_checked_at = NOW() - interval '5 minutes'")
              (Text -> Only Text
forall a. a -> Only a
Only (Text
"catchup-nobackfill" :: Text))

        now <- IO UTCTime
getCurrentTime
        runM env $ catchUpAt schema table [cron] now

        jobs <- runM env $ HL.listJobs 100 0 :: IO [JobRead payload]
        length jobs `shouldBe` 1

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"fires only the current minute when last_checked_at is null" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        -- A fresh schedule with no last_checked_at replays no earlier ticks,
        -- whatever the policy.
        let Right CronJob payload
base =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"fresh"
                Text
"* * * * *"
                OverlapPolicy
AllowOverlap
                (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"fresh"))
            cron :: CronJob payload
cron = CronJob payload
base {backfill = Backfill 3600}
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]

        now <- IO UTCTime
getCurrentTime
        runM env $ catchUpAt schema table [cron] now

        jobs <- runM env $ HL.listJobs 100 0 :: IO [JobRead payload]
        length jobs `shouldBe` 1

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"sets last_checked_at to currentTick" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        -- Under slow processing the watermark equals the 'now' passed in.
        let Right CronJob payload
cron =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"watermark" Text
"* * * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"x"))
            currentTickPast :: UTCTime
currentTickPast = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
0 Int
0
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] UTCTime
currentTickPast

        rows <- env -> m [CronScheduleRow] -> IO [CronScheduleRow]
forall a. env -> m a -> IO a
runM env
env (m [CronScheduleRow] -> IO [CronScheduleRow])
-> m [CronScheduleRow] -> IO [CronScheduleRow]
forall a b. (a -> b) -> a -> b
$ Text -> Maybe Text -> m [CronScheduleRow]
forall (m :: * -> *).
MonadArbiter m =>
Text -> Maybe Text -> m [CronScheduleRow]
Ops.listCronSchedules Text
schema Maybe Text
forall a. Maybe a
Nothing
        case lookup "watermark" [(CS.name row, row) | row <- rows] of
          Just CronScheduleRow
row -> CronScheduleRow -> Maybe UTCTime
CS.lastCheckedAt CronScheduleRow
row Maybe UTCTime -> Maybe UTCTime -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` UTCTime -> Maybe UTCTime
forall a. a -> Maybe a
Just UTCTime
currentTickPast
          Maybe CronScheduleRow
Nothing -> HasCallStack => String -> IO ()
String -> IO ()
expectationFailure String
"watermark schedule missing"

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"does not advance last_checked_at backwards" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        -- GREATEST guard for concurrent pools with skewed clocks.
        let Right CronJob payload
cron =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"monotonic" Text
"* * * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"x"))
            later :: UTCTime
later = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
5 Int
0
            earlier :: UTCTime
earlier = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
0 Int
0
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] UTCTime
later
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] UTCTime
earlier

        rows <- env -> m [CronScheduleRow] -> IO [CronScheduleRow]
forall a. env -> m a -> IO a
runM env
env (m [CronScheduleRow] -> IO [CronScheduleRow])
-> m [CronScheduleRow] -> IO [CronScheduleRow]
forall a b. (a -> b) -> a -> b
$ Text -> Maybe Text -> m [CronScheduleRow]
forall (m :: * -> *).
MonadArbiter m =>
Text -> Maybe Text -> m [CronScheduleRow]
Ops.listCronSchedules Text
schema Maybe Text
forall a. Maybe a
Nothing
        case lookup "monotonic" [(CS.name row, row) | row <- rows] of
          Just CronScheduleRow
row -> CronScheduleRow -> Maybe UTCTime
CS.lastCheckedAt CronScheduleRow
row Maybe UTCTime -> Maybe UTCTime -> IO ()
forall a. (HasCallStack, Show a, Eq a) => a -> a -> IO ()
`shouldBe` UTCTime -> Maybe UTCTime
forall a. a -> Maybe a
Just UTCTime
later
          Maybe CronScheduleRow
Nothing -> HasCallStack => String -> IO ()
String -> IO ()
expectationFailure String
"monotonic schedule missing"

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"does not re-fire ticks whose jobs were already processed" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        -- After firing and processing the live tick, a second call at the
        -- same minute produces no duplicate. The dedup row is gone once the
        -- job is acked. The watermark blocks the re-fire.
        let Right CronJob payload
cron =
              Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob
                Text
"no-duplicate"
                Text
"* * * * *"
                OverlapPolicy
AllowOverlap
                (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"once"))
            tick :: UTCTime
tick = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
0 Int
0
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] UTCTime
tick

        claimed <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
100 NominalDiffTime
60 :: IO [JobRead payload]
        length claimed `shouldBe` 1
        runM env $ traverse_ (void . HL.ackJob) claimed

        runM env $ catchUpAt schema table [cron] tick
        afterRetry <- runM env $ HL.listJobs 100 0 :: IO [JobRead payload]
        afterRetry `shouldBe` []

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"DST fall-back: a fixed-time schedule fires once after its first job is acked" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        -- 01:30 America/New_York on 2025-11-02 reads at 05:30Z and 06:30Z.
        let Right CronJob payload
cron =
              Text
-> Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJobInTimezone
                Text
"fall-back"
                Text
"America/New_York"
                Text
"30 1 * * *"
                OverlapPolicy
AllowOverlap
                (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"once"))
            firstPass :: UTCTime
firstPass = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
11 Int
2 Int
5 Int
30 Int
0
            secondPass :: UTCTime
secondPass = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
11 Int
2 Int
6 Int
30 Int
0
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
          LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
          Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] UTCTime
firstPass

        claimed <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> NominalDiffTime -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> NominalDiffTime -> m [JobRead payload]
HL.claimNextVisibleJobs Int
100 NominalDiffTime
60 :: IO [JobRead payload]
        length claimed `shouldBe` 1
        runM env $ traverse_ (void . HL.ackJob) claimed

        runM env $ catchUpAt schema table [cron] secondPass
        afterSecond <- runM env $ HL.listJobs 100 0 :: IO [JobRead payload]
        afterSecond `shouldBe` []

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"gate prevents double-fire when last_fired_at already covers the minute" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        -- Simulates a fast pool that already fired and acked 12:00. The gate
        -- blocks a slow pool retrying the same minute.
        let Right CronJob payload
cron = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"skew-race" Text
"* * * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"skew"))
            tick :: UTCTime
tick = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
0 Int
0
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
        ByteString -> (Connection -> IO ()) -> IO ()
forall a. ByteString -> (Connection -> IO a) -> IO a
withConn ByteString
connStr ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn ->
          IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$
            Connection -> Query -> (UTCTime, Text) -> IO Int64
forall q. ToRow q => Connection -> Query -> q -> IO Int64
PG.execute
              Connection
conn
              (Text -> Text -> Query
cronUpdate Text
schema Text
"last_fired_at = ?")
              (UTCTime
tick, Text
"skew-race" :: Text)
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] UTCTime
tick
        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> Int -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> Int -> m [JobRead payload]
HL.listJobs Int
100 Int
0 :: IO [JobRead payload]
        length jobs `shouldBe` 0

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"gate lets the next minute through after firing the previous one" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"skew-advance" Text
"* * * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"advance"))
            tickPrev :: UTCTime
tickPrev = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
0 Int
0
            tickNext :: UTCTime
tickNext = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
1 Int
0
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]
        ByteString -> (Connection -> IO ()) -> IO ()
forall a. ByteString -> (Connection -> IO a) -> IO a
withConn ByteString
connStr ((Connection -> IO ()) -> IO ()) -> (Connection -> IO ()) -> IO ()
forall a b. (a -> b) -> a -> b
$ \Connection
conn ->
          IO Int64 -> IO ()
forall (f :: * -> *) a. Functor f => f a -> f ()
void (IO Int64 -> IO ()) -> IO Int64 -> IO ()
forall a b. (a -> b) -> a -> b
$
            Connection -> Query -> (UTCTime, Text) -> IO Int64
forall q. ToRow q => Connection -> Query -> q -> IO Int64
PG.execute
              Connection
conn
              (Text -> Text -> Query
cronUpdate Text
schema Text
"last_fired_at = ?")
              (UTCTime
tickPrev, Text
"skew-advance" :: Text)
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ Text -> Text -> [CronJob payload] -> UTCTime -> m ()
forall (m :: * -> *) payload.
QueueOperation m payload =>
Text -> Text -> [CronJob payload] -> UTCTime -> m ()
catchUpAt Text
schema Text
table [CronJob payload
cron] UTCTime
tickNext
        jobs <- env -> m [JobRead payload] -> IO [JobRead payload]
forall a. env -> m a -> IO a
runM env
env (m [JobRead payload] -> IO [JobRead payload])
-> m [JobRead payload] -> IO [JobRead payload]
forall a b. (a -> b) -> a -> b
$ Int -> Int -> m [JobRead payload]
forall payload (m :: * -> *).
QueueOperation m payload =>
Int -> Int -> m [JobRead payload]
HL.listJobs Int
100 Int
0 :: IO [JobRead payload]
        length jobs `shouldBe` 1

    String -> SpecWith env -> SpecWith env
forall a. HasCallStack => String -> SpecWith a -> SpecWith a
describe String
"cron concurrency primitives" (SpecWith env -> SpecWith env) -> SpecWith env -> SpecWith env
forall a b. (a -> b) -> a -> b
$ do
      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"touchCronChecked advances last_checked_at monotonically and matches by name" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"touch-test" Text
"* * * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"x"))
            tEarly :: UTCTime
tEarly = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
0 Int
0
            tLate :: UTCTime
tLate = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
5 Int
0
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]

        nLate <- env -> m Int64 -> IO Int64
forall a. env -> m a -> IO a
runM env
env (m Int64 -> IO Int64) -> m Int64 -> IO Int64
forall a b. (a -> b) -> a -> b
$ Text -> [Text] -> UTCTime -> m Int64
forall (m :: * -> *).
MonadArbiter m =>
Text -> [Text] -> UTCTime -> m Int64
Ops.touchCronChecked Text
schema [Text
"touch-test"] UTCTime
tLate
        nLate `shouldBe` 1
        Just rowLate <- runM env $ Ops.getCronScheduleByName schema "touch-test"
        CS.lastCheckedAt rowLate `shouldBe` Just tLate

        -- An earlier watermark matches the row and leaves it in place.
        nEarly <- runM env $ Ops.touchCronChecked schema ["touch-test"] tEarly
        nEarly `shouldBe` 1
        Just rowEarly <- runM env $ Ops.getCronScheduleByName schema "touch-test"
        CS.lastCheckedAt rowEarly `shouldBe` Just tLate

        -- An unknown name matches nothing.
        nMiss <- runM env $ Ops.touchCronChecked schema ["no-such-schedule"] tLate
        nMiss `shouldBe` 0

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"tryFireCronGate fires once per minute floor" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let Right CronJob payload
cron = Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
forall payload.
Text
-> Text
-> OverlapPolicy
-> (TickKind -> UTCTime -> JobWrite payload)
-> Either String (CronJob payload)
cronJob Text
"gate-test" Text
"* * * * *" OverlapPolicy
AllowOverlap (\TickKind
_ UTCTime
_ -> payload -> JobWrite payload
forall payload. payload -> JobWrite payload
defaultJob (Text -> payload
mkSimple Text
"x"))
            minuteZero :: UTCTime
minuteZero = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
0 Int
0
            minuteOne :: UTCTime
minuteOne = Integer -> Int -> Int -> Int -> Int -> Int -> UTCTime
mkTime Integer
2025 Int
6 Int
15 Int
12 Int
1 Int
0
        env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ LogConfig -> Text -> [CronJob payload] -> m ()
forall (m :: * -> *) payload.
MonadArbiter m =>
LogConfig -> Text -> [CronJob payload] -> m ()
initCronSchedules LogConfig
silentLogConfig Text
table [CronJob payload
cron]

        firstFire <- env -> m Bool -> IO Bool
forall a. env -> m a -> IO a
runM env
env (m Bool -> IO Bool) -> m Bool -> IO Bool
forall a b. (a -> b) -> a -> b
$ Text -> Text -> UTCTime -> m Bool
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> UTCTime -> m Bool
Ops.tryFireCronGate Text
schema Text
"gate-test" UTCTime
minuteZero
        firstFire `shouldBe` True
        -- The same minute floor cannot fire twice.
        secondFire <- runM env $ Ops.tryFireCronGate schema "gate-test" minuteZero
        secondFire `shouldBe` False
        -- A later minute floor fires again.
        nextFire <- runM env $ Ops.tryFireCronGate schema "gate-test" minuteOne
        nextFire `shouldBe` True

      String -> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a.
(HasCallStack, Example a) =>
String -> a -> SpecWith (Arg a)
it String
"tryAcquireCronLeader is mutually exclusive per (schema, queue, name) and releases on commit" ((env -> IO ()) -> SpecWith (Arg (env -> IO ())))
-> (env -> IO ()) -> SpecWith (Arg (env -> IO ()))
forall a b. (a -> b) -> a -> b
$ \env
env -> do
        let acquire :: Text -> m Bool
acquire = Text -> Text -> Text -> m Bool
forall (m :: * -> *).
MonadArbiter m =>
Text -> Text -> Text -> m Bool
Ops.tryAcquireCronLeader Text
schema Text
table
        held <- IO (MVar Bool)
forall (m :: * -> *) a. MonadIO m => m (MVar a)
newEmptyMVar
        release <- newEmptyMVar
        -- The advisory lock is transaction-scoped. The holder's transaction stays open to hold it.
        let holdLeader = env -> m () -> IO ()
forall a. env -> m a -> IO a
runM env
env (m () -> IO ()) -> (m () -> m ()) -> m () -> IO ()
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 () -> IO ()) -> m () -> IO ()
forall a b. (a -> b) -> a -> b
$ do
              got <- Text -> m Bool
acquire Text
"leader"
              liftIO (putMVar held got)
              liftIO (takeMVar release)
        withAsync holdLeader $ \Async ()
holder -> do
          got1 <- MVar Bool -> IO Bool
forall (m :: * -> *) a. MonadIO m => MVar a -> m a
takeMVar MVar Bool
held
          got1 `shouldBe` True

          -- The holder still has "leader". A second transaction loses on the same key.
          -- A different schedule name is independent.
          (got2, got3) <- runM env . withDbTransaction $ (,) <$> acquire "leader" <*> acquire "other"
          got2 `shouldBe` False
          got3 `shouldBe` True

          -- Once the holder commits, "leader" is free for a fresh transaction.
          putMVar release ()
          wait holder
          got4 <- runM env . withDbTransaction $ acquire "leader"
          got4 `shouldBe` True