+ ^{formatTimeW SelFormatDateTime genTime} +
| + $case match + $of MatchAsap + _{MsgCronMatchAsap} + $of MatchNone + _{MsgCronMatchNone} + $of MatchAt t + ^{formatTimeW SelFormatDateTime t} + |
+
+ #{encodePrettyToTextBuilder job}
+ $nothing
+ _{MsgAdminCrontabNotGenerated}
+ |]
diff --git a/src/Jobs.hs b/src/Jobs.hs
index b917354f0..bdf6b847f 100644
--- a/src/Jobs.hs
+++ b/src/Jobs.hs
@@ -97,6 +97,7 @@ handleJobs foundation@UniWorX{..}
jobCrontab <- liftIO $ newTVarIO HashMap.empty
jobConfirm <- liftIO $ newTVarIO HashMap.empty
jobShutdown <- liftIO newEmptyTMVarIO
+ jobCurrentCrontab <- liftIO $ newTVarIO Nothing
atomically $ putTMVar appJobState JobState
{ jobContext = JobContext{..}
, ..
@@ -109,12 +110,12 @@ manageCrontab :: forall m.
=> UniWorX -> (forall a. m a -> m a) -> m ()
manageCrontab foundation@UniWorX{..} unmask = do
ch <- allocateLinkedAsync $ do
- context <- atomically . fmap jobContext $ readTMVar appJobState
+ jState <- atomically $ readTMVar appJobState
liftIO . unsafeHandler foundation . void $ do
atomically . assertM_ (not . Map.null . jobWorkers) $ readTMVar appJobState
runReaderT ?? foundation $
writeJobCtlBlock JobCtlDetermineCrontab
- void $ evalRWST (forever execCrontab) context HashMap.empty
+ void $ evalRWST (forever execCrontab) jState HashMap.empty
let awaitTermination = guardM $
readTMVar appJobState >>= fmap not . isEmptyTMVar . jobShutdown
@@ -252,7 +253,7 @@ stopJobCtl UniWorX{appJobState} = do
, jobCron jSt'
] ++ workers
-execCrontab :: RWST JobContext () (HashMap JobCtl (Max UTCTime)) (HandlerFor UniWorX) ()
+execCrontab :: RWST JobState () (HashMap JobCtl (Max UTCTime)) (HandlerFor UniWorX) ()
-- ^ Keeping a `HashMap` of the latest execution times of `JobCtl`s we have
-- seen, wait for the time of the next job and fire it
execCrontab = do
@@ -276,7 +277,7 @@ execCrontab = do
refT <- liftIO getCurrentTime
settings <- getsYesod appSettings'
(currentCrontab, (jobCtl, nextMatch), currentState) <- mapRWST (liftIO . atomically) $ do
- crontab <- liftBase . readTVar =<< asks jobCrontab
+ crontab <- liftBase . readTVar =<< asks (jobCrontab . jobContext)
State.modify . HashMap.filterWithKey $ \k _ -> HashMap.member k crontab
prevExec <- State.get
@@ -288,13 +289,16 @@ execCrontab = do
do
lastTimes <- State.get
now <- liftIO getCurrentTime
- $logDebugS "Crontab" . intercalate "\n" . map tshow . sortOn fst . flip map (HashMap.toList currentCrontab) $ \(job, cron) -> (,job) $ nextCronMatch appTZ (getMax <$> HashMap.lookup job lastTimes) (debouncingAcc settings job) now cron
+ let currentCrontab' = sortOn fst . flip map (HashMap.toList currentCrontab) $ \(job, cron) -> (,job) $ nextCronMatch appTZ (getMax <$> HashMap.lookup job lastTimes) (debouncingAcc settings job) now cron
+ crontabTVar <- asks jobCurrentCrontab
+ atomically . writeTVar crontabTVar $ Just (now, currentCrontab')
+ $logDebugS "Crontab" . intercalate "\n" $ "Current crontab:" : map tshow currentCrontab'
let doJob = mapRWST (liftHandler . runDBJobs . setSerializable) $ do
newCrontab <- lift $ hoist lift determineCrontab'
when (newCrontab /= currentCrontab) $
mapRWST (liftIO . atomically) $
- liftBase . void . flip swapTVar newCrontab =<< asks jobCrontab
+ liftBase . void . flip swapTVar newCrontab =<< asks (jobCrontab . jobContext)
mergeState
newState <- State.get
@@ -315,11 +319,11 @@ execCrontab = do
MatchAsap -> doJob
MatchNone -> return ()
MatchAt nextTime -> do
- JobContext{jobCrontab} <- ask
+ crontab <- asks $ jobCrontab . jobContext
nextTime' <- applyJitter jobCtl nextTime
$logDebugS "Cron" [st|Waiting until #{tshow (utcToLocalTimeTZ appTZ nextTime')} to execute #{tshow jobCtl}|]
logFunc <- askLoggerIO
- whenM (liftIO . flip runLoggingT logFunc $ waitUntil jobCrontab currentCrontab nextTime')
+ whenM (liftIO . flip runLoggingT logFunc $ waitUntil crontab currentCrontab nextTime')
doJob
where
diff --git a/src/Jobs/Types.hs b/src/Jobs/Types.hs
index 7b36c2801..a7b56be8d 100644
--- a/src/Jobs/Types.hs
+++ b/src/Jobs/Types.hs
@@ -17,6 +17,7 @@ module Jobs.Types
, showWorkerId, newWorkerId
, JobQueue, jqInsert, jqDequeue
, JobPriority(..), prioritiseJob
+ , module Cron
) where
import Import.NoFoundation hiding (Unique, state)
@@ -37,6 +38,8 @@ import Utils.Metrics (withJobWorkerStateLbls)
import qualified Prometheus (Label4)
+import Cron (CronNextMatch(..), _MatchAsap, _MatchAt, _MatchNone)
+
data Job = JobSendNotification { jRecipient :: UserId, jNotification :: Notification }
| JobSendTestEmail { jEmail :: Email, jMailContext :: MailContext }
@@ -253,6 +256,7 @@ data JobState = JobState
, jobPoolManager :: Async ()
, jobCron :: Async ()
, jobShutdown :: TMVar ()
+ , jobCurrentCrontab :: TVar (Maybe (UTCTime, [(CronNextMatch UTCTime, JobCtl)]))
}
jobWorkerNames :: JobState -> Set JobWorkerId
|