diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,14 @@ -## [*Unreleased*](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.1.0...main) +## [*Unreleased*](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.2.0...main) None + +## [v1.1.2.0](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.1.0...v1.1.2.0) + +- Add `reserveFor` and `jobReserveForMicroseconds` for setting `ACK` window for + individual jobs. +- Timeout jobs that have exceeded their `reserve_for` setting. Jobs without an + explicit `reserve_for` will default to Faktory's 1800 second timeout. +- Allow configuration of default job options via `settingsDefaultJobOptions`. ## [v1.1.1.0](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.0.1...v1.1.1.0) diff --git a/README.lhs b/README.lhs --- a/README.lhs +++ b/README.lhs @@ -1,6 +1,6 @@ # faktory\_worker\_haskell -[![CircleCI](https://circleci.com/gh/freckle/faktory_worker_haskell.svg?style=svg)](https://circleci.com/gh/freckle/faktory_worker_haskell) +[![CI](https://github.com/freckle/faktory_worker_haskell/actions/workflows/ci.yml/badge.svg)](https://github.com/freckle/faktory_worker_haskell/actions/workflows/ci.yml) Haskell client and worker process for the Faktory background job server. diff --git a/faktory.cabal b/faktory.cabal --- a/faktory.cabal +++ b/faktory.cabal @@ -1,6 +1,6 @@ cabal-version: 1.18 name: faktory -version: 1.1.1.0 +version: 1.1.2.0 license: MIT license-file: LICENSE copyright: 2018 Freckle Education @@ -65,6 +65,7 @@ Faktory.Producer Faktory.Protocol Faktory.Settings + Faktory.Settings.Queue Faktory.Worker hs-source-dirs: library diff --git a/library/Faktory/Job.hs b/library/Faktory/Job.hs --- a/library/Faktory/Job.hs +++ b/library/Faktory/Job.hs @@ -5,6 +5,7 @@ , perform , retry , once + , reserveFor , queue , jobtype , at @@ -16,6 +17,7 @@ , jobArg , jobOptions , jobRetriesRemaining + , jobReserveForMicroseconds ) where import Faktory.Prelude @@ -70,13 +72,17 @@ -- | Construct a 'Job' and apply options and Producer settings buildJob :: JobOptions -> Producer -> arg -> IO (Job arg) -buildJob options producer arg = applyOptions namespace options =<< newJob arg +buildJob options producer arg = applyOptions namespace (applyDefaults options) + =<< newJob arg where namespace = connectionInfoNamespace $ settingsConnection $ clientSettings $ producerClient producer + applyDefaults = + mappend $ settingsDefaultJobOptions $ clientSettings $ producerClient + producer -- | Construct a 'Job' with default 'JobOptions' newJob :: arg -> IO (Job arg) @@ -101,6 +107,12 @@ enqueuedRetry = maybe faktoryDefaultRetry getLast $ joRetry $ jobOptions job attemptCount = maybe 0 ((+ 1) . jfRetryCount) $ jobFailure job +jobReserveForMicroseconds :: Job arg -> Int +jobReserveForMicroseconds = + maybe faktoryDefaultReserveFor (secondToMicrosecond . fromIntegral . getLast) + . joReserveFor + . jobOptions + instance ToJSON args => ToJSON (Job args) where toJSON = object . toPairs toEncoding = pairs . mconcat . toPairs @@ -114,6 +126,7 @@ , "retry" .= joRetry jobOptions , "queue" .= joQueue jobOptions , "custom" .= joCustom jobOptions + , "reserve_for" .= joReserveFor jobOptions ] -- brittany-disable-next-binding @@ -134,3 +147,9 @@ -- faktoryDefaultRetry :: Int faktoryDefaultRetry = 25 + +faktoryDefaultReserveFor :: Int +faktoryDefaultReserveFor = secondToMicrosecond 1800 + +secondToMicrosecond :: Int -> Int +secondToMicrosecond n = n * (10 ^ (6 :: Int)) diff --git a/library/Faktory/JobOptions.hs b/library/Faktory/JobOptions.hs --- a/library/Faktory/JobOptions.hs +++ b/library/Faktory/JobOptions.hs @@ -6,6 +6,7 @@ -- * Modifiers , retry , once + , reserveFor , queue , jobtype , at @@ -24,9 +25,10 @@ import Data.Semigroup.Generic import Data.Time import Faktory.Job.Custom -import Faktory.Settings (Namespace, Queue) -import qualified Faktory.Settings as Settings +import Faktory.Settings.Queue (Namespace, Queue) +import qualified Faktory.Settings.Queue as Settings import GHC.Generics +import Numeric.Natural (Natural) -- | Options for the execution of a job -- @@ -48,8 +50,9 @@ , joQueue :: Maybe (Last Queue) , joSchedule :: Maybe (Last (Either UTCTime NominalDiffTime)) , joCustom :: Maybe Custom + , joReserveFor :: Maybe (Last Natural) } - deriving stock Generic + deriving stock (Eq, Show, Generic) deriving (Semigroup, Monoid) via GenericSemigroupMonoid JobOptions -- brittany-disable-next-binding @@ -62,6 +65,7 @@ <*> o .:? "queue" <*> (fmap (Last . Left) <$> o .:? "at") <*> o .:? "custom" + <*> o .:? "reserve_for" getAtFromSchedule :: JobOptions -> IO (Maybe UTCTime) getAtFromSchedule options = for (getLast <$> joSchedule options) $ \case @@ -72,6 +76,9 @@ namespaceQueue namespace options = case joQueue options of Nothing -> options Just (Last q) -> options <> queue (Settings.namespaceQueue namespace q) + +reserveFor :: Natural -> JobOptions +reserveFor n = mempty { joReserveFor = Just $ Last n } retry :: Int -> JobOptions retry n = mempty { joRetry = Just $ Last n } diff --git a/library/Faktory/Settings.hs b/library/Faktory/Settings.hs --- a/library/Faktory/Settings.hs +++ b/library/Faktory/Settings.hs @@ -20,10 +20,9 @@ import Faktory.Prelude import Data.Aeson -import Data.ByteString.Lazy (ByteString, fromStrict) -import Data.String -import Data.Text.Encoding (encodeUtf8) import Faktory.Connection +import Faktory.JobOptions (JobOptions) +import Faktory.Settings.Queue import System.Environment (lookupEnv) import System.IO (hPutStrLn, stderr) import System.Random @@ -32,6 +31,7 @@ { settingsConnection :: ConnectionInfo , settingsLogDebug :: String -> IO () , settingsLogError :: String -> IO () + , settingsDefaultJobOptions :: JobOptions } defaultSettings :: Settings @@ -39,6 +39,7 @@ { settingsConnection = defaultConnectionInfo , settingsLogDebug = \_msg -> pure () , settingsLogError = hPutStrLn stderr . ("[ERROR]: " <>) + , settingsDefaultJobOptions = mempty } -- | Defaults, but read @'Connection'@ from the environment @@ -71,19 +72,6 @@ { settingsQueue = maybe defaultQueue (Queue . pack) mQueue , settingsId = WorkerId <$> mWorkerId } - -newtype Queue = Queue Text - deriving stock (Eq, Show) - deriving newtype (IsString, FromJSON, ToJSON) - -namespaceQueue :: Namespace -> Queue -> Queue -namespaceQueue (Namespace n) (Queue q) = Queue $ mappend n q - -queueArg :: Queue -> ByteString -queueArg (Queue q) = fromStrict $ encodeUtf8 q - -defaultQueue :: Queue -defaultQueue = "default" newtype WorkerId = WorkerId String deriving newtype (FromJSON, ToJSON) diff --git a/library/Faktory/Settings/Queue.hs b/library/Faktory/Settings/Queue.hs new file mode 100644 --- /dev/null +++ b/library/Faktory/Settings/Queue.hs @@ -0,0 +1,28 @@ +module Faktory.Settings.Queue + ( Queue(..) + , namespaceQueue + , queueArg + , defaultQueue + , Namespace(..) + ) where + +import Faktory.Prelude + +import Data.Aeson +import Data.ByteString.Lazy (ByteString, fromStrict) +import Data.String +import Data.Text.Encoding (encodeUtf8) +import Faktory.Connection + +newtype Queue = Queue Text + deriving stock (Eq, Show) + deriving newtype (IsString, FromJSON, ToJSON) + +namespaceQueue :: Namespace -> Queue -> Queue +namespaceQueue (Namespace n) (Queue q) = Queue $ mappend n q + +queueArg :: Queue -> ByteString +queueArg (Queue q) = fromStrict $ encodeUtf8 q + +defaultQueue :: Queue +defaultQueue = "default" diff --git a/library/Faktory/Worker.hs b/library/Faktory/Worker.hs --- a/library/Faktory/Worker.hs +++ b/library/Faktory/Worker.hs @@ -18,10 +18,11 @@ import Data.Aeson.Casing import qualified Data.Text as T import Faktory.Client -import Faktory.Job (Job, JobId, jobArg, jobJid) +import Faktory.Job (Job, JobId, jobArg, jobJid, jobReserveForMicroseconds) import Faktory.Settings import GHC.Generics import GHC.Stack +import System.Timeout (timeout) -- | If processing functions @'throw'@ this, @'runWorker'@ will exit data WorkerHalt = WorkerHalt @@ -90,8 +91,10 @@ let namespace = connectionInfoNamespace $ settingsConnection settings processAndAck job = do - f job - ackJob client job + mResult <- timeout (jobReserveForMicroseconds job) $ f job + case mResult of + Nothing -> settingsLogError settings "Job reservation period expired." + Just () -> ackJob client job emJob <- fetchJob client $ namespaceQueue namespace $ settingsQueue workerSettings diff --git a/tests/Faktory/JobSpec.hs b/tests/Faktory/JobSpec.hs --- a/tests/Faktory/JobSpec.hs +++ b/tests/Faktory/JobSpec.hs @@ -10,6 +10,8 @@ import Data.Aeson.QQ import Data.Time (getCurrentTime) import Faktory.Job +import Faktory.Producer +import Faktory.Settings import Test.Hspec spec :: Spec @@ -123,6 +125,38 @@ |] jobRetriesRemaining job `shouldBe` 0 + + it "handles reserve_for" $ do + job <- decodeJob [aesonQQ| + { "jid": "abc" + , "args": [""] + , "reserve_for": 3600 + } + |] + + jobReserveForMicroseconds job `shouldBe` 3600000000 + + describe "buildJob" $ do + it "defaults to jobtype 'default'" $ do + bracket (newProducer defaultSettings) closeProducer $ \producer -> do + job <- buildJob @Text mempty producer "text" + + jobOptions job `shouldBe` jobtype "Default" + + it "adds job option defaults from settings" $ do + let settings = defaultSettings { settingsDefaultJobOptions = retry 5 } + bracket (newProducer settings) closeProducer $ \producer -> do + job <- buildJob @Text mempty producer "text" + + jobOptions job `shouldBe` (jobtype "Default" <> retry 5) + + it "doesn't prefers explicit job options over defaults in settings" $ do + let settings = defaultSettings { settingsDefaultJobOptions = retry 5 } + bracket (newProducer settings) closeProducer $ \producer -> do + job <- buildJob @Text (retry 88) producer "text" + + jobOptions job `shouldBe` (jobtype "Default" <> retry 88) + decodeJob :: Value -> IO (Job Text) decodeJob v = case fromJSON v of diff --git a/tests/Faktory/Test.hs b/tests/Faktory/Test.hs --- a/tests/Faktory/Test.hs +++ b/tests/Faktory/Test.hs @@ -18,6 +18,7 @@ import Faktory.Producer as X import Test.Hspec as X +import Control.Concurrent (threadDelay) import Control.Concurrent.Async import Control.Concurrent.MVar import Faktory.Settings @@ -57,6 +58,7 @@ runWorker settings workerSettings $ \faktoryJob -> do let job = jobArg faktoryJob + when (job == "WAIT") $ threadDelay 3000000 modifyMVar_ processedJobs $ pure . (job :) when (job == "BOOM") $ throw $ userError "BOOM" when (job == "HALT") $ throw WorkerHalt diff --git a/tests/FaktorySpec.hs b/tests/FaktorySpec.hs --- a/tests/FaktorySpec.hs +++ b/tests/FaktorySpec.hs @@ -51,3 +51,15 @@ threadDelay $ 2 * 1000000 + 250000 jobs `shouldMatchList` ["HALT"] + + it "does not process jobs when reserve_for timeout expires" $ do + jobs <- workerTestCase $ \producer -> do + void $ perform @Text (reserveFor 1) producer "WAIT" + + jobs `shouldMatchList` ["HALT"] + + it "processes jobs within reserve_for window" $ do + jobs <- workerTestCase $ \producer -> do + void $ perform @Text (reserveFor 4) producer "WAIT" + + jobs `shouldMatchList` ["WAIT", "HALT"]