git.iko.soy/odd-jobs
on master help sign in
Expose runner env and pollRunJob for deterministic tests

This allows you to re-use the default hasRunner instance. Making it easier to for example run the jobpoller function, allowing you to get deterministic tests. The test suite resource tests are an example of more determininistic tests. Rather then using timeouts we can wait for async handles to complete. This tends to be faster and more reliable. Furthermroe we can make more fine grained assertions about time, because we can explicetly state how many jobs runs should happen in between. squash log: ---- This allows you to re-use the default has runner instance Expose poll run job Fix merge conflicts fix hlint use join undo join? join Add draft runSingleJobFromQueue I just want to make these deterniminstic before to ensure I didn't make any mistakes. rewrite how job queue uses resources in the tests add some comments on what these functions do wait for the job to start if it's running a job Fix delay acction issues Introduce explicit waiting where neccisary fix hlint fix warning Bumping this number has no effect, the test has still a 1 in 30 chance of failing. but I guess running it again will make it pass for my PR attempt at fixing this test fix signature of test modify test to wait for job to start with help of the callback this appears to be reliable

commitb873da36a3c6e3a5d283c7fcc60ee0c30794b104 authorJappie Klooster <jappieklooster@hotmail.com> dateMon 14 Nov 2022, 15:59 (4 years ago) parent0a61575
2 files changed+173−125
compare with master
unifiedsplit
src/OddJobs/Job.hs+66−55 test/Test.hs+107−70
src/OddJobs/Job.hs+66−55
@@ -38,6 +38,7 @@ module OddJobs.Job
3838 , AllJobTypes(..)
3939 , ResourceId(..)
4040 , FunctionName
41+ , RunnerEnv(..)
4142
4243 -- ** Structured logging
4344 --
@@ -52,6 +53,7 @@ module OddJobs.Job
5253 , jobEventListener
5354 , jobPoller
5455 , jobPollingSql
56+ , pollRunJob
5557 , JobRunner
5658 , HasJobRunner (..)
5759
@@ -374,6 +376,9 @@ updateJobHelper tname conn (newStatus, existingStates, mRunAt, jid) =
374376data TimeoutException = TimeoutException deriving (Eq, Show)
375377instance Exception TimeoutException
376378
379+-- | this spawns the job on a a seperate thread, and
380+-- also spawns a monitoring thread, whichever finishes
381+-- first kills the other.
377382runJobWithTimeout :: (HasJobRunner m)
378383 => Seconds
379384 -> Job
@@ -397,6 +402,7 @@ runJobWithTimeout timeoutSec job = do
397402 (atomicModifyIORef' threadsRef $ \threads -> (DL.delete a threads, ()))
398403
399404
405+-- | runs a job, blocks for as long as it's in progress
400406runJob :: (HasJobRunner m) => JobId -> m ()
401407runJob jid = do
402408 findJobById jid >>= \case
@@ -559,60 +565,67 @@ jobPollingIO pollerDbConn processName tname lockTimeout = do
559565jobPoller :: (HasJobRunner m) => m ()
560566jobPoller = do
561567 processName <- liftIO jobWorkerName
562− pool <- getDbPool
563− tname <- getTableName
564− lockTimeout <- getDefaultJobTimeout
565− traceM "jobPoller just before log statement"
566568 log LevelInfo $ LogText $ toS $ "Starting the job monitor via DB polling with processName=" <> processName
567−
568− let poll conn mResCfg = join $ mask_ $ do
569− log LevelDebug $ LogText $ toS $ "[" <> processName <> "] Polling the job queue.."
570− t <- liftIO getCurrentTime
571− r <- case mResCfg of
572− Nothing -> liftIO $
573− PGS.query conn jobPollingSql
574− ( tname
575− , Locked
576− , t
577− , processName
578− , tname
579− , t
580− , In [Queued, Retry]
581− , Locked
582− , addUTCTime (fromIntegral $ negate $ unSeconds lockTimeout) t)
583− Just ResourceCfg{..} -> liftIO $
584− PGS.query conn jobPollingWithResourceSql
585− ( tname
586− , Locked
587− , t
588− , processName
589− , tname
590− , t
591− , In [Queued, Retry]
592− , Locked
593− , addUTCTime (fromIntegral $ negate $ unSeconds lockTimeout) t
594− , resCfgCheckResourceFunction
595− )
596− case r of
597− -- When we don't have any jobs to run, we can relax a bit...
598− [] -> pure delayAction
599−
600− -- When we find a job to run, fork and try to find the next job without any delay...
601− [Only (jid :: JobId)] -> do
602− void $ async $ runJob jid
603− pure noDelayAction
604−
605− x -> error $ "WTF just happened? I was supposed to get only a single row, but got: " ++ show x
606−
607569 concurrencyControlFn <- getConcurrencyControlFn
608− withResource pool $ \pollerDbConn -> forever $ concurrencyControlFn pollerDbConn >>= \case
609− DontPoll -> do
610− log LevelWarn $ LogText "NOT polling the job queue due to concurrency control"
611− -- If we can't run any jobs ATM, relax and wait for resources to free up
612− delayAction
613− PollAny -> poll pollerDbConn Nothing
614− PollWithResources resCfg -> poll pollerDbConn (Just resCfg)
615−
570+ pool <- getDbPool
571+ forever $ do
572+ concurencyPolicy <- withResource pool concurrencyControlFn
573+ case concurencyPolicy of
574+ DontPoll -> log LevelWarn $ LogText "NOT polling the job queue due to concurrency control"
575+ PollAny -> void $ pollRunJob processName Nothing
576+ PollWithResources resCfg -> void $ pollRunJob processName (Just resCfg)
577+
578+-- | Polls a job and runs it, or executes a delay action if no job was found
579+--
580+-- returns an async to the task that was started, allowing you
581+-- to block until it's finished
582+pollRunJob :: (HasJobRunner m) => String -> Maybe ResourceCfg -> m (Maybe (Async ()))
583+pollRunJob processName mResCfg = do
584+ tname <- getTableName
585+ -- note it's better to use fine grained pool connection,
586+ -- since data.pool already has it's internal resource cache.
587+ -- this way the user can configure how long a connection
588+ -- needs to remain open.
589+ pool <- getDbPool
590+ lockTimeout <- getDefaultJobTimeout
591+ join $ withResource pool $ \pollerDbConn -> mask_ $ do
592+ log LevelDebug $ LogText $ toS $ "[" <> processName <> "] Polling the job queue.."
593+ t <- liftIO getCurrentTime
594+ r <- case mResCfg of
595+ Nothing -> liftIO $
596+ PGS.query pollerDbConn jobPollingSql
597+ ( tname
598+ , Locked
599+ , t
600+ , processName
601+ , tname
602+ , t
603+ , In [Queued, Retry]
604+ , Locked
605+ , addUTCTime (fromIntegral $ negate $ unSeconds lockTimeout) t)
606+ Just ResourceCfg{..} -> liftIO $
607+ PGS.query pollerDbConn jobPollingWithResourceSql
608+ ( tname
609+ , Locked
610+ , t
611+ , processName
612+ , tname
613+ , t
614+ , In [Queued, Retry]
615+ , Locked
616+ , addUTCTime (fromIntegral $ negate $ unSeconds lockTimeout) t
617+ , resCfgCheckResourceFunction
618+ )
619+ case r of
620+ -- When we don't have any jobs to run, we can relax a bit...
621+ [] -> pure (Nothing <$ delayAction)
622+
623+ -- When we find a job to run, fork and try to find the next job without any delay...
624+ [Only (jid :: JobId)] -> do
625+ x <- async $ runJob jid
626+ pure $ Just x <$ noDelayAction
627+
628+ x -> error $ "WTF just happened? I was supposed to get only a single row, but got: " ++ show x
616629 where
617630 delayAction = delaySeconds =<< getPollingInterval
618631 noDelayAction = pure ()
@@ -740,7 +753,7 @@ scheduleJob conn tname payload runAt = do
740753 case rs of
741754 [] -> Prelude.error . (<> "Not expecting a blank result set when creating a job. Query=") <$> queryFormatter
742755 [r] -> pure r
743− _ -> Prelude.error . (<> "Not expecting multiple rows when creating a single job. Query=") <$> queryFormatter
756+ _ -> Prelude.error . (<> "Not expecting multiple rows when creating a single job. Query=") <$> queryFormatter
744757
745758type ResourceList = [(ResourceId, Int)]
746759
@@ -786,7 +799,6 @@ scheduleJobWithResources conn tname ResourceCfg{..} payload resources runAt = do
786799
787800 pure job
788801
789−
790802-- getRunnerEnv :: (HasJobRunner m) => m RunnerEnv
791803-- getRunnerEnv = ask
792804
@@ -840,4 +852,3 @@ fetchAllJobRunners :: (MonadIO m)
840852 -> m [JobRunnerName]
841853fetchAllJobRunners UIConfig{uicfgTableName, uicfgDbPool} = liftIO $ withResource uicfgDbPool $ \conn -> do
842854 mapMaybe fromOnly <$> PGS.query conn "select distinct locked_by from ?" (Only uicfgTableName)
843−
test/Test.hs+107−70
@@ -47,6 +47,7 @@ import Data.Maybe (fromMaybe)
4747import qualified OddJobs.ConfigBuilder as Job
4848import UnliftIO
4949import Control.Exception (ArithException)
50+import Data.Bifunctor(first)
5051
5152$(Aeson.deriveJSON Aeson.defaultOptions ''Seconds)
5253
@@ -137,7 +138,7 @@ jobRunner Job{jobPayload, jobAttempts} = case fromJSON jobPayload of
137138 Success (j :: JobPayload) ->
138139 let recur pload idx = case pload of
139140 PayloadAlwaysFail delay -> delaySeconds delay >> error ("Forced error after " <> show delay <> " seconds")
140− PayloadSucceed delay -> void $ delaySeconds delay
141+ PayloadSucceed delay -> delaySeconds delay
141142 PayloadFail delay innerpload -> if idx<jobAttempts
142143 then recur innerpload (idx + 1)
143144 else delaySeconds delay >> error ("Forced error after " <> show delay <> " seconds. step=" <> show idx)
@@ -235,8 +236,35 @@ withRandomTable jobPool action = do
235236 (liftIO $ Pool.withResource jobPool (\conn -> liftIO $ Migrations.createJobTable conn tname) >> runInIO (action tname))
236237 (liftIO $ Pool.withResource jobPool $ \conn -> liftIO $ void $ PGS.execute conn "drop table if exists ?" (Only tname))
237238
238−withRandomResourceTables :: (MonadBaseControl
239− IO m, MonadUnliftIO m) => Int -> Pool Connection -> Job.TableName -> (Job.ResourceCfg -> m a) -> m a
239+runSingleJobFromQueueBlock :: Job.Config -> IO (Maybe (Either SomeException ()))
240+runSingleJobFromQueueBlock cfg =
241+ traverse waitCatch =<< runSingleJobFromQueue cfg
242+
243+-- | waits untill the job has started
244+runSingleJobFromQueue :: Job.Config -> IO (Maybe (Async ()))
245+runSingleJobFromQueue config' = do
246+ r <- liftIO $ newIORef mempty
247+ waitTillJobStart <- newEmptyMVar
248+ let config = config' {
249+ Job.cfgPollingInterval = 0
250+ , Job.cfgJobRunner = \job -> do
251+ putMVar waitTillJobStart ()
252+ jobRunner job
253+ }
254+ let monitorEnv = Job.RunnerEnv
255+ { Job.envConfig = config
256+ , Job.envJobThreadsRef = r
257+ }
258+ result <- runReaderT (Job.pollRunJob "runSingleJobFromQueue" $ readResourceConfig config) monitorEnv
259+ forM_ result $ const $ readMVar waitTillJobStart
260+ pure result
261+
262+readResourceConfig :: Job.Config -> Maybe Job.ResourceCfg
263+readResourceConfig cfg = case Job.cfgConcurrencyControl cfg of
264+ Job.ResourceLimits res -> Just res
265+ _ -> Nothing
266+
267+withRandomResourceTables :: (MonadBaseControl IO m, MonadUnliftIO m) => Int -> Pool Connection -> Job.TableName -> (Job.ResourceCfg -> m a) -> m a
240268withRandomResourceTables defaultLimit jobPool tname action = do
241269 resCfgResourceTable <- liftIO $ fromString . ("resources_" <>) <$> replicateM 10 (R.randomRIO ('a', 'z'))
242270 resCfgUsageTable <- liftIO $ fromString . ("usage_" <>) <$> replicateM 10 (R.randomRIO ('a', 'z'))
@@ -262,7 +290,13 @@ withNewJobMonitor jobPool actualTest = do
262290 withRandomTable jobPool $ \tname -> do
263291 withNamedJobMonitor tname jobPool Prelude.id (actualTest tname)
264292
265−withNamedJobMonitor tname jobPool cfgFn actualTest = do
293+withNamedJobMonitor :: Job.TableName -> Pool Connection -> (Job.Config -> Job.Config) -> (IORef [Job.LogEvent] -> IO a) -> IO a
294+withNamedJobMonitor tname jobPool cfgFn actualTest =
295+ withConfig tname jobPool cfgFn $ \logRef cfg -> do
296+ withAsync (Job.startJobRunner cfg) (const $ actualTest logRef)
297+
298+withConfig :: Job.TableName -> Pool Connection -> (Job.Config -> Job.Config) -> (IORef [Job.LogEvent] -> Job.Config -> IO a) -> IO a
299+withConfig tname jobPool cfgFn configConsumer = do
266300 logRef :: IORef [Job.LogEvent] <- newIORef []
267301 tcache <- newTimeCache simpleTimeFormat'
268302 withTimedFastLogger tcache LogNone $ \tlogger -> do
@@ -270,7 +304,7 @@ withNamedJobMonitor tname jobPool cfgFn actualTest = do
270304 tlogger $ \t -> toLogStr t <> " | " <> Job.defaultLogStr Job.defaultJobType logLevel logEvent
271305 atomicModifyIORef' logRef (\logs -> (logEvent:logs, ()))
272306 cfg = Job.mkConfig flogger tname jobPool Job.UnlimitedConcurrentJobs jobRunner (\cfg' -> cfgFn $ cfg'{Job.cfgDefaultMaxAttempts=testMaxAttempts})
273− withAsync (Job.startJobRunner cfg) (const $ actualTest logRef)
307+ configConsumer logRef cfg
274308
275309payloadGen :: MonadGen m => m JobPayload
276310payloadGen = Gen.recursive Gen.choice nonRecursive recursive
@@ -300,43 +334,22 @@ testEnsureShutdown appPool jobPool = testCase "ensure shutdown" $ do
300334 assertJobIdStatus conn tname logRef "Job is scheduled in future, should still be queueud" Job.Queued jobId
301335 pure (jobId, logRef)
302336
303−testGracefulShutdown appPool jobPool = testCase "ensure graceful shutdown" $ do
304− withRandomTable jobPool $ \tname -> do
305− setupPreconditions 3 tname >>= \case
306− Nothing -> fail "Unable to setup preconditions for this test, even after trying thrice"
307− Just (j1, j2, logRef) -> do
337+testGracefulShutdown :: Pool Connection -> Pool Connection -> TestTree
338+testGracefulShutdown appPool jobPool =
339+ testCase "ensure wait for all jobs is correct" $
340+ withRandomTable jobPool $ \tname -> do
341+ waitTillJobStart <- newEmptyMVar
308342 Pool.withResource appPool $ \conn -> do
343+ j1 <- Job.createJob conn tname (PayloadSucceed $ 4 * Job.defaultPollingInterval)
344+ time<- getCurrentTime
345+ j2 <- Job.scheduleJob conn tname (PayloadSucceed 0) (addUTCTime (fromIntegral $ unSeconds $ Job.defaultPollingInterval * 2) time)
346+ logRef <- withNamedJobMonitor tname jobPool (\cfg -> cfg { Job.cfgOnJobStart = \ _ -> do
347+ putMVar waitTillJobStart ()
348+ } ) $ \logRef -> do
349+ readMVar waitTillJobStart -- as soon as the job has started we quit the monitor:w
350+ pure logRef -- eg closing the withNamedJobMonitor closure shutsdwon the monitor
309351 assertJobIdStatus conn tname logRef "Expecting the first job to be completed successfully if graceful shutdown is implemented correctly" Job.Success (jobId j1)
310352 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)
311− where
312− -- When this block completes execution, the graceful shutdown should have happened.
313− setupPreconditions i tname =
314− if i == (0 :: Integer)
315− then pure Nothing
316− else do
317− withNamedJobMonitor tname jobPool Prelude.id $ \logRef -> do
318− Pool.withResource appPool $ \conn -> do
319− t <- getCurrentTime
320− j1 <- Job.createJob conn tname (PayloadSucceed $ 4 * Job.defaultPollingInterval)
321− j2 <- Job.scheduleJob conn tname (PayloadSucceed 0) (addUTCTime (fromIntegral $ unSeconds $ Job.defaultPollingInterval * 2) t)
322− let waitForPrecondition = do
323− k1 <- Job.findJobByIdIO conn tname (jobId j1)
324− k2 <- Job.findJobByIdIO conn tname (jobId j2)
325− case (jobStatus <$> k1, jobStatus <$> k2) of
326− (Nothing, _) -> do
327− traceM "Not expecting first job to be completed so quickly. Trying again."
328− setupPreconditions (i-1) tname
329− (_, Nothing) -> do
330− traceM "Not expecting second job to be completed so quickly. Trying again."
331− setupPreconditions (i-1) tname
332− (Just Job.Locked, Just Job.Queued) -> pure $ Just (j1, j2, logRef) -- precondition met
333− (Just Job.Success, _) -> setupPreconditions (i-1) tname
334− (_, Just Job.Success) -> setupPreconditions (i-1) tname
335− (Just Job.Queued, Just Job.Queued) -> do
336− delaySeconds 1
337− waitForPrecondition
338− x -> fail $ "Unexpeted job statuses while setting up the test: " <> show x
339− waitForPrecondition
340353
341354testJobScheduling appPool jobPool = testCase "job scheduling" $ do
342355 withNewJobMonitor jobPool $ \tname logRef -> do
@@ -447,8 +460,8 @@ testRetryBackoff appPool jobPool = testCase "retry backoff" $ do
447460 modifyRetryBackoff cfg
448461 = cfg { Job.cfgDefaultRetryBackoff = pure . backoff }
449462
450−testResourceLimitedScheduling appPool jobPool = testGroup "concurrency control by resources"
451− [ testCase "resources control how jobs run" $ setup $ \tname resCfg logRef conn -> do
463+testResourceLimitedScheduling _appPool jobPool = testGroup "concurrency control by resources"
464+ [ testCase "resources control how jobs run" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
452465 let firstResource = [(Job.ResourceId "first", 1)]
453466 secondResource = [(Job.ResourceId "second", 1)]
454467
@@ -456,22 +469,25 @@ testResourceLimitedScheduling appPool jobPool = testGroup "concurrency control b
456469 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
457470 Job{jobId = thirdJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) secondResource
458471
459− delaySeconds $ Job.defaultPollingInterval + Seconds 2
472+ _jone <- runSingleJobFromQueue cfg
473+ _jtwo <- runSingleJobFromQueue cfg
474+ _jthree <- runSingleJobFromQueue cfg
460475
461476 assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
462477 assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued secondJob
463478 assertJobIdStatus conn tname logRef "Job uses different resources - it should be running" Job.Locked thirdJob
464479
465− , testCase "resource-less jobs run normally" $ setup $ \tname resCfg logRef conn -> do
480+ , testCase "resource-less jobs run normally" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
466481 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) []
467482 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) []
468483
469− delaySeconds $ Job.defaultPollingInterval + Seconds 2
484+ _jone <- runSingleJobFromQueue cfg
485+ _jtwo <- runSingleJobFromQueue cfg
470486
471487 assertJobIdStatus conn tname logRef "First job uses no resources - it should be running" Job.Locked firstJob
472488 assertJobIdStatus conn tname logRef "Second job uses no resources - it should be running" Job.Locked secondJob
473489
474− , testCase "jobs can hold multiple resources" $ setup $ \tname resCfg logRef conn -> do
490+ , testCase "jobs can hold multiple resources" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
475491 let firstResource = [(Job.ResourceId "first", 1)]
476492 secondResource = [(Job.ResourceId "second", 1)]
477493
@@ -479,57 +495,70 @@ testResourceLimitedScheduling appPool jobPool = testGroup "concurrency control b
479495 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
480496 Job{jobId = thirdJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) secondResource
481497
482− delaySeconds $ Job.defaultPollingInterval + Seconds 2
498+ jone <- runSingleJobFromQueue cfg
499+ jtwo <- runSingleJobFromQueue cfg
500+ jthree <- runSingleJobFromQueue cfg
501+ assertEqual "first job claimed the work" (void jone) (Just ())
502+ assertEqual "second thread was blocked" (void jtwo) Nothing
503+ assertEqual "third thread was blocked " (void jthree) Nothing
483504
484505 assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
485506 assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued secondJob
486507 assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued thirdJob
487508
488− , testCase "resources are freed when jobs succeed" $ setup $ \tname resCfg logRef conn -> do
509+ , testCase "resources are freed when jobs succeed" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
489510 let firstResource = [(Job.ResourceId "first", 1)]
490511
491512 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
492513 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
493514
494− delaySeconds $ Job.defaultPollingInterval + Seconds 2
515+ jone <- runSingleJobFromQueue cfg
516+ _jtwo <- runSingleJobFromQueue cfg
495517
496518 assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
497519 assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued secondJob
498520
499− delaySeconds $ (Job.defaultPollingInterval * 2) + Seconds 8
521+ forM_ jone wait
522+
523+ _three <- runSingleJobFromQueue cfg
500524
501525 assertJobIdStatus conn tname logRef "First job should have succeeded by now" Job.Success firstJob
502526 assertJobIdStatus conn tname logRef "Second job should be running after first job suceeded" Job.Locked secondJob
503527
504− , testCase "resources are freed when jobs fail" $ setup $ \tname resCfg logRef conn -> do
528+ , testCase "resources are freed when jobs fail" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
505529 let firstResource = [(Job.ResourceId "first", 1)]
506530
507531 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadAlwaysFail 10) firstResource
508532 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
509533
510− delaySeconds $ Job.defaultPollingInterval + Seconds 2
534+ jone <- runSingleJobFromQueue cfg
535+ _jtwo <- runSingleJobFromQueue cfg
511536
512537 assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
513538 assertJobIdStatus conn tname logRef "Job uses same resources as first - it shouldn't be running" Job.Queued secondJob
514539
515− delaySeconds $ (Job.defaultPollingInterval * 2) + Seconds 8
540+ jobRes <- forM jone waitCatch
541+ assertEqual "should've thrown an exception" (Just (Left ())) (first (const ()) <$> jobRes)
542+
543+ _three <- runSingleJobFromQueue cfg
516544
517545 assertJobIdStatus conn tname logRef "First job should have failed by now" Job.Failed firstJob
518546 assertJobIdStatus conn tname logRef "Second job should be running after first job failed" Job.Locked secondJob
519547
520− , testCase "resources allow for multiple usage" $ setup $ \tname resCfg logRef conn -> do
548+ , testCase "resources allow for multiple usage" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
521549 let firstResource = [(Job.ResourceId "first", 1)]
522550 resCfg' = resCfg { Job.resCfgDefaultLimit = 2 }
523551
524552 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg' (PayloadSucceed 10) firstResource
525553 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg' (PayloadSucceed 10) firstResource
526554
527− delaySeconds $ Job.defaultPollingInterval + Seconds 2
555+ _jone <- runSingleJobFromQueue cfg
556+ _jtwo <- runSingleJobFromQueue cfg
528557
529558 assertJobIdStatus conn tname logRef "First job uses some resources - it should be running" Job.Locked firstJob
530559 assertJobIdStatus conn tname logRef "Second job uses some resources - it should be running" Job.Locked secondJob
531560
532− , testCase "resource usage can differ by job" $ setup $ \tname resCfg logRef conn -> do
561+ , testCase "resource usage can differ by job" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
533562 let firstResource n = [(Job.ResourceId "first", n)]
534563 resCfg' = resCfg { Job.resCfgDefaultLimit = 5 }
535564
@@ -537,57 +566,65 @@ testResourceLimitedScheduling appPool jobPool = testGroup "concurrency control b
537566 Job{jobId = secondJob} <- Job.createJobWithResources conn tname resCfg' (PayloadSucceed 10) (firstResource 3)
538567 Job{jobId = thirdJob} <- Job.createJobWithResources conn tname resCfg' (PayloadSucceed 10) (firstResource 2)
539568
540− delaySeconds $ Job.defaultPollingInterval + Seconds 2
569+ jone <- runSingleJobFromQueue cfg
570+ _jtwo <- runSingleJobFromQueue cfg
571+ _jthree <- runSingleJobFromQueue cfg
541572
573+ -- TODO, so it's not locked but already succeeded, which is werid, but the rest is valid?
542574 assertJobIdStatus conn tname logRef "Job was created first - it should be running" Job.Locked firstJob
543575 assertJobIdStatus conn tname logRef "Job doesn't fit alongside first job - it shouldn't be running" Job.Queued secondJob
544576 assertJobIdStatus conn tname logRef "Job doesn't fit alongside first job - it shouldn't be running" Job.Queued thirdJob
545577
546− delaySeconds $ (Job.defaultPollingInterval * 2) + Seconds 8
578+ forM_ jone wait
579+
580+ _jfour <- runSingleJobFromQueue cfg
581+ _jfive <- runSingleJobFromQueue cfg
547582
548583 assertJobIdStatus conn tname logRef "First job should have succeeded by now" Job.Success firstJob
549584 assertJobIdStatus conn tname logRef "Second job should be running after first job suceeded" Job.Locked secondJob
550585 assertJobIdStatus conn tname logRef "Third job fits alongside second job - it should be running after first job suceeded" Job.Locked thirdJob
551586
552− , testCase "resource limits permanently block too-big jobs" $ setup $ \tname resCfg logRef conn -> do
587+ , testCase "resource limits permanently block too-big jobs" $ setup jobPool $ \tname resCfg logRef conn cfg -> do
553588 let firstResource = [(Job.ResourceId "first", 2)]
554589
555590 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
556591
557− delaySeconds $ Job.defaultPollingInterval + Seconds 2
592+ void $ runSingleJobFromQueueBlock cfg
558593
559594 assertJobIdStatus conn tname logRef "Job requires too many resources - shouldn't be running" Job.Queued firstJob
560595
561− , testCase "resources with 0 limit block jobs" $ setup' 0 $ \tname resCfg logRef conn -> do
596+ , testCase "resources with 0 limit block jobs" $ setup' jobPool 0 $ \tname resCfg logRef conn cfg -> do
562597 let firstResource = [(Job.ResourceId "first", 1)]
563598
564599 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
565600
566− delaySeconds $ Job.defaultPollingInterval + Seconds 2
601+ void $ runSingleJobFromQueueBlock cfg
567602
568603 assertJobIdStatus conn tname logRef "Job can't access 0-limited resources - shouldn't be running" Job.Queued firstJob
569604
570− , testCase "resources with negative limit block jobs" $ setup' (-5) $ \tname resCfg logRef conn -> do
605+ , testCase "resources with negative limit block jobs" $ setup' jobPool (-5) $ \tname resCfg logRef conn cfg -> do
571606 let firstResource = [(Job.ResourceId "first", 1)]
572607
573608 Job{jobId = firstJob} <- Job.createJobWithResources conn tname resCfg (PayloadSucceed 10) firstResource
574609
575− delaySeconds $ Job.defaultPollingInterval + Seconds 2
610+ void $ runSingleJobFromQueueBlock cfg
576611
577612 assertJobIdStatus conn tname logRef "Job can't access negative-limited resources - shouldn't be running" Job.Queued firstJob
578613 ]
579− where
580− setup = setup' 1
581614
582− setup' defaultLimit action =
583− withRandomTable jobPool $ \tname ->
584− withRandomResourceTables defaultLimit jobPool tname $ \resCfg ->
585− withNamedJobMonitor tname jobPool (cfgFn resCfg) $ \logRef -> do
586− Pool.withResource appPool $ \conn -> action tname resCfg logRef conn
615+setup jobPool = setup' jobPool 1
587616
617+setup' jobPool defaultLimit action =
618+ withRandomTable jobPool $ \tname ->
619+ withRandomResourceTables defaultLimit jobPool tname $ \resCfg ->
620+ withConfig tname jobPool (cfgFn resCfg) $ \logRef cfg -> do
621+ Pool.withResource jobPool $ \conn -> action tname resCfg logRef conn cfg
622+
623+ where
588624 cfgFn resCfg cfg = cfg
589625 { Job.cfgDefaultMaxAttempts = 1 -- Simplifies some tests where we have a failing job
590626 , Job.cfgConcurrencyControl = Job.ResourceLimits resCfg
627+ , Job.cfgDeleteSuccessfulJobs = False
591628 }
592629
593630data JobEvent = JobStart