@@ -1,6 +1,11 @@
1−{-# LANGUAGE TypeSynonymInstances, FlexibleInstances, NamedFieldPuns, DeriveGeneric, FlexibleContexts, TypeFamilies, StandaloneDeriving #-}
1+{-# LANGUAGE FlexibleInstances, NamedFieldPuns, DeriveGeneric, FlexibleContexts, TypeFamilies, StandaloneDeriving, RankNTypes #-}
2+{-# LANGUAGE CPP #-}
3+{-# LANGUAGE PartialTypeSignatures #-}
4+{-# LANGUAGE RecordWildCards #-}
5+{-# OPTIONS_GHC -Wno-missing-signatures -Wno-partial-type-signatures #-}
26module Test where
37
8+import Control.Monad.Trans.Control
49import Test.Tasty as Tasty
510import qualified OddJobs.Migrations as Migrations
611import qualified OddJobs.Job as Job
@@ -9,15 +14,18 @@ import Data.Functor (void)
914import Data.Pool as Pool
1015import Test.Tasty.HUnit
1116import Debug.Trace
12−import Control.Exception.Lifted (finally, catch, bracket)
17+-- import Control.Exception.Lifted (finally, catch, bracket)
18+import Control.Monad (void)
1319import Control.Monad.Logger
1420import Control.Monad.Reader
1521import Data.Aeson as Aeson
1622import Data.Aeson.TH as Aeson
17−import Control.Concurrent.Lifted
18−import Control.Concurrent.Async.Lifted
23+-- import Control.Concurrent.Lifted
24+-- import Control.Concurrent.Async.Lifted
1925import OddJobs.Job (Job(..), JobId, delaySeconds, Seconds(..))
20−import System.Log.FastLogger (fromLogStr, withFastLogger, LogType(..), defaultBufSize, FastLogger, FileLogSpec(..), newTimedFastLogger)
26+import System.Log.FastLogger ( fromLogStr, withFastLogger, LogType'(..)
27+ , defaultBufSize, FastLogger, FileLogSpec(..), newTimedFastLogger
28+ , withTimedFastLogger)
2129import System.Log.FastLogger.Date (newTimeCache, simpleTimeFormat')
2230import Data.String.Conv (toS)
2331import Data.Time
@@ -29,7 +37,6 @@ import Test.Tasty.Hedgehog
2937import qualified System.Random as R
3038import Data.String (fromString)
3139import qualified Data.IntMap.Strict as Map
32−import Control.Monad.Trans.Control (liftWith, restoreT)
3340import Control.Monad.Morph (hoist)
3441import Data.List as DL
3542import OddJobs.Web as Web
@@ -37,6 +44,10 @@ import qualified Data.Time.Convenience as Time
3744import qualified Data.Text as T
3845import Data.Ord (comparing, Down(..))
3946import Data.Maybe (fromMaybe)
47+import qualified OddJobs.ConfigBuilder as Job
48+import UnliftIO
49+import Control.Exception (ArithException)
50+import Data.Bifunctor(first)
4051
4152$(Aeson.deriveJSON Aeson.defaultOptions ''Seconds)
4253
@@ -54,16 +65,14 @@ main = do
5465 , connectDatabase = "jobs_test"
5566 }
5667
57− createAppPool = createPool
58− (PGS.connect connInfo) -- cretea a new resource
59− (PGS.close) -- destroy resource
60− 1 -- stripes
68+ createAppPool = Pool.newPool $ Pool.defaultPoolConfig
69+ (PGS.connect connInfo) -- create a new resource
70+ PGS.close -- destroy resource
6171 (fromRational 10) -- number of seconds unused resources are kept around
6272 45
63− createJobPool = createPool
64− (PGS.connect connInfo) -- cretea a new resource
65− (PGS.close) -- destroy resource
66− 1 -- stripes
73+ createJobPool = Pool.newPool $ Pool.defaultPoolConfig
74+ (PGS.connect connInfo) -- create a new resource
75+ PGS.close -- destroy resource
6776 (fromRational 10) -- number of seconds unused resources are kept around
6877 45
6978
@@ -73,11 +82,21 @@ tests appPool jobPool = testGroup "All tests"
7382 , testJobScheduling appPool jobPool
7483 , testJobFailure appPool jobPool
7584 , testEnsureShutdown appPool jobPool
76− , testGracefuleShutdown appPool jobPool
77− ]
78− , testGroup "property tests" [ testEverything appPool jobPool
79− -- , propFilterJobs appPool jobPool
85+ , testGracefulShutdown appPool jobPool
86+ , testPushFailedJobEndQueue jobPool
87+ , testRetryBackoff appPool jobPool
88+ , testGroup "callback tests"
89+ [ testOnJobFailed appPool jobPool
90+ , testOnJobStart appPool jobPool
91+ , testOnJobSuccess appPool jobPool
92+ , testOnJobTimeout appPool jobPool
8093 ]
94+ , testResourceLimitedScheduling appPool jobPool
95+ , testKillJob appPool jobPool
96+ ]
97+ -- , testGroup "property tests" [ testEverything appPool jobPool
98+ -- -- , propFilterJobs appPool jobPool
99+ -- ]
81100 ]
82101
83102myTestCase
@@ -115,20 +134,24 @@ testPayload :: Value
115134testPayload = toJSON (10 :: Int)
116135
117136jobRunner :: Job.Job -> IO ()
118−jobRunner Job{jobPayload, jobAttempts} = case (fromJSON jobPayload) of
137+jobRunner Job{jobPayload, jobAttempts} = case fromJSON jobPayload of
119138 Aeson.Error e -> error e
120139 Success (j :: JobPayload) ->
121140 let recur pload idx = case pload of
122− PayloadAlwaysFail delay -> (delaySeconds delay) >> (error $ "Forced error after " <> show delay <> " seconds")
123− PayloadSucceed delay -> (delaySeconds delay) >> pure ()
141+ PayloadAlwaysFail delay -> delaySeconds delay >> error ("Forced error after " <> show delay <> " seconds")
142+ PayloadSucceed delay -> delaySeconds delay
124143 PayloadFail delay innerpload -> if idx<jobAttempts
125144 then recur innerpload (idx + 1)
126− else (delaySeconds delay) >> (error $ "Forced error after " <> show delay <> " seconds. step=" <> show idx)
145+ else delaySeconds delay >> error ("Forced error after " <> show delay <> " seconds. step=" <> show idx)
146+ PayloadThrowStringException s -> throwString s
147+ PayloadThrowDivideByZero -> seq (1 `div` 0 :: Integer) (pure ())
127148 in recur j 0
128149
129150data JobPayload = PayloadSucceed Seconds
130151 | PayloadFail Seconds JobPayload
131152 | PayloadAlwaysFail Seconds
153+ | PayloadThrowStringException String
154+ | PayloadThrowDivideByZero
132155 deriving (Eq, Show, Generic)
133156
134157instance ToJSON JobPayload where
@@ -138,93 +161,495 @@ instance FromJSON JobPayload where
138161 parseJSON = genericParseJSON Aeson.defaultOptions
139162
140163
141−assertJobIdStatus :: (HasCallStack) => Connection -> Job.TableName -> String -> Job.Status -> JobId -> Assertion
142−assertJobIdStatus conn tname msg st jid = Job.findJobByIdIO conn tname jid >>= \case
143− Nothing -> assertFailure $ "Not expecting job to be deleted. JobId=" <> show jid
144− Just (Job{jobStatus}) -> assertEqual msg st jobStatus
164+logEventToJob :: Job.LogEvent -> Maybe Job.Job
165+logEventToJob le = case le of
166+ Job.LogJobStart j -> Just j
167+ Job.LogJobSuccess j _ -> Just j
168+ Job.LogJobFailed j _ _ _ -> Just j
169+ Job.LogJobTimeout j -> Just j
170+ Job.LogKillJobSuccess j -> Just j
171+ Job.LogKillJobFailed j -> Just j
172+ Job.LogPoll -> Nothing
173+ Job.LogWebUIRequest -> Nothing
174+ Job.LogText _ -> Nothing
175+
176+assertJobIdStatus :: HasCallStack
177+ => Connection
178+ -> Job.TableName
179+ -> IORef [Job.LogEvent]
180+ -> String
181+ -> Job.Status
182+ -> JobId
183+ -> Assertion
184+assertJobIdStatus conn tname logRef msg st jid = do
185+ logs <- readIORef logRef
186+ let _mjid = Just jid
187+ case st of
188+ Job.Success ->
189+ assertBool (msg <> ": Success event not found in job-logs for JobId=" <> show jid) $
190+ flip DL.any logs $ \case
191+ Job.LogJobSuccess j _ -> jid == Job.jobId j
192+ _ -> False
193+
194+ Job.Queued ->
195+ assertBool (msg <> ": Not expecting to find a queued job in the the job-logs JobId=" <> show jid) $
196+ not $ flip DL.any logs $ \case
197+ Job.LogJobStart j -> jid == Job.jobId j
198+ Job.LogJobSuccess j _ -> jid == Job.jobId j
199+ Job.LogJobFailed j _ _ _ -> jid == Job.jobId j
200+ Job.LogJobTimeout j -> jid == Job.jobId j
201+ Job.LogKillJobSuccess _ -> False
202+ Job.LogKillJobFailed _ -> False
203+ Job.LogWebUIRequest -> False
204+ Job.LogText _ -> False
205+ Job.LogPoll -> False
206+
207+ Job.Failed ->
208+ assertBool (msg <> ": Failed event not found in job-logs for JobId=" <> show jid) $
209+ flip DL.any logs $ \case
210+ Job.LogJobFailed j _ _ _ -> jid == Job.jobId j
211+ _ -> False
212+
213+ Job.Cancelled ->
214+ assertBool (msg <> ": Job killed event not found in job-logs for JobId=" <> show jid) $
215+ flip DL.any logs $ \case
216+ Job.LogKillJobSuccess j -> jid == Job.jobId j
217+ _ -> False
145218
146−ensureJobId :: (HasCallStack) => Connection -> Job.TableName -> JobId -> IO Job
219+ Job.Retry ->
220+ assertBool (msg <> ": Failed event not found in job-logs for JobId=" <> show jid) $
221+ flip DL.any logs $ \case
222+ Job.LogJobFailed j _ _ _ -> jid == Job.jobId j
223+ _ -> False
224+
225+ Job.Locked ->
226+ assertBool (msg <> ": Start event should be present in the log for a locked job JobId=" <> show jid) $
227+ flip DL.any logs $ \case
228+ Job.LogJobStart j -> jid == Job.jobId j
229+ _ -> False
230+
231+ when (st /= Job.Success) $ do
232+ Job.findJobByIdIO conn tname jid >>= \case
233+ Nothing -> assertFailure $ "Not expecting job to be deleted. JobId=" <> show jid
234+ Just (Job{jobStatus}) -> assertEqual msg st jobStatus
235+
236+ensureJobId :: HasCallStack => Connection -> Job.TableName -> JobId -> IO Job
147237ensureJobId conn tname jid = Job.findJobByIdIO conn tname jid >>= \case
148238 Nothing -> error $ "Not expecting job to be deleted. JobId=" <> show jid
149239 Just j -> pure j
150240
151−-- withRandomTable :: (MonadIO m) => Pool Connection -> (Job.TableName -> m a) -> m a
241+withRandomTable :: (MonadBaseControl
242+ IO m, MonadUnliftIO m) => Pool Connection -> (Job.TableName -> m a) -> m a
152243withRandomTable jobPool action = do
153− (tname :: Job.TableName) <- liftIO ((("jobs_" <>) . fromString) <$> (replicateM 10 (R.randomRIO ('a', 'z'))))
244+ runInIO <- askRunInIO
245+ (tname :: Job.TableName) <- liftIO (fromString . ("jobs_" <>) <$> replicateM 10 (R.randomRIO ('a', 'z')))
154246 finally
155− ((Pool.withResource jobPool $ \conn -> (liftIO $ Migrations.createJobTable conn tname)) >> (action tname))
156− (Pool.withResource jobPool $ \conn -> liftIO $ void $ PGS.execute_ conn ("drop table if exists " <> tname <> ";"))
247+ (liftIO $ Pool.withResource jobPool (\conn -> liftIO $ Migrations.createJobTable conn tname) >> runInIO (action tname))
248+ (liftIO $ Pool.withResource jobPool $ \conn -> liftIO $ void $ PGS.execute conn "drop table if exists ?" (Only tname))
249+
250+runSingleJobFromQueueBlock :: Job.Config -> IO (Maybe (Either SomeException ()))
251+runSingleJobFromQueueBlock cfg =
252+ traverse waitCatch =<< runSingleJobFromQueue cfg
253+
254+-- | waits untill the job has started
255+runSingleJobFromQueue :: Job.Config -> IO (Maybe (Async ()))
256+runSingleJobFromQueue config' = do
257+ r <- liftIO $ newIORef mempty
258+ waitTillJobStart <- newEmptyMVar
259+ let config = config' {
260+ Job.cfgPollingInterval = 0
261+ , Job.cfgJobRunner = \job -> do
262+ putMVar waitTillJobStart ()
263+ jobRunner job
264+ }
265+ let monitorEnv = Job.RunnerEnv
266+ { Job.envConfig = config
267+ , Job.envJobThreadsRef = r
268+ }
269+ result <- runReaderT (Job.pollRunJob "runSingleJobFromQueue" $ readResourceConfig config) monitorEnv
270+ forM_ result $ const $ readMVar waitTillJobStart
271+ pure result
272+
273+readResourceConfig :: Job.Config -> Maybe Job.ResourceCfg
274+readResourceConfig cfg = case Job.cfgConcurrencyControl cfg of
275+ Job.ResourceLimits res -> Just res
276+ _ -> Nothing
277+
278+withRandomResourceTables :: (MonadBaseControl IO m, MonadUnliftIO m) => Int -> Pool Connection -> Job.TableName -> (Job.ResourceCfg -> m a) -> m a
279+withRandomResourceTables defaultLimit jobPool tname action = do
280+ resCfgResourceTable <- liftIO $ fromString . ("resources_" <>) <$> replicateM 10 (R.randomRIO ('a', 'z'))
281+ resCfgUsageTable <- liftIO $ fromString . ("usage_" <>) <$> replicateM 10 (R.randomRIO ('a', 'z'))
282+ resCfgCheckResourceFunction <- liftIO $ fromString . ("check_resource_fn_" <>) <$> replicateM 10 (R.randomRIO ('a', 'z'))
283+ let resCfg = Job.ResourceCfg
284+ { Job.resCfgDefaultLimit = defaultLimit
285+ , ..
286+ }
157287
158−-- withNewJobMonitor :: (Pool Connection) -> (TableName -> Assertion) -> Assertion
159−withNewJobMonitor jobPool actualTest = withRandomTable jobPool $ \tname -> withNamedJobMonitor tname jobPool (actualTest tname)
160−
161−withNamedJobMonitor tname jobPool actualTest = do
162− (defaults, cleanup) <- Test.defaultJobMonitor tname jobPool
163− let jobMonitorSettings = defaults{ Job.monitorJobRunner = jobRunner
164− , Job.monitorDefaultMaxAttempts = 3
165− }
166288 finally
167− (withAsync (Job.runJobMonitor jobMonitorSettings) (const actualTest))
168− (cleanup)
289+ (do
290+ liftIO $ Pool.withResource jobPool $ \conn -> liftIO $ Migrations.createResourceTables conn tname resCfg
291+ action resCfg)
292+ (liftIO $ Pool.withResource jobPool $ \conn -> liftIO $ void $ PGS.execute conn
293+ "drop function if exists ?; drop table if exists ?; drop table if exists ?"
294+ (resCfgCheckResourceFunction, resCfgUsageTable, resCfgResourceTable))
295+
296+testMaxAttempts :: Int
297+testMaxAttempts = 3
169298
299+-- withNewJobMonitor :: (Pool Connection) -> (TableName -> Assertion) -> Assertion
300+withNewJobMonitor jobPool actualTest = do
301+ withRandomTable jobPool $ \tname -> do
302+ withNamedJobMonitor tname jobPool Prelude.id (actualTest tname)
303+
304+withNamedJobMonitor :: Job.TableName -> Pool Connection -> (Job.Config -> Job.Config) -> (IORef [Job.LogEvent] -> IO a) -> IO a
305+withNamedJobMonitor tname jobPool cfgFn actualTest =
306+ withConfig tname jobPool cfgFn $ \logRef cfg -> do
307+ withAsync (Job.startJobRunner cfg) (const $ actualTest logRef)
308+
309+withConfig :: Job.TableName -> Pool Connection -> (Job.Config -> Job.Config) -> (IORef [Job.LogEvent] -> Job.Config -> IO a) -> IO a
310+withConfig tname jobPool cfgFn configConsumer = do
311+ logRef :: IORef [Job.LogEvent] <- newIORef []
312+ tcache <- newTimeCache simpleTimeFormat'
313+ withTimedFastLogger tcache LogNone $ \tlogger -> do
314+ let flogger logLevel logEvent = do
315+ tlogger $ \t -> toLogStr t <> " | " <> Job.defaultLogStr Job.defaultJobType logLevel logEvent
316+ atomicModifyIORef' logRef (\logs -> (logEvent:logs, ()))
317+ cfg = Job.mkConfig flogger tname jobPool Job.UnlimitedConcurrentJobs jobRunner (\cfg' -> cfgFn $ cfg'{Job.cfgDefaultMaxAttempts=testMaxAttempts})
318+ configConsumer logRef cfg
170319
171320payloadGen :: MonadGen m => m JobPayload
172321payloadGen = Gen.recursive Gen.choice nonRecursive recursive
173322 where
174323 nonRecursive = [ PayloadAlwaysFail <$> Gen.element [1, 2, 3]
175324 , PayloadSucceed <$> Gen.element [1, 2, 3]]
176− recursive = [ PayloadFail <$> (Gen.element [1, 2, 3]) <*> payloadGen ]
177−
178−testJobCreation appPool jobPool = testCase "job creation" $ withNewJobMonitor jobPool $ \tname -> Pool.withResource appPool $ \conn -> do
179− Job{jobId} <- Job.createJob conn tname (PayloadSucceed 0)
180− delaySeconds $ Seconds 6
181− assertJobIdStatus conn tname "Expecting job to tbe successful by now" Job.Success jobId
182−
183−testEnsureShutdown appPool jobPool = testCase "ensure shutdown" $ withRandomTable jobPool $ \tname -> do
184− jid <- scheduleJob tname
185− delaySeconds (2 * Job.defaultPollingInterval)
186− Pool.withResource appPool $ \conn ->
187− assertJobIdStatus conn tname "Job should still be in queued state if job-monitor is no longer running" Job.Queued jid
325+ recursive = [ PayloadFail <$> Gen.element [1, 2, 3] <*> payloadGen ]
326+
327+testJobCreation appPool jobPool = testCase "job creation" $ do
328+ withNewJobMonitor jobPool $ \tname logRef -> do
329+ Pool.withResource appPool $ \conn -> do
330+ Job{jobId} <- Job.createJob conn tname (PayloadSucceed 0)
331+ delaySeconds $ Seconds 6
332+ assertJobIdStatus conn tname logRef "Expecting job to be successful by now" Job.Success jobId
333+
334+testEnsureShutdown appPool jobPool = testCase "ensure shutdown" $ do
335+ withRandomTable jobPool $ \tname -> do
336+ (jid, logRef) <- scheduleJob tname
337+ delaySeconds (2 * Job.defaultPollingInterval)
338+ Pool.withResource appPool $ \conn -> do
339+ assertJobIdStatus conn tname logRef "Job should still be in queued state if job-monitor is no longer running" Job.Queued jid
188340 where
189− scheduleJob tname = withNamedJobMonitor tname jobPool $ do
341+ scheduleJob tname = withNamedJobMonitor tname jobPool Prelude.id $ \logRef -> do
342+ t <- getCurrentTime
343+ Pool.withResource appPool $ \conn -> do
344+ Job{jobId} <- Job.scheduleJob conn tname (PayloadSucceed 0) (addUTCTime (fromIntegral (2 * unSeconds Job.defaultPollingInterval)) t)
345+ assertJobIdStatus conn tname logRef "Job is scheduled in future, should still be queueud" Job.Queued jobId
346+ pure (jobId, logRef)
347+
348+testGracefulShutdown :: Pool Connection -> Pool Connection -> TestTree
349+testGracefulShutdown appPool jobPool =
350+ testCase "ensure wait for all jobs is correct" $
351+ withRandomTable jobPool $ \tname -> do
352+ waitTillJobStart <- newEmptyMVar
353+ Pool.withResource appPool $ \conn -> do
354+ j1 <- Job.createJob conn tname (PayloadSucceed $ 4 * Job.defaultPollingInterval)
355+ time<- getCurrentTime
356+ j2 <- Job.scheduleJob conn tname (PayloadSucceed 0) (addUTCTime (fromIntegral $ unSeconds $ Job.defaultPollingInterval * 2) time)
357+ logRef <- withNamedJobMonitor tname jobPool (\cfg -> cfg { Job.cfgOnJobStart = \ _ -> do
358+ putMVar waitTillJobStart ()
359+ } ) $ \logRef -> do
360+ readMVar waitTillJobStart -- as soon as the job has started we quit the monitor:w
361+ pure logRef -- eg closing the withNamedJobMonitor closure shutsdwon the monitor
362+ assertJobIdStatus conn tname logRef "Expecting the first job to be completed successfully if graceful shutdown is implemented correctly" Job.Success (jobId j1)
363+ assertJobIdStatus conn tname logRef "Expecting the second job to be queued because no new job should be picked up during graceful shutdown" Job.Queued (jobId j2)
364+
365+testJobScheduling appPool jobPool = testCase "job scheduling" $ do
366+ withNewJobMonitor jobPool $ \tname logRef -> do
367+ Pool.withResource appPool $ \conn -> do
190368 t <- getCurrentTime
369+ job@Job{jobId} <- Job.scheduleJob conn tname (PayloadSucceed 0) (addUTCTime (fromIntegral (3600 :: Integer)) t)
370+ delaySeconds $ Seconds 2
371+ assertJobIdStatus conn tname logRef "Job is scheduled in the future. It should NOT have been successful by now" Job.Queued jobId
372+ _ <- Job.saveJobIO conn tname job{jobRunAt = addUTCTime (fromIntegral (-1 :: Integer)) t}
373+ delaySeconds (Job.defaultPollingInterval + Seconds 2)
374+ assertJobIdStatus conn tname logRef "Job had a runAt date in the past. It should have been successful by now" Job.Success jobId
375+
376+testJobFailure appPool jobPool = testCase "job failure" $ do
377+ withNewJobMonitor jobPool $ \tname _logRef -> do
378+ Pool.withResource appPool $ \conn -> do
379+ Job{jobId} <- Job.createJob conn tname (PayloadAlwaysFail 0)
380+ delaySeconds $ calculateRetryDelay testMaxAttempts + Seconds 3
381+ Job{jobAttempts, jobStatus} <- ensureJobId conn tname jobId
382+ assertEqual "Exepcting job to be in Failed status" Job.Failed jobStatus
383+ assertEqual ("Expecting job attempts to be 3. Found " <> show jobAttempts) 3 jobAttempts
384+
385+testPushFailedJobEndQueue :: Pool Connection -> TestTree
386+testPushFailedJobEndQueue jobPool = testCase "testPushFailedJobEndQueue" $ do
387+ withRandomTable jobPool $ \tname -> do
388+ Pool.withResource jobPool $ \conn -> do
389+ job1 <- Job.createJob conn tname (PayloadAlwaysFail 0)
390+ job2 <- Job.createJob conn tname (PayloadAlwaysFail 0)
391+ _ <- Job.saveJobIO conn tname (job1 {jobAttempts = 1})
392+ [Only resId] <- Job.jobPollingIO conn "testPushFailedJobEndQueue" tname 5
393+ assertEqual
394+ "Expecting the current job to be 2 since job 1 has been modified"
395+ (jobId job2) resId
396+
397+testOnJobFailed appPool jobPool = testCase "job error handler" $ do
398+ withRandomTable jobPool $ \tname -> do
399+ Pool.withResource appPool $ \conn -> do
400+ _ <- Job.createJob conn tname (PayloadThrowStringException "forced exception")
401+ _ <- Job.createJob conn tname PayloadThrowDivideByZero
402+ mvar1 <- newMVar False
403+ mvar2 <- newMVar False
404+ let addErrHandlers cfg = cfg { Job.cfgOnJobFailed = [ Job.JobErrHandler errHandler2
405+ , Job.JobErrHandler errHandler1
406+ ]
407+ }
408+ errHandler1 (_ :: StringException) _ _ = swapMVar mvar1 True
409+ errHandler2 (_ :: ArithException) _ _ = swapMVar mvar2 True
410+ withNamedJobMonitor tname jobPool addErrHandlers $ \_logRef -> do
411+ delaySeconds (Job.defaultPollingInterval * 3)
412+ readMVar mvar1 >>= assertEqual "Error Handler 1 doesn't seem to have run" True
413+ readMVar mvar2 >>= assertEqual "Error Handler 2 doesn't seem to have run" True
414+
415+testOnJobStart appPool jobPool = testCase "onJobStart" $ do
416+ withRandomTable jobPool $ \tname -> do
417+ callbackRef <- newIORef False
418+ let callback = const $ modifyIORef' callbackRef (const True)
419+ withNamedJobMonitor tname jobPool (\cfg -> cfg{Job.cfgOnJobStart=callback}) $ \_logRef -> do
420+ Pool.withResource appPool $ \conn -> do
421+ _ <- Job.createJob conn tname (PayloadSucceed 0)
422+ delaySeconds (2 * Job.defaultPollingInterval)
423+ readIORef callbackRef >>= assertBool "It seems that onJobStart callback has not been called"
424+
425+testOnJobSuccess appPool jobPool = testCase "onJobSuccess" $ do
426+ withRandomTable jobPool $ \tname -> do
427+ callbackRef <- newIORef False
428+ let callback = const $ modifyIORef' callbackRef (const True)
429+ withNamedJobMonitor tname jobPool (\cfg -> cfg{Job.cfgOnJobSuccess=callback}) $ \_logRef -> do
430+ Pool.withResource appPool $ \conn -> do
431+ _ <- Job.createJob conn tname (PayloadSucceed 0)
432+ delaySeconds (2 * Job.defaultPollingInterval)
433+ readIORef callbackRef >>= assertBool "It seems that onJobSuccess callback has not been called"
434+
435+testOnJobTimeout appPool jobPool = testCase "onJobTimeout" $ do
436+ withRandomTable jobPool $ \tname -> do
437+ callbackRef <- newIORef False
438+ let callback = const $ modifyIORef' callbackRef (const True)
439+ withNamedJobMonitor tname jobPool (\cfg -> cfg{Job.cfgOnJobTimeout=callback, Job.cfgDefaultJobTimeout=Seconds 2}) $ \_logRef -> do
440+ Pool.withResource appPool $ \conn -> do
441+ _ <- Job.createJob conn tname (PayloadSucceed 10)
442+ delaySeconds (2 * Job.defaultPollingInterval)
443+ readIORef callbackRef >>= assertBool "It seems that onJobTimeout callback has not been called"
444+
445+testRetryBackoff appPool jobPool = testCase "retry backoff" $ do
446+ withRandomTable jobPool $ \tname -> do
447+ withNamedJobMonitor tname jobPool modifyRetryBackoff $ \_logRef -> do
191448 Pool.withResource appPool $ \conn -> do
192− Job{jobId} <- Job.scheduleJob conn tname (PayloadSucceed 0) (addUTCTime (fromIntegral (2 * (unSeconds Job.defaultPollingInterval))) t)
193− assertJobIdStatus conn tname "Job is scheduled in future, should still be queueud" Job.Queued jobId
194− pure jobId
195−
196−testGracefuleShutdown appPool jobPool = testCase "ensure graceful shutdown" $ withRandomTable jobPool $ \tname -> do
197− (j1, j2) <- withNamedJobMonitor tname jobPool $ Pool.withResource appPool $ \conn -> do
198− t <- getCurrentTime
199− j1 <- Job.createJob conn tname (PayloadSucceed $ 2 * Job.defaultPollingInterval)
200− j2 <- Job.scheduleJob conn tname (PayloadSucceed 0) (addUTCTime (fromIntegral $ unSeconds $ Job.defaultPollingInterval) t)
201− liftIO $ putStrLn "created"
202− pure (j1, j2)
203− Pool.withResource appPool $ \conn -> do
204− delaySeconds 1
205− assertJobIdStatus conn tname "Expecting the first job to be in locked state because it should be running" Job.Locked (jobId j1)
206− assertJobIdStatus conn tname "Expecting the second job to be queued because no new job should be picked up during graceful shutdown" Job.Queued (jobId j2)
207− delaySeconds $ 3 * Job.defaultPollingInterval
208− assertJobIdStatus conn tname "Expecting the first job to be completed successfully if graceful shutdown is implemented correctly" Job.Success (jobId j1)
209− assertJobIdStatus conn tname "Expecting the second job to be queued because no new job should be picked up during graceful shutdown" Job.Queued (jobId j2)
210−
211− pure ()
212−
213−testJobScheduling appPool jobPool = testCase "job scheduling" $ withNewJobMonitor jobPool $ \tname -> Pool.withResource appPool $ \conn -> do
214− t <- getCurrentTime
215− job@Job{jobId} <- Job.scheduleJob conn tname (PayloadSucceed 0) (addUTCTime (fromIntegral 3600) t)
216− delaySeconds $ Seconds 2
217− assertJobIdStatus conn tname "Job is scheduled in the future. It should NOT have been successful by now" Job.Queued jobId
218− j <- Job.saveJobIO conn tname job{jobRunAt = (addUTCTime (fromIntegral (-1)) t)}
219− delaySeconds (Job.defaultPollingInterval + (Seconds 2))
220− assertJobIdStatus conn tname "Job had a runAt date in the past. It should have been successful by now" Job.Success jobId
221−
222−testJobFailure appPool jobPool = testCase "job retry" $ withNewJobMonitor jobPool $ \tname -> Pool.withResource appPool $ \conn -> do
223− Job{jobId} <- Job.createJob conn tname (PayloadAlwaysFail 0)
224− delaySeconds $ Seconds 15
225− Job{jobAttempts, jobStatus} <- ensureJobId conn tname jobId
226− assertEqual "Exepcting job to be in Failed status" Job.Failed jobStatus
227− assertEqual ("Expecting job attempts to be 3. Found " <> show jobAttempts) 3 jobAttempts
449+ jobQueueTime <- getCurrentTime
450+ Job{jobId} <- Job.createJob conn tname (PayloadAlwaysFail 0)
451+ delaySeconds (2 * Job.defaultPollingInterval)
452+
453+ let assertBackedOff = do
454+ Job{jobAttempts, jobStatus, jobRunAt} <- ensureJobId conn tname jobId
455+ assertEqual "Exepcting job to be in Retry status" Job.Retry jobStatus
456+ assertBool "Expecting job runAt to be in the future"
457+ (jobRunAt >= addUTCTime (fromIntegral $ unSeconds $ backoff jobAttempts) jobQueueTime)
458+
459+ assertBackedOff
460+
461+ -- Run it for another attempt and check the backoff scales
462+ job <- ensureJobId conn tname jobId
463+ _ <- Job.saveJobIO conn tname (job {jobRunAt = jobQueueTime})
464+ delaySeconds (2 * Job.defaultPollingInterval)
465+
466+ assertBackedOff
467+ where
468+
469+ backoff attempts = Seconds $ 300 * attempts
470+
471+ modifyRetryBackoff cfg
472+ = cfg { Job.cfgDefaultRetryBackoff = pure . backoff }
473+
474+testResourceLimitedScheduling _appPool jobPool = testGroup "concurrency control by resources"
475+ [ testCase "resources control how jobs run" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
476+ let firstResource = [(Job.ResourceId "first", 1)]
477+ secondResource = [(Job.ResourceId "second", 1)]
478+
479+ Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
480+ Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
481+ Job{jobId = thirdJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) secondResource
482+
483+ _jone <- runSingleJobFromQueue cfg
484+ _jtwo <- runSingleJobFromQueue cfg
485+ _jthree <- runSingleJobFromQueue cfg
486+
487+ assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
488+ assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued secondJob
489+ assertJobIdStatus conn tname logRef "Job uses different resources - it should be running" Job.Locked thirdJob
490+
491+ , testCase "resource-less jobs run normally" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
492+ Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) []
493+ Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) []
494+
495+ _jone <- runSingleJobFromQueue cfg
496+ _jtwo <- runSingleJobFromQueue cfg
497+
498+ assertJobIdStatus conn tname logRef "First job uses no resources - it should be running" Job.Locked firstJob
499+ assertJobIdStatus conn tname logRef "Second job uses no resources - it should be running" Job.Locked secondJob
500+
501+ , testCase "jobs can hold multiple resources" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
502+ let firstResource = [(Job.ResourceId "first", 1)]
503+ secondResource = [(Job.ResourceId "second", 1)]
504+
505+ Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) (firstResource <> secondResource)
506+ Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
507+ Job{jobId = thirdJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) secondResource
508+
509+ jone <- runSingleJobFromQueue cfg
510+ jtwo <- runSingleJobFromQueue cfg
511+ jthree <- runSingleJobFromQueue cfg
512+ assertEqual "first job claimed the work" (void jone) (Just ())
513+ assertEqual "second thread was blocked" (void jtwo) Nothing
514+ assertEqual "third thread was blocked " (void jthree) Nothing
515+
516+ assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
517+ assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued secondJob
518+ assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued thirdJob
519+
520+ , testCase "resources are freed when jobs succeed" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
521+ let firstResource = [(Job.ResourceId "first", 1)]
522+
523+ Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
524+ Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
525+
526+ jone <- runSingleJobFromQueue cfg
527+ _jtwo <- runSingleJobFromQueue cfg
528+
529+ assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
530+ assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued secondJob
531+
532+ forM_ jone wait
533+
534+ _three <- runSingleJobFromQueue cfg
535+
536+ assertJobIdStatus conn tname logRef "First job should have succeeded by now" Job.Success firstJob
537+ assertJobIdStatus conn tname logRef "Second job should be running after first job suceeded" Job.Locked secondJob
538+
539+ , testCase "resources are freed when jobs fail" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
540+ let firstResource = [(Job.ResourceId "first", 1)]
541+
542+ Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadAlwaysFail 10) firstResource
543+ Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
544+
545+ jone <- runSingleJobFromQueue cfg
546+ _jtwo <- runSingleJobFromQueue cfg
547+
548+ assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
549+ assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued secondJob
550+
551+ jobRes <- forM jone waitCatch
552+ assertEqual "should've thrown an exception" (Just (Left ())) (first (const ()) <$> jobRes)
553+
554+ _three <- runSingleJobFromQueue cfg
555+
556+ assertJobIdStatus conn tname logRef "First job should have failed by now" Job.Failed firstJob
557+ assertJobIdStatus conn tname logRef "Second job should be running after first job failed" Job.Locked secondJob
558+
559+ , testCase "resources allow for multiple usage" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
560+ let firstResource = [(Job.ResourceId "first", 1)]
561+ resCfg' = resCfg { Job.resCfgDefaultLimit = 2 }
562+
563+ Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg' (PayloadSucceed 10) firstResource
564+ Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg' (PayloadSucceed 10) firstResource
565+
566+ _jone <- runSingleJobFromQueue cfg
567+ _jtwo <- runSingleJobFromQueue cfg
568+
569+ assertJobIdStatus conn tname logRef "First job uses some resources - it should be running" Job.Locked firstJob
570+ assertJobIdStatus conn tname logRef "Second job uses some resources - it should be running" Job.Locked secondJob
571+
572+ , testCase "resource usage can differ by job" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
573+ let firstResource n = [(Job.ResourceId "first", n)]
574+ resCfg' = resCfg { Job.resCfgDefaultLimit = 5 }
575+
576+ Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg' (PayloadSucceed 10) (firstResource 4)
577+ Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg' (PayloadSucceed 10) (firstResource 3)
578+ Job{jobId = thirdJob} <- Job.createJobWithResources conn tname resCfg' (PayloadSucceed 10) (firstResource 2)
579+
580+ jone <- runSingleJobFromQueue cfg
581+ _jtwo <- runSingleJobFromQueue cfg
582+ _jthree <- runSingleJobFromQueue cfg
583+
584+ -- TODO, so it's not locked but already succeeded, which is werid, but the rest is valid?
585+ assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
586+ assertJobIdStatus conn tname logRef "Job doesn't fit alongside first job - it shouldn't be running" Job.Queued secondJob
587+ assertJobIdStatus conn tname logRef "Job doesn't fit alongside first job - it shouldn't be running" Job.Queued thirdJob
588+
589+ forM_ jone wait
590+
591+ _jfour <- runSingleJobFromQueue cfg
592+ _jfive <- runSingleJobFromQueue cfg
593+
594+ assertJobIdStatus conn tname logRef "First job should have succeeded by now" Job.Success firstJob
595+ assertJobIdStatus conn tname logRef "Second job should be running after first job suceeded" Job.Locked secondJob
596+ assertJobIdStatus conn tname logRef "Third job fits alongside second job - it should be running after first job suceeded" Job.Locked thirdJob
597+
598+ , testCase "resource limits permanently block too-big jobs" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
599+ let firstResource = [(Job.ResourceId "first", 2)]
600+
601+ Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
602+
603+ void $ runSingleJobFromQueueBlock cfg
604+
605+ assertJobIdStatus conn tname logRef "Job requires too many resources - shouldn't be running" Job.Queued firstJob
606+
607+ , testCase "resources with 0 limit block jobs" $ setup' jobPool 0 $ \tname resCfg logRef conn cfg -> do
608+ let firstResource = [(Job.ResourceId "first", 1)]
609+
610+ Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
611+
612+ void $ runSingleJobFromQueueBlock cfg
613+
614+ assertJobIdStatus conn tname logRef "Job can't access 0-limited resources - shouldn't be running" Job.Queued firstJob
615+
616+ , testCase "resources with negative limit block jobs" $ setup' jobPool (-5) $ \tname resCfg logRef conn cfg -> do
617+ let firstResource = [(Job.ResourceId "first", 1)]
618+
619+ Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
620+
621+ void $ runSingleJobFromQueueBlock cfg
622+
623+ assertJobIdStatus conn tname logRef "Job can't access negative-limited resources - shouldn't be running" Job.Queued firstJob
624+ ]
625+
626+setup jobPool = setup' jobPool 1
627+
628+setup' jobPool defaultLimit action =
629+ withRandomTable jobPool $ \tname ->
630+ withRandomResourceTables defaultLimit jobPool tname $ \resCfg ->
631+ withConfig tname jobPool (cfgFn resCfg) $ \logRef cfg -> do
632+ Pool.withResource jobPool $ \conn -> action tname resCfg logRef conn cfg
633+
634+ where
635+ cfgFn resCfg cfg = cfg
636+ { Job.cfgDefaultMaxAttempts = 1 -- Simplifies some tests where we have a failing job
637+ , Job.cfgConcurrencyControl = Job.ResourceLimits resCfg
638+ , Job.cfgDeleteSuccessfulJobs = False
639+ }
640+
641+testKillJob appPool jobPool = testCase "killing a ongoing job" $ do
642+ withRandomTable jobPool $ \tname -> withNamedJobMonitor tname jobPool Prelude.id $ \logRef -> do
643+ Pool.withResource appPool $ \conn -> do
644+ Job{jobId = jid} <- Job.createJob conn tname (PayloadSucceed 60)
645+ delaySeconds $ Job.defaultPollingInterval + Seconds 2
646+
647+ assertJobIdStatus conn tname logRef "Job should be running" Job.Locked jid
648+
649+ _ <- Job.killJobIO conn tname jid
650+ delaySeconds $ Job.defaultPollingInterval + Seconds 2
651+
652+ assertJobIdStatus conn tname logRef "Job is cancelled and the job thread should be killed" Job.Cancelled jid
228653
229654data JobEvent = JobStart
230655 | JobRetry
@@ -232,110 +657,111 @@ data JobEvent = JobStart
232657 | JobFailed
233658 deriving (Eq, Show)
234659
235−testEverything appPool jobPool = testProperty "test everything" $ property $ do
660+-- testEverything appPool jobPool = testProperty "test everything" $ property $ do
236661
237− jobPayloads <- forAll $ Gen.list (Range.linear 300 1000) payloadGen
238− jobsMVar <- liftIO $ newMVar (Map.empty :: Map.IntMap [(JobEvent, Job.Job)])
662+-- jobPayloads <- forAll $ Gen.list (Range.linear 300 1000) payloadGen
663+-- jobsMVar <- liftIO $ newMVar (Map.empty :: Map.IntMap [(JobEvent, Job.Job)])
239664
240− let maxDelay = sum $ map (payloadDelay jobPollingInterval) jobPayloads
241− completeRun = ((unSeconds maxDelay) `div` concurrencyFactor) + (2 * (unSeconds testPollingInterval))
665+-- let maxDelay = sum $ map (payloadDelay jobPollingInterval) jobPayloads
666+-- completeRun = ((unSeconds maxDelay) `div` concurrencyFactor) + (2 * (unSeconds testPollingInterval))
242667
243− shutdownAfter <- forAll $ Gen.choice [ pure $ Seconds completeRun -- Either we allow all jobs to be completed properly
244− , (Seconds <$> (Gen.int $ Range.linear 1 completeRun)) -- Or, we shutdown early after a random number of seconds
245− ]
668+-- shutdownAfter <- forAll $ Gen.choice
669+-- [ pure $ Seconds completeRun -- Either we allow all jobs to be completed properly
670+-- , (Seconds <$> (Gen.int $ Range.linear 1 completeRun)) -- Or, we shutdown early after a random number of seconds
671+-- ]
246672
247− test $ withRandomTable jobPool $ \tname -> do
248− (defaults, cleanup) <- liftIO $ Test.defaultJobMonitor tname jobPool
249− let jobMonitorSettings = defaults { Job.monitorJobRunner = jobRunner
250− , Job.monitorTableName = tname
251− , Job.monitorOnJobStart = onJobEvent JobStart jobsMVar
252− , Job.monitorOnJobFailed = onJobEvent JobRetry jobsMVar
253− , Job.monitorOnJobPermanentlyFailed = onJobEvent JobFailed jobsMVar
254− , Job.monitorOnJobSuccess = onJobEvent JobSuccess jobsMVar
255− , Job.monitorDefaultMaxAttempts = 3
256− , Job.monitorPollingInterval = jobPollingInterval
257− }
673+-- test $ withRandomTable jobPool $ \tname -> do
674+-- (defaults, cleanup) <- liftIO $ Test.defaultJobMonitor tname jobPool
675+-- let jobMonitorSettings = defaults { Job.monitorJobRunner = jobRunner
676+-- , Job.monitorTableName = tname
677+-- , Job.monitorOnJobStart = onJobEvent JobStart jobsMVar
678+-- , Job.monitorOnJobFailed = onJobEvent JobRetry jobsMVar
679+-- , Job.monitorOnJobPermanentlyFailed = onJobEvent JobFailed jobsMVar
680+-- , Job.monitorOnJobSuccess = onJobEvent JobSuccess jobsMVar
681+-- , Job.monitorDefaultMaxAttempts = 3
682+-- , Job.monitorPollingInterval = jobPollingInterval
683+-- }
258684
259− (jobs :: [Job]) <- withAsync
260− (liftIO $ Job.runJobMonitor jobMonitorSettings)
261− (const $ finally
262− (liftIO $ actualTest shutdownAfter jobPayloads tname jobsMVar)
263− (liftIO cleanup))
685+-- (jobs :: [Job]) <- withAsync
686+-- (liftIO $ Job.runJobMonitor jobMonitorSettings)
687+-- (const $ finally
688+-- (liftIO $ actualTest shutdownAfter jobPayloads tname jobsMVar)
689+-- (liftIO cleanup))
264690
265− jobAudit <- takeMVar jobsMVar
691+-- jobAudit <- takeMVar jobsMVar
266692
267− [(Only (lockedJobCount :: Int))] <- liftIO $ Pool.withResource appPool $ \conn ->
268− PGS.query conn ("SELECT coalesce(count(id), 0) FROM " <> tname <> " where status=?") (Only Job.Locked)
693+-- [(Only (lockedJobCount :: Int))] <- liftIO $ Pool.withResource appPool $ \conn ->
694+-- PGS.query conn ("SELECT coalesce(count(id), 0) FROM " <> tname <> " where status=?") (Only Job.Locked)
269695
270− -- ALL jobs should show up in the audit, which means they should have
271− -- been attempted /at least/ once
272− (DL.sort $ map jobId jobs) === (DL.sort $ Map.keys jobAudit)
696+-- -- ALL jobs should show up in the audit, which means they should have
697+-- -- been attempted /at least/ once
698+-- (DL.sort $ map jobId jobs) === (DL.sort $ Map.keys jobAudit)
273699
274− -- No job should be in a locked state
275− 0 === lockedJobCount
700+-- -- No job should be in a locked state
701+-- 0 === lockedJobCount
276702
277− -- No job should've been simultaneously picked-up by more than one
278− -- worker
279− True === (Map.foldl (\m js -> m && noRaceCondition js) True jobAudit)
703+-- -- No job should've been simultaneously picked-up by more than one
704+-- -- worker
705+-- True === (Map.foldl (\m js -> m && noRaceCondition js) True jobAudit)
280706
281− liftIO $ print $ "Test passed with job-count = " <> show (length jobPayloads)
707+-- liftIO $ print $ "Test passed with job-count = " <> show (length jobPayloads)
282708
283− where
709+-- where
284710
285− testPollingInterval = 5
286− concurrencyFactor = 5
287−
288− noRaceCondition js = DL.foldl (&&) True $ DL.zipWith (\(x,_) (y,_) -> not $ x==JobStart && y==JobStart) js (tail js)
289−
290− jobPollingInterval = Seconds 2
291−
292− onJobEvent evt jobsMVar job@Job{jobId} = void $ modifyMVar_ jobsMVar $ \jobMap -> do
293− pure $ Map.insertWith (++) jobId [(evt, job)] jobMap
294−
295− actualTest :: Seconds -> [JobPayload] -> Job.TableName -> MVar (Map.IntMap [(JobEvent, Job.Job)]) -> IO [Job]
296− actualTest shutdownAfter jobPayloads tname jobsMVar = do
297− jobs <- forConcurrently jobPayloads $ \pload ->
298− Pool.withResource appPool $ \conn ->
299− liftIO $ Job.createJob conn tname pload
300− let poller nextAction = case nextAction of
301− Left s -> pure $ Left s
302− Right remaining ->
303− if remaining == Seconds 0
304− then pure (Right $ Seconds 0)
305− else do delaySeconds testPollingInterval
306− print $ "------- Polling (remaining = " <> show (unSeconds remaining) <> " sec)------"
307− x <- withMVar jobsMVar $ \jobMap ->
308− if (Map.foldl (\m js -> m && (isJobTerminalState js)) True jobMap)
309− then pure (Right $ Seconds 0)
310− else if remaining < testPollingInterval
311− then pure (Left $ "Timeout. Job count=" <> show (length jobPayloads) <> " shutdownAfter=" <> show shutdownAfter)
312− else pure $ Right (remaining - testPollingInterval)
313− poller x
314−
315− poller (Right shutdownAfter) >>= \case
316− Left s -> pure jobs -- Prelude.error s
317− Right _ -> pure jobs
318−
319− isJobTerminalState js = case js of
320− [] -> False
321− (_, j):_ -> case (Job.jobStatus j) of
322− Job.Failed -> True
323− Job.Success -> True
324− _ -> False
711+-- testPollingInterval = 5
712+-- concurrencyFactor = 5
713+
714+-- noRaceCondition js = DL.foldl (&&) True $ DL.zipWith (\(x,_) (y,_) -> not $ x==JobStart && y==JobStart) js (tail js)
715+
716+-- jobPollingInterval = Seconds 2
717+
718+-- onJobEvent evt jobsMVar job@Job{jobId} = void $ modifyMVar_ jobsMVar $ \jobMap -> do
719+-- pure $ Map.insertWith (++) jobId [(evt, job)] jobMap
720+
721+-- actualTest :: Seconds -> [JobPayload] -> Job.TableName -> MVar (Map.IntMap [(JobEvent, Job.Job)]) -> IO [Job]
722+-- actualTest shutdownAfter jobPayloads tname jobsMVar = do
723+-- jobs <- forConcurrently jobPayloads $ \pload ->
724+-- Pool.withResource appPool $ \conn ->
725+-- liftIO $ Job.createJob conn tname pload
726+-- let poller nextAction = case nextAction of
727+-- Left s -> pure $ Left s
728+-- Right remaining ->
729+-- if remaining == Seconds 0
730+-- then pure (Right $ Seconds 0)
731+-- else do delaySeconds testPollingInterval
732+-- print $ "------- Polling (remaining = " <> show (unSeconds remaining) <> " sec)------"
733+-- x <- withMVar jobsMVar $ \jobMap ->
734+-- if (Map.foldl (\m js -> m && (isJobTerminalState js)) True jobMap)
735+-- then pure (Right $ Seconds 0)
736+-- else if remaining < testPollingInterval
737+-- then pure (Left $ "Timeout. Job count=" <> show (length jobPayloads) <> " shutdownAfter=" <> show shutdownAfter)
738+-- else pure $ Right (remaining - testPollingInterval)
739+-- poller x
740+
741+-- poller (Right shutdownAfter) >>= \case
742+-- Left s -> pure jobs -- Prelude.error s
743+-- Right _ -> pure jobs
744+
745+-- isJobTerminalState js = case js of
746+-- [] -> False
747+-- (_, j):_ -> case (Job.jobStatus j) of
748+-- Job.Failed -> True
749+-- Job.Success -> True
750+-- _ -> False
325751
326752
327753
328754payloadDelay :: Seconds -> JobPayload -> Seconds
329−payloadDelay jobPollingInterval pload = payloadDelay_ (Seconds 0) pload
755+payloadDelay jobPollingInterval = payloadDelay_ (Seconds 0)
330756 where
331− payloadDelay_ total p = case p of
332− PayloadAlwaysFail x -> total + x + jobPollingInterval
333− PayloadSucceed x -> total + x + jobPollingInterval
334− PayloadFail x ip -> payloadDelay_ (total + x + jobPollingInterval) ip
335−
336−
337−
338−-- TODO: test to ensure that errors in callback do not affect the running of jobs
757+ payloadDelay_ total p =
758+ let defaultDelay x = total + x + jobPollingInterval
759+ in case p of
760+ PayloadAlwaysFail x -> defaultDelay x
761+ PayloadSucceed x -> defaultDelay x
762+ PayloadFail x ip -> payloadDelay_ (defaultDelay x) ip
763+ PayloadThrowStringException _ -> defaultDelay 0
764+ PayloadThrowDivideByZero -> defaultDelay 0
339765
340766deriving instance Enum Time.Unit
341767deriving instance Enum Time.Direction
@@ -348,7 +774,7 @@ timeGen t d = do
348774 pure $ Time.timeSince t (fromInteger i) u d
349775
350776anyTimeGen :: MonadGen m => UTCTime -> m UTCTime
351−anyTimeGen t = (Gen.element (enumFrom Time.Ago)) >>= (timeGen t)
777+anyTimeGen t = Gen.element (enumFrom Time.Ago) >>= timeGen t
352778
353779futureTimeGen :: MonadGen m => UTCTime -> m UTCTime
354780futureTimeGen t = timeGen t Time.FromThat
@@ -400,8 +826,8 @@ genFilter t = do
400826 updatedBefore <- case updatedAfter of
401827 Nothing -> Gen.maybe (anyTimeGen t)
402828 Just x -> Gen.maybe (futureTimeGen x)
403− orderClause <- Gen.maybe ((,) <$> (Gen.element $ enumFrom Web.OrdCreatedAt) <*> (Gen.element $ enumFrom Web.Asc))
404− limitOffset <- Gen.maybe ((,) <$> (Gen.int (Range.constant 5 10)) <*> (Gen.int (Range.constant 0 30)))
829+ orderClause <- Gen.maybe ((,) <$> Gen.element (enumFrom Web.OrdCreatedAt) <*> Gen.element (enumFrom Web.Asc))
830+ limitOffset <- Gen.maybe ((,) <$> Gen.int (Range.constant 5 10) <*> Gen.int (Range.constant 0 30))
405831 runAfter <- Gen.maybe (futureTimeGen t)
406832 pure Web.blankFilter
407833 { filterStatuses = statuses
@@ -419,23 +845,23 @@ filterJobs :: Filter -> [Job] -> [Job]
419845filterJobs Web.Filter{filterStatuses, filterCreatedAfter, filterCreatedBefore, filterUpdatedAfter, filterUpdatedBefore, filterOrder, filterPage, filterRunAfter} js =
420846 applyLimitOffset $
421847 applyOrdering (fromMaybe (Web.OrdUpdatedAt, Web.Desc) filterOrder) $
422− (flip DL.filter) js $ \j -> (filterByStatus j) &&
423− (filterByCreatedAfter j) &&
424− (filterByCreatedBefore j) &&
425− (filterByUpdatedAfter j) &&
426− (filterByUpdatedBefore j) &&
427− (filterByRunAfter j)
848+ flip DL.filter js $ \j -> filterByStatus j &&
849+ filterByCreatedAfter j &&
850+ filterByCreatedBefore j &&
851+ filterByUpdatedAfter j &&
852+ filterByUpdatedBefore j &&
853+ filterByRunAfter j
428854 where
429− applyLimitOffset = maybe Prelude.id (\(l, o) -> (Prelude.take l). (Prelude.drop o)) filterPage
855+ applyLimitOffset = maybe Prelude.id (\(l, o) -> Prelude.take l . Prelude.drop o) filterPage
430856
431857 applyOrdering (fld, dir) lst =
432858 let comparer = resultOrder $ case fld of
433− Web.OrdCreatedAt -> (comparing jobCreatedAt)
434− Web.OrdUpdatedAt -> (comparing jobUpdatedAt)
435− Web.OrdLockedAt -> (comparing jobLockedAt)
436− Web.OrdStatus -> (comparing jobStatus)
437− Web.OrdJobType -> comparing Job.jobType
438− resultOrder fn = \x y -> case fn x y of
859+ Web.OrdCreatedAt -> comparing jobCreatedAt
860+ Web.OrdUpdatedAt -> comparing jobUpdatedAt
861+ Web.OrdLockedAt -> comparing jobLockedAt
862+ Web.OrdStatus -> comparing jobStatus
863+ Web.OrdJobType -> comparing Job.defaultJobType
864+ resultOrder fn x y = case fn x y of
439865 EQ -> compare (Down $ jobId x) (Down $ jobId y)
440866 LT -> case dir of
441867 Web.Asc -> LT
@@ -445,22 +871,26 @@ filterJobs Web.Filter{filterStatuses, filterCreatedAfter, filterCreatedBefore, f
445871 Web.Desc -> LT
446872 in sortBy comparer lst
447873
448− filterByStatus Job.Job{jobStatus} = if Prelude.null filterStatuses
449− then True
450− else jobStatus `elem` filterStatuses
874+ filterByStatus Job.Job{jobStatus} = Prelude.null filterStatuses
875+ || (jobStatus `elem` filterStatuses)
451876 filterByCreatedAfter Job.Job{jobCreatedAt} = maybe True (<= jobCreatedAt) filterCreatedAfter
452877 filterByCreatedBefore Job.Job{jobCreatedAt} = maybe True (> jobCreatedAt) filterCreatedBefore
453878 filterByUpdatedAfter Job.Job{jobUpdatedAt} = maybe True (<= jobUpdatedAt) filterUpdatedAfter
454879 filterByUpdatedBefore Job.Job{jobUpdatedAt} = maybe True (> jobUpdatedAt) filterUpdatedBefore
455880 filterByRunAfter Job.Job{jobRunAt} = maybe True (< jobRunAt) filterRunAfter
456881
457−defaultJobMonitor :: Job.TableName
458− -> Pool Connection
459− -> IO (Job.JobMonitor, IO ())
460−defaultJobMonitor tname pool = do
461− tcache <- newTimeCache simpleTimeFormat'
462− (tlogger, cleanup) <- newTimedFastLogger tcache LogNone
463− let flogger loc lsource llevel lstr = tlogger $ \t -> toLogStr t <> " | " <> defaultLogStr loc lsource llevel lstr
464− pure ( Job.defaultJobMonitor flogger tname pool
465− , cleanup
466− )
882+-- defaultJobMonitor :: Job.TableName
883+-- -> Pool Connection
884+-- -> (Job -> IO ())
885+-- -> IO (Job.JobMonitor, IO ())
886+-- defaultJobMonitor tname pool = do
887+-- tcache <- newTimeCache simpleTimeFormat'
888+-- (tlogger, cleanup) <- newTimedFastLogger tcache LogNone
889+-- let flogger loc lsource llevel lstr = tlogger $ \t -> toLogStr t <> " | " <> defaultLogStr loc lsource llevel lstr
890+-- Job.mkConfig _loggingFn tname pool Job.UnlimitedConcurrentJobs (pure ())
891+-- pure ( Job.defaultJobMonitor flogger tname pool
892+-- , cleanup
893+-- )
894+
895+calculateRetryDelay :: Int -> Seconds
896+calculateRetryDelay n = Seconds $ foldl' (\s x -> s + (2 ^ x)) 0 [1..n]