+ ^{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/Handler/Admin/Test.hs b/src/Handler/Admin/Test.hs
index c31fd691b..645152b0e 100644
--- a/src/Handler/Admin/Test.hs
+++ b/src/Handler/Admin/Test.hs
@@ -86,11 +86,9 @@ postAdminTestR = do
((emailResult, emailWidget), emailEnctype) <- runFormPost . identifyForm ("email" :: Text) $ renderAForm FormStandard emailTestForm
formResultModal emailResult AdminTestR $ \(email, ls) -> do
- jId <- mapWriterT runDB $ do
- jId <- queueJob $ JobSendTestEmail email ls
- tell . pure $ Message Success [shamlet|Email-test gestartet (Job ##{tshow (fromSqlKey jId)})|] (Just IconEmail)
- return jId
- runReaderT (writeJobCtl $ JobCtlPerform jId) =<< getYesod
+ mapWriterT runDBJobs $ do
+ lift . queueDBJob $ JobSendTestEmail email ls
+ tell . pure $ Message Success [shamlet|Email-test gestartet|] (Just IconEmail)
addMessage Warning [shamlet|Inkorrekt ausgegebener Alert|] -- For testing alert handling when short circuiting; for proper (not fallback-solution) handling always use `tell` within `formResultModal`
let emailWidget' = wrapForm emailWidget def
diff --git a/src/Handler/Utils/Exam.hs b/src/Handler/Utils/Exam.hs
index 8ba5f5584..cc8405762 100644
--- a/src/Handler/Utils/Exam.hs
+++ b/src/Handler/Utils/Exam.hs
@@ -1,4 +1,4 @@
-{-# OPTIONS_GHC -fno-warn-deprecations -fno-warn-incomplete-uni-patterns #-}
+{-# OPTIONS_GHC -fno-warn-incomplete-uni-patterns #-}
module Handler.Utils.Exam
( fetchExamAux
@@ -519,7 +519,7 @@ examAutoOccurrence (hash -> seed) rule ExamAutoOccurrenceConfig{..} occurrences
)
postprocess result = (resultAscList, resultUsers)
where
- resultAscList = pad . Map.fromListWith Set.union . accRes (pure <$> Set.lookupMin rangeAlphabet) $ (\r -> traceShow (over (traverse . _2 . traverse . traverse) CI.original r) r) result
+ resultAscList = pad . Map.fromListWith Set.union $ accRes (pure <$> Set.lookupMin rangeAlphabet) result
where
accRes _ [] = []
accRes prevEnd ((occA, nsA) : (occB, nsB) : xs)
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/Queue.hs b/src/Jobs/Queue.hs
index 045649ed1..18c85be59 100644
--- a/src/Jobs/Queue.hs
+++ b/src/Jobs/Queue.hs
@@ -80,22 +80,28 @@ writeJobCtlBlock :: (MonadThrow m, MonadIO m, MonadReader UniWorX m) => JobCtl -
-- | Pass an instruction to the `Job`-Workers and block until it was acted upon
writeJobCtlBlock = writeJobCtlBlock' writeJobCtl
-queueJobUnsafe :: Bool -> Job -> YesodDB UniWorX QueuedJobId
+queueJobUnsafe :: Bool -> Job -> YesodDB UniWorX (Maybe QueuedJobId)
queueJobUnsafe queuedJobWriteLastExec job = do
$logInfoS "queueJob" $ tshow job
- queuedJobCreationTime <- liftIO getCurrentTime
- queuedJobCreationInstance <- getsYesod appInstanceID
- insert QueuedJob
- { queuedJobContent = toJSON job
- , queuedJobLockInstance = Nothing
- , queuedJobLockTime = Nothing
- , ..
- }
- -- We should not immediately notify a worker; instead wait for the transaction to finish first
- -- writeJobCtl $ JobCtlPerform jId -- FIXME: Should do fancy load balancing across instances (or something)
- -- return jId
+
+ doQueue <- fmap not . and2M (return $ jobNoQueueSame job) $ exists [ QueuedJobContent ==. toJSON job ]
+
+ if
+ | doQueue -> Just <$> do
+ queuedJobCreationTime <- liftIO getCurrentTime
+ queuedJobCreationInstance <- getsYesod appInstanceID
+ insert QueuedJob
+ { queuedJobContent = toJSON job
+ , queuedJobLockInstance = Nothing
+ , queuedJobLockTime = Nothing
+ , ..
+ }
+ -- We should not immediately notify a worker; instead wait for the transaction to finish first
+ -- writeJobCtl $ JobCtlPerform jId -- FIXME: Should do fancy load balancing across instances (or something)
+ -- return jId
+ | otherwise -> return Nothing
-queueJob :: (MonadHandler m, HandlerSite m ~ UniWorX) => Job -> m QueuedJobId
+queueJob :: (MonadHandler m, HandlerSite m ~ UniWorX) => Job -> m (Maybe QueuedJobId)
-- ^ Queue a job for later execution
--
-- Makes no guarantees as to when it will be executed (`queueJob'`) and does not interact with any running database transactions (`runDBJobs`)
@@ -105,15 +111,15 @@ queueJob' :: (MonadThrow m, MonadHandler m, HandlerSite m ~ UniWorX) => Job -> m
-- ^ `queueJob` followed by `writeJobCtl` `JobCtlPerform` to ensure, that it is executed asap
queueJob' job = do
app <- getYesod
- queueJob job >>= flip runReaderT app . writeJobCtl . JobCtlPerform
+ queueJob job >>= maybe (return ()) (flip runReaderT app . writeJobCtl . JobCtlPerform)
-- | Slightly modified Version of `DB` for `runDBJobs`
type JobDB = YesodJobDB UniWorX
queueDBJob, queueDBJobCron :: Job -> YesodJobDB UniWorX ()
-- | Queue a job as part of a database transaction and execute it after the transaction succeeds
-queueDBJob job = mapReaderT lift (queueJobUnsafe False job) >>= tell . Set.singleton
-queueDBJobCron job = mapReaderT lift (queueJobUnsafe True job) >>= tell . Set.singleton
+queueDBJob job = mapReaderT lift (queueJobUnsafe False job) >>= tell . maybe Set.empty Set.singleton
+queueDBJobCron job = mapReaderT lift (queueJobUnsafe True job) >>= tell . maybe Set.empty Set.singleton
sinkDBJobs :: ConduitT Job Void (YesodJobDB UniWorX) ()
-- | Queue many jobs as part of a database transaction and execute them after the transaction passes
diff --git a/src/Jobs/Types.hs b/src/Jobs/Types.hs
index 7b36c2801..11fe8b12e 100644
--- a/src/Jobs/Types.hs
+++ b/src/Jobs/Types.hs
@@ -17,6 +17,8 @@ module Jobs.Types
, showWorkerId, newWorkerId
, JobQueue, jqInsert, jqDequeue
, JobPriority(..), prioritiseJob
+ , jobNoQueueSame
+ , module Cron
) where
import Import.NoFoundation hiding (Unique, state)
@@ -37,6 +39,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 }
@@ -232,6 +236,19 @@ prioritiseJob (JobCtlGenerateHealthReport _) = JobPrioRealtime
prioritiseJob JobCtlDetermineCrontab = JobPrioRealtime
prioritiseJob _ = JobPrioBatch
+jobNoQueueSame :: Job -> Bool
+jobNoQueueSame = \case
+ JobSendPasswordReset{} -> True
+ JobTruncateTransactionLog{} -> True
+ JobPruneInvitations{} -> True
+ JobDeleteTransactionLogIPs{} -> True
+ JobSynchroniseLdapUser{} -> True
+ JobChangeUserDisplayEmail{} -> True
+ JobPruneSessionFiles{} -> True
+ JobPruneUnreferencedFiles{} -> True
+ JobInjectFiles{} -> True
+ _ -> False
+
newtype JobQueue = JobQueue { getJobQueue :: MaxPQueue JobPriority JobCtl }
deriving (Eq, Ord, Read, Show)
@@ -253,6 +270,7 @@ data JobState = JobState
, jobPoolManager :: Async ()
, jobCron :: Async ()
, jobShutdown :: TMVar ()
+ , jobCurrentCrontab :: TVar (Maybe (UTCTime, [(CronNextMatch UTCTime, JobCtl)]))
}
jobWorkerNames :: JobState -> Set JobWorkerId
|