faktory 1.1.1.0 → 1.1.2.0
raw patch · 11 files changed
+128/−26 lines, 11 files
Files
- CHANGELOG.md +9/−1
- README.lhs +1/−1
- faktory.cabal +2/−1
- library/Faktory/Job.hs +20/−1
- library/Faktory/JobOptions.hs +10/−3
- library/Faktory/Settings.hs +4/−16
- library/Faktory/Settings/Queue.hs +28/−0
- library/Faktory/Worker.hs +6/−3
- tests/Faktory/JobSpec.hs +34/−0
- tests/Faktory/Test.hs +2/−0
- tests/FaktorySpec.hs +12/−0
CHANGELOG.md view
@@ -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)
README.lhs view
@@ -1,6 +1,6 @@ # faktory\_worker\_haskell -[](https://circleci.com/gh/freckle/faktory_worker_haskell)+[](https://github.com/freckle/faktory_worker_haskell/actions/workflows/ci.yml) Haskell client and worker process for the Faktory background job server.
faktory.cabal view
@@ -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
library/Faktory/Job.hs view
@@ -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))
library/Faktory/JobOptions.hs view
@@ -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 }
library/Faktory/Settings.hs view
@@ -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)
+ library/Faktory/Settings/Queue.hs view
@@ -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"
library/Faktory/Worker.hs view
@@ -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
tests/Faktory/JobSpec.hs view
@@ -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
tests/Faktory/Test.hs view
@@ -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
tests/FaktorySpec.hs view
@@ -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"]