@@ -233,9 +233,17 @@ data JobEvent = JobStart
233233 deriving (Eq, Show)
234234
235235testEverything appPool jobPool = testProperty "test everything" $ property $ do
236+
236237 jobPayloads <- forAll $ Gen.list (Range.linear 300 1000) payloadGen
237238 jobsMVar <- liftIO $ newMVar (Map.empty :: Map.IntMap [(JobEvent, Job.Job)])
238239
240+ let maxDelay = sum $ map (payloadDelay jobPollingInterval) jobPayloads
241+ completeRun = ((unSeconds maxDelay) `div` concurrencyFactor) + (2 * (unSeconds testPollingInterval))
242+
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+ ]
246+
239247 test $ withRandomTable jobPool $ \tname -> do
240248 (defaults, cleanup) <- liftIO $ Test.defaultJobMonitor tname jobPool
241249 let jobMonitorSettings = defaults { Job.monitorJobRunner = jobRunner
@@ -251,15 +259,21 @@ testEverything appPool jobPool = testProperty "test everything" $ property $ do
251259 (jobs :: [Job]) <- withAsync
252260 (liftIO $ Job.runJobMonitor jobMonitorSettings)
253261 (const $ finally
254− (liftIO $ actualTest jobPayloads tname jobsMVar)
262+ (liftIO $ actualTest shutdownAfter jobPayloads tname jobsMVar)
255263 (liftIO cleanup))
256264
257265 jobAudit <- takeMVar jobsMVar
258266
259− -- All jobs should show up in the audit, which means they should have
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)
269+
270+ -- ALL jobs should show up in the audit, which means they should have
260271 -- been attempted /at least/ once
261272 (DL.sort $ map jobId jobs) === (DL.sort $ Map.keys jobAudit)
262273
274+ -- No job should be in a locked state
275+ 0 === lockedJobCount
276+
263277 -- No job should've been simultaneously picked-up by more than one
264278 -- worker
265279 True === (Map.foldl (\m js -> m && noRaceCondition js) True jobAudit)
@@ -268,6 +282,9 @@ testEverything appPool jobPool = testProperty "test everything" $ property $ do
268282
269283 where
270284
285+ testPollingInterval = 5
286+ concurrencyFactor = 5
287+
271288 noRaceCondition js = DL.foldl (&&) True $ DL.zipWith (\(x,_) (y,_) -> not $ x==JobStart && y==JobStart) js (tail js)
272289
273290 jobPollingInterval = Seconds 2
@@ -275,16 +292,12 @@ testEverything appPool jobPool = testProperty "test everything" $ property $ do
275292 onJobEvent evt jobsMVar job@Job{jobId} = void $ modifyMVar_ jobsMVar $ \jobMap -> do
276293 pure $ Map.insertWith (++) jobId [(evt, job)] jobMap
277294
278− actualTest :: [JobPayload] -> Job.TableName -> MVar (Map.IntMap [(JobEvent, Job.Job)]) -> IO [Job]
279− actualTest jobPayloads tname jobsMVar = do
295+ actualTest :: Seconds -> [JobPayload] -> Job.TableName -> MVar (Map.IntMap [(JobEvent, Job.Job)]) -> IO [Job]
296+ actualTest shutdownAfter jobPayloads tname jobsMVar = do
280297 jobs <- forConcurrently jobPayloads $ \pload ->
281298 Pool.withResource appPool $ \conn ->
282299 liftIO $ Job.createJob conn tname pload
283− let concurrencyFactor = 5
284− maxDelay = sum $ map (payloadDelay jobPollingInterval) jobPayloads
285− timeout = Seconds $ ((unSeconds maxDelay) `div` concurrencyFactor) + (2 * (unSeconds testPollingInterval))
286− testPollingInterval = 5
287− poller nextAction = case nextAction of
300+ let poller nextAction = case nextAction of
288301 Left s -> pure $ Left s
289302 Right remaining ->
290303 if remaining == Seconds 0
@@ -295,12 +308,12 @@ testEverything appPool jobPool = testProperty "test everything" $ property $ do
295308 if (Map.foldl (\m js -> m && (isJobTerminalState js)) True jobMap)
296309 then pure (Right $ Seconds 0)
297310 else if remaining < testPollingInterval
298− then pure (Left $ "Timeout. Job count=" <> show (length jobPayloads) <> " timeout=" <> show timeout <> " payload delay=" <> show maxDelay)
311+ then pure (Left $ "Timeout. Job count=" <> show (length jobPayloads) <> " shutdownAfter=" <> show shutdownAfter)
299312 else pure $ Right (remaining - testPollingInterval)
300313 poller x
301314
302− poller (Right timeout) >>= \case
303− Left s -> Prelude.error s
315+ poller (Right shutdownAfter) >>= \case
316+ Left s -> pure jobs -- Prelude.error s
304317 Right _ -> pure jobs
305318
306319 isJobTerminalState js = case js of