odd-jobs 0.1.0 → 0.2.0
raw patch · 12 files changed
+1665/−1006 lines, 12 filesdep +generic-derivingdep +servant-static-thdep +wainew-component:exe:odd-jobs-cli-example
Dependencies added: generic-deriving, servant-static-th, wai
Files
- README.md +5/−28
- dev/DevelMain.hs +33/−32
- examples/OddJobsCliExample.lhs +162/−0
- odd-jobs.cabal +67/−3
- src/OddJobs/Cli.hs +121/−40
- src/OddJobs/ConfigBuilder.hs +294/−0
- src/OddJobs/Endpoints.hs +145/−389
- src/OddJobs/Job.hs +141/−395
- src/OddJobs/Links.hs +0/−14
- src/OddJobs/Types.hs +316/−0
- src/OddJobs/Web.hs +351/−27
- test/Test.hs +30/−78
README.md view
@@ -1,31 +1,8 @@ # Introduction -Please read the **detailed tutorial** available at [Odd Jobs Tutorial](https://www.haskelltutorials.com/odd-jobs). Broadly, here is the **most common** way to get started with this library:+- [Odd Jobs home page](https://www.haskelltutorials.com/odd-jobs) - contains a description of top-level features of this library+- [Getting started & implementation guide for Odd Jobs](https://www.haskelltutorials.com/odd-jobs/guide.html)+- [Haskell Job Queues: An Ultimate Guide](https://www.haskelltutorials.com/odd-jobs/haskell-job-queues-ultimate-guide.html) - A detailed writeup on why we built Odd Jobs and how it compares against other similar libraries.+- [Start reading Hackage documentation from `OddJobs.Job`](https://hackage.haskell.org/package/odd-jobs-0.2.0/docs/OddJobs-Job.html) (**Note:** Please ensure you're reading docs for the correct version of the library)+- Open an issue on [Odd Jobs Github repo](https://github.com/saurabhnanda/odd-jobs) if you need help, or want to collaborate. -- (For a working version of the steps described below, please check the [simple-example](https://github.com/saurabhnanda/odd-jobs/tree/master/simple-example) directory in the project repo).-- Create a DB table to hold your jobs using the `OddJobs.Migrations.createJobTable` function.-- Create a sum-type to represent your job payloads, for example:- ```- data MyJob = SendWelcomeEmail Int- | SendPasswordResetEmail Text- | SetupSampleData Int- deriving (Eq, Show, Generic, ToJSON, FromJSON)- ```-- Ensure that you use a "tagged" JSON encoding for this data-type for best results (this is the default behaviour in Aeson). For example:- ```- Aeson.encode (SendWelcomeEmail 10) == "{\"tag\":\"SendWelcomeEmail\", \"contents\":10}"- ```- -- Create you "core" job-runner function:- ```- myJobRunner :: Job -> IO ()- myJobRunner Job{jobPayload} = - case jobPayload of- SendWelcomeEmail userId -> sendWelcomeEmail userId- SendPasswordResetEmail tkn -> sendPasswordResetEmail tkn- SetupSampleData userId -> sendPasswordResetEmail userId- ```- -- Use `OddJobs.Job.createJob` and `OddJobs.Job.scheduleJob` within your app whenever you want to enqueue a job.-- Use `OddJobs.Cli` to wrap all of this together into a nice CLI that can fork itself as a background daemon.-- Hook this up to your [deployment scripts](https://www.haskelltutorials.com/odd-jobs/deployment.html) and you're good to go!
dev/DevelMain.hs view
@@ -30,47 +30,48 @@ import Network.Wai.Handler.Warp (defaultSettings, runSettings, setPort) -import OddJobs.Endpoints (startApp, stopApp)+-- import OddJobs.Endpoints (startApp, stopApp) -- import qualified ElmCodeGen -- | Start or restart the server. -- newStore is from foreign-store. -- A Store holds onto some data across ghci reloads update :: IO ()-update = do- mtidStore <- lookupStore tidStoreNum- case mtidStore of- -- no server running- Nothing -> do- done <- storeAction doneStore newEmptyMVar- tid <- start done- _ <- storeAction (Store tidStoreNum) (newIORef tid)- return ()- -- server is already running- Just tidStore -> restartAppInNewThread tidStore- where- doneStore :: Store (MVar ())- doneStore = Store 0+update = undefined+-- update = do+-- mtidStore <- lookupStore tidStoreNum+-- case mtidStore of+-- -- no server running+-- Nothing -> do+-- done <- storeAction doneStore newEmptyMVar+-- tid <- start done+-- _ <- storeAction (Store tidStoreNum) (newIORef tid)+-- return ()+-- -- server is already running+-- Just tidStore -> restartAppInNewThread tidStore+-- where+-- doneStore :: Store (MVar ())+-- doneStore = Store 0 - -- shut the server down with killThread and wait for the done signal- restartAppInNewThread :: Store (IORef ThreadId) -> IO ()- restartAppInNewThread tidStore = modifyStoredIORef tidStore $ \tid -> do- killThread tid- withStore doneStore takeMVar- readStore doneStore >>= start+-- -- shut the server down with killThread and wait for the done signal+-- restartAppInNewThread :: Store (IORef ThreadId) -> IO ()+-- restartAppInNewThread tidStore = modifyStoredIORef tidStore $ \tid -> do+-- killThread tid+-- withStore doneStore takeMVar+-- readStore doneStore >>= start - -- | Start the server in a separate thread.- start :: MVar () -- ^ Written to when the thread is killed.- -> IO ThreadId- start done = do- -- (port, config, app) <- initialize- -- ElmCodeGen.updatea- forkIO (finally startApp- -- Note that this implies concurrency- -- between shutdownApp and the next app that is starting.- -- Normally this should be fine- (putMVar done () >> stopApp))+-- -- | Start the server in a separate thread.+-- start :: MVar () -- ^ Written to when the thread is killed.+-- -> IO ThreadId+-- start done = do+-- -- (port, config, app) <- initialize+-- -- ElmCodeGen.updatea+-- forkIO (finally startApp+-- -- Note that this implies concurrency+-- -- between shutdownApp and the next app that is starting.+-- -- Normally this should be fine+-- (putMVar done () >> stopApp)) -- | kill the server shutdown :: IO ()
+ examples/OddJobsCliExample.lhs view
@@ -0,0 +1,162 @@+=== 1. Create a table to store jobs++In this example, our jobs table will be called `jobs_test`++<div class="lhs-code">+```+ghci> import Datasbe.PostgreSQL.Simple (connectPostgreSQL)+ghci> import OddJobs.Migrations+ghci> conn <- connectPostgreSQL "dbname=jobs_test user=jobs_test password=jobs_test host=localhost"+ghci> createJobTable conn "jobs_test"+```+</div>++=== 2. Create a module for your job-runner++Ideally, this module should be compiled into a separate executable and should depend on your application's library module. Please refer to TODO for an example of how to setup your `package.yaml` (or cabal) file.++| If you do not wish to deploy odd-jobs as an independent executable, you may embed it within your main application's executable as well. This is described in TODO.++\begin{code}+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE NamedFieldPuns #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE LambdaCase #-}++module OddJobsCliExample where++import OddJobs.Job (Job(..), ConcurrencyControl(..), Config(..), throwParsePayload)+import OddJobs.ConfigBuilder (mkConfig, withConnectionPool, defaultTimedLogger, defaultLogStr, defaultJobType)+import OddJobs.Cli (defaultMain)++-- Note: It is not necessary to use fast-logger. You can use any logging library+-- that can give you a logging function in the IO monad.+import System.Log.FastLogger(withTimedFastLogger, LogType'(..), defaultBufSize)+import System.Log.FastLogger.Date (newTimeCache, simpleTimeFormat)++import Data.Text (Text)+import Data.Aeson as Aeson+import GHC.Generics++-- This example is using these functions to introduce an artificial delay of a+-- few seconds in one of the jobs. Otherwise it is not really needed.+import OddJobs.Types (delaySeconds, Seconds(..))+\end{code} ++=== 3. Set-up a Haskell type to represent your job-payload++- Ideally, this data-type should be defined _inside_ your application's code and the module containing this type-definition should be part of the `exposed-modules` stanza. Again, please look at TODO for an example of how to setup your `package.yaml` (or cabal) file.+- To work with all the default settings provided by 'OddJobs.ConfigBuilder' this data-type should have a **"tagged" JSON serialisation,** i.e.:++ ```json+ {"tag": "SendWelcomEmail", "contents": 10}+ ```++ In case your JSON payload does not conform to this structure, please look at TODO.++- In this example, we are _blindly_ deriving `ToJSON` and `FromJSON` instances because the default behaviour of Aeson is to generate a tagged JSON as-per the example given above.++\begin{code}+data MyJob+ = SendWelcomeEmail Int+ | SendPasswordResetEmail Text+ | SetupSampleData Int+ deriving (Eq, Show, Generic, ToJSON, FromJSON)+\end{code}++=== 4. Write the core job-runner function++In this example, the core job-runner function is in the `IO` monad. In all probability, you application's code will be in a custom monad, and not IO. Pleae refer to TODO, on how to work with custom monads.++\begin{code}+myJobRunner :: Job -> IO ()+myJobRunner job = do+ (throwParsePayload job) >>= \case+ SendWelcomeEmail userId -> do+ putStrLn $ "This should call the function that actually sends the welcome email. " <>+ "\nWe are purposely waiting 60 seconds before completing this job so that graceful shutdown can be demonstrated."+ delaySeconds (Seconds 60)+ putStrLn "60 second wait is now over..."+ SendPasswordResetEmail tkn ->+ putStrLn "This should call the function that actually sends the password-reset email"+ SetupSampleData userId -> do+ Prelude.error "User onboarding is incomplete"+ putStrLn "This should call the function that actually sets up sample data in a newly registered user's account"+\end{code}++=== 5. Write the main function using `OddJobs.Cli`++\begin{code}+main :: IO ()+main = do+ defaultMain startJobMonitor+ where+ -- A callback-within-callback function. If the commands-line args contain a+ -- `start` command, this function will be called. Once this function has+ -- constructed the 'Config' (which requires setting up a logging function,+ -- and a DB pool) it needs to execute the `callback` function that is passed+ -- to it.+ startJobMonitor callback =++ -- a utility function provided by `OddJobs.ConfigBuilder` which ensures+ -- that the DB pool is gracefully destroyed upon shutdown.+ withConnectionPool (Left "dbname=jobs_test user=jobs_test password=jobs_test host=localhost")$ \dbPool -> do++ -- Boilerplate code to setup a TimedFastLogger (from the fast-logger library)+ tcache <- newTimeCache simpleTimeFormat+ withTimedFastLogger tcache (LogFileNoRotate "oddjobs.log" defaultBufSize) $ \logger -> do++ -- Using the default string-based logging provided by+ -- `OddJobs.ConfigBuilder`. If you want to actually use+ -- structured-logging you'll need to define your own logging function.+ let jobLogger = defaultTimedLogger logger (defaultLogStr defaultJobType)+ cfg = mkConfig jobLogger "jobs" dbPool (MaxConcurrentJobs 50) myJobRunner Prelude.id++ -- Finally, executing the callback function that was passed to me...+ callback cfg+\end{code}++=== 6. Compile and start the Odd Jobs runner++<div class="lhs-code">+```+$ stack install <your-package-name>:exe:odd-jobs-cli+$ odd-jobs-cli start --daemonize --web-ui-basic-auth=oddjobs --web-ui-basic-password=awesome+```+</div>+++=== 7. Enqueue some jobs from within your application's code++<div class="lhs-code">+```+ghci> import OddJobs.Job (createJob)+ghci> import Database.PostgreSQL.Simple+ghci> conn <- connectPostgreSQL "dbname=jobs_test user=jobs_test password=jobs_test host=localhost"+ghci> createJob conn $ SendWelcomeEmail 10+ghci> createJob conn $ SetupSampleData 10+```+</div>++=== 8. Check-out the awesome web UI++Visit [http://localhost:7777](http://localhost:7777) (`username=oddjobs` / `password=awesome` as configured earlier).++=== 9. Check-out the log file to see what Odd Jobs is doing++<div class="lhs-code">+```+$ tail -f oddjobs.log+```+</div>++=== 10. Finally, shutdown Odd Jobs _gracefully_++Please read [graceful shutdown](#graceful-shutdown) to know more.++<div class="lhs-code">+```+$ odd-jobs-cli stop --timeout 65+```+</div>
odd-jobs.cabal view
@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: 4ca0abdf0789c7f27fe483c4da9d3dca5a1c1d05d10eb630e77c370297e77368+-- hash: 55fcabc52c82895fc22a613b9b0536420144ff36e138ed38153b02b1babde122 name: odd-jobs-version: 0.1.0+version: 0.2.0 synopsis: A full-featured PostgreSQL-backed job queue (with an admin UI) description: - Background jobs library for Haskell. - Extracted from production code at [Vacation Labs](https://www.vacationlabs.com).@@ -38,8 +38,8 @@ OddJobs.Endpoints OddJobs.Cli OddJobs.Types+ OddJobs.ConfigBuilder other-modules:- OddJobs.Links UI Paths_odd_jobs hs-source-dirs:@@ -56,6 +56,7 @@ , fast-logger , filepath , friendly-time+ , generic-deriving , hostname , lucid , monad-control@@ -68,6 +69,7 @@ , servant , servant-lucid , servant-server+ , servant-static-th , string-conv , text , text-conversions@@ -77,15 +79,25 @@ , unliftio , unliftio-core , unordered-containers+ , wai , warp default-language: Haskell2010 executable devel main-is: DevelMain.hs other-modules:+ OddJobs.Cli+ OddJobs.ConfigBuilder+ OddJobs.Endpoints+ OddJobs.Job+ OddJobs.Migrations+ OddJobs.Types+ OddJobs.Web+ UI Paths_odd_jobs hs-source-dirs: dev+ src default-extensions: NamedFieldPuns LambdaCase TemplateHaskell ScopedTypeVariables GeneralizedNewtypeDeriving QuasiQuotes OverloadedStrings ghc-options: -Wall -fno-warn-orphans -fno-warn-unused-imports -fno-warn-dodgy-exports -Werror=missing-fields -threaded -with-rtsopts=-N -main-is DevelMain build-depends:@@ -99,6 +111,7 @@ , filepath , foreign-store , friendly-time+ , generic-deriving , hostname , lucid , monad-control@@ -112,6 +125,7 @@ , servant , servant-lucid , servant-server+ , servant-static-th , string-conv , text , text-conversions@@ -121,9 +135,56 @@ , unliftio , unliftio-core , unordered-containers+ , wai , warp default-language: Haskell2010 +executable odd-jobs-cli-example+ main-is: OddJobsCliExample.lhs+ other-modules:+ Paths_odd_jobs+ hs-source-dirs:+ examples+ default-extensions: NamedFieldPuns LambdaCase TemplateHaskell ScopedTypeVariables GeneralizedNewtypeDeriving QuasiQuotes OverloadedStrings+ ghc-options: -Wall -fno-warn-orphans -fno-warn-unused-imports -fno-warn-dodgy-exports -Werror=missing-fields -threaded -with-rtsopts=-N -main-is OddJobsCliExample+ build-depends:+ aeson+ , base >=4.7 && <5+ , bytestring+ , direct-daemonize+ , directory+ , either+ , fast-logger+ , filepath+ , friendly-time+ , generic-deriving+ , hostname+ , lucid+ , monad-control+ , monad-logger+ , mtl+ , odd-jobs+ , optparse-applicative+ , postgresql-simple+ , resource-pool+ , safe+ , servant+ , servant-lucid+ , servant-server+ , servant-static-th+ , string-conv+ , text+ , text-conversions+ , time+ , timing-convenience+ , unix+ , unliftio+ , unliftio-core+ , unordered-containers+ , wai+ , warp+ default-language: Haskell2010+ test-suite jobrunner type: exitcode-stdio-1.0 main-is: Test.hs@@ -147,6 +208,7 @@ , fast-logger , filepath , friendly-time+ , generic-deriving , hedgehog , hostname , lifted-async@@ -165,6 +227,7 @@ , servant , servant-lucid , servant-server+ , servant-static-th , string-conv , tasty , tasty-discover@@ -178,5 +241,6 @@ , unliftio , unliftio-core , unordered-containers+ , wai , warp default-language: Haskell2010
src/OddJobs/Cli.hs view
@@ -1,9 +1,12 @@ {-# LANGUAGE RecordWildCards #-}+{-# LANGUAGE PartialTypeSignatures #-}+{-# LANGUAGE TypeOperators #-}+{-# LANGUAGE DataKinds #-} module OddJobs.Cli where import Options.Applicative as Opts import Data.Text-import OddJobs.Job (startJobRunner, Config(..), defaultLockTimeout)+import OddJobs.Job (startJobRunner, Config(..)) import System.Daemonize (DaemonOptions(..), daemonize) import System.FilePath (FilePath) import System.Posix.Process (getProcessID)@@ -13,7 +16,13 @@ import OddJobs.Types (Seconds(..), delaySeconds) import qualified System.Posix.Signals as Sig import qualified UnliftIO.Async as Async-+import qualified OddJobs.Endpoints as UI+import Servant.Server as Servant+import Servant.API+import Data.Proxy+import Data.Text.Encoding (decodeUtf8)+import Network.Wai.Handler.Warp as Warp+import Debug.Trace -- * Introduction --@@ -31,11 +40,15 @@ -- It is __highly recommended__ that you read the following links before putting -- odd-jobs into production. ----- * A [simple-example](https://github.com/saurabhnanda/odd-jobs/blob/master/simple-example/src/Main.hs)--- of how to use the 'defaultMain' function, that should make the callback-within-callback--- more understandable.+-- * The [getting started+-- guide](https://www.haskelltutorials.com/odd-jobs/guide.html#getting-started)+-- which will walk you through a bare-bones example of using the+-- 'defaultMain' function. It should make the callback-within-callback+-- clearer. ----- * TODO: link-off to \"Deployment\" in the tutorial+-- * The [section on+-- deployment](https://www.haskelltutorials.com/odd-jobs/guide.html#deployment)+-- in the guide. -- * Default behaviour --@@ -64,15 +77,16 @@ All of these resource allocations need to be bracketed so that when the job-runner exits, they may be cleaned-up gracefully. -Please take a look at @simple-example@ for how to use this function in-practice. (TODO: link-off to the example).+Please take a look at the [getting started+guide](https://www.haskelltutorials.com/odd-jobs/guide.html#getting-started) for+an example of how to use this function. -} defaultMain :: ((Config -> IO ()) -> IO ()) -- ^ A callback function that will be executed once the dameon has -- forked into the background. -> IO () defaultMain startFn = do- Args{argsCommand} <- customExecParser defaultCliParserPrefs (defaultCliInfo defaultLockTimeout)+ Args{argsCommand} <- customExecParser defaultCliParserPrefs defaultCliInfo case argsCommand of Start cmdArgs -> do defaultStartCommand cmdArgs startFn@@ -95,15 +109,16 @@ -- ^ the same callback-within-callback function described in -- 'defaultMain' -> IO ()-defaultStartCommand StartArgs{..} startFn = do+defaultStartCommand args@StartArgs{..} startFn = do progName <- getProgName case startDaemonize of False -> do- startFn startJobRunner+ startFn coreStartupFn True -> do (Dir.doesPathExist startPidFile) >>= \case True -> do- putStrLn $ "PID file already exists. Please check if " <> progName <> " is still running in the background." <>+ putStrLn $+ "PID file already exists. Please check if " <> progName <> " is still running in the background." <> " If not, you can safely delete this file and start " <> progName <> " again: " <> startPidFile Exit.exitWith (Exit.ExitFailure 1) False -> do@@ -111,8 +126,33 @@ pid <- getProcessID writeFile startPidFile (show pid) putStrLn $ "Started " <> progName <> " in background with PID=" <> show pid <> ". PID written to " <> startPidFile- startFn $ \jm -> startJobRunner jm{cfgPidFile = Just startPidFile}+ startFn $ \cfg -> coreStartupFn cfg{cfgPidFile = Just startPidFile}+ where+ coreStartupFn cfg = do+ Async.withAsync (defaultWebUi args cfg) $ \_ -> do+ startJobRunner cfg ++defaultWebUi :: StartArgs+ -> Config+ -> IO ()+defaultWebUi StartArgs{..} cfg@Config{..} = do+ env <- UI.mkEnv cfg ("/" <>)+ case startWebUiAuth of+ Nothing -> pure ()+ Just AuthNone ->+ let app = UI.server cfg env Prelude.id+ in Warp.run startWebUiPort $+ Servant.serve (Proxy :: Proxy UI.FinalAPI) app+ Just (AuthBasic u p) ->+ let api = Proxy :: Proxy (BasicAuth "OddJobs Admin UI" OddJobsUser :> UI.FinalAPI)+ ctx = defaultBasicAuth (u, p) :. EmptyContext+ -- Now the app will receive an extra argument for OddJobsUser,+ -- which we aren't really interested in.+ app _ = UI.server cfg env Prelude.id+ in Warp.run startWebUiPort $+ Servant.serveWithContext api ctx app+ {-| Used by 'defaultMain' if 'Stop' command is issued via the CLI. Sends a @SIGINT@ signal to the process indicated by 'shutPidFile'. Waits for a maximum of 'shutTimeout' seconds (controller by @--timeout@) for the daemon to shutdown@@ -166,11 +206,8 @@ -- | The top-level command-line parser-argParser :: Seconds- -- ^ the default value for 'shutTimeout'- -> Parser Args-argParser defaultTimeout = Args- <$> (commandParser defaultTimeout)+argParser :: Parser Args+argParser = Args <$> commandParser -- ** Top-level command parser @@ -182,11 +219,10 @@ deriving (Eq, Show) -- Parser for 'argsCommand'-commandParser :: Seconds -- ^ default value for 'shutTimeout'- -> Parser Command-commandParser defaultTimeout = hsubparser+commandParser :: Parser Command+commandParser = hsubparser ( command "start" (info startParser (progDesc "start the odd-jobs runner")) <>- command "stop" (info (stopParser defaultTimeout) (progDesc "stop the odd-jobs runner")) <>+ command "stop" (info stopParser (progDesc "stop the odd-jobs runner")) <> command "status" (info statusParser (progDesc "print status of all active jobs")) ) @@ -195,8 +231,13 @@ -- | @start@ command is parsed into this data-structure by 'startParser' data StartArgs = StartArgs {- -- | Switch to enable/disable the web UI (the web UI is still WIP)- startWebUiEnable :: !Bool+ -- | Switch to pick the authentication mechanism for web UI. __Note:__ We+ -- don't have a separate switch to enable\/disable the web UI. Picking an+ -- authentication mechanism by specifying the relevant options,+ -- automatically enables the web UI. Ref: 'webUiAuthParser'+ startWebUiAuth :: !(Maybe WebUiAuth)+ -- | Port on which the web UI will run.+ , startWebUiPort :: !Int -- | You'll need to pass the @--daemonize@ switch to fork the job-runner as -- a background daemon, else it will keep running as a foreground process. , startDaemonize :: !Bool@@ -206,22 +247,51 @@ startParser :: Parser Command startParser = fmap Start $ StartArgs- <$> switch ( long "web-ui-enable" <>- help "Please look at other web-ui-* options to configure the Web UI"- )+ <$> webUiAuthParser+ <*> option auto ( long "web-ui-port" <>+ metavar "PORT" <>+ value 7777 <>+ showDefault <>+ help "The port on which the Web UI listens. Please note, to actually enable the Web UI you need to pick one of the available auth schemes"+ ) <*> switch ( long "daemonize" <> help "Fork the job-runner as a background daemon. If omitted, the job-runner remains in the foreground." ) <*> pidFileParser +data WebUiAuth+ = AuthNone+ | AuthBasic !Text !Text+ deriving (Eq, Show) +-- | Pick one of the following auth mechanisms for the web UI:+--+-- * No auth - @--web-ui-no-auth@ __NOT RECOMMENDED__+-- * Basic auth - @--web-ui-basic-auth-user <USER>@ and+-- @--web-ui-basic-auth-password <PASS>@+webUiAuthParser :: Parser (Maybe WebUiAuth)+webUiAuthParser = basicAuthParser <|> noAuthParser <|> (pure Nothing)+ where+ basicAuthParser = fmap Just $ AuthBasic+ <$> strOption ( long "web-ui-basic-auth-user" <>+ metavar "USER" <>+ help "Username for basic auth"+ )+ <*> strOption ( long "web-ui-basic-auth-password" <>+ metavar "PASS" <>+ help "Password for basic auth"+ )+ noAuthParser = flag' (Just AuthNone)+ ( long "web-ui-no-auth" <>+ help "Start the web UI with any authentication. NOT RECOMMENDED."+ )+ -- ** Stop command -- | @stop@ command is parsed into this data-structure by 'stopParser'. Please -- note, that this command first sends a @SIGINT@ to the daemon and waits for--- 'shutTimeout' seconds (which defaults to 'defaultLockTimeout'). If the daemon--- doesn't shut down cleanly within that time, it sends a @SIGKILL@ to kill--- immediately.+-- 'shutTimeout' seconds. If the daemon doesn't shut down cleanly within that+-- time, it sends a @SIGKILL@ to kill immediately. data StopArgs = StopArgs { -- | After sending a @SIGINT@, how many seconds to wait before sending a -- @SIGKILL@@@ -230,13 +300,13 @@ , shutPidFile :: !FilePath } deriving (Eq, Show) -stopParser :: Seconds -> Parser Command-stopParser defaultTimeout = fmap Stop $ StopArgs+stopParser :: Parser Command+stopParser = fmap Stop $ StopArgs <$> option (Seconds <$> auto) ( long "timeout" <> metavar "TIMEOUT" <>- help "Maximum seconds to wait before force-killing the background daemon." <>- value defaultTimeout <>- showDefaultWith (show . unSeconds)+ help "Maximum seconds to wait before force-killing the background daemon."+ -- value defaultTimeout <>+ -- showDefaultWith (show . unSeconds) ) <*> pidFileParser @@ -265,11 +335,9 @@ showHelpOnError <> showHelpOnEmpty -defaultCliInfo :: Seconds- -- ^ default value for 'shutTimeout'- -> ParserInfo Args-defaultCliInfo defaultTimeout =- info ((argParser defaultTimeout) <**> helper) fullDesc+defaultCliInfo :: ParserInfo Args+defaultCliInfo =+ info (argParser <**> helper) fullDesc defaultDaemonOptions :: DaemonOptions defaultDaemonOptions = DaemonOptions@@ -281,3 +349,16 @@ } +-- ** Auth implementations for the default Web UI++-- *** Basic Auth++data OddJobsUser = OddJobsUser !Text !Text deriving (Eq, Show)++defaultBasicAuth :: (Text, Text) -> BasicAuthCheck OddJobsUser+defaultBasicAuth (user, pass) = BasicAuthCheck $ \b ->+ let u = decodeUtf8 (basicAuthUsername b)+ p = decodeUtf8 (basicAuthPassword b)+ in if u==user && p==pass+ then pure (Authorized $ OddJobsUser u p)+ else pure BadPassword
+ src/OddJobs/ConfigBuilder.hs view
@@ -0,0 +1,294 @@+{-# LANGUAGE RankNTypes #-}+{-# LANGUAGE FlexibleContexts #-}++module OddJobs.ConfigBuilder where++import OddJobs.Types+import Database.PostgreSQL.Simple as PGS+import Data.Pool+import Control.Monad.Logger (LogLevel(..), LogStr, toLogStr)+import Data.Text (Text)+import Lucid (Html, toHtml, class_, div_, span_, br_, button_, a_, href_, onclick_)+import Data.Maybe (fromMaybe)+import Data.List as DL+import Data.Aeson as Aeson hiding (Success)+import qualified Data.Text as T+import qualified Data.HashMap.Lazy as HM+import GHC.Generics+import Data.Proxy (Proxy(..))+import Generics.Deriving.ConNames+import Control.Monad+import Data.String.Conv+import GHC.Exts (toList)+import qualified Data.ByteString as BS+import UnliftIO (MonadUnliftIO, withRunInIO, bracket, liftIO)+import qualified System.Log.FastLogger as FLogger+++-- | This function gives you a 'Config' with a bunch of sensible defaults+-- already applied. It requires the bare minimum configuration parameters that+-- this library cannot assume on your behalf.+--+-- It makes a few __important assumptions__ about your 'jobPayload 'JSON, which+-- are documented in 'defaultJobType'.+mkConfig :: (LogLevel -> LogEvent -> IO ())+ -- ^ "Structured logging" function. Ref: 'cfgLogger'+ -> TableName+ -- ^ DB table which holds your jobs. Ref: 'cfgTableName'+ -> Pool Connection+ -- ^ DB connection-pool to be used by job-runner. Ref: 'cfgDbPool'+ -> ConcurrencyControl+ -- ^ Concurrency configuration. Ref: 'cfgConcurrencyControl'+ -> (Job -> IO ())+ -- ^ The actual "job runner" which contains your application code. Ref: 'cfgJobRunner'+ -> (Config -> Config)+ -- ^ A function that allows you to modify the \"interim config\". The+ -- \"interim config\" will cotain a bunch of in-built default config+ -- params, along with the config params that you\'ve just provided+ -- (i.e. logging function, table name, DB pool, etc). You can use this+ -- function to override values in the \"interim config\". If you do not+ -- wish to modify the \"interim config\" just pass 'Prelude.id' as an+ -- argument to this parameter. __Note:__ it is strongly recommended+ -- that you __do not__ modify the generated 'Config' outside of this+ -- function, unless you know what you're doing.+ -> Config+ -- ^ The final 'Config' that can be used to start various job-runners+mkConfig logger tname dbpool ccControl jrunner configOverridesFn =+ let cfg = configOverridesFn $ Config+ { cfgPollingInterval = defaultPollingInterval+ , cfgOnJobSuccess = (const $ pure ())+ , cfgOnJobFailed = []+ , cfgJobRunner = jrunner+ , cfgLogger = logger+ , cfgDbPool = dbpool+ , cfgOnJobStart = (const $ pure ())+ , cfgDefaultMaxAttempts = 10+ , cfgTableName = tname+ , cfgOnJobTimeout = (const $ pure ())+ , cfgConcurrencyControl = ccControl+ , cfgPidFile = Nothing+ , cfgJobType = defaultJobType+ , cfgDefaultJobTimeout = Seconds 600+ , cfgJobToHtml = defaultJobToHtml (cfgJobType cfg)+ , cfgAllJobTypes = (defaultDynamicJobTypes (cfgTableName cfg) (cfgJobTypeSql cfg))+ , cfgJobTypeSql = defaultJobTypeSql+ }+ in cfg++++-- | If you aren't interested in structured logging, you can use this function+-- to emit plain-text logs (or define your own).+defaultLogStr :: (Job -> Text)+ -> LogLevel+ -> LogEvent+ -> LogStr+defaultLogStr jobTypeFn logLevel logEvent =+ (toLogStr $ show logLevel) <> " | " <> str+ where+ jobToLogStr job@Job{jobId} =+ "JobId=" <> (toLogStr $ show jobId) <> " JobType=" <> (toLogStr $ jobTypeFn job)++ str = case logEvent of+ LogJobStart j ->+ "Started | " <> jobToLogStr j+ LogJobFailed j e fm t ->+ let tag = case fm of+ FailWithRetry -> "Failed (retry)"+ FailPermanent -> "Failed (permanent)"+ in tag <> " | " <> jobToLogStr j <> " | runtime=" <> (toLogStr $ show t) <> " | error=" <> (toLogStr $ show e)+ LogJobSuccess j t ->+ "Success | " <> (jobToLogStr j) <> " | runtime=" <> (toLogStr $ show t)+ LogJobTimeout j@Job{jobLockedAt, jobLockedBy} ->+ "Timeout | " <> jobToLogStr j <> " | lockedBy=" <> (toLogStr $ maybe "unknown" unJobRunnerName jobLockedBy) <>+ " lockedAt=" <> (toLogStr $ maybe "unknown" show jobLockedAt)+ LogPoll ->+ "Polling jobs table"+ LogWebUIRequest ->+ "WebUIRequest (TODO: Log the actual request)"+ LogText t ->+ toLogStr t++defaultJobToHtml :: (Job -> Text)+ -> [Job]+ -> IO [Html ()]+defaultJobToHtml jobType js =+ pure $ DL.map jobToHtml js+ where+ jobToHtml :: Job -> Html ()+ jobToHtml j = do+ div_ [ class_ "job" ] $ do+ div_ [ class_ "job-type" ] $ do+ toHtml $ jobType j+ div_ [ class_ "job-payload" ] $ do+ defaultPayloadToHtml $ defaultJobContent $ jobPayload j+ case jobLastError j of+ Nothing -> mempty+ Just e -> do+ div_ [ class_ "job-error collapsed" ] $ do+ a_ [ href_ "javascript: void(0);", onclick_ "toggleError(this)" ] $ do+ span_ [ class_ "badge badge-secondary error-expand" ] "+ Last error"+ span_ [ class_ "badge badge-secondary error-collapse d-none" ] "- Last error"+ " "+ defaultErrorToHtml e+++defaultErrorToHtml :: Value -> Html ()+defaultErrorToHtml e =+ case e of+ Aeson.String s -> handleLineBreaks s+ Aeson.Bool b -> toHtml $ show b+ Aeson.Number n -> toHtml $ show n+ Aeson.Null -> toHtml ("(null)" :: Text)+ Aeson.Object o -> toHtml $ show o -- TODO: handle this properly+ Aeson.Array a -> toHtml $ show a -- TODO: handle this properly+ where+ handleLineBreaks s = do+ forM_ (T.splitOn "\n" s) $ \x -> do+ toHtml x+ br_ []++defaultJobContent :: Value -> Value+defaultJobContent v = case v of+ Aeson.Object o -> case HM.lookup "contents" o of+ Nothing -> v+ Just c -> c+ _ -> v++defaultPayloadToHtml :: Value -> Html ()+defaultPayloadToHtml v = case v of+ Aeson.Object o -> do+ toHtml ("{ " :: Text)+ forM_ (HM.toList o) $ \(k, v2) -> do+ span_ [ class_ " key-value-pair " ] $ do+ span_ [ class_ "key" ] $ toHtml $ k <> ":"+ span_ [ class_ "value" ] $ defaultPayloadToHtml v2+ toHtml (" }" :: Text)+ Aeson.Array a -> do+ toHtml ("[" :: Text)+ forM_ (toList a) $ \x -> do+ defaultPayloadToHtml x+ toHtml (", " :: Text)+ toHtml ("]" :: Text)+ Aeson.String t -> toHtml t+ Aeson.Number n -> toHtml $ show n+ Aeson.Bool b -> toHtml $ show b+ Aeson.Null -> toHtml ("null" :: Text)++defaultJobTypeSql :: PGS.Query+defaultJobTypeSql = "payload->>'tag'"++defaultConstantJobTypes :: forall a . (Generic a, ConNames (Rep a))+ => Proxy a+ -> AllJobTypes+defaultConstantJobTypes _ =+ AJTFixed $ DL.map toS $ conNames (undefined :: a)++defaultDynamicJobTypes :: TableName+ -> PGS.Query+ -> AllJobTypes+defaultDynamicJobTypes tname jobTypeSql = AJTSql $ \conn -> do+ fmap (DL.map ((fromMaybe "(unknown)") . fromOnly)) $ PGS.query_ conn $ "select distinct(" <> jobTypeSql <> ") from " <> tname <> " order by 1 nulls last"++-- | This makes __two important assumptions__. First, this /assumes/ that jobs+-- in your app are represented by a sum-type. For example:+--+-- @+-- data MyJob = SendWelcomeEmail Int+-- | SendPasswordResetEmail Text+-- | SetupSampleData Int+-- @+--+-- Second, it /assumes/ that the JSON representatin of this sum-type is+-- "tagged". For example, the following...+--+-- > let pload = SendWelcomeEmail 10+--+-- ...when converted to JSON, would look like...+--+-- > {"tag":"SendWelcomeEmail", "contents":10}+--+-- It uses this assumption to extract the "job type" from a 'Data.Aeson.Value'+-- (which would be @SendWelcomeEmail@ in the example given above). This is used+-- in logging and the admin UI.+--+-- Even if tihs assumption is violated, the job-runner /should/ continue to+-- function. It's just that you won't get very useful log messages.+--+-- __Note:__ If your job payload does not conform to the structure described+-- above, please read the section on [customising the job payload's+-- structure](https://www.haskelltutorials.com/odd-jobs/guide.html#custom-payload-structure)+-- in the implementation guide.+defaultJobType :: Job -> Text+defaultJobType Job{jobPayload} =+ case jobPayload of+ Aeson.Object hm -> case HM.lookup "tag" hm of+ Just (Aeson.String t) -> t+ _ -> "unknown"+ _ -> "unknown"+++-- | As the name says. Ref: 'cfgPollingInterval'+defaultPollingInterval :: Seconds+defaultPollingInterval = Seconds 5++-- | Convenience function to create a DB connection-pool with some sensible+-- defaults. Please see the source-code of this function to understand what it's+-- doing.+withConnectionPool :: (MonadUnliftIO m)+ => Either BS.ByteString PGS.ConnectInfo+ -> (Pool PGS.Connection -> m a)+ -> m a+withConnectionPool connConfig action = withRunInIO $ \runInIO -> do+ bracket poolCreator destroyAllResources (runInIO . action)+ where+ poolCreator = liftIO $+ case connConfig of+ Left connString ->+ createPool (PGS.connectPostgreSQL connString) PGS.close 1 (fromIntegral $ 2 * (unSeconds defaultPollingInterval)) 8+ Right connInfo ->+ createPool (PGS.connect connInfo) PGS.close 1 (fromIntegral $ 2 * (unSeconds defaultPollingInterval)) 8++-- | A convenience function to help you define a timed-logger with some sensible+-- defaults.+defaultTimedLogger :: FLogger.TimedFastLogger+ -> (LogLevel -> LogEvent -> LogStr)+ -> LogLevel+ -> LogEvent+ -> IO ()+defaultTimedLogger logger logStrFn logLevel logEvent =+ if logLevel == LevelDebug+ then pure ()+ else logger $ \t -> (toLogStr t) <> " | " <>+ (logStrFn logLevel logEvent) <>+ "\n"+++defaultJsonLogEvent :: LogEvent -> Aeson.Value+defaultJsonLogEvent logEvent =+ case logEvent of+ LogJobStart job ->+ Aeson.object [ "tag" Aeson..= ("LogJobStart" :: Text)+ , "contents" Aeson..= (defaultJsonJob job) ]+ LogJobSuccess job runTime ->+ Aeson.object [ "tag" Aeson..= ("LogJobSuccess" :: Text)+ , "contents" Aeson..= (defaultJsonJob job, runTime) ]+ LogJobFailed job e fm runTime ->+ Aeson.object [ "tag" Aeson..= ("LogJobFailed" :: Text)+ , "contents" Aeson..= (defaultJsonJob job, show e, defaultJsonFailureMode fm, runTime) ]+ LogJobTimeout job ->+ Aeson.object [ "tag" Aeson..= ("LogJobTimeout" :: Text)+ , "contents" Aeson..= (defaultJsonJob job) ]+ LogPoll ->+ Aeson.object [ "tag" Aeson..= ("LogJobPoll" :: Text)]+ LogWebUIRequest ->+ Aeson.object [ "tag" Aeson..= ("LogWebUIRequest" :: Text)]+ LogText t ->+ Aeson.object [ "tag" Aeson..= ("LogText" :: Text)+ , "contents" Aeson..= t ]++defaultJsonJob :: Job -> Aeson.Value+defaultJsonJob job = genericToJSON Aeson.defaultOptions job++defaultJsonFailureMode :: FailureMode -> Aeson.Value+defaultJsonFailureMode fm = genericToJSON Aeson.defaultOptions fm
src/OddJobs/Endpoints.hs view
@@ -1,8 +1,12 @@-{-# LANGUAGE TypeOperators, DeriveGeneric, NamedFieldPuns, DataKinds, StandaloneDeriving, FlexibleContexts #-}+{-# LANGUAGE TypeOperators, DeriveGeneric, NamedFieldPuns, DataKinds, StandaloneDeriving, FlexibleContexts, RecordWildCards, RankNTypes #-} ++-- | TODO: Rename this to OddJobs.Servant+ module OddJobs.Endpoints where -import OddJobs.Web as Web+import OddJobs.Web as Web hiding (Routes(..))+import qualified OddJobs.Web as Web import OddJobs.Job as Job import OddJobs.Types import GHC.Generics@@ -25,20 +29,22 @@ import Data.String.Conv (toS) import Control.Monad.Except import Data.Time as Time-import Data.Time.Format.Human (humanReadableTime') import Data.Aeson as Aeson import qualified Data.HashMap.Strict as HM import GHC.Exts (toList)-import Data.Maybe (fromMaybe)-import Data.Text.Conversions (fromText, toText)+import Data.Maybe (fromMaybe, mapMaybe) import Control.Applicative ((<|>))-import Data.Time.Convenience (timeSince, Unit(..), Direction(..))-import qualified OddJobs.Links as Links+-- import qualified OddJobs.Links as Links import Data.List ((\\)) import qualified System.Log.FastLogger as FLogger import qualified System.Log.FastLogger.Date as FLogger import Control.Monad.Logger as MLogger import qualified Data.ByteString.Lazy as BSL+import qualified Data.List as DL+import UnliftIO.IORef+import Debug.Trace+import qualified OddJobs.ConfigBuilder as Builder+import Servant.Static.TH (createApiAndServerDecs) -- startApp :: IO () -- startApp = undefined@@ -46,410 +52,160 @@ -- stopApp :: IO () -- stopApp = undefined -tname :: TableName-tname = "jobs_aqgrqtaowi"+$(createApiAndServerDecs "StaticAssetRoutes" "staticAssetServer" "assets") -startApp :: IO ()-startApp = do- let connInfo = ConnectInfo- { connectHost = "localhost"- , connectPort = fromIntegral 5432- , connectUser = "jobs_test"- , connectPassword = "jobs_test"- , connectDatabase = "jobs_test"- }- dbPool <- createPool- (PGS.connect connInfo) -- cretea a new resource- (PGS.close) -- destroy resource- 1 -- stripes- (fromRational 10) -- number of seconds unused resources are kept around- 5 -- maximum open connections+data Routes route = Routes+ { rFilterResults :: route :- QueryParam "filters" Web.Filter :> Get '[HTML] (Html ())+ , rEnqueue :: route :- "enqueue" :> Capture "jobId" JobId :> Post '[HTML] NoContent+ , rRunNow :: route :- "run" :> Capture "jobId" JobId :> Post '[HTML] NoContent+ , rCancel :: route :- "cancel" :> Capture "jobId" JobId :> Post '[HTML] NoContent+ , rRefreshJobTypes :: route :- "refresh-job-types" :> Post '[HTML] NoContent+ , rRefreshJobRunners :: route :- "refresh-job-runners" :> Post '[HTML] NoContent+ } deriving (Generic) - tcache <- FLogger.newTimeCache FLogger.simpleTimeFormat'- (tlogger, cleanup) <- FLogger.newTimedFastLogger tcache (FLogger.LogStdout FLogger.defaultBufSize)- let flogger = Job.defaultTimedLogger tlogger (Job.defaultLogStr (Job.defaultJobToText Job.defaultJobType))- jm = Job.defaultConfig flogger tname dbPool Job.UnlimitedConcurrentJobs (const $ pure ()) - let nt :: ReaderT Job.Config IO a -> Servant.Handler a- nt action = (liftIO $ try $ runReaderT action jm) >>= \case- Left (e :: SomeException) -> Servant.Handler $ ExceptT $ pure $ Left $ err500 { errBody = toS $ show e }- Right a -> Servant.Handler $ ExceptT $ pure $ Right a- appProxy = (Proxy :: Proxy (ToServant Routes AsApi))-- finally- (run 8080 $ genericServe (server dbPool))- (cleanup >> (Pool.destroyAllResources dbPool))--stopApp :: IO ()-stopApp = pure ()+type FinalAPI =+ (ToServant Routes AsApi) :<|>+ "assets" :> StaticAssetRoutes -server :: Pool Connection- -> Routes AsServer-server dbPool = Routes- { rFilterResults = (\mFilter -> filterResults dbPool mFilter)- , rStaticAssets = serveDirectoryFileServer "assets"+data Env = Env+ { envRoutes :: Web.Routes+ , envJobTypesRef :: IORef [Text]+ , envJobRunnersRef :: IORef [JobRunnerName] } ---- withDbConnection :: HasJobMonitor m--- => (Connection -> m a)--- -> m a--- withDbConnection fn = getDbPool >>= \pool -> Pool.withResource pool fn--filterResults :: Pool Connection- -> Maybe Filter- -> Handler (Html ())-filterResults dbPool mFilter = do- let filters = fromMaybe mempty mFilter- (jobs, runningCount) <- liftIO $ Pool.withResource dbPool $ \conn -> (,)- <$> (filterJobs conn tname filters)- <*> (countJobs conn tname filters{ filterStatuses = [Job.Locked] })- t <- liftIO getCurrentTime- pure $ pageLayout $ do- searchBar t filters- resultsPanel t filters jobs runningCount-+mkEnv :: (MonadIO m) => Config -> (Text -> Text) -> m Env+mkEnv cfg@Config{..} linksFn = do+ allJobTypes <- fetchAllJobTypes cfg+ allJobRunners <- fetchAllJobRunners cfg+ envJobTypesRef <- newIORef allJobTypes+ envJobRunnersRef <- newIORef allJobRunners+ let envRoutes = routes linksFn -pageNav :: Html ()-pageNav = do- div_ $ nav_ [ class_ "navbar navbar-default navigation-clean" ] $ div_ [ class_ "container" ] $ do- div_ [ class_ "navbar-header" ] $ do- a_ [ class_ "navbar-brand navbar-link", href_ "#", style_ "padding: 0px;" ] $ img_ [ src_ "/assets/odd-jobs-color-logo.png", title_ "Odd Jobs Logo" ]- button_ [ class_ "navbar-toggle collapsed", data_ "toggle" "collapse", data_ "target" "#navcol-1" ] $ do- span_ [ class_ "sr-only" ] $ "Toggle navigation"- span_ [ class_ "icon-bar" ] $ ""- span_ [ class_ "icon-bar" ] $ ""- span_ [ class_ "icon-bar" ] $ ""- -- div_ [ class_ "collapse navbar-collapse", id_ "navcol-1" ] $ ul_ [ class_ "nav navbar-nav navbar-right" ] $ do- -- li_ [ class_ "active", role_ "presentation" ] $ a_ [ href_ "#" ] $ "First Item"- -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Second Item"- -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Third Item"- -- li_ [ class_ "dropdown" ] $ do- -- a_ [ class_ "dropdown-toggle", data_ "toggle" "dropdown", ariaExpanded_ "false", href_ "#" ] $ do- -- "Dropdown"- -- span_ [ class_ "caret" ] $ ""- -- ul_ [ class_ "dropdown-menu", role_ "menu" ] $ do- -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "First Item"- -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Second Item"- -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Third Item"+ pure Env{..}+ -- TODO: remove hard-coded port+ -- run 8080 $ genericServe (server cfg Env{..}) -pageLayout :: Html () -> Html ()-pageLayout inner = do- doctype_- html_ $ do- head_ $ do- meta_ [ charset_ "utf-8" ]- meta_ [ name_ "viewport", content_ "width=device-width, initial-scale=1.0" ]- title_ "haskell-pg-queue"- link_ [ rel_ "stylesheet", href_ "assets/bootstrap/css/bootstrap.min.css" ]- link_ [ rel_ "stylesheet", href_ "https://fonts.googleapis.com/css?family=Lato:100i,300,300i,400,700,900" ]- link_ [ rel_ "stylesheet", href_ "assets/css/logo-slider.css" ]- link_ [ rel_ "stylesheet", href_ "assets/css/Navigation-Clean1.css" ]- link_ [ rel_ "stylesheet", href_ "assets/css/styles.css" ]- body_ $ do- pageNav- div_ $ div_ [ class_ "container", style_ "/*background-color:#f2f2f2;*/" ] $ div_ [ class_ "row" ] $ div_ [ class_ "col-md-12" ] $ do- inner- script_ [ src_ "assets/js/jquery.min.js" ] $ ("" :: Text)- script_ [ src_ "assets/bootstrap/js/bootstrap.min.js" ] $ ("" :: Text)- script_ [ src_ "https://cdnjs.cloudflare.com/ajax/libs/slick-carousel/1.6.0/slick.js" ] $ ("" :: Text)- script_ [ src_ "assets/js/logo-slider.js" ] $ ("" :: Text)+ -- let nt :: ReaderT Job.Config IO a -> Servant.Handler a+ -- nt action = (liftIO $ try $ runReaderT action jm) >>= \case+ -- Left (e :: SomeException) -> Servant.Handler $ ExceptT $ pure $ Left $ err500 { errBody = toS $ show e }+ -- Right a -> Servant.Handler $ ExceptT $ pure $ Right a+ -- appProxy = (Proxy :: Proxy (ToServant Routes AsApi)) -searchBar :: UTCTime -> Filter -> Html ()-searchBar t filter@Filter{filterStatuses, filterCreatedAfter, filterCreatedBefore, filterUpdatedAfter, filterUpdatedBefore, filterJobTypes, filterRunAfter} = do- form_ [ style_ "padding-top: 2em;" ] $ do- div_ [ class_ "form-group" ] $ do- div_ [ class_ "search-container" ] $ do- ul_ [ class_ "list-inline search-bar" ] $ do- forM_ filterStatuses $ \s -> renderFilter "Status" (toText s) (Links.rFilterResults $ Just filter{filterStatuses = filterStatuses \\ [s]})- maybe mempty (\x -> renderFilter "Created after" (showText x) (Links.rFilterResults $ Just filter{filterCreatedAfter = Nothing})) filterCreatedAfter- maybe mempty (\x -> renderFilter "Created before" (showText x) (Links.rFilterResults $ Just filter{filterCreatedBefore = Nothing})) filterCreatedBefore- maybe mempty (\x -> renderFilter "Updated after" (showText x) (Links.rFilterResults $ Just filter{filterUpdatedAfter = Nothing})) filterUpdatedAfter- maybe mempty (\x -> renderFilter "Updated before" (showText x) (Links.rFilterResults $ Just filter{filterUpdatedBefore = Nothing})) filterUpdatedBefore- maybe mempty (\x -> renderFilter "Run after" (showText x) (Links.rFilterResults $ Just filter{filterRunAfter = Nothing})) filterRunAfter- forM_ filterJobTypes $ \x -> renderFilter "Job type" x (Links.rFilterResults $ Just filter{filterJobTypes = filterJobTypes \\ [x]})+ -- finally+ -- (run 8080 $ genericServe (server jm dbPool jobTypesRef jobRunnerRef))+ -- (cleanup >> (Pool.destroyAllResources dbPool)) - button_ [ class_ "btn btn-default search-button", type_ "button" ] $ "Search"- ul_ [ class_ "list-inline" ] $ do- li_ $ span_ $ strong_ "Common searches:"- li_ $ a_ [ href_ (Links.rFilterResults $ Just mempty) ] $ "All jobs"- li_ $ a_ [ href_ (Links.rFilterResults $ Just $ filter{ filterStatuses = [Job.Locked] }) ] $ "Currently running"- li_ $ a_ [ href_ (Links.rFilterResults $ Just $ filter{ filterStatuses = [Job.Success] }) ] $ "Successful"- li_ $ a_ [ href_ (Links.rFilterResults $ Just $ filter{ filterStatuses = [Job.Failed] }) ] $ "Failed"- li_ $ a_ [ href_ (Links.rFilterResults $ Just $ filter{ filterRunAfter = Just t }) ] $ "Future"- -- li_ $ a_ [ href_ "#" ] $ "Retried"- li_ $ a_ [ href_ (Links.rFilterResults $ Just $ filter{ filterStatuses = [Job.Queued] }) ] $ "Queued"- li_ $ a_ [ href_ (Links.rFilterResults $ Just $ filter{ filterUpdatedAfter = Just $ timeSince t 10 Minutes Ago }) ] $ "Last 10 mins"- li_ $ a_ [ href_ (Links.rFilterResults $ Just $ filter{ filterCreatedAfter = Just $ timeSince t 10 Minutes Ago }) ] $ "Recently created"- where- renderFilter :: Text -> Text -> Text -> Html ()- renderFilter k v u = do- li_ [ class_ "search-filter" ] $ do- span_ [ class_ "filter-name" ] $ toHtml k- span_ [ class_ "filter-value" ] $ do- toHtml v- a_ [ href_ u, class_ "text-danger" ] $ i_ [ class_ "glyphicon glyphicon-remove" ] $ ""+stopApp :: IO ()+stopApp = pure () -timeDuration :: UTCTime -> UTCTime -> (Int, String)-timeDuration from to = (diff, str)+server :: forall m . (MonadIO m)+ => Config+ -> Env+ -> (forall a . Handler a -> m a)+ -> ServerT FinalAPI m+server cfg env nt =+ (toServant routeServer) :<|> staticAssetServer where- str = if diff <= 0- then "under 1s"- else (if d>0 then (show d) <> "d" else "") <>- (if m>0 then (show m) <> "m" else "") <>- (if s>0 then (show s) <> "s" else "")- diff = (abs $ round $ diffUTCTime from to)- (m', s) = diff `divMod` 60- (h', m) = m' `divMod` 60- (d, h) = h' `divMod` 24--showText :: (Show a) => a -> Text-showText a = toS $ show a--jobContent :: Value -> Value-jobContent v = case v of- Aeson.Object o -> case HM.lookup "contents" o of- Nothing -> v- Just c -> c- _ -> v+ routeServer :: Routes (AsServerT m)+ routeServer = Routes+ { rFilterResults = nt . (filterResults cfg env)+ , rEnqueue = nt . (enqueueJob cfg env)+ , rCancel = nt . (cancelJob cfg env)+ , rRunNow = nt . (runJobNow cfg env)+ , rRefreshJobTypes = nt $ refreshJobTypes cfg env+ , rRefreshJobRunners = nt $ refreshJobRunners cfg env+ } -rowSuccess :: UTCTime -> Job -> Html ()-rowSuccess t job@Job{jobStatus, jobCreatedAt, jobUpdatedAt, jobPayload, jobAttempts, jobRunAt} = do- tr_ $ do- td_ [ class_ "job-type" ] $ case jobStatus of- Job.Success -> statusSuccess- Job.Failed -> statusFailed- Job.Queued -> if jobRunAt > t- then statusFuture- else statusWaiting- Job.Retry -> statusRetry- Job.Locked -> statusLocked+server2 :: Config+ -> Env+ -> Routes AsServer+server2 cfg env = Routes+ { rFilterResults = (filterResults cfg env)+ , rEnqueue = (enqueueJob cfg env)+ , rCancel = (cancelJob cfg env)+ , rRunNow = (runJobNow cfg env)+ , rRefreshJobTypes = refreshJobTypes cfg env+ , rRefreshJobRunners = refreshJobRunners cfg env+ } - -- span_ [ class_ "label label-success" ] $ "Success"- -- span_ [ class_ "job-run-time" ] $ "Completed 23h ago. Took 3 sec."-- td_ $ toHtml $ Job.defaultJobType job -- TODO: this needs to be changed- td_ $ div_ [ class_ "job-payload" ] $ payloadToHtml $ jobContent jobPayload- -- span_ [ class_ "key-value-pair" ] $ do- -- span_ [ class_ "key" ] $ "args"- -- span_ [ class_ "value" ] $ do- -- "[\"flexi_payment_reminder\", 3432423,"- -- a_ [ href_ "#", class_ "json-ellipsis" ] $ "\8230"- -- "]"- td_ "Text"- td_ $ case jobStatus of- Job.Success -> mempty- Job.Failed -> actionsFailed- Job.Queued -> if jobRunAt > t- then actionsFuture- else mempty- Job.Retry -> actionsRetry- Job.Locked -> mempty- where-- actionsFailed = do- button_ [ class_ "btn btn-default", type_ "button" ] $ "Retry again"-- actionsRetry = do- button_ [ class_ "btn btn-default", type_ "button" ] $ "Retry now"-- actionsFuture = do- button_ [ class_ "btn btn-default", type_ "button" ] $ "Run now"-- payloadToHtml :: Value -> Html ()- payloadToHtml v = case v of- Aeson.Object o -> do- toHtml ("{ " :: Text)- forM_ (HM.toList o) $ \(k, v) -> do- span_ [ class_ " key-value-pair " ] $ do- span_ [ class_ "key" ] $ toHtml $ k <> ":"- span_ [ class_ "value" ] $ payloadToHtml v- toHtml (" }" :: Text)- Aeson.Array a -> do- toHtml ("[" :: Text)- forM_ (toList a) $ \x -> do- payloadToHtml x- toHtml (", " :: Text)- toHtml ("]" :: Text)- Aeson.String t -> toHtml t- Aeson.Number n -> toHtml $ show n- Aeson.Bool b -> toHtml $ show b- Aeson.Null -> toHtml ("null" :: Text)-- statusSuccess = do- span_ [ class_ "label label-success" ] $ "Success"- span_ [ class_ "job-run-time" ] $ do- let (d, s) = timeDuration jobCreatedAt jobUpdatedAt- abbr_ [ title_ (showText jobUpdatedAt) ] $ toHtml $ "Completed " <> humanReadableTime' t jobUpdatedAt <> ". "- abbr_ [ title_ (showText d <> " seconds")] $ toHtml $ "Took " <> s-- statusFailed = do- span_ [ class_ "label label-danger" ] $ "Failed"- span_ [ class_ "job-run-time" ] $ do- abbr_ [ title_ (showText jobUpdatedAt) ] $ toHtml $ "Failed " <> humanReadableTime' t jobUpdatedAt <> " after " <> show jobAttempts <> " attempts"-- statusFuture = do- span_ [ class_ "label" ] $ "Future"- span_ [ class_ "job-run-time" ] $ do- abbr_ [ title_ (showText jobRunAt) ] $ toHtml $ humanReadableTime' t jobRunAt-- statusWaiting = do- span_ [ class_ "label label-warning" ] $ "Waiting"- -- span_ [ class_ "job-run-time" ] ("Waiting to be picked up" :: Text)-- statusRetry = do- span_ [ class_ "label label-info" ] $ toHtml $ "Retries (" <> show jobAttempts <> ")"- span_ [ class_ "job-run-time" ] $ do- abbr_ [ title_ (showText jobUpdatedAt) ] $ toHtml $ "Retried " <> humanReadableTime' t jobUpdatedAt <> ". "- abbr_ [ title_ (showText jobRunAt)] $ toHtml $ "Next retry in " <> humanReadableTime' t jobRunAt-- statusLocked = do- span_ [ class_ "label label-warning" ] $ toHtml ("Locked" :: Text)- -- span_ [ class_ "job-run-time" ] $ do- -- abbr_ [ title_ (showText jobUpdatedAt) ] $ toHtml $ "Retried " <> humanReadableTime' t jobUpdatedAt <> ". "- -- abbr_ [ title_ (showText jobRunAt)] $ toHtml $ "Next retry in " <> humanReadableTime' t jobRunAt+refreshJobRunners :: Config+ -> Env+ -> Handler NoContent+refreshJobRunners cfg@Config{..} Env{envRoutes=Web.Routes{..}, envJobRunnersRef} = do+ allJobRunners <- fetchAllJobRunners cfg+ atomicModifyIORef' envJobRunnersRef (\_ -> (allJobRunners, ()))+ throwError $ err302{errHeaders=[("Location", toS $ rFilterResults Nothing)]} +refreshJobTypes :: Config+ -> Env+ -> Handler NoContent+refreshJobTypes cfg Env{envRoutes=Web.Routes{..}, envJobTypesRef} = do+ allJobTypes <- fetchAllJobTypes cfg+ atomicModifyIORef' envJobTypesRef (\_ -> (allJobTypes, ()))+ throwError $ err302{errHeaders=[("Location", toS $ rFilterResults Nothing)]} -rowRetry :: Html ()-rowRetry = do- tr_ $ do- td_ [ class_ "job-type" ] $ do- span_ [ class_ "label label-info" ] $ "Retried (5)"- span_ [ class_ "job-run-time" ] $ "23 mins ago. Next retry in 90 min."- td_ "Queued Mail"- td_ $ div_ [ class_ "job-payload" ] $ do- span_ [ class_ "key-value-pair" ] $ do- span_ [ class_ "key" ] $ "client_id"- span_ [ class_ "value" ] $ "456"- span_ [ class_ "key-value-pair" ] $ do- span_ [ class_ "key" ] $ "user_id"- span_ [ class_ "value" ] $ "123"- span_ [ class_ "key-value-pair" ] $ do- span_ [ class_ "key" ] $ "args"- span_ [ class_ "value" ] $ do- "[\"flexi_payment_reminder\", 3432423,"- a_ [ href_ "#", class_ "json-ellipsis" ] $ "\8230"- "]"- td_ "Text"- td_ $ div_ [ class_ "btn-group" ] $ do- button_ [ class_ "btn btn-default", type_ "button" ] $ "Retry now"- button_ [ class_ "btn btn-default dropdown-toggle", data_ "toggle" "dropdown", ariaExpanded_ "false", type_ "button" ] $ span_ [ class_ "caret" ] $ ""- ul_ [ class_ "dropdown-menu", role_ "menu" ] $ do- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "First Item"- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Second Item"- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Third Item"+cancelJob :: Config+ -> Env+ -> JobId+ -> Handler NoContent+cancelJob Config{..} env jid = do+ liftIO $ withResource cfgDbPool $ \conn -> void $ cancelJobIO conn cfgTableName jid+ redirectToHome env +runJobNow :: Config+ -> Env+ -> JobId+ -> Handler NoContent+runJobNow Config{..} env jid = do+ liftIO $ withResource cfgDbPool $ \conn -> void $ runJobNowIO conn cfgTableName jid+ redirectToHome env -rowFailed :: Html ()-rowFailed = do- tr_ $ do- td_ [ class_ "job-type" ] $ do- span_ [ class_ "label label-danger" ] $ "Failed"- span_ [ class_ "job-run-time" ] $ "23 mins ago. After 25 attempts."- td_ "Queued Mail"- td_ $ div_ [ class_ "job-payload" ] $ do- span_ [ class_ "key-value-pair" ] $ do- span_ [ class_ "key" ] $ "client_id"- span_ [ class_ "value" ] $ "456"- span_ [ class_ "key-value-pair" ] $ do- span_ [ class_ "key" ] $ "user_id"- span_ [ class_ "value" ] $ "123"- span_ [ class_ "key-value-pair" ] $ do- span_ [ class_ "key" ] $ "args"- span_ [ class_ "value" ] $ do- "[\"flexi_payment_reminder\", 3432423,"- a_ [ href_ "#", class_ "json-ellipsis" ] $ "\8230"- "]"- td_ "Text"- td_ $ button_ [ class_ "btn btn-default", type_ "button" ] $ "Retry again"+enqueueJob :: Config+ -> Env+ -> JobId+ -> Handler NoContent+enqueueJob Config{..} env jid = do+ liftIO $ withResource cfgDbPool $ \conn -> do+ void $ unlockJobIO conn cfgTableName jid+ void $ runJobNowIO conn cfgTableName jid+ redirectToHome env -rowFuture :: Html ()-rowFuture = do- tr_ $ do- td_ [ class_ "job-type" ] $ do- span_ [ class_ "label label-default" ] $ "Future"- span_ [ class_ "job-run-time" ] $ "37 mins from now"- td_ "Queued Mail"- td_ $ div_ [ class_ "job-payload" ] $ do- span_ [ class_ "key-value-pair" ] $ do- span_ [ class_ "key" ] $ "client_id"- span_ [ class_ "value" ] $ "456"- span_ [ class_ "key-value-pair" ] $ do- span_ [ class_ "key" ] $ "user_id"- span_ [ class_ "value" ] $ "123"- span_ [ class_ "key-value-pair" ] $ do- span_ [ class_ "key" ] $ "args"- span_ [ class_ "value" ] $ do- "[\"flexi_payment_reminder\", 3432423,"- a_ [ href_ "#", class_ "json-ellipsis" ] $ "\8230"- "]"- td_ "Text"- td_ $ div_ [ class_ "btn-group" ] $ do- button_ [ class_ "btn btn-default", type_ "button" ] $ "Run now"- button_ [ class_ "btn btn-default dropdown-toggle", data_ "toggle" "dropdown", ariaExpanded_ "false", type_ "button" ] $ span_ [ class_ "caret" ] $ ""- ul_ [ class_ "dropdown-menu", role_ "menu" ] $ do- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "First Item"- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Second Item"- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Third Item"+redirectToHome :: Env -> Handler NoContent+redirectToHome Env{envRoutes=Web.Routes{..}} = do+ throwError $ err301{errHeaders=[("Location", toS $ rFilterResults Nothing)]} -rowLocked :: Html ()-rowLocked = do- tr_ $ do- td_ [ class_ "job-type" ] $ do- span_ [ class_ "label label-warning" ] $ "Locked"- span_ [ class_ "job-run-time" ] $ "Since 2min by"- span_ [ class_ "job-runner-name" ] $ "hostname:3242"- td_ "Queued Mail"- td_ $ div_ [ class_ "job-payload" ] $ do- span_ [ class_ "key-value-pair" ] $ do- span_ [ class_ "key" ] $ "client_id"- span_ [ class_ "value" ] $ "456"- span_ [ class_ "key-value-pair" ] $ do- span_ [ class_ "key" ] $ "user_id"- span_ [ class_ "value" ] $ "123"- span_ [ class_ "key-value-pair" ] $ do- span_ [ class_ "key" ] $ "args"- span_ [ class_ "value" ] $ do- "[\"flexi_payment_reminder\", 3432423,"- a_ [ href_ "#", class_ "json-ellipsis" ] $ "\8230"- "]"- td_ "Text"- td_ $ button_ [ class_ "btn btn-default", type_ "button" ] $ "Unlock"--resultsPanel :: UTCTime -> Filter -> [Job] -> Int -> Html ()-resultsPanel t filter@Filter{filterPage} jobs runningCount = do- div_ [ class_ "panel panel-default" ] $ do- div_ [ class_ "panel-heading" ] $ h3_ [ class_ "panel-title" ] $ do- "Currently running "- span_ [ class_ "badge" ] $ toHtml (show runningCount)- div_ [ class_ "panel-body" ] $ div_ [ class_ "currently-running" ] $ div_ [ class_ "table-responsive" ] $ table_ [ class_ "table" ] $ do- thead_ $ tr_ $ do- th_ "Job status"- th_ "Job type"- th_ "Job payload"- th_ "Last error"- th_ "Actions"- tbody_ $ do- forM_ jobs $ \j -> case jobStatus j of- Job.Success -> rowSuccess t j- _ -> rowSuccess t j- -- rowLocked- -- rowSuccess- -- rowFuture- -- rowRetry- -- rowFailed- div_ [ class_ "panel-footer" ] $ nav_ $ ul_ [ class_ "pager", style_ "margin:0px;" ] $ do- li_ [ class_ "previous" ] $ case filterPage of- Nothing -> a_ [ disabled_ "disabled" ] $ "Prev"- Just (l, 0) -> a_ [ disabled_ "disabled" ] $ "Prev"- Just (l, o) -> a_ [ href_ (Links.rFilterResults $ Just $ filter {filterPage = Just (l, max 0 $ o - l)}) ] $ "Prev"- li_ [ class_ "next" ] $ case filterPage of- Nothing -> a_ [ href_ (Links.rFilterResults $ Just $ filter {filterPage = Just (10, 10)}) ] $ "Next"- Just (l, o) -> a_ [ href_ (Links.rFilterResults $ Just $ filter {filterPage = Just (l, o + l)}) ] $ "Next"+filterResults :: Config+ -> Env+ -> Maybe Filter+ -> Handler (Html ())+filterResults cfg@Config{cfgJobToHtml, cfgDbPool} Env{..} mFilter = do+ let filters = fromMaybe mempty mFilter+ (jobs, runningCount) <- liftIO $ Pool.withResource cfgDbPool $ \conn -> (,)+ <$> (filterJobs cfg conn filters)+ <*> (countJobs cfg conn filters{ filterStatuses = [Job.Locked] })+ t <- liftIO getCurrentTime+ js <- liftIO $ fmap (DL.zip jobs) $ cfgJobToHtml jobs+ allJobTypes <- readIORef envJobTypesRef+ let navHtml = Web.sideNav envRoutes allJobTypes [] t filters+ bodyHtml = Web.resultsPanel envRoutes t filters js runningCount+ pure $ Web.pageLayout envRoutes navHtml bodyHtml +routes :: (Text -> Text) -> Web.Routes+routes linkFn = Web.Routes+ { Web.rFilterResults = rFilterResults+ , Web.rEnqueue = rEnqueue+ , Web.rRunNow = rRunNow+ , Web.rCancel = rCancel+ , Web.rRefreshJobTypes = rRefreshJobTypes+ , Web.rRefreshJobRunners = rRefreshJobRunners+ , Web.rStaticAsset = linkFn+ }+ where+ OddJobs.Endpoints.Routes{..} = allFieldLinks' (linkFn . toS . show . linkURI) :: OddJobs.Endpoints.Routes (AsLink Text) -ariaExpanded_ :: Text -> Attribute-ariaExpanded_ v = makeAttribute "aria-expanded" v+-- absText :: Link -> Text+-- absText l = "/" <> (toS $ show $ linkURI l)
src/OddJobs/Job.hs view
@@ -1,4 +1,7 @@ {-# LANGUAGE RankNTypes, FlexibleInstances, FlexibleContexts, PartialTypeSignatures, TupleSections, DeriveGeneric, UndecidableInstances #-}+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE ScopedTypeVariables #-}+ module OddJobs.Job ( -- * Starting the job-runner@@ -9,21 +12,9 @@ -- * Configuring the job-runner -- -- $config- , Config(..)+ , Config(..) , ConcurrencyControl(..) - -- ** Configuration helpers- --- -- $configHelpers- , defaultConfig- , defaultJobToText- , defaultJobType- , defaultTimedLogger- , defaultLogStr- , defaultPollingInterval- , defaultLockTimeout- , withConnectionPool- -- * Creating/scheduling jobs -- -- $createJobs@@ -36,9 +27,13 @@ , Job(..) , JobId , Status(..)+ , JobRunnerName(..) , TableName , delaySeconds , Seconds(..)+ , JobErrHandler(..)+ , AllJobTypes(..)+ -- ** Structured logging -- -- $logging@@ -58,12 +53,15 @@ -- * Database helpers -- -- $dbHelpers- , findJobById , findJobByIdIO- , saveJob , saveJobIO+ , runJobNowIO+ , unlockJobIO+ , cancelJobIO , jobDbColumns , concatJobDbColumns+ , fetchAllJobTypes+ , fetchAllJobRunners -- * JSON helpers --@@ -80,9 +78,6 @@ import Data.Text as T import Database.PostgreSQL.Simple as PGS import Database.PostgreSQL.Simple.Notification-import Database.PostgreSQL.Simple.FromField as FromField-import Database.PostgreSQL.Simple.ToField as ToField-import Database.PostgreSQL.Simple.FromRow as FromRow import UnliftIO.Async import UnliftIO.Concurrent (threadDelay, myThreadId) import Data.String@@ -90,7 +85,7 @@ import Network.HostName (getHostName) import UnliftIO.MVar import Debug.Trace-import Control.Monad.Logger as MLogger (LogLevel(..), LogStr(..), toLogStr)+import Control.Monad.Logger as MLogger (LogLevel(..), LogStr, toLogStr) import UnliftIO.IORef import UnliftIO.Exception ( SomeException(..), try, catch, finally , catchAny, bracket, Exception(..), throwIO@@ -107,7 +102,7 @@ import Data.String.Conv (StringConv(..), toS) import Data.Functor (void) import Control.Monad (forever)-import Data.Maybe (isNothing, maybe, fromMaybe)+import Data.Maybe (isNothing, maybe, fromMaybe, listToMaybe, mapMaybe) import Data.Either (either) import Control.Monad.Reader import GHC.Generics@@ -118,9 +113,10 @@ import System.FilePath (FilePath) import qualified System.Directory as Dir import Data.Aeson.Internal (iparse, IResult(..), formatError)-import qualified System.Log.FastLogger as FLogger import Prelude hiding (log)-+import GHC.Exts (toList)+import Database.PostgreSQL.Simple.Types as PGS (Identifier(..))+import Database.PostgreSQL.Simple.ToField as PGS (toField) -- | The documentation of odd-jobs currently promotes 'startJobRunner', which -- expects a fairly detailed 'Config' record, as a top-level function for@@ -128,237 +124,38 @@ -- an enviroment for a 'ReaderT', and almost all functions are written in this -- 'ReaderT' monad which impleents an instance of the 'HasJobRunner' type-class. ----- **In future,** this /internal/ implementation detail will allow us to offer a+-- __In future,__ this /internal/ implementation detail will allow us to offer a -- type-class based interface as well (similar to what -- 'Yesod.JobQueue.YesodJobQueue' provides). class (MonadUnliftIO m, MonadBaseControl IO m) => HasJobRunner m where getPollingInterval :: m Seconds onJobSuccess :: Job -> m ()- onJobFailed :: Job -> m ()- onJobPermanentlyFailed :: Job -> m ()+ onJobFailed :: forall a . m [JobErrHandler a] getJobRunner :: m (Job -> IO ()) getDbPool :: m (Pool Connection) getTableName :: m TableName onJobStart :: Job -> m () getDefaultMaxAttempts :: m Int- onJobTimeout :: Job -> m () getRunnerEnv :: m RunnerEnv getConcurrencyControl :: m ConcurrencyControl getPidFile :: m (Maybe FilePath)- getJobToText :: m (Job -> Text) log :: LogLevel -> LogEvent -> m ()+ getDefaultJobTimeout :: m Seconds -- $logging ----- TODO: Complete the prose here---data LogEvent- -- | Emitted when a job starts execution- = LogJobStart !Job- -- | Emitted when a job succeeds along with the time taken for execution.- | LogJobSuccess !Job !NominalDiffTime- -- | Emitted when a job fails (but will be retried) along with the time taken for- -- /this/ attempt- | LogJobFailed !Job !NominalDiffTime- -- | Emitted when a job fails permanently (and will no longer be retried) along- -- with the time taken for /this/ attempt (i.e. final attempt)- | LogJobPermanentlyFailed !Job !NominalDiffTime- -- | Emitted when a job times out and is picked-up again for execution- | LogJobTimeout !Job- -- | Emitted whenever 'jobPoller' polls the DB table- | LogPoll- -- | Emitted whenever any other event occurs- | LogText !Text- deriving (Eq, Show, Generic)----- | While odd-jobs is highly configurable and the 'Config' data-type might seem--- daunting at first, it is not necessary to tweak every single configuration--- parameter by hand. Please start-off by using the sensible defaults provided--- by the [configuration helpers](#configHelpers), and tweaking--- config parameters on a case-by-case basis.-data Config = Config- { -- | The DB table which holds your jobs. Please note, this should have been- -- created by the 'OddJobs.Migrations.createJobTable' function.- cfgTableName :: TableName-- -- | The actualy "job-runner" that __you__ need to provide. Please look at- -- the examples/tutorials if your applicaton's code is not in the @IO@- -- monad.- , cfgJobRunner :: Job -> IO ()-- -- | The number of times a failing job is retried before it is considered is- -- "permanently failed" and ignored by the job-runner. This config parameter- -- is called "/default/ max attempts" because, in the future, it would be- -- possible to specify the number of retry-attemps on a per-job basis- -- (__Note:__ per-job retry-attempts has not been implemented yet)- , cfgDefaultMaxAttempts :: Int-- -- | Controls how many jobs can be run concurrently by /this instance/ of- -- the job-runner. __Please note,__ this is NOT the global concurrency of- -- entire job-queue. It is possible to have job-runners running on multiple- -- machines, and each will apply the concurrency control independnt of other- -- job-runners. TODO: Link-off to relevant section in the tutorial.- , cfgConcurrencyControl :: ConcurrencyControl--- -- | The DB connection-pool to use for the job-runner. __Note:__ in case- -- your jobs require a DB connection, please create a separate- -- connection-pool for them. This pool will be used ONLY for monitoring jobs- -- and changing their status. We need to have _at least 4 connections__ in- -- this connection-pool for the job-runner to work as expected. (TODO:- -- Link-off to tutorial)- , cfgDbPool :: Pool Connection-- -- | How frequently should the 'jobPoller' check for jobs where the Job's- -- 'jobRunAt' field indicates that it's time for the job to be executed.- -- TODO: link-off to the tutorial.- , cfgPollingInterval :: Seconds-- -- | User-defined callback function that is called whenever a job succeeds.- , cfgOnJobSuccess :: Job -> IO ()-- -- | User-defined callback function that is called whenever a job fails. This- -- does not indicate permanent failure and means the job will be retried. It- -- is a good idea to log the failures to Airbrake, NewRelic, Sentry, or some- -- other error monitoring tool.- , cfgOnJobFailed :: Job -> IO ()-- -- | User-defined callback function that is called whenever a job fails- -- /permanently/ (i.e. number of retry-attempts have crossed the configured- -- threshold). It is a good idea to log the failures to Airbrake, NewRelic,- -- Sentry, or some other error monitoring tool.- , cfgOnJobPermanentlyFailed :: Job -> IO ()-- -- | User-defined callback function that is called whenever a job starts- -- execution.- , cfgOnJobStart :: Job -> IO ()-- -- | User-defined callback function that is called whenever a job times-out.- , cfgOnJobTimeout :: Job -> IO ()-- -- | File to store the PID of the job-runner process. This is used only when- -- invoking the job-runner as an independent background deemon (the usual mode- -- of deployment). (TODO: Link-off to tutorial).- , cfgPidFile :: Maybe FilePath-- -- | A "structured logging" function that __you__ need to provide. The- -- @odd-jobs@ library does NOT use the standard logging interface provided by- -- 'monad-logger' on purpose. TODO: link-off to tutorial. Please also read- -- 'cfgJobToText' and 'cfgJobType'- , cfgLogger :: LogLevel -> LogEvent -> IO ()-- -- | When emitting certain text messages in logs, how should the 'Job' be- -- summarized in a textual format? Related: 'defaultJobToText'- , cfgJobToText :: Job -> Text-- -- | How to extract the "job type" from a 'Job'. Related: 'defaultJobType'- , cfgJobType :: Job -> Text- }--data ConcurrencyControl- -- | The maximum number of concurrent jobs that /this instance/ of the- -- job-runner can execute. TODO: Link-off to tutorial.- = MaxConcurrentJobs Int- -- | __Not recommended:__ Please do not use this in production unless you know- -- what you're doing. No machine can support unlimited concurrency. If your- -- jobs are doing anything worthwhile, running a sufficiently large number- -- concurrently is going to max-out /some/ resource of the underlying machine,- -- such as, CPU, memory, disk IOPS, or network bandwidth.- | UnlimitedConcurrentJobs-- -- | Use this to dynamically determine if the next job should be picked-up, or- -- not. This is useful to write custom-logic to determine whether a limited- -- resource is below a certain usage threshold (eg. CPU usage is below 80%).- -- __Caveat:__ This feature has not been tested in production, yet. TODO:- -- Link-off to tutorial.- | DynamicConcurrency (IO Bool)--instance Show ConcurrencyControl where- show cc = case cc of- MaxConcurrentJobs n -> "MaxConcurrentJobs " <> show n- UnlimitedConcurrentJobs -> "UnlimitedConcurrentJobs"- DynamicConcurrency _ -> "DynamicConcurrency (IO Bool)"----- $configHelpers+-- OddJobs uses \"structured logging\" for important events that occur during+-- the life-cycle of a job-runner. This is useful if you're using JSON\/XML for+-- aggegating logs of your entire application to something like Kibana, AWS+-- CloudFront, GCP StackDriver Logging, etc. ----- #configHelpers#---- | This function gives you a 'Config' with a bunch of sensible defaults--- already applied. It requies the bare minimum arguments that this library--- cannot assume on your behalf.+-- If you're not interested in using structured logging, look at+-- 'OddJobs.ConfigBuilder.defaultLogStr' to output plain-text logs (or you can+-- write your own function, as well). ----- It makes a few __important assumptions__ about your 'jobPayload 'JSON, which--- are documented in 'defaultJobType'.-defaultConfig :: (LogLevel -> LogEvent -> IO ()) -- ^ "Structured logging" function. Ref: 'cfgLogger'- -> TableName -- ^ DB table which holds your jobs. Ref: 'cfgTableName'- -> Pool Connection -- ^ DB connection-pool to be used by job-runner. Ref: 'cfgDbPool'- -> ConcurrencyControl -- ^ Concurrency configuration. Ref: 'cfgConcurrencyControl'- -> (Job -> IO ()) -- ^ The actual "job runner" which contains your application code. Ref: 'cfgJobRunner'- -> Config-defaultConfig logger tname dbpool ccControl jrunner =- let cfg = Config- { cfgPollingInterval = defaultPollingInterval- , cfgOnJobSuccess = (const $ pure ())- , cfgOnJobFailed = (const $ pure ())- , cfgOnJobPermanentlyFailed = (const $ pure ())- , cfgJobRunner = jrunner- , cfgLogger = logger- , cfgDbPool = dbpool- , cfgOnJobStart = (const $ pure ())- , cfgDefaultMaxAttempts = 10- , cfgTableName = tname- , cfgOnJobTimeout = (const $ pure ())- , cfgConcurrencyControl = ccControl- , cfgPidFile = Nothing- , cfgJobToText = defaultJobToText (cfgJobType cfg)- , cfgJobType = defaultJobType- }- in cfg --- | Used only by 'defaultLogStr' now. TODO: Is this even required anymore?--- Should this be removed?-defaultJobToText :: (Job -> Text) -> Job -> Text-defaultJobToText jobTypeFn job@Job{jobId} =- "JobId=" <> (toS $ show jobId) <> " JobType=" <> jobTypeFn job --- | This makes __two important assumptions__. First, this /assumes/ that jobs--- in your app are represented by a sum-type. For example:------ @--- data MyJob = SendWelcomeEmail Int--- | SendPasswordResetEmail Text--- | SetupSampleData Int--- @------ Second, it /assumes/ that the JSON representatin of this sum-type is--- "tagged". For example, the following...------ > let pload = SendWelcomeEmail 10------ ...when converted to JSON, would look like...------ > {"tag":"SendWelcomeEmail", "contents":10}------ It uses this assumption to extract the "job type" from a 'Data.Aeson.Value'--- (which would be @SendWelcomeEmail@ in the example given above). This is used--- in logging and the admin UI.------ Even if tihs assumption is violated, the job-runner /should/ continue to--- function. It's just that you won't get very useful log messages.-defaultJobType :: Job -> Text-defaultJobType Job{jobPayload} =- case jobPayload of- Aeson.Object hm -> case HM.lookup "tag" hm of- Just (Aeson.String t) -> t- _ -> "unknown"- _ -> "unknown"--- data RunnerEnv = RunnerEnv { envConfig :: !Config , envJobThreadsRef :: !(IORef [Async ()])@@ -371,15 +168,10 @@ instance HasJobRunner RunnerM where getPollingInterval = cfgPollingInterval . envConfig <$> ask- onJobFailed job = do- fn <- cfgOnJobFailed . envConfig <$> ask- logCallbackErrors (jobId job) "onJobFailed" $ liftIO $ fn job+ onJobFailed = cfgOnJobFailed . envConfig <$> ask onJobSuccess job = do fn <- cfgOnJobSuccess . envConfig <$> ask logCallbackErrors (jobId job) "onJobSuccess" $ liftIO $ fn job- onJobPermanentlyFailed job = do- fn <- cfgOnJobPermanentlyFailed . envConfig <$> ask- logCallbackErrors (jobId job) "onJobPermanentlyFailed" $ liftIO $ fn job getJobRunner = cfgJobRunner . envConfig <$> ask getDbPool = cfgDbPool . envConfig <$> ask getTableName = cfgTableName . envConfig <$> ask@@ -387,10 +179,6 @@ fn <- cfgOnJobStart . envConfig <$> ask logCallbackErrors (jobId job) "onJobStart" $ liftIO $ fn job - onJobTimeout job = do- fn <- cfgOnJobTimeout . envConfig <$> ask- logCallbackErrors (jobId job) "onJobTimeout" $ liftIO $ fn job- getDefaultMaxAttempts = cfgDefaultMaxAttempts . envConfig <$> ask getRunnerEnv = ask@@ -398,12 +186,13 @@ getConcurrencyControl = (cfgConcurrencyControl . envConfig <$> ask) getPidFile = cfgPidFile . envConfig <$> ask- getJobToText = cfgJobToText . envConfig <$> ask log logLevel logEvent = do loggerFn <- cfgLogger . envConfig <$> ask liftIO $ loggerFn logLevel logEvent + getDefaultJobTimeout = cfgDefaultJobTimeout . envConfig <$> ask+ -- | Start the job-runner in the /current/ thread, i.e. you'll need to use -- 'forkIO' or 'async' manually, if you want the job-runner to run in the -- background. Consider using 'OddJobs.Cli' to rapidly build your own@@ -417,107 +206,18 @@ } runReaderT jobMonitor monitorEnv --- | Convenience function to create a DB connection-pool with some sensible--- defaults. Please see the source-code of this function to understand what it's--- doing. TODO: link-off to tutorial.-withConnectionPool :: (MonadUnliftIO m)- => Either BS.ByteString PGS.ConnectInfo- -> (Pool PGS.Connection -> m a)- -> m a-withConnectionPool connConfig action = withRunInIO $ \runInIO -> do- bracket poolCreator destroyAllResources (runInIO . action)- where- poolCreator = liftIO $- case connConfig of- Left connString ->- createPool (PGS.connectPostgreSQL connString) PGS.close 1 (fromIntegral $ 2 * (unSeconds defaultPollingInterval)) 5- Right connInfo ->- createPool (PGS.connect connInfo) PGS.close 1 (fromIntegral $ 2 * (unSeconds defaultPollingInterval)) 5---- | As the name says. Ref: 'cfgPollingInterval'-defaultPollingInterval :: Seconds-defaultPollingInterval = Seconds 5--type JobId = Int--data Status = Success- | Queued- | Failed- | Retry- | Locked- deriving (Eq, Show, Generic, Enum)--instance Ord Status where- compare x y = compare (toText x) (toText y)--data Job = Job- { jobId :: JobId- , jobCreatedAt :: UTCTime- , jobUpdatedAt :: UTCTime- , jobRunAt :: UTCTime- , jobStatus :: Status- , jobPayload :: Aeson.Value- , jobLastError :: Maybe Value- , jobAttempts :: Int- , jobLockedAt :: Maybe UTCTime- , jobLockedBy :: Maybe Text- } deriving (Eq, Show)--instance ToText Status where- toText s = case s of- Success -> "success"- Queued -> "queued"- Retry -> "retry"- Failed -> "failed"- Locked -> "locked"--instance (StringConv Text a) => FromText (Either a Status) where- fromText t = case t of- "success" -> Right Success- "queued" -> Right Queued- "failed" -> Right Failed- "retry" -> Right Retry- "locked" -> Right Locked- x -> Left $ toS $ "Unknown job status: " <> x--instance FromField Status where- fromField f mBS = (fromText <$> (fromField f mBS)) >>= \case- Left e -> FromField.returnError PGS.ConversionFailed f e- Right s -> pure s--instance ToField Status where- toField s = toField $ toText s--instance FromRow Job where- fromRow = Job- <$> field -- jobId- <*> field -- createdAt- <*> field -- updatedAt- <*> field -- runAt- <*> field -- status- <*> field -- payload- <*> field -- lastError- <*> field -- attempts- <*> field -- lockedAt- <*> field -- lockedBy---- TODO: Add a sum-type for return status which can signal the monitor about--- whether the job needs to be retried, marked successfull, or whether it has--- completed failed.-type JobRunner = Job -> IO ()-- jobWorkerName :: IO String jobWorkerName = do pid <- getProcessID hname <- getHostName pure $ hname ++ ":" ++ (show pid) --- | TODO: Make this configurable for the job-runner, why is this still--- hard-coded?-defaultLockTimeout :: Seconds-defaultLockTimeout = Seconds 600-+-- | If you are writing SQL queries where you want to return ALL columns from+-- the jobs table it is __recommended__ that you do not issue a @SELECT *@ or+-- @RETURNIG *@. List out specific DB columns using 'jobDbColumns' and+-- 'concatJobDbColumns' instead. This will insulate you from runtime errors+-- caused by addition of new columns to 'cfgTableName' in future versions of+-- OddJobs. jobDbColumns :: (IsString s, Semigroup s) => [s] jobDbColumns = [ "id"@@ -532,6 +232,11 @@ , "locked_by" ] +-- | All 'jobDbColumns' joined together with commas. Useful for constructing SQL+-- queries, eg:+--+-- @'query_' conn $ "SELECT " <> concatJobDbColumns <> "FROM jobs"@+ concatJobDbColumns :: (IsString s, Semigroup s) => s concatJobDbColumns = concatJobDbColumns_ jobDbColumns "" where@@ -550,6 +255,18 @@ pool <- getDbPool withResource pool action +--+-- $dbHelpers+--+-- A bunch of functions that help you query 'cfgTableName' and change the status+-- of individual jobs. Most of these functions are in @IO@ and you /might/ want+-- to write wrappers that lift them into you application's custom monad.+--+-- __Note:__ When passing a 'Connection' to these function, it is+-- __recommended__ __to not__ take a connection from 'cfgDbPool'. Use your+-- application\'s database pool instead.+--+ findJobById :: (HasJobRunner m) => JobId -> m (Maybe Job)@@ -567,6 +284,8 @@ saveJobQuery :: TableName -> PGS.Query saveJobQuery tname = "UPDATE " <> tname <> " set run_at = ?, status = ?, payload = ?, last_error = ?, attempts = ?, locked_at = ?, locked_by = ? WHERE id = ? RETURNING " <> concatJobDbColumns +deleteJobQuery :: TableName -> PGS.Query+deleteJobQuery tname = "delete " <> tname <> " Where id = ?" saveJob :: (HasJobRunner m) => Job -> m Job saveJob j = do@@ -590,6 +309,45 @@ [j] -> pure j js -> Prelude.error $ "Not expecting multiple rows to ber returned when updating job id=" <> (show jobId) +deleteJob :: (HasJobRunner m) => JobId -> m ()+deleteJob jid = do+ tname <- getTableName+ withDbConnection $ \conn -> liftIO $ deleteJobIO conn tname jid++deleteJobIO :: Connection -> TableName -> JobId -> IO ()+deleteJobIO conn tname jid = do+ void $ PGS.execute conn (deleteJobQuery tname) (Only jid)++runJobNowIO :: Connection -> TableName -> JobId -> IO (Maybe Job)+runJobNowIO conn tname jid = do+ t <- getCurrentTime+ updateJobHelper tname conn (Queued, [Queued, Retry, Failed], Just t, jid)++-- | TODO: First check in all job-runners if this job is still running, or not,+-- and somehow send an uninterruptibleCancel to that thread.+unlockJobIO :: Connection -> TableName -> JobId -> IO (Maybe Job)+unlockJobIO conn tname jid = do+ fmap listToMaybe $ PGS.query conn q (Retry, jid, In [Locked])+ where+ q = "update " <> tname <> " set status=?, run_at=now(), locked_at=null, locked_by=null where id=? and status in ? returning " <> concatJobDbColumns++cancelJobIO :: Connection -> TableName -> JobId -> IO (Maybe Job)+cancelJobIO conn tname jid =+ updateJobHelper tname conn (Failed, [Queued, Retry], Nothing, jid)++updateJobHelper :: TableName+ -> Connection+ -> (Status, [Status], Maybe UTCTime, JobId)+ -> IO (Maybe Job)+updateJobHelper tname conn (newStatus, existingStates, mRunAt, jid) =+ fmap listToMaybe $ PGS.query conn q (newStatus, runAt, jid, PGS.In existingStates)+ where+ q = "update " <> tname <> " set attempts=0, status=?, run_at=? where id=? and status in ? returning " <> concatJobDbColumns+ runAt = case mRunAt of+ Nothing -> PGS.toField $ PGS.Identifier "run_at"+ Just t -> PGS.toField t++ data TimeoutException = TimeoutException deriving (Eq, Show) instance Exception TimeoutException @@ -623,42 +381,44 @@ Nothing -> Prelude.error $ "Could not find job id=" <> show jid Just job -> do startTime <- liftIO getCurrentTime+ lockTimeout <- getDefaultJobTimeout (flip catches) [Handler $ timeoutHandler job startTime, Handler $ exceptionHandler job startTime] $ do- runJobWithTimeout defaultLockTimeout job+ runJobWithTimeout lockTimeout job endTime <- liftIO getCurrentTime- newJob <- saveJob job{jobStatus=Success, jobLockedBy=Nothing, jobLockedAt=Nothing}+ deleteJob jid+ let newJob = job{jobStatus=Success, jobLockedBy=Nothing, jobLockedAt=Nothing} log LevelInfo $ LogJobSuccess newJob (diffUTCTime endTime startTime) onJobSuccess newJob pure () where- timeoutHandler job startTime (e :: TimeoutException) = retryOrFail (show e) job onJobTimeout onJobPermanentlyFailed startTime- exceptionHandler job startTime (e :: SomeException) = retryOrFail (show e) job onJobFailed onJobPermanentlyFailed startTime- retryOrFail errStr job@Job{jobAttempts} onFail onPermanentFail startTime = do+ timeoutHandler job startTime (e :: TimeoutException) = retryOrFail (toException e) job startTime+ exceptionHandler job startTime (e :: SomeException) = retryOrFail (toException e) job startTime+ retryOrFail e job@Job{jobAttempts} startTime = do endTime <- liftIO getCurrentTime defaultMaxAttempts <- getDefaultMaxAttempts- jobToText <- getJobToText let runTime = diffUTCTime endTime startTime- (newStatus, action, logAction) = if jobAttempts >= defaultMaxAttempts- then ( Failed- , onPermanentFail- , log LevelError $ LogJobPermanentlyFailed job runTime- )- else ( Retry- , onFail- , log LevelWarn $ LogJobFailed job runTime- )+ (newStatus, failureMode, logLevel) = if jobAttempts >= defaultMaxAttempts+ then ( Failed, FailPermanent, LevelError )+ else ( Retry, FailWithRetry, LevelWarn ) t <- liftIO getCurrentTime newJob <- saveJob job{ jobStatus=newStatus , jobLockedBy=Nothing , jobLockedAt=Nothing- , jobLastError=(Just $ toJSON errStr) -- TODO: convert errors to json properly+ , jobLastError=(Just $ toJSON $ show e) -- TODO: convert errors to json properly , jobRunAt=(addUTCTime (fromIntegral $ (2::Int) ^ jobAttempts) t) }- logAction- void $ action newJob- pure ()+ case fromException e :: Maybe TimeoutException of+ Nothing -> log logLevel $ LogJobFailed newJob e failureMode runTime+ Just _ -> log logLevel $ LogJobTimeout newJob + let tryHandler (JobErrHandler handler) res = case fromException e of+ Nothing -> res+ Just e_ -> handler e_ newJob failureMode+ handlers <- onJobFailed+ liftIO $ void $ Prelude.foldr tryHandler (throwIO e) handlers+ pure () +-- TODO: This might have a resource leak. restartUponCrash :: (HasJobRunner m, Show a) => Text -> m a -> m () restartUponCrash name_ action = do a <- async action@@ -704,7 +464,7 @@ [] -> log LevelInfo $ LogText "All job-threads exited" as -> do tid <- myThreadId- (a, _) <- waitAnyCatch as+ void $ waitAnyCatch as log LevelDebug $ LogText $ toS $ "Waiting for " <> show (DL.length as) <> " jobs to complete before shutting down. myThreadId=" <> (show tid) delaySeconds (Seconds 1) waitForJobs@@ -738,6 +498,7 @@ processName <- liftIO jobWorkerName pool <- getDbPool tname <- getTableName+ lockTimeout <- getDefaultJobTimeout log LevelInfo $ LogText $ toS $ "Starting the job monitor via DB polling with processName=" <> processName concurrencyControlFn <- getConcurrencyControlFn withResource pool $ \pollerDbConn -> forever $ concurrencyControlFn >>= \case@@ -748,7 +509,7 @@ t <- liftIO getCurrentTime r <- liftIO $ PGS.query pollerDbConn (jobPollingSql tname)- (Locked, t, processName, t, (In [Queued, Retry]), Locked, (addUTCTime (fromIntegral $ negate $ unSeconds defaultLockTimeout) t))+ (Locked, t, processName, t, (In [Queued, Retry]), Locked, (addUTCTime (fromIntegral $ negate $ unSeconds lockTimeout) t)) case r of -- When we don't have any jobs to run, we can relax a bit... [] -> pure delayAction@@ -902,41 +663,26 @@ either throwString (pure . Prelude.id) (eitherParsePayloadWith parser job) --- | TODO: Should the library be doing this?-defaultTimedLogger :: FLogger.TimedFastLogger- -> (LogLevel -> LogEvent -> LogStr)- -> LogLevel- -> LogEvent- -> IO ()-defaultTimedLogger logger logStrFn logLevel logEvent =- if logLevel == LevelDebug- then pure ()- else logger $ \t -> (toLogStr t) <> " | " <>- (logStrFn logLevel logEvent) <>- "\n"+-- | Used by the web\/admin UI to fetch a \"master list\" of all known+-- job-types. Ref: 'cfgAllJobTypes'+fetchAllJobTypes :: (MonadIO m)+ => Config+ -> m [Text]+fetchAllJobTypes Config{cfgAllJobTypes, cfgDbPool} = liftIO $ do+ case cfgAllJobTypes of+ AJTFixed jts -> pure jts+ AJTSql fn -> withResource cfgDbPool fn+ AJTCustom fn -> fn - -- TODO; Should the library be doing this?-defaultLogStr :: (Job -> Text)- -> LogLevel- -> LogEvent- -> LogStr-defaultLogStr jobToText logLevel logEvent =- (toLogStr $ show logLevel) <> " | " <> str- where- jobToLogStr j = toLogStr $ jobToText j- str = case logEvent of- LogJobStart j ->- "Started | " <> jobToLogStr j- LogJobFailed j t ->- "Failed (retry) | " <> jobToLogStr j <> " | runtime=" <> (toLogStr $ show t)- LogJobSuccess j t ->- "Success | " <> (jobToLogStr j) <> " | runtime=" <> (toLogStr $ show t)- LogJobPermanentlyFailed j t ->- "Failed (permanent) | " <> jobToLogStr j <> " | runtime=" <> (toLogStr $ show t)- LogJobTimeout j@Job{jobLockedAt, jobLockedBy} ->- "Timeout | " <> jobToLogStr j <> " | lockedBy=" <> (toLogStr $ fromMaybe "unknown" jobLockedBy) <>- " lockedAt=" <> (toLogStr $ maybe "unknown" show jobLockedAt)- LogPoll ->- "Polling jobs table"- LogText t ->- toLogStr t+-- | Used by web\/admin IO to fetch a \"master list\" of all known job-runners.+-- There is a known issue with the way this has been implemented:+--+-- * Since this looks at the 'jobLockedBy' column of 'cfgTableName', it will+-- discover only those job-runners that are actively executing at least one+-- job at the time this function is executed.+fetchAllJobRunners :: (MonadIO m)+ => Config+ -> m [JobRunnerName]+fetchAllJobRunners Config{cfgTableName, cfgDbPool} = liftIO $ withResource cfgDbPool $ \conn -> do+ fmap (mapMaybe fromOnly) $ PGS.query_ conn $ "select distinct locked_by from " <> cfgTableName+
− src/OddJobs/Links.hs
@@ -1,14 +0,0 @@-{-# LANGUAGE RecordWildCards, OverloadedStrings #-}--module OddJobs.Links where--import Servant-import Servant.API.Generic-import OddJobs.Web (Routes(..))-import Data.String.Conv-import Data.Text--Routes{..} = allFieldLinks' absText :: Routes (AsLink Text)--absText :: Link -> Text-absText l = "/" <> (toS $ show $ linkURI l)
src/OddJobs/Types.hs view
@@ -1,8 +1,27 @@+{-# LANGUAGE ExistentialQuantification #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE FlexibleContexts #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE RankNTypes #-}+ module OddJobs.Types where import Database.PostgreSQL.Simple as PGS import UnliftIO (MonadIO) import UnliftIO.Concurrent (threadDelay)+import Data.Text.Conversions+import Database.PostgreSQL.Simple.FromField as FromField+import Database.PostgreSQL.Simple.ToField as ToField+import Database.PostgreSQL.Simple.FromRow as FromRow+import Data.Time+import UnliftIO.Exception+import Data.Text (Text)+import GHC.Generics+import Data.Aeson as Aeson hiding (Success)+import Data.String.Conv+import Lucid (Html(..))+import Data.Pool (Pool)+import Control.Monad.Logger (LogLevel) -- | An alias for 'Query' type. Since this type has an instance of 'IsString' -- you do not need to do anything special to create a value for this type. Just@@ -29,3 +48,300 @@ oneSec :: Int oneSec = 1000000 ++data LogEvent+ -- | Emitted when a job starts execution+ = LogJobStart !Job+ -- | Emitted when a job succeeds along with the time taken for execution.+ | LogJobSuccess !Job !NominalDiffTime+ -- | Emitted when a job fails (but will be retried) along with the time taken for+ -- /this/ attempt+ | LogJobFailed !Job !SomeException !FailureMode !NominalDiffTime+ -- | Emitted when a job times out and is picked-up again for execution+ | LogJobTimeout !Job+ -- | Emitted whenever 'OddJobs.Job.jobPoller' polls the DB table+ | LogPoll+ -- | TODO+ | LogWebUIRequest+ -- | Emitted whenever any other event occurs+ | LogText !Text+ deriving (Show, Generic)++-- | Used by 'JobErrHandler' and 'LogEvent' to indicate the nature of failure.+data FailureMode+ -- The job failed, but will be retried in the future.+ = FailWithRetry+ -- | The job failed and will no longer be retried (probably because it has+ -- been tried 'cfgDefaultMaxAttempts' times already).+ | FailPermanent deriving (Eq, Show, Generic)++-- | Exception handler for jobs. This is conceptually very similar to how+-- 'Control.Exception.Handler' and 'Control.Exception.catches' (from+-- 'Control.Exception') work in-tandem. Using 'cfgOnJobFailed' you can install+-- /multiple/ exception handlers, where each handler is responsible for one type+-- of exception. OddJobs will execute the correct exception handler on the basis+-- of the type of runtime exception raised. For example:+--+-- @+-- cfgOnJobFailed =+-- [ JobErrHandler $ \(e :: HttpException) job failMode -> ...+-- , JobErrHandler $ \(e :: SqlException) job failMode -> ...+-- , JobErrHandler $ \(e :: ) job failMode -> ...+-- ]+-- @+--+-- __Note:__ Please refer to the section on [alerts and+-- notifications](https://www.haskelltutorials.com/odd-jobs/guide.html#alerts)+-- in the implementation guide to understand how to use the machinery provided+-- by 'JobErrHandler' and 'cfgOnJobFailed'.+data JobErrHandler a = forall e . (Exception e) => JobErrHandler (e -> Job -> FailureMode -> IO a)+++-- | __Note:__ Please read the section on [controlling+-- concurrency](https://www.haskelltutorials.com/odd-jobs/guide.html#controlling-concurrency)+-- in the implementation guide to understand the implications of each option+-- given by the data-type.+data ConcurrencyControl+ -- | The maximum number of concurrent jobs that /this instance/ of the+ -- job-runner can execute.+ = MaxConcurrentJobs Int+ -- | __Not recommended:__ Please do not use this in production unless you know+ -- what you're doing. No machine can support unlimited concurrency. If your+ -- jobs are doing anything worthwhile, running a sufficiently large number+ -- concurrently is going to max-out /some/ resource of the underlying machine,+ -- such as, CPU, memory, disk IOPS, or network bandwidth.+ | UnlimitedConcurrentJobs++ -- | Use this to dynamically determine if the next job should be picked-up, or+ -- not. This is useful to write custom-logic to determine whether a limited+ -- resource is below a certain usage threshold (eg. CPU usage is below 80%).+ -- __Caveat:__ This feature has not been tested in production, yet.+ | DynamicConcurrency (IO Bool)++instance Show ConcurrencyControl where+ show cc = case cc of+ MaxConcurrentJobs n -> "MaxConcurrentJobs " <> show n+ UnlimitedConcurrentJobs -> "UnlimitedConcurrentJobs"+ DynamicConcurrency _ -> "DynamicConcurrency (IO Bool)"++type JobId = Int++data Status+ -- | In the current version of odd-jobs you /should not/ find any jobs having+ -- the 'Success' status, because successful jobs are immediately deleted.+ -- However, in the future, we may keep them around for a certain time-period+ -- before removing them from the jobs table.+ = Success+ -- | Jobs in 'Queued' status /may/ be picked up by the job-runner on the basis+ -- of the 'jobRunAt' field.+ | Queued+ -- | Jobs in 'Failed' status will will not be retried by the job-runner.+ | Failed+ -- | Jobs in 'Retry' status will be retried by the job-runner on the basis of+ -- the 'jobRunAt' field.+ | Retry+ -- | Jobs in 'Locked' status are currently being executed by a job-runner,+ -- which is identified by the 'jobLockedBy' field. The start of job-execution+ -- is indicated by the 'jobLocketAt' field.+ | Locked+ deriving (Eq, Show, Generic, Enum, Bounded)++instance Ord Status where+ compare x y = compare (toText x) (toText y)++instance ToJSON Status where+ toJSON s = toJSON $ toText s++instance FromJSON Status where+ parseJSON = withText "Expecting text to convert into Job.Status" $ \t -> do+ case (fromText t :: Either String Status) of+ Left e -> fail e+ Right r -> pure r+++newtype JobRunnerName = JobRunnerName { unJobRunnerName :: Text } deriving (Eq, Show, FromField, ToField, Generic, ToJSON, FromJSON)++data Job = Job+ { jobId :: JobId+ , jobCreatedAt :: UTCTime+ , jobUpdatedAt :: UTCTime+ , jobRunAt :: UTCTime+ , jobStatus :: Status+ , jobPayload :: Aeson.Value+ , jobLastError :: Maybe Value+ , jobAttempts :: Int+ , jobLockedAt :: Maybe UTCTime+ , jobLockedBy :: Maybe JobRunnerName+ } deriving (Eq, Show, Generic)++instance ToText Status where+ toText s = case s of+ Success -> "success"+ Queued -> "queued"+ Retry -> "retry"+ Failed -> "failed"+ Locked -> "locked"++instance (StringConv Text a) => FromText (Either a Status) where+ fromText t = case t of+ "success" -> Right Success+ "queued" -> Right Queued+ "failed" -> Right Failed+ "retry" -> Right Retry+ "locked" -> Right Locked+ x -> Left $ toS $ "Unknown job status: " <> x++instance FromField Status where+ fromField f mBS = (fromText <$> (fromField f mBS)) >>= \case+ Left e -> FromField.returnError PGS.ConversionFailed f e+ Right s -> pure s++instance ToField Status where+ toField s = toField $ toText s++instance FromRow Job where+ fromRow = Job+ <$> field -- jobId+ <*> field -- createdAt+ <*> field -- updatedAt+ <*> field -- runAt+ <*> field -- status+ <*> field -- payload+ <*> field -- lastError+ <*> field -- attempts+ <*> field -- lockedAt+ <*> field -- lockedBy++-- TODO: Add a sum-type for return status which can signal the monitor about+-- whether the job needs to be retried, marked successfull, or whether it has+-- completed failed.+type JobRunner = Job -> IO ()++-- | The web\/admin UI needs to know a \"master list\" of all job-types to be+-- able to power the \"filter by job-type\" feature. This data-type helps in+-- letting odd-jobs know /how/ to get such a master-list. The function specified+-- by this type is run once when the job-runner starts (and stored in an+-- internal @IORef@). After that the list of job-types needs to be updated+-- manually by pressing the appropriate \"refresh\" link in the admin\/web UI.+data AllJobTypes+ -- | A fixed-list of job-types. If you don't want to increase boilerplate,+ -- consider using 'OddJobs.ConfigBuilder.defaultConstantJobTypes' which will+ -- automatically generate the list of available job-types based on a sum-type+ -- that represents your job payload.+ = AJTFixed [Text]+ -- | Construct the list of job-types dynamically by looking at the actual+ -- payloads in 'cfgTableName' (using an SQL query).+ | AJTSql (Connection -> IO [Text])+ -- | A custom 'IO' action for fetching the list of job-types.+ | AJTCustom (IO [Text])++-- | While odd-jobs is highly configurable and the 'Config' data-type might seem+-- daunting at first, it is not necessary to tweak every single configuration+-- parameter by hand.+--+-- __Recommendation:__ Please start-off by building a 'Config' by using the+-- 'OddJobs.ConfigBuilder.mkConfig' function (to get something with sensible+-- defaults) and then tweaking config parameters on a case-by-case basis.+data Config = Config+ { -- | The DB table which holds your jobs. Please note, this should have been+ -- created by the 'OddJobs.Migrations.createJobTable' function.+ cfgTableName :: TableName++ -- | The actualy "job-runner" that __you__ need to provide. If this function+ -- throws a runtime exception, the job will be retried+ -- 'cfgDefaultMaxAttempts' times. Please look at the examples/tutorials if+ -- your applicaton's code is not in the @IO@ monad.+ , cfgJobRunner :: Job -> IO ()++ -- | The number of times a failing job is retried before it is considered is+ -- "permanently failed" and ignored by the job-runner. This config parameter+ -- is called "/default/ max attempts" because, in the future, it would be+ -- possible to specify the number of retry-attemps on a per-job basis+ -- (__Note:__ per-job retry-attempts has not been implemented yet)+ , cfgDefaultMaxAttempts :: Int++ -- | Controls how many jobs can be run concurrently by /this instance/ of+ -- the job-runner. __Please note,__ this is NOT the global concurrency of+ -- entire job-queue. It is possible to have job-runners running on multiple+ -- machines, and each will apply the concurrency control independnt of other+ -- job-runners. __Ref:__ Section on [controllng+ -- concurrency](https://www.haskelltutorials.com/odd-jobs/guide.html#controlling-concurrency)+ -- in the implementtion guide.+ , cfgConcurrencyControl :: ConcurrencyControl++ -- | The DB connection-pool to use for the job-runner. __Note:__ in case+ -- your jobs require a DB connection, please create a separate+ -- connection-pool for them. This pool will be used ONLY for monitoring jobs+ -- and changing their status. We need to have __at least 4 connections__ in+ -- this connection-pool for the job-runner to work as expected.+ , cfgDbPool :: Pool Connection++ -- | How frequently should the 'jobPoller' check for jobs where the Job's+ -- 'jobRunAt' field indicates that it's time for the job to be executed.+ -- __Ref:__ Please read the section on [how Odd Jobs works+ -- (architecture)](https://www.haskelltutorials.com/odd-jobs/guide.html#architecture)+ -- to find out more.+ , cfgPollingInterval :: Seconds++ -- | User-defined callback function that is called whenever a job succeeds.+ , cfgOnJobSuccess :: Job -> IO ()++ -- | User-defined error-handler that is called whenever a job fails (indicated+ -- by 'cfgJobRunner' throwing an unhandled runtime exception). Please refer to+ -- 'JobErrHandler' for documentation on how to use this.+ , cfgOnJobFailed :: forall a . [JobErrHandler a]++ -- | User-defined callback function that is called whenever a job starts+ -- execution.+ , cfgOnJobStart :: Job -> IO ()++ -- | User-defined callback function that is called whenever a job times-out.+ -- Also check 'cfgDefaultJobTimeout'+ , cfgOnJobTimeout :: Job -> IO ()++ -- | File to store the PID of the job-runner process. This is used only when+ -- invoking the job-runner as an independent background deemon (the usual mode+ -- of deployment).+ , cfgPidFile :: Maybe FilePath++ -- | A "structured logging" function that __you__ need to provide. The+ -- @odd-jobs@ library does NOT use the standard logging interface provided by+ -- 'monad-logger' on purpose. Also look at 'cfgJobType' and 'defaultLogStr'+ --+ -- __Note:__ Please take a look at the section on [structured+ -- logging](https://www.haskelltutorials.com/odd-jobs/guide.html#structured-logging)+ -- to find out how to use this to log in JSON.+ , cfgLogger :: LogLevel -> LogEvent -> IO ()++ -- | How to extract the "job type" from a 'Job'. If you are overriding this,+ -- please consider overriding 'cfgJobTypeSql' as well. Related:+ -- 'OddJobs.ConfigBuilder.defaultJobType'+ , cfgJobType :: Job -> Text++ -- | How to extract the \"job type\" directly in SQL. There are many places,+ -- especially in the web\/admin UI, where we need to know a job's type+ -- directly in SQL (because transferrring the entire @payload@ column to+ -- Haskell, and then parsing it into JSON, and then applying the+ -- 'cfgJobType' function on it would be too inefficient). Ref:+ -- 'OddJobs.ConfigBuilder.defaultJobTypeSql' and 'cfgJobType'+ , cfgJobTypeSql :: PGS.Query++ -- | How long can a job run after which it is considered to be "crashed" and+ -- picked up for execution again+ , cfgDefaultJobTimeout :: Seconds++ -- | How to convert a list of 'Job's to a list of HTML fragments. This is+ -- used in the Web\/Admin UI. This function accepts a /list/ of jobs and+ -- returns a /list/ of 'Html' fragments, because, in case, you need to query+ -- another table to fetch some metadata (eg. convert a primary-key to a+ -- human-readable name), you can do it efficiently instead of resulting in+ -- an N+1 SQL bug. Ref: 'defaultJobToHtml'+ , cfgJobToHtml :: [Job] -> IO [Html ()]++ -- | How to get a list of all known job-types? This is used by the+ -- Web\/Admin UI to power the \"filter by job-type\" functionality. The+ -- default value for this is 'OddJobs.ConfigBuilder.defaultDynamicJobTypes'+ -- which does a @SELECT DISTINCT payload ->> ...@ to get a list of job-types+ -- directly from the DB.+ , cfgAllJobTypes :: AllJobTypes+ }
src/OddJobs/Web.hs view
@@ -1,12 +1,13 @@-{-# LANGUAGE DeriveGeneric, NamedFieldPuns, TypeOperators, DataKinds #-}+{-# LANGUAGE DeriveGeneric, NamedFieldPuns, TypeOperators, DataKinds, RecordWildCards, DeriveAnyClass #-} module OddJobs.Web where import OddJobs.Types import OddJobs.Job as Job import Data.Time import Data.Aeson as Aeson-import Data.Text as T-import GHC.Generics+import qualified Data.Text as T+import Data.Text (Text)+import GHC.Generics hiding (from, to) import Database.PostgreSQL.Simple as PGS import Database.PostgreSQL.Simple.ToRow as PGS import Database.PostgreSQL.Simple.ToField as PGS@@ -20,7 +21,16 @@ import Servant.API.Generic import Servant.HTML.Lucid import Lucid+import Lucid.Html5+import Lucid.Base import Data.String.Conv+import qualified Data.HashMap.Strict as HM+import Data.List as DL hiding (filter, and)+import Control.Monad+import Data.Time.Format.Human (humanReadableTime')+import Data.Time.Convenience (timeSince, Unit(..), Direction(..))+import Data.Text.Conversions (fromText, toText)+import Prelude hiding (filter, and) data OrderDirection = Asc | Desc deriving (Eq, Show, Generic, Enum) @@ -41,6 +51,7 @@ , filterOrder :: Maybe (OrderByField, OrderDirection) , filterPage :: Maybe (Int, Int) , filterRunAfter :: Maybe UTCTime+ , filterJobRunner :: [JobRunnerName] } deriving (Eq, Show, Generic) instance Semigroup Filter where@@ -54,6 +65,7 @@ , filterOrder = filterOrder b <|> filterOrder a , filterPage = filterPage b <|> filterPage a , filterRunAfter = filterRunAfter b <|> filterRunAfter a+ , filterJobRunner = nub (filterJobRunner b <> filterJobRunner a) } instance Monoid Filter where@@ -70,12 +82,9 @@ , filterOrder = Nothing , filterPage = Just (10, 0) , filterRunAfter = Nothing+ , filterJobRunner = [] } --instance ToJSON Status-instance FromJSON Status- instance ToJSON OrderDirection instance FromJSON OrderDirection instance ToJSON OrderByField@@ -95,15 +104,31 @@ instance ToHttpApiData Filter where toQueryParam x = toS $ Aeson.encode x -data Routes route = Routes- { rFilterResults :: route :- QueryParam "filters" Filter :> Get '[HTML] (Html ())- , rStaticAssets :: route :- "assets" :> Raw- } deriving (Generic)+-- data Routes route = Routes+-- { rFilterResults :: route :- QueryParam "filters" Filter :> Get '[HTML] (Html ())+-- , rStaticAssets :: route :- "assets" :> Raw+-- , rEnqueue :: route :- "enqueue" :> Capture "jobId" JobId :> Post '[HTML] NoContent+-- , rRunNow :: route :- "run" :> Capture "jobId" JobId :> Post '[HTML] NoContent+-- , rCancel :: route :- "cancel" :> Capture "jobId" JobId :> Post '[HTML] NoContent+-- , rRefreshJobTypes :: route :- "refresh-job-types" :> Post '[HTML] NoContent+-- , rRefreshJobRunners :: route :- "refresh-job-runners" :> Post '[HTML] NoContent+-- } deriving (Generic) -filterJobsQuery :: TableName -> Filter -> (PGS.Query, [Action])-filterJobsQuery tname Filter{filterStatuses, filterCreatedBefore, filterCreatedAfter, filterUpdatedBefore, filterUpdatedAfter, filterJobTypes, filterOrder, filterPage, filterRunAfter} =- ( "SELECT " <> Job.concatJobDbColumns <> " FROM " <> tname <> whereClause <> " " <> (orderClause $ fromMaybe (OrdUpdatedAt, Desc) filterOrder) <> " " <> limitOffsetClause+data Routes = Routes+ { rFilterResults :: Maybe Filter -> Text+ , rEnqueue :: JobId -> Text+ , rRunNow :: JobId -> Text+ , rCancel :: JobId -> Text+ , rRefreshJobTypes :: Text+ , rRefreshJobRunners :: Text+ , rStaticAsset :: Text -> Text+ }+++filterJobsQuery :: Config -> Filter -> (PGS.Query, [Action])+filterJobsQuery Config{cfgTableName, cfgJobTypeSql} Filter{..} =+ ( "SELECT " <> Job.concatJobDbColumns <> " FROM " <> cfgTableName <> whereClause <> " " <> (orderClause $ fromMaybe (OrdUpdatedAt, Desc) filterOrder) <> " " <> limitOffsetClause , whereActions ) where@@ -124,9 +149,14 @@ Nothing -> mempty Just (l, o) -> "LIMIT " <> fromString (show l) <> " OFFSET " <> fromString (show o) - (whereClause, whereActions) = case statusClause `and` createdAfterClause `and` createdBeforeClause `and` updatedBeforeClause `and` updatedAfterClause `and` jobTypeClause `and` runAfterClause of- Nothing -> (mempty, toRow ())- Just (q, as) -> (" WHERE " <> q, as)+ (whereClause, whereActions) =+ let finalClause = statusClause `and` createdAfterClause `and`+ createdBeforeClause `and` updatedBeforeClause `and`+ updatedAfterClause `and` jobTypeClause `and`+ runAfterClause `and` jobRunnerClause+ in case finalClause of+ Nothing -> (mempty, toRow ())+ Just (q, as) -> (" WHERE " <> q, as) statusClause = if Prelude.null filterStatuses then Nothing@@ -142,14 +172,18 @@ jobTypeClause = case filterJobTypes of [] -> Nothing xs ->- let qFragment = "payload @> ?"- valBuilder v = toField $ Aeson.object ["tag" .= v]+ let qFragment = "(" <> cfgJobTypeSql <> ")=?" build ys (q, vs) = case ys of [] -> (q, vs)- (y:[]) -> (qFragment <> q, (valBuilder y):vs)- (y:ys_) -> build ys_ (" OR " <> qFragment <> q, (valBuilder y):vs)+ (y:[]) -> (qFragment <> q, (toField y):vs)+ (y:ys_) -> build ys_ (" OR " <> qFragment <> q, (toField y):vs) in Just $ build xs (mempty, []) + jobRunnerClause :: Maybe (Query, [Action])+ jobRunnerClause = case filterJobRunner of+ [] -> Nothing+ xs -> Just ("locked_by in ?", toRow $ Only $ In xs)+ and :: Maybe (Query, [PGS.Action]) -> Maybe (Query, [PGS.Action]) -> Maybe (Query, [PGS.Action]) and Nothing Nothing = Nothing and Nothing (Just (q, as)) = Just (q, as)@@ -158,14 +192,14 @@ -- orderClause = _ -filterJobs :: Connection -> TableName -> Filter -> IO [Job]-filterJobs conn tname f = do- let (q, queryArgs) = filterJobsQuery tname f+filterJobs :: Config -> Connection -> Filter -> IO [Job]+filterJobs cfg conn f = do+ let (q, queryArgs) = filterJobsQuery cfg f PGS.query conn q queryArgs -countJobs :: Connection -> TableName -> Filter -> IO Int-countJobs conn tname f = do- let (q, queryArgs) = filterJobsQuery tname f+countJobs :: Config -> Connection -> Filter -> IO Int+countJobs cfg conn f = do+ let (q, queryArgs) = filterJobsQuery cfg f finalqry = "SELECT count(*) FROM (" <> q <> ") a" [Only r] <- PGS.query conn finalqry queryArgs pure r@@ -177,3 +211,293 @@ -- { filterStatuses = [Job.Success, Queued] -- , filterJobTypes = ["QueuedMail", "ConfirmBooking"] -- }++++pageNav :: Routes -> Html ()+pageNav Routes{..} = do+ div_ $ nav_ [ class_ "navbar navbar-default navigation-clean" ] $ div_ [ class_ "container-fluid" ] $ do+ div_ [ class_ "navbar-header" ] $ do+ a_ [ class_ "navbar-brand navbar-link", href_ "#", style_ "margin-left: 2px; padding: 0px;" ] $ img_ [ src_ $ rStaticAsset "assets/odd-jobs-color-logo.png", title_ "Odd Jobs Logo" ]+ button_ [ class_ "navbar-toggle collapsed", data_ "toggle" "collapse", data_ "target" "#navcol-1" ] $ do+ span_ [ class_ "sr-only" ] $ "Toggle navigation"+ span_ [ class_ "icon-bar" ] $ ""+ span_ [ class_ "icon-bar" ] $ ""+ span_ [ class_ "icon-bar" ] $ ""+ -- div_ [ class_ "collapse navbar-collapse", id_ "navcol-1" ] $ ul_ [ class_ "nav navbar-nav navbar-right" ] $ do+ -- li_ [ class_ "active", role_ "presentation" ] $ a_ [ href_ "#" ] $ "First Item"+ -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Second Item"+ -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Third Item"+ -- li_ [ class_ "dropdown" ] $ do+ -- a_ [ class_ "dropdown-toggle", data_ "toggle" "dropdown", ariaExpanded_ "false", href_ "#" ] $ do+ -- "Dropdown"+ -- span_ [ class_ "caret" ] $ ""+ -- ul_ [ class_ "dropdown-menu", role_ "menu" ] $ do+ -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "First Item"+ -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Second Item"+ -- li_ [ role_ "presentation" ] $ a_ [ href_ "#" ] $ "Third Item"++pageLayout :: Routes -> Html() -> Html () -> Html ()+pageLayout routes@Routes{..} navHtml bodyHtml = do+ doctype_+ html_ $ do+ head_ $ do+ meta_ [ charset_ "utf-8" ]+ meta_ [ name_ "viewport", content_ "width=device-width, initial-scale=1.0" ]+ title_ "haskell-pg-queue"+ link_ [ rel_ "stylesheet", href_ $ rStaticAsset "assets/bootstrap/css/bootstrap.min.css" ]+ link_ [ rel_ "stylesheet", href_ $ rStaticAsset "https://fonts.googleapis.com/css?family=Lato:100i,300,300i,400,700,900" ]+ link_ [ rel_ "stylesheet", href_ $ rStaticAsset "assets/css/styles.css" ]+ body_ $ do+ pageNav routes+ div_ $ div_ [ class_ "container-fluid", style_ "/*background-color:#f2f2f2;*/" ] $ div_ [ class_ "row" ] $ do+ div_ [ class_ "d-none d-md-block col-md-2" ] navHtml+ div_ [ class_ "col-12 col-md-10" ] bodyHtml+ script_ [ src_ $ rStaticAsset "assets/js/jquery.min.js" ] $ ("" :: Text)+ script_ [ src_ $ rStaticAsset "assets/bootstrap/js/bootstrap.min.js" ] $ ("" :: Text)+ script_ [ src_ $ rStaticAsset "assets/js/custom.js" ] $ ("" :: Text)++sideNav :: Routes -> [Text] -> [JobRunnerName] -> UTCTime -> Filter -> Html ()+sideNav Routes{..} jobTypes jobRunnerNames t filter@Filter{..} = do+ div_ [ class_ "filters mt-3" ] $ do+ jobStatusFilters+ jobTypeFilters+ jobRunnerFilters+ where+ jobStatusFilters = do+ h6_ "Filter by job status"+ div_ [ class_ "card" ] $ do+ ul_ [ class_ "list-group list-group-flush" ] $ do+ li_ [ class_ ("list-group-item " <> if filterStatuses == [] then "active-nav" else "") ] $ do+ let lnk = (rFilterResults $ Just filter{filterStatuses = [], filterPage = (OddJobs.Web.filterPage blankFilter)})+ a_ [ href_ lnk ] $ do+ "all"+ -- span_ [ class_ "badge badge-pill badge-secondary float-right" ] "12"+ forM_ ((\\) (enumFrom minBound) [Job.Success]) $ \st -> do+ li_ [ class_ ("list-group-item " <> if (st `elem` filterStatuses) then "active-nav" else "") ] $ do+ let lnk = (rFilterResults $ Just filter{filterStatuses = [st], filterPage = Nothing})+ a_ [ href_ lnk ] $ do+ toHtml $ toText st+ -- span_ [ class_ "badge badge-pill badge-secondary float-right" ] "12"++ jobRunnerFilters = do+ h6_ [ class_ "mt-3" ] $ do+ "Filter by job-runner"+ form_ [ method_ "post", action_ rRefreshJobRunners, class_ "d-inline"] $ do+ button_ [ type_ "submit", class_ "btn btn-link m-0 p-0 ml-1 float-right"] $ do+ small_ "refresh"++ div_ [ class_ "card" ] $ do+ ul_ [ class_ "list-group list-group-flush" ] $ do+ li_ [ class_ ("list-group-item " <> if filterJobRunner == [] then "active-nav" else "") ] $ do+ let lnk = (rFilterResults $ Just filter{filterJobRunner = [], filterPage = (OddJobs.Web.filterPage blankFilter)})+ a_ [ href_ lnk ] "all"+ forM_ jobRunnerNames $ \jr -> do+ li_ [ class_ ("list-group-item" <> if (jr `elem` filterJobRunner) then " active-nav" else "")] $ do+ a_ [ href_ "#" ] $ toHtml $ unJobRunnerName jr++ jobTypeFilters = do+ h6_ [ class_ "mt-3" ] $ do+ "Filter by job-type"+ form_ [ method_ "post", action_ rRefreshJobTypes, class_ "d-inline"] $ do+ button_ [ type_ "submit", class_ "btn btn-link m-0 p-0 ml-1 float-right"] $ do+ small_ "refresh"++ div_ [ class_ "card" ] $ do+ ul_ [ class_ "list-group list-group-flush" ] $ do+ li_ [ class_ ("list-group-item " <> if filterJobTypes == [] then "active-nav" else "") ] $ do+ let lnk = (rFilterResults $ Just filter{filterJobTypes = [], filterPage = (OddJobs.Web.filterPage blankFilter)})+ a_ [ href_ lnk ] "all"+ forM_ jobTypes $ \jt -> do+ li_ [ class_ ("list-group-item" <> if (jt `elem` filterJobTypes) then " active-nav" else "")] $ do+ a_ [ href_ (rFilterResults $ Just filter{filterJobTypes=[jt]}) ] $ toHtml jt++searchBar :: Routes -> UTCTime -> Filter -> Html ()+searchBar Routes{..} t filter@Filter{filterStatuses, filterCreatedAfter, filterCreatedBefore, filterUpdatedAfter, filterUpdatedBefore, filterJobTypes, filterRunAfter} = do+ form_ [ style_ "padding-top: 2em;" ] $ do+ div_ [ class_ "form-group" ] $ do+ div_ [ class_ "search-container" ] $ do+ ul_ [ class_ "list-inline search-bar" ] $ do+ forM_ filterStatuses $ \s -> renderFilter "Status" (toText s) (rFilterResults $ Just filter{filterStatuses = filterStatuses \\ [s]})+ maybe mempty (\x -> renderFilter "Created after" (showText x) (rFilterResults $ Just filter{filterCreatedAfter = Nothing})) filterCreatedAfter+ maybe mempty (\x -> renderFilter "Created before" (showText x) (rFilterResults $ Just filter{filterCreatedBefore = Nothing})) filterCreatedBefore+ maybe mempty (\x -> renderFilter "Updated after" (showText x) (rFilterResults $ Just filter{filterUpdatedAfter = Nothing})) filterUpdatedAfter+ maybe mempty (\x -> renderFilter "Updated before" (showText x) (rFilterResults $ Just filter{filterUpdatedBefore = Nothing})) filterUpdatedBefore+ maybe mempty (\x -> renderFilter "Run after" (showText x) (rFilterResults $ Just filter{filterRunAfter = Nothing})) filterRunAfter+ forM_ filterJobTypes $ \x -> renderFilter "Job type" x (rFilterResults $ Just filter{filterJobTypes = filterJobTypes \\ [x]})++ button_ [ class_ "btn btn-default search-button", type_ "button" ] $ "Search"+ -- ul_ [ class_ "list-inline" ] $ do+ -- li_ $ span_ $ strong_ "Common searches:"+ -- li_ $ a_ [ href_ (rFilterResults $ Just mempty) ] $ "All jobs"+ -- li_ $ a_ [ href_ (rFilterResults $ Just $ filter{ filterStatuses = [Job.Locked] }) ] $ "Currently running"+ -- li_ $ a_ [ href_ (rFilterResults $ Just $ filter{ filterStatuses = [Job.Success] }) ] $ "Successful"+ -- li_ $ a_ [ href_ (rFilterResults $ Just $ filter{ filterStatuses = [Job.Failed] }) ] $ "Failed"+ -- li_ $ a_ [ href_ (rFilterResults $ Just $ filter{ filterRunAfter = Just t }) ] $ "Future"+ -- -- li_ $ a_ [ href_ "#" ] $ "Retried"+ -- li_ $ a_ [ href_ (rFilterResults $ Just $ filter{ filterStatuses = [Job.Queued] }) ] $ "Queued"+ -- li_ $ a_ [ href_ (rFilterResults $ Just $ filter{ filterUpdatedAfter = Just $ timeSince t 10 Minutes Ago }) ] $ "Last 10 mins"+ -- li_ $ a_ [ href_ (rFilterResults $ Just $ filter{ filterCreatedAfter = Just $ timeSince t 10 Minutes Ago }) ] $ "Recently created"+ where+ renderFilter :: Text -> Text -> Text -> Html ()+ renderFilter k v u = do+ li_ [ class_ "search-filter" ] $ do+ span_ [ class_ "filter-name" ] $ toHtml k+ span_ [ class_ "filter-value" ] $ do+ toHtml v+ a_ [ href_ u, class_ "text-danger" ] $ i_ [ class_ "glyphicon glyphicon-remove" ] $ ""+++timeDuration :: UTCTime -> UTCTime -> (Int, String)+timeDuration from to = (diff, str)+ where+ str = if diff <= 0+ then "under 1s"+ else (if d>0 then (show d) <> "d" else "") <>+ (if m>0 then (show m) <> "m" else "") <>+ (if s>0 then (show s) <> "s" else "")+ diff = (abs $ round $ diffUTCTime from to)+ (m', s) = diff `divMod` 60+ (h', m) = m' `divMod` 60+ (d, h) = h' `divMod` 24++showText :: (Show a) => a -> Text+showText a = toS $ show a++jobContent :: Value -> Value+jobContent v = case v of+ Aeson.Object o -> case HM.lookup "contents" o of+ Nothing -> v+ Just c -> c+ _ -> v++jobRow :: Routes -> UTCTime -> (Job, Html ()) -> Html ()+jobRow routes t (job@Job{..}, jobHtml) = do+ tr_ $ do+ td_ [ class_ "job-type" ] $ do+ let statusFn = case jobStatus of+ Job.Success -> statusSuccess+ Job.Failed -> statusFailed+ Job.Queued -> if jobRunAt > t+ then statusFuture+ else statusWaiting+ Job.Retry -> statusRetry+ Job.Locked -> statusLocked+ statusFn t job++ td_ jobHtml+ td_ $ do+ let actionsFn = case jobStatus of+ Job.Success -> (const mempty)+ Job.Failed -> actionsFailed+ Job.Queued -> if jobRunAt > t+ then actionsFuture+ else actionsWaiting+ Job.Retry -> actionsRetry+ Job.Locked -> (const mempty)+ actionsFn routes job+++actionsFailed :: Routes -> Job -> Html ()+actionsFailed Routes{..} Job{..} = do+ form_ [ action_ (rEnqueue jobId), method_ "post" ] $ do+ button_ [ class_ "btn btn-secondary", type_ "submit" ] $ "Enqueue again"++actionsRetry :: Routes -> Job -> Html ()+actionsRetry Routes{..} Job{..} = do+ form_ [ action_ (rRunNow jobId), method_ "post" ] $ do+ button_ [ class_ "btn btn-secondary", type_ "submit" ] $ "Run now"++actionsFuture :: Routes -> Job -> Html ()+actionsFuture Routes{..} Job{..} = do+ form_ [ action_ (rRunNow jobId), method_ "post" ] $ do+ button_ [ class_ "btn btn-secondary", type_ "submit" ] $ "Run now"++actionsWaiting :: Routes -> Job -> Html ()+actionsWaiting Routes{..} Job{..} = do+ form_ [ action_ (rCancel jobId), method_ "post" ] $ do+ button_ [ class_ "btn btn-danger", type_ "submit" ] $ "Cancel"++statusSuccess :: UTCTime -> Job -> Html ()+statusSuccess t Job{..} = do+ span_ [ class_ "badge badge-success" ] $ "Success"+ span_ [ class_ "job-run-time" ] $ do+ let (d, s) = timeDuration jobCreatedAt jobUpdatedAt+ abbr_ [ title_ (showText jobUpdatedAt) ] $ toHtml $ "Completed " <> humanReadableTime' t jobUpdatedAt <> ". "+ abbr_ [ title_ (showText d <> " seconds")] $ toHtml $ "Took " <> s++statusFailed :: UTCTime -> Job -> Html ()+statusFailed t Job{..} = do+ span_ [ class_ "badge badge-danger" ] $ "Failed"+ span_ [ class_ "job-run-time" ] $ do+ abbr_ [ title_ (showText jobUpdatedAt) ] $ toHtml $ "Failed " <> humanReadableTime' t jobUpdatedAt <> " after " <> show jobAttempts <> " attempts"++statusFuture :: UTCTime -> Job -> Html ()+statusFuture t Job{..} = do+ span_ [ class_ "badge badge-secondary" ] $ "Future"+ span_ [ class_ "job-run-time" ] $ do+ abbr_ [ title_ (showText jobRunAt) ] $ toHtml $ humanReadableTime' t jobRunAt++statusWaiting :: UTCTime -> Job -> Html ()+statusWaiting t Job{..} = do+ span_ [ class_ "badge badge-warning" ] $ "Waiting"+ -- span_ [ class_ "job-run-time" ] ("Waiting to be picked up" :: Text)++statusRetry :: UTCTime -> Job -> Html ()+statusRetry t Job{..} = do+ span_ [ class_ "badge badge-warning" ] $ toHtml $ "Retries (" <> show jobAttempts <> ")"+ span_ [ class_ "job-run-time" ] $ do+ abbr_ [ title_ (showText jobUpdatedAt) ] $ toHtml $ "Retried " <> humanReadableTime' t jobUpdatedAt <> ". "+ abbr_ [ title_ (showText jobRunAt)] $ toHtml $ "Next retry in " <> humanReadableTime' t jobRunAt++statusLocked :: UTCTime -> Job -> Html ()+statusLocked t Job{..} = do+ span_ [ class_ "badge badge-info" ] $ toHtml ("Locked" :: Text)+ -- span_ [ class_ "job-run-time" ] $ do+ -- abbr_ [ title_ (showText jobUpdatedAt) ] $ toHtml $ "Retried " <> humanReadableTime' t jobUpdatedAt <> ". "+ -- abbr_ [ title_ (showText jobRunAt)] $ toHtml $ "Next retry in " <> humanReadableTime' t jobRunAt++resultsPanel :: Routes -> UTCTime -> Filter -> [(Job, Html ())] -> Int -> Html ()+resultsPanel routes@Routes{..} t filter@Filter{filterPage} js runningCount = do+ div_ [ class_ "card mt-3" ] $ do+ div_ [ class_ "card-header bg-secondary text-white" ] $ do+ "Currently running "+ span_ [ class_ "badge badge-primary badge-primary" ] $ toHtml (show runningCount)+ div_ [ class_ "currently-running" ] $ div_ [ class_ "" ] $ table_ [ class_ "table table-striped table-hover" ] $ do+ thead_ [ class_ "thead-dark"] $ do+ tr_ $ do+ th_ "Job status"+ th_ "Job"+ th_ [ style_ "min-width: 12em;" ] "Actions"+ tbody_ $ do+ forM_ js (jobRow routes t)+ div_ [ class_ "card-footer" ] $ do+ nav_ $ do+ ul_ [ class_ "pagination" ] $ do+ prevLink+ nextLink+ where+ prevLink = do+ let (extraClass, lnk) = case filterPage of+ Nothing -> ("disabled", "")+ Just (l, 0) -> ("disabled", "")+ Just (l, o) -> ("", rFilterResults $ Just $ filter {filterPage = Just (l, max 0 $ o - l)})+ li_ [ class_ ("page-item previous " <> extraClass) ] $ do+ a_ [ class_ "page-link", href_ lnk ] $ "Prev"++ nextLink = do+ let (extraClass, lnk) = case filterPage of+ Nothing ->+ if (DL.length js) < 10+ then ("disabled", "")+ else ("", (rFilterResults $ Just $ filter {filterPage = Just (10, 10)}))+ Just (l, o) ->+ if (DL.length js) < l+ then ("disabled", "")+ else ("", (rFilterResults $ Just $ filter {filterPage = Just (l, o + l)}))+ li_ [ class_ ("page-item next " <> extraClass) ] $ do+ a_ [ class_ "page-link", href_ lnk ] $ "Next"++ariaExpanded_ :: Text -> Attribute+ariaExpanded_ v = makeAttribute "aria-expanded" v+
test/Test.hs view
@@ -37,9 +37,7 @@ import qualified Data.Text as T import Data.Ord (comparing, Down(..)) import Data.Maybe (fromMaybe)-import qualified System.Random as R-import Data.IORef (IORef(..), atomicModifyIORef', newIORef, readIORef)-import Data.Either (isRight)+ $(Aeson.deriveJSON Aeson.defaultOptions ''Seconds) main :: IO ()@@ -200,6 +198,7 @@ t <- getCurrentTime j1 <- Job.createJob conn tname (PayloadSucceed $ 2 * Job.defaultPollingInterval) j2 <- Job.scheduleJob conn tname (PayloadSucceed 0) (addUTCTime (fromIntegral $ unSeconds $ Job.defaultPollingInterval) t)+ liftIO $ putStrLn "created" pure (j1, j2) Pool.withResource appPool $ \conn -> do delaySeconds 1@@ -222,9 +221,9 @@ testJobFailure appPool jobPool = testCase "job retry" $ withNewJobMonitor jobPool $ \tname -> Pool.withResource appPool $ \conn -> do Job{jobId} <- Job.createJob conn tname (PayloadAlwaysFail 0)- delaySeconds $ Seconds 20+ delaySeconds $ Seconds 15 Job{jobAttempts, jobStatus} <- ensureJobId conn tname jobId- assertEqual "Expecteding job to be in Failed status" Job.Failed jobStatus+ assertEqual "Exepcting job to be in Failed status" Job.Failed jobStatus assertEqual ("Expecting job attempts to be 3. Found " <> show jobAttempts) 3 jobAttempts data JobEvent = JobStart@@ -235,29 +234,12 @@ testEverything appPool jobPool = testProperty "test everything" $ property $ do - jobPayloads <- forAll $ Gen.list (Range.linear 1 1000) payloadGen- jobsRef <- liftIO $ newIORef (Right 0 :: Either String Int, Map.empty :: Map.IntMap [(JobEvent, Job.Job)])- ccRef <- liftIO $ newIORef True- ccControl <- forAll $ genConcurrencyControl ccRef+ jobPayloads <- forAll $ Gen.list (Range.linear 300 1000) payloadGen+ jobsMVar <- liftIO $ newMVar (Map.empty :: Map.IntMap [(JobEvent, Job.Job)]) let maxDelay = sum $ map (payloadDelay jobPollingInterval) jobPayloads completeRun = ((unSeconds maxDelay) `div` concurrencyFactor) + (2 * (unSeconds testPollingInterval)) - onJobEvent :: JobEvent -> (Int -> Int) -> Job.Job -> IO ()- onJobEvent evt jobCountFn job@Job{jobId} =- void $ atomicModifyIORef' jobsRef $ \(ccViolated, jobMap) ->- let newCCViolated = (jobCountFn <$> ccViolated) >>= \activeJobCount ->- case ccControl of- Job.MaxConcurrentJobs mj ->- if activeJobCount > mj- then Left $ "Concurrent jobs should NOT be more than " <> show mj <> ". Found: " <> show activeJobCount- else Right activeJobCount- Job.UnlimitedConcurrentJobs -> Right activeJobCount- Job.DynamicConcurrency _ -> Right activeJobCount- in ( (newCCViolated, Map.insertWith (++) jobId [(evt, job)] jobMap)- , ()- )- shutdownAfter <- forAll $ Gen.choice [ pure $ Seconds completeRun -- Either we allow all jobs to be completed properly , (Seconds <$> (Gen.int $ Range.linear 1 completeRun)) -- Or, we shutdown early after a random number of seconds ]@@ -266,48 +248,37 @@ (defaults, cleanup) <- liftIO $ Test.defaultJobMonitor tname jobPool let jobMonitorSettings = defaults { Job.monitorJobRunner = jobRunner , Job.monitorTableName = tname- , Job.monitorOnJobStart = onJobEvent JobStart (1+)- , Job.monitorOnJobFailed = onJobEvent JobRetry (\x -> x-1)- , Job.monitorOnJobPermanentlyFailed = onJobEvent JobFailed (\x -> x-1)- , Job.monitorOnJobSuccess = onJobEvent JobSuccess (\x -> x-1)+ , Job.monitorOnJobStart = onJobEvent JobStart jobsMVar+ , Job.monitorOnJobFailed = onJobEvent JobRetry jobsMVar+ , Job.monitorOnJobPermanentlyFailed = onJobEvent JobFailed jobsMVar+ , Job.monitorOnJobSuccess = onJobEvent JobSuccess jobsMVar , Job.monitorDefaultMaxAttempts = 3 , Job.monitorPollingInterval = jobPollingInterval- , Job.monitorConcurrencyControl = ccControl } (jobs :: [Job]) <- withAsync- (liftIO $ Job.runJobMonitor jobMonitorSettings)- (const $ finally- (liftIO $ actualTest shutdownAfter jobPayloads tname jobsRef)- (liftIO cleanup))+ (liftIO $ Job.runJobMonitor jobMonitorSettings)+ (const $ finally+ (liftIO $ actualTest shutdownAfter jobPayloads tname jobsMVar)+ (liftIO cleanup)) - (ccViolated, jobAudit) <- liftIO $ readIORef jobsRef+ jobAudit <- takeMVar jobsMVar [(Only (lockedJobCount :: Int))] <- liftIO $ Pool.withResource appPool $ \conn -> PGS.query conn ("SELECT coalesce(count(id), 0) FROM " <> tname <> " where status=?") (Only Job.Locked) - annotate $ "Job-payload count: " <> show (DL.length jobPayloads)- annotate $ "Concurrency control: " <> show ccControl- annotate $ "Shutdown delay (sec): " <> show shutdownAfter- annotate $ "Complete-run time estimate (sec): " <> show completeRun-- -- No job should be in a locked state- 0 === lockedJobCount- -- ALL jobs should show up in the audit, which means they should have -- been attempted /at least/ once- -- when (shutdownAfter >= completeRun) $ (DL.sort $ map jobId jobs) === (DL.sort $ Map.keys jobAudit) + -- No job should be in a locked state+ 0 === lockedJobCount -- No job should've been simultaneously picked-up by more than one -- worker True === (Map.foldl (\m js -> m && noRaceCondition js) True jobAudit) - -- Concurrency control should not have been violated- Hedgehog.assert $ isRight ccViolated-- -- liftIO $ print $ "Test passed with job-count = " <> show (length jobPayloads)+ liftIO $ print $ "Test passed with job-count = " <> show (length jobPayloads) where @@ -318,18 +289,11 @@ jobPollingInterval = Seconds 2 - -- onJobEvent :: JobEvent- -- -> Job.ConcurrencyControl- -- -> IORef (Either String Int)- -- -> IORef (Int, Map.IntMap [(JobEvent, Job.Job)]) -> (Int -> Int) -> Job.Job -> IO ()- -- onJobEvent evt ccViolatedRef jobsRef countFn job@Job{jobId} = do- -- void $ atomicModifyIORef' jobsRef $ \(activeJobCount, jobMap) ->- -- ( ( countFn activeJobCount, Map.insertWith (++) jobId [(evt, job)] jobMap )- -- , ()- -- )+ onJobEvent evt jobsMVar job@Job{jobId} = void $ modifyMVar_ jobsMVar $ \jobMap -> do+ pure $ Map.insertWith (++) jobId [(evt, job)] jobMap - actualTest :: Seconds -> [JobPayload] -> Job.TableName -> IORef (Either String Int, Map.IntMap [(JobEvent, Job.Job)]) -> IO [Job]- actualTest shutdownAfter jobPayloads tname jobsRef = do+ actualTest :: Seconds -> [JobPayload] -> Job.TableName -> MVar (Map.IntMap [(JobEvent, Job.Job)]) -> IO [Job]+ actualTest shutdownAfter jobPayloads tname jobsMVar = do jobs <- forConcurrently jobPayloads $ \pload -> Pool.withResource appPool $ \conn -> liftIO $ Job.createJob conn tname pload@@ -339,13 +303,13 @@ if remaining == Seconds 0 then pure (Right $ Seconds 0) else do delaySeconds testPollingInterval- -- print $ "------- Polling (remaining = " <> show (unSeconds remaining) <> " sec)------"- (_, jobMap) <- liftIO $ readIORef jobsRef- x <- if (Map.foldl (\m js -> m && (isJobTerminalState js)) True jobMap)- then pure (Right $ Seconds 0)- else if remaining < testPollingInterval- then pure (Left $ "Timeout. Job count=" <> show (length jobPayloads) <> " shutdownAfter=" <> show shutdownAfter)- else pure $ Right (remaining - testPollingInterval)+ print $ "------- Polling (remaining = " <> show (unSeconds remaining) <> " sec)------"+ x <- withMVar jobsMVar $ \jobMap ->+ if (Map.foldl (\m js -> m && (isJobTerminalState js)) True jobMap)+ then pure (Right $ Seconds 0)+ else if remaining < testPollingInterval+ then pure (Left $ "Timeout. Job count=" <> show (length jobPayloads) <> " shutdownAfter=" <> show shutdownAfter)+ else pure $ Right (remaining - testPollingInterval) poller x poller (Right shutdownAfter) >>= \case@@ -450,19 +414,7 @@ , filterRunAfter = runAfter } -dynamicConcurrencyFn :: IORef Bool -> IO Bool-dynamicConcurrencyFn ccRef = do- f <- R.randomIO- atomicModifyIORef' ccRef $ const (f, f) -genConcurrencyControl :: MonadGen m => IORef Bool -> m Job.ConcurrencyControl-genConcurrencyControl ccRef = Gen.choice- [ Job.MaxConcurrentJobs <$> (Gen.int $ Range.linear 1 20)- , pure Job.UnlimitedConcurrentJobs- , pure $ Job.DynamicConcurrency $ dynamicConcurrencyFn ccRef- ]-- filterJobs :: Filter -> [Job] -> [Job] filterJobs Web.Filter{filterStatuses, filterCreatedAfter, filterCreatedBefore, filterUpdatedAfter, filterUpdatedBefore, filterOrder, filterPage, filterRunAfter} js = applyLimitOffset $@@ -509,6 +461,6 @@ tcache <- newTimeCache simpleTimeFormat' (tlogger, cleanup) <- newTimedFastLogger tcache LogNone let flogger loc lsource llevel lstr = tlogger $ \t -> toLogStr t <> " | " <> defaultLogStr loc lsource llevel lstr- pure ( Job.defaultJobMonitor flogger tname pool Job.UnlimitedConcurrentJobs+ pure ( Job.defaultJobMonitor flogger tname pool , cleanup )