@@ -1,39 +1,77 @@
11{-# LANGUAGE RankNTypes, FlexibleInstances, FlexibleContexts, PartialTypeSignatures, TupleSections, DeriveGeneric, UndecidableInstances #-}
22module OddJobs.Job
3− ( jobMonitor
4− , jobEventListener
5− , jobPoller
3+ (
4+ -- * Starting the job-runner
5+ --
6+ -- $startRunner
7+ startJobRunner
8+
9+ -- * Configuring the job-runner
10+ --
11+ -- $config
12+ , Config(..)
13+ , ConcurrencyControl(..)
14+
15+ -- ** Configuration helpers
16+ --
17+ -- $configHelpers
18+ , defaultConfig
19+ , defaultJobToText
20+ , defaultJobType
21+ , defaultTimedLogger
22+ , defaultLogStr
23+ , defaultPollingInterval
24+ , defaultLockTimeout
25+ , withConnectionPool
26+
27+ -- * Creating/scheduling jobs
28+ --
29+ -- $createJobs
630 , createJob
731 , scheduleJob
32+
33+ -- * @Job@ and associated data-types
34+ --
35+ -- $dataTypes
836 , Job(..)
37+ , JobId
38+ , Status(..)
39+ , TableName
40+ , delaySeconds
41+ , Seconds(..)
42+ -- ** Structured logging
43+ --
44+ -- $logging
45+ , LogEvent(..)
46+ , LogLevel(..)
47+
48+ -- * Job-runner interals
49+ --
50+ -- $internals
51+ , jobMonitor
52+ , jobEventListener
53+ , jobPoller
54+ , jobPollingSql
955 , JobRunner
1056 , HasJobRunner (..)
11− , Status(..)
57+
58+ -- * Database helpers
59+ --
60+ -- $dbHelpers
1261 , findJobById
1362 , findJobByIdIO
14− , JobId
1563 , saveJob
1664 , saveJobIO
17− , defaultPollingInterval
18− , Config(..)
19− , defaultConfig
20− , startJobRunner
21− , TableName
2265 , jobDbColumns
2366 , concatJobDbColumns
24− , delaySeconds
25− , Seconds(..)
26− , ConcurrencyControl(..)
27− , withConnectionPool
28− , lockTimeout
67+
68+ -- * JSON helpers
69+ --
70+ -- $jsonHelpers
2971 , eitherParsePayload
3072 , throwParsePayload
3173 , eitherParsePayloadWith
3274 , throwParsePayloadWith
33− , defaultJobToText
34− , defaultJobType
35− , defaultTimedLogger
36− , defaultLogStr
3775 )
3876where
3977
@@ -83,13 +121,22 @@ import Data.Aeson.Internal (iparse, IResult(..), formatError)
83121import qualified System.Log.FastLogger as FLogger
84122import Prelude hiding (log)
85123
124+
125+-- | The documentation of odd-jobs currently promotes 'startJobRunner', which
126+-- expects a fairly detailed 'Config' record, as a top-level function for
127+-- initiating a job-runner. However, internally, this 'Config' record is used as
128+-- an enviroment for a 'ReaderT', and almost all functions are written in this
129+-- 'ReaderT' monad which impleents an instance of the 'HasJobRunner' type-class.
130+--
131+-- **In future,** this /internal/ implementation detail will allow us to offer a
132+-- type-class based interface as well (similar to what
133+-- 'Yesod.JobQueue.YesodJobQueue' provides).
86134class (MonadUnliftIO m, MonadBaseControl IO m) => HasJobRunner m where
87135 getPollingInterval :: m Seconds
88136 onJobSuccess :: Job -> m ()
89137 onJobFailed :: Job -> m ()
90138 onJobPermanentlyFailed :: Job -> m ()
91139 getJobRunner :: m (Job -> IO ())
92− getMaxAttempts :: m Int
93140 getDbPool :: m (Pool Connection)
94141 getTableName :: m TableName
95142 onJobStart :: Job -> m ()
@@ -101,37 +148,133 @@ class (MonadUnliftIO m, MonadBaseControl IO m) => HasJobRunner m where
101148 getJobToText :: m (Job -> Text)
102149 log :: LogLevel -> LogEvent -> m ()
103150
104−data LogEvent = LogJobStart !Job
105− | LogJobFailed !Job !NominalDiffTime
106− | LogJobSuccess !Job !NominalDiffTime
107− | LogJobPermanentlyFailed !Job !NominalDiffTime
108− | LogJobTimeout !Job
109− | LogPoll
110− | LogText !Text
111− deriving (Eq, Show, Generic)
112151
152+-- $logging
153+--
154+-- TODO: Complete the prose here
155+
156+
157+data LogEvent
158+ -- | Emitted when a job starts execution
159+ = LogJobStart !Job
160+ -- | Emitted when a job succeeds along with the time taken for execution.
161+ | LogJobSuccess !Job !NominalDiffTime
162+ -- | Emitted when a job fails (but will be retried) along with the time taken for
163+ -- /this/ attempt
164+ | LogJobFailed !Job !NominalDiffTime
165+ -- | Emitted when a job fails permanently (and will no longer be retried) along
166+ -- with the time taken for /this/ attempt (i.e. final attempt)
167+ | LogJobPermanentlyFailed !Job !NominalDiffTime
168+ -- | Emitted when a job times out and is picked-up again for execution
169+ | LogJobTimeout !Job
170+ -- | Emitted whenever 'jobPoller' polls the DB table
171+ | LogPoll
172+ -- | Emitted whenever any other event occurs
173+ | LogText !Text
174+ deriving (Eq, Show, Generic)
175+
176+
177+-- | While odd-jobs is highly configurable and the 'Config' data-type might seem
178+-- daunting at first, it is not necessary to tweak every single configuration
179+-- parameter by hand. Please start-off by using the sensible defaults provided
180+-- by the [configuration helpers](#configHelpers), and tweaking
181+-- config parameters on a case-by-case basis.
113182data Config = Config
114− { cfgPollingInterval :: Seconds
183+ { -- | The DB table which holds your jobs. Please note, this should have been
184+ -- created by the 'OddJobs.Migrations.createJobsTable' function.
185+ cfgTableName :: TableName
186+
187+ -- | The actualy "job-runner" that __you__ need to provide. Please look at
188+ -- the examples/tutorials if your applicaton's code is not in the @IO@
189+ -- monad.
190+ , cfgJobRunner :: Job -> IO ()
191+
192+ -- | The number of times a failing job is retried before it is considered is
193+ -- "permanently failed" and ignored by the job-runner. This config parameter
194+ -- is called "/default/ max attempts" because, in the future, it would be
195+ -- possible to specify the number of retry-attemps on a per-job basis
196+ -- (__Note:__ per-job retry-attempts has not been implemented yet)
197+ , cfgDefaultMaxAttempts :: Int
198+
199+ -- | Controls how many jobs can be run concurrently by /this instance/ of
200+ -- the job-runner. __Please note,__ this is NOT the global concurrency of
201+ -- entire job-queue. It is possible to have job-runners running on multiple
202+ -- machines, and each will apply the concurrency control independnt of other
203+ -- job-runners. TODO: Link-off to relevant section in the tutorial.
204+ , cfgConcurrencyControl :: ConcurrencyControl
205+
206+
207+ -- | The DB connection-pool to use for the job-runner. __Note:__ in case
208+ -- your jobs require a DB connection, please create a separate
209+ -- connection-pool for them. This pool will be used ONLY for monitoring jobs
210+ -- and changing their status. We need to have _at least 4 connections__ in
211+ -- this connection-pool for the job-runner to work as expected. (TODO:
212+ -- Link-off to tutorial)
213+ , cfgDbPool :: Pool Connection
214+
215+ -- | How frequently should the 'jobPoller' check for jobs where the Job's
216+ -- 'jobRunAt' field indicates that it's time for the job to be executed.
217+ -- TODO: link-off to the tutorial.
218+ , cfgPollingInterval :: Seconds
219+
220+ -- | User-defined callback function that is called whenever a job succeeds.
115221 , cfgOnJobSuccess :: Job -> IO ()
222+
223+ -- | User-defined callback function that is called whenever a job fails. This
224+ -- does not indicate permanent failure and means the job will be retried. It
225+ -- is a good idea to log the failures to Airbrake, NewRelic, Sentry, or some
226+ -- other error monitoring tool.
116227 , cfgOnJobFailed :: Job -> IO ()
228+
229+ -- | User-defined callback function that is called whenever a job fails
230+ -- /permanently/ (i.e. number of retry-attempts have crossed the configured
231+ -- threshold). It is a good idea to log the failures to Airbrake, NewRelic,
232+ -- Sentry, or some other error monitoring tool.
117233 , cfgOnJobPermanentlyFailed :: Job -> IO ()
234+
235+ -- | User-defined callback function that is called whenever a job starts
236+ -- execution.
118237 , cfgOnJobStart :: Job -> IO ()
238+
239+ -- | User-defined callback function that is called whenever a job times-out.
119240 , cfgOnJobTimeout :: Job -> IO ()
120− , cfgJobRunner :: Job -> IO ()
121− , cfgMaxAttempts :: Int
122− , cfgLogger :: LogLevel -> LogEvent -> IO ()
123− , cfgDbPool :: Pool Connection
124− , cfgTableName :: TableName
125− , cfgDefaultMaxAttempts :: Int
126− , cfgConcurrencyControl :: ConcurrencyControl
241+
242+ -- | File to store the PID of the job-runner process. This is used only when
243+ -- invoking the job-runner as an independent background deemon (the usual mode
244+ -- of deployment). (TODO: Link-off to tutorial).
127245 , cfgPidFile :: Maybe FilePath
246+
247+ -- | A "structured logging" function that __you__ need to provide. The
248+ -- @odd-jobs@ library does NOT use the standard logging interface provided by
249+ -- 'monad-logger' on purpose. TODO: link-off to tutorial. Please also read
250+ -- 'cfgJobToText' and 'cfgJobType'
251+ , cfgLogger :: LogLevel -> LogEvent -> IO ()
252+
253+ -- | When emitting certain text messages in logs, how should the 'Job' be
254+ -- summarized in a textual format? Related: 'defaultJobToText'
128255 , cfgJobToText :: Job -> Text
256+
257+ -- | How to extract the "job type" from a 'Job'. Related: 'defaultJobType'
129258 , cfgJobType :: Job -> Text
130259 }
131260
132−data ConcurrencyControl = MaxConcurrentJobs Int
133− | UnlimitedConcurrentJobs
134− | DynamicConcurrency (IO Bool)
261+data ConcurrencyControl
262+ -- | The maximum number of concurrent jobs that /this instance/ of the
263+ -- job-runner can execute. TODO: Link-off to tutorial.
264+ = MaxConcurrentJobs Int
265+ -- | __Not recommended:__ Please do not use this in production unless you know
266+ -- what you're doing. No machine can support unlimited concurrency. If your
267+ -- jobs are doing anything worthwhile, running a sufficiently large number
268+ -- concurrently is going to max-out /some/ resource of the underlying machine,
269+ -- such as, CPU, memory, disk IOPS, or network bandwidth.
270+ | UnlimitedConcurrentJobs
271+
272+ -- | Use this to dynamically determine if the next job should be picked-up, or
273+ -- not. This is useful to write custom-logic to determine whether a limited
274+ -- resource is below a certain usage threshold (eg. CPU usage is below 80%).
275+ -- __Caveat:__ This feature has not been tested in production, yet. TODO:
276+ -- Link-off to tutorial.
277+ | DynamicConcurrency (IO Bool)
135278
136279instance Show ConcurrencyControl where
137280 show cc = case cc of
@@ -139,6 +282,83 @@ instance Show ConcurrencyControl where
139282 UnlimitedConcurrentJobs -> "UnlimitedConcurrentJobs"
140283 DynamicConcurrency _ -> "DynamicConcurrency (IO Bool)"
141284
285+
286+-- $configHelpers
287+--
288+-- #configHelpers#
289+
290+-- | This function gives you a 'Config' with a bunch of sensible defaults
291+-- already applied. It requies the bare minimum arguments that this library
292+-- cannot assume on your behalf.
293+--
294+-- It makes a few __important assumptions__ about your 'jobPayload 'JSON, which
295+-- are documented in 'defaultJobType'.
296+defaultConfig :: (LogLevel -> LogEvent -> IO ()) -- ^ "Structured logging" function. Ref: 'cfgLogger'
297+ -> TableName -- ^ DB table which holds your jobs. Ref: 'cfgTableName'
298+ -> Pool Connection -- ^ DB connection-pool to be used by job-runner. Ref: 'cfgDbPool'
299+ -> ConcurrencyControl -- ^ Concurrency configuration. Ref: 'cfgConcurrencyControl'
300+ -> (Job -> IO ()) -- ^ The actual "job runner" which contains your application code. Ref: 'cfgJobRunner'
301+ -> Config
302+defaultConfig logger tname dbpool ccControl jrunner =
303+ let cfg = Config
304+ { cfgPollingInterval = defaultPollingInterval
305+ , cfgOnJobSuccess = (const $ pure ())
306+ , cfgOnJobFailed = (const $ pure ())
307+ , cfgOnJobPermanentlyFailed = (const $ pure ())
308+ , cfgJobRunner = jrunner
309+ , cfgLogger = logger
310+ , cfgDbPool = dbpool
311+ , cfgOnJobStart = (const $ pure ())
312+ , cfgDefaultMaxAttempts = 10
313+ , cfgTableName = tname
314+ , cfgOnJobTimeout = (const $ pure ())
315+ , cfgConcurrencyControl = ccControl
316+ , cfgPidFile = Nothing
317+ , cfgJobToText = defaultJobToText (cfgJobType cfg)
318+ , cfgJobType = defaultJobType
319+ }
320+ in cfg
321+
322+-- | Used only by 'defaultLogStr' now. TODO: Is this even required anymore?
323+-- Should this be removed?
324+defaultJobToText :: (Job -> Text) -> Job -> Text
325+defaultJobToText jobTypeFn job@Job{jobId} =
326+ "JobId=" <> (toS $ show jobId) <> " JobType=" <> jobTypeFn job
327+
328+-- | This makes __two important assumptions__. First, this /assumes/ that jobs
329+-- in your app are represented by a sum-type. For example:
330+--
331+-- @
332+-- data MyJob = SendWelcomeEmail Int
333+-- | SendPasswordResetEmail Text
334+-- | SetupSampleData Int
335+-- @
336+--
337+-- Second, it /assumes/ that the JSON representatin of this sum-type is
338+-- "tagged". For example, the following...
339+--
340+-- > let pload = SendWelcomeEmail 10
341+--
342+-- ...when converted to JSON, would look like...
343+--
344+-- > {"tag":"SendWelcomeEmail", "contents":10}
345+--
346+-- It uses this assumption to extract the "job type" from a 'Data.Aeson.Value'
347+-- (which would be @SendWelcomeEmail@ in the example given above). This is used
348+-- in logging and the admin UI.
349+--
350+-- Even if tihs assumption is violated, the job-runner /should/ continue to
351+-- function. It's just that you won't get very useful log messages.
352+defaultJobType :: Job -> Text
353+defaultJobType Job{jobPayload} =
354+ case jobPayload of
355+ Aeson.Object hm -> case HM.lookup "tag" hm of
356+ Just (Aeson.String t) -> t
357+ _ -> "unknown"
358+ _ -> "unknown"
359+
360+
361+
142362data RunnerEnv = RunnerEnv
143363 { envConfig :: !Config
144364 , envJobThreadsRef :: !(IORef [Async ()])
@@ -161,7 +381,6 @@ instance HasJobRunner RunnerM where
161381 fn <- cfgOnJobPermanentlyFailed . envConfig <$> ask
162382 logCallbackErrors (jobId job) "onJobPermanentlyFailed" $ liftIO $ fn job
163383 getJobRunner = cfgJobRunner . envConfig <$> ask
164− getMaxAttempts = cfgMaxAttempts . envConfig <$> ask
165384 getDbPool = cfgDbPool . envConfig <$> ask
166385 getTableName = cfgTableName . envConfig <$> ask
167386 onJobStart job = do
@@ -185,6 +404,10 @@ instance HasJobRunner RunnerM where
185404 loggerFn <- cfgLogger . envConfig <$> ask
186405 liftIO $ loggerFn logLevel logEvent
187406
407+-- | Start the job-runner in the /current/ thread, i.e. you'll need to use
408+-- 'forkIO' or 'async' manually, if you want the job-runner to run in the
409+-- background. Consider using 'OddJobs.Cli' to rapidly build your own
410+-- standalone daemon.
188411startJobRunner :: Config -> IO ()
189412startJobRunner jm = do
190413 r <- newIORef []
@@ -194,45 +417,9 @@ startJobRunner jm = do
194417 }
195418 runReaderT jobMonitor monitorEnv
196419
197−defaultConfig :: (LogLevel -> LogEvent -> IO ())
198− -> TableName
199− -> Pool Connection
200− -> ConcurrencyControl
201− -> Config
202−defaultConfig logger tname dbpool ccControl =
203− let cfg = Config
204− { cfgPollingInterval = defaultPollingInterval
205− , cfgOnJobSuccess = (const $ pure ())
206− , cfgOnJobFailed = (const $ pure ())
207− , cfgOnJobPermanentlyFailed = (const $ pure ())
208− , cfgJobRunner = (const $ pure ())
209− , cfgMaxAttempts = 25
210− , cfgLogger = logger
211− , cfgDbPool = dbpool
212− , cfgOnJobStart = (const $ pure ())
213− , cfgDefaultMaxAttempts = 10
214− , cfgTableName = tname
215− , cfgOnJobTimeout = (const $ pure ())
216− , cfgConcurrencyControl = ccControl
217− , cfgPidFile = Nothing
218− , cfgJobToText = defaultJobToText (cfgJobType cfg)
219− , cfgJobType = defaultJobType
220− }
221− in cfg
222−
223−defaultJobToText :: (Job -> Text) -> Job -> Text
224−defaultJobToText jobTypeFn job@Job{jobId} =
225− "JobId=" <> (toS $ show jobId) <> " JobType=" <> jobTypeFn job
226−
227−defaultJobType :: Job -> Text
228−defaultJobType Job{jobPayload} =
229− case jobPayload of
230− Aeson.Object hm -> case HM.lookup "tag" hm of
231− Just (Aeson.String t) -> t
232− _ -> "unknown"
233− _ -> "unknown"
234−
235−
420+-- | Convenience function to create a DB connection-pool with some sensible
421+-- defaults. Please see the source-code of this function to understand what it's
422+-- doing. TODO: link-off to tutorial.
236423withConnectionPool :: (MonadUnliftIO m)
237424 => Either BS.ByteString PGS.ConnectInfo
238425 -> (Pool PGS.Connection -> m a)
@@ -247,6 +434,7 @@ withConnectionPool connConfig action = withRunInIO $ \runInIO -> do
247434 Right connInfo ->
248435 createPool (PGS.connect connInfo) PGS.close 1 (fromIntegral $ 2 * (unSeconds defaultPollingInterval)) 5
249436
437+-- | As the name says. Ref: 'cfgPollingInterval'
250438defaultPollingInterval :: Seconds
251439defaultPollingInterval = Seconds 5
252440
@@ -325,9 +513,10 @@ jobWorkerName = do
325513 hname <- getHostName
326514 pure $ hname ++ ":" ++ (show pid)
327515
328−-- TODO: Make this configurable based on a per-job basis
329−lockTimeout :: Seconds
330−lockTimeout = Seconds 600
516+-- | TODO: Make this configurable for the job-runner, why is this still
517+-- hard-coded?
518+defaultLockTimeout :: Seconds
519+defaultLockTimeout = Seconds 600
331520
332521jobDbColumns :: (IsString s, Semigroup s) => [s]
333522jobDbColumns =
@@ -435,7 +624,7 @@ runJob jid = do
435624 Just job -> do
436625 startTime <- liftIO getCurrentTime
437626 (flip catches) [Handler $ timeoutHandler job startTime, Handler $ exceptionHandler job startTime] $ do
438− runJobWithTimeout lockTimeout job
627+ runJobWithTimeout defaultLockTimeout job
439628 endTime <- liftIO getCurrentTime
440629 newJob <- saveJob job{jobStatus=Success, jobLockedBy=Nothing, jobLockedAt=Nothing}
441630 log LevelInfo $ LogJobSuccess newJob (diffUTCTime endTime startTime)
@@ -483,6 +672,10 @@ restartUponCrash name_ action = do
483672 Right r -> log LevelError $ LogText $ name_ <> " seems to have exited with the folloing result: " <> toS (show r) <> ". Restaring."
484673 restartUponCrash name_ action
485674
675+-- | Spawns 'jobPoller' and 'jobEventListener' in separate threads and restarts
676+-- them in the off-chance they happen to crash. Also responsible for
677+-- implementing graceful shutdown, i.e. waiting for all jobs already being
678+-- executed to finish execution before exiting the main thread.
486679jobMonitor :: forall m . (HasJobRunner m) => m ()
487680jobMonitor = do
488681 a1 <- async $ restartUponCrash "Job poller" jobPoller
@@ -498,47 +691,8 @@ jobMonitor = do
498691 Just f -> do
499692 log LevelInfo $ LogText $ "Removing PID file: " <> toS f
500693 liftIO $ Dir.removePathForcibly f
501− -- liftIO $ putStrLn "STOPPED jobPoller and jobEventListener threads."
502−
503− -- threadsRef <- newIORef []
504− -- jobMonitor_ threadsRef Nothing Nothing Nothing
505− -- where
506− -- jobMonitor_ :: IORef [Async ()] -> Maybe (Async ()) -> Maybe (Async ()) -> Maybe (Async ()) -> m ()
507− -- jobMonitor_ threadsRef mPollerAsync mEventAsync mThreadMonitorAsync = do
508− -- pollerAsync <- maybe (async $ jobPoller threadsRef) pure mPollerAsync
509− -- eventAsync <- maybe (async $ jobEventListener threadsRef) pure mEventAsync
510− -- threadMonitorAsync <- maybe (async $ threadMonitor threadsRef) pure mThreadMonitorAsync
511− -- finally
512− -- (restartUponCrash threadsRef pollerAsync eventAsync threadMonitorAsync)
513− -- (do logInfoN "Received shutdown event. Cancelling job-poller and event-listener threads"
514− -- cancel eventAsync
515− -- cancel pollerAsync
516− -- cancel threadMonitorAsync
517− -- )
518−
519−
520− -- restartUponCrash threadsRef pollerAsync eventAsync threadMonitorAsync = do
521− -- (t, result) <- waitAnyCatch [pollerAsync, eventAsync, threadMonitorAsync]
522− -- if t==pollerAsync
523− -- then do either
524− -- (\(SomeException e) -> logErrorN $ "Job poller seems to have crashed. Respawning: " <> toS (show e))
525− -- (\x -> logErrorN $ "Job poller seems to have escaped the `forever` loop. Respawning: " <> toS (show x))
526− -- result
527− -- jobMonitor_ threadsRef Nothing (Just eventAsync) (Just threadMonitorAsync)
528− -- else if t==eventAsync
529− -- then do either
530− -- (\(SomeException e) -> logErrorN $ "Event listener seems to have crashed. Respawning: " <> toS (show e))
531− -- (\x -> logErrorN $ "Event listener seems to have escaped the `forever` loop. Respawning: " <> toS (show x))
532− -- result
533− -- jobMonitor_ threadsRef (Just pollerAsync) Nothing (Just threadMonitorAsync)
534− -- else if t==threadMonitorAsync
535− -- then do either
536− -- (\(SomeException e) -> logErrorN $ "Thread monitor seems to have crashed. Respawning: " <> toS (show e))
537− -- (\x -> logErrorN $ "Thread monitor seems to have escaped the `forever` loop. Respawning: " <> toS (show x))
538− -- result
539− -- jobMonitor_ threadsRef (Just pollerAsync) (Just eventAsync) Nothing
540− -- else logErrorN "Impossible happened. One of the three top-level threads/asyncs crashed but we were unable to figure out which one."
541694
695+-- | Ref: 'jobPoller'
542696jobPollingSql :: TableName -> Query
543697jobPollingSql tname = "update " <> tname <> " set status = ?, locked_at = ?, locked_by = ?, attempts=attempts+1 WHERE id in (select id from " <> tname <> " where (run_at<=? AND ((status in ?) OR (status = ? and locked_at<?))) ORDER BY run_at ASC LIMIT 1 FOR UPDATE) RETURNING id"
544698
@@ -555,49 +709,6 @@ waitForJobs = do
555709 delaySeconds (Seconds 1)
556710 waitForJobs
557711
558−
559− -- finally threadMonitor_ $ do
560− -- timeoutThread <- async timeout
561− -- waitForCompletion timeoutThread
562− -- where
563− -- waitForCompletion timeoutThread = readIORef threadsRef >>= \case
564− -- [] -> do
565− -- logDebugN "No job threads running."
566− -- threads -> do
567− -- logDebugN $ toS $ "Waiting for " <> show (DL.length threads) <> " to complete..."
568− -- (thread, _) <- waitAnyCatch threads
569− -- removeThreadRef thread
570− -- waitForCompletion timeoutThread
571−
572− -- timeout = do
573− -- delaySeconds lockTimeout
574− -- logDebugN "===> Timeout has expired. Forcefulling cancelling all job-threads now. <==="
575− -- mapM_ uninterruptibleCancel =<< (readIORef threadsRef)
576−
577− -- threadMonitor_ = forever $ readIORef threadsRef >>= \case
578− -- [] -> delaySeconds =<< getPollingInterval
579− -- threads -> do
580− -- logDebugN $ toS $ "Waiting on job threads: " <> show (DL.length threads) <> " threads"
581− -- (thread, ret) <- waitAnyCatch threads
582− -- let tid = asyncThreadId thread
583− -- logDebugN $ toS $ "Thread finished " <> show tid <> ". Result: " <> show ret
584− -- removeThreadRef thread
585−
586− -- removeThreadRef thread = atomicModifyIORef' threadsRef $ \threads -> (DL.delete thread threads, ())
587−
588−-- withThreadAccounting :: (HasJobRunner m)
589−-- => m a
590−-- -> m (Async a)
591−-- withThreadAccounting action = do
592−-- threadsRef <- envJobThreadsRef <$> getRunnerEnv
593−-- mv <- newEmptyMVar
594−-- a <- async $ do
595−-- takeMVar mv
596−-- action
597−-- atomicModifyIORef' threadsRef $ \threads -> (a:threads, ())
598−-- putMVar mv ()
599−-- pure a
600−
601712getConcurrencyControlFn :: (HasJobRunner m)
602713 => m (m Bool)
603714getConcurrencyControlFn = getConcurrencyControl >>= \case
@@ -607,6 +718,21 @@ getConcurrencyControlFn = getConcurrencyControl >>= \case
607718 pure $ (DL.length curJobs) < maxJobs
608719 DynamicConcurrency fn -> pure $ liftIO fn
609720
721+-- | Executes 'jobPollingSql' every 'cfgPollingInterval' seconds to pick up jobs
722+-- for execution. Uses @UPDATE@ along with @SELECT...FOR UPDATE@ to efficiently
723+-- find a job that matches /all/ of the following conditions:
724+--
725+-- * 'jobRunAt' should be in the past
726+-- * /one of the following/ conditions match:
727+--
728+-- * 'jobStatus' should be 'Queued' or 'Retry'
729+--
730+-- * 'jobStatus' should be 'Locked' and 'jobLockedAt' should be
731+-- 'defaultLockTimeout' seconds in the past, thus indicating that the
732+-- job was picked up execution, but didn't complete on time (possible
733+-- because the thread/process executing it crashed without being able to
734+-- update the DB)
735+
610736jobPoller :: (HasJobRunner m) => m ()
611737jobPoller = do
612738 processName <- liftIO jobWorkerName
@@ -622,7 +748,7 @@ jobPoller = do
622748 t <- liftIO getCurrentTime
623749 r <- liftIO $
624750 PGS.query pollerDbConn (jobPollingSql tname)
625− (Locked, t, processName, t, (In [Queued, Retry]), Locked, (addUTCTime (fromIntegral $ negate $ unSeconds lockTimeout) t))
751+ (Locked, t, processName, t, (In [Queued, Retry]), Locked, (addUTCTime (fromIntegral $ negate $ unSeconds defaultLockTimeout) t))
626752 case r of
627753 -- When we don't have any jobs to run, we can relax a bit...
628754 [] -> pure delayAction
@@ -638,6 +764,8 @@ jobPoller = do
638764 delayAction = delaySeconds =<< getPollingInterval
639765 noDelayAction = pure ()
640766
767+-- | Uses PostgreSQL's LISTEN/NOTIFY to be immediately notified of newly created
768+-- jobs.
641769jobEventListener :: (HasJobRunner m)
642770 => m ()
643771jobEventListener = do
@@ -697,12 +825,42 @@ jobEventListener = do
697825createJobQuery :: TableName -> PGS.Query
698826createJobQuery tname = "INSERT INTO " <> tname <> "(run_at, status, payload, last_error, attempts, locked_at, locked_by) VALUES (?, ?, ?, ?, ?, ?, ?) RETURNING " <> concatJobDbColumns
699827
700−createJob :: ToJSON p => Connection -> TableName -> p -> IO Job
828+-- $createJobs
829+--
830+-- Ideally you'd want to create wrappers for 'createJob' and 'scheduleJob' in
831+-- your application so that instead of being in @IO@ they can be in your
832+-- application's monad @m@ instead (this saving you from a @liftIO@ every time
833+-- you want to enqueue a job
834+
835+-- | Create a job for immediate execution.
836+--
837+-- Internally calls 'scheduleJob' passing it the current time. Read
838+-- 'scheduleJob' for further documentation.
839+createJob :: ToJSON p
840+ => Connection
841+ -> TableName
842+ -> p
843+ -> IO Job
701844createJob conn tname payload = do
702845 t <- getCurrentTime
703846 scheduleJob conn tname payload t
704847
705−scheduleJob :: ToJSON p => Connection -> TableName -> p -> UTCTime -> IO Job
848+-- | Create a job for execution at the given time.
849+--
850+-- * If time has already past, 'jobEventListener' is going to pick this up
851+-- for execution immediately.
852+--
853+-- * If time is in the future, 'jobPoller' is going to pick this up with an
854+-- error of +/- 'cfgPollingInterval' seconds. Please do not expect very high
855+-- accuracy of when the job is actually executed.
856+scheduleJob :: ToJSON p
857+ => Connection -- ^ DB connection to use. __Note:__ This should
858+ -- /ideally/ come out of your application's DB pool,
859+ -- not the 'cfgDbPool' you used in the job-runner.
860+ -> TableName -- ^ DB-table which holds your jobs
861+ -> p -- ^ Job payload
862+ -> UTCTime -- ^ when should the job be executed
863+ -> IO Job
706864scheduleJob conn tname payload runAt = do
707865 let args = ( runAt, Queued, toJSON payload, Nothing :: Maybe Value, 0 :: Int, Nothing :: Maybe Text, Nothing :: Maybe Text )
708866 queryFormatter = toS <$> (PGS.formatQuery conn (createJobQuery tname) args)
@@ -744,6 +902,7 @@ throwParsePayloadWith parser job =
744902 either throwString (pure . Prelude.id) (eitherParsePayloadWith parser job)
745903
746904
905+-- | TODO: Should the library be doing this?
747906defaultTimedLogger :: FLogger.TimedFastLogger
748907 -> (LogLevel -> LogEvent -> LogStr)
749908 -> LogLevel
@@ -756,6 +915,7 @@ defaultTimedLogger logger logStrFn logLevel logEvent =
756915 (logStrFn logLevel logEvent) <>
757916 "\n"
758917
918+ -- TODO; Should the library be doing this?
759919defaultLogStr :: (Job -> Text)
760920 -> LogLevel
761921 -> LogEvent