faktory 1.1.2.6 → 1.1.2.7

raw patch · 9 files changed

+53/−19 lines, 9 filesPVP: major bump suggested

API removals or changes: PVP suggests a major version bump

API changes (from Hackage documentation)

Files

• CHANGELOG.md +6/−1
• faktory.cabal +1/−1
• library/Faktory/Job.hs +1/−0
• library/Faktory/JobFailure.hs +1/−0
• library/Faktory/Settings.hs +2/−0
• library/Faktory/Worker.hs +9/−4
• tests/Faktory/Ent/BatchSpec.hs +2/−2
• tests/Faktory/Test.hs +19/−11
• tests/FaktorySpec.hs +12/−0

@@ -1,4 +1,9 @@-## [_Unreleased_](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.2.6...main)+## [_Unreleased_](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.2.7...main)++## [v1.1.2.7](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.2.6...v1.1.2.7)++- Fix handling decoding errors of the Job payload. Previously, it would log an+  error and neither `ACK` nor `FAIL` the Job. Now it `FAIL`s the Job.  ## [v1.1.2.6](https://github.com/frontrowed/faktory_worker_haskell/compare/v1.1.2.5...v1.1.2.6) 

@@ -1,6 +1,6 @@ cabal-version:   1.18 name:            faktory-version:         1.1.2.6+version:         1.1.2.7 license:         MIT license-file:    LICENSE copyright:       2018 Freckle Education

@@ -48,6 +48,7 @@   , jobOptions :: JobOptions   , jobFailure :: Maybe JobFailure   }+  deriving stock (Show, Functor, Foldable, Traversable)  -- | Perform a Job with the given options --

@@ -15,6 +15,7 @@   , jfErrorType :: Maybe Text   , jfBacktrace :: Maybe [Text]   }+  deriving stock (Show)  -- brittany-disable-next-binding 

@@ -55,6 +55,7 @@   { settingsQueue :: Queue   , settingsId :: Maybe WorkerId   , settingsIdleDelay :: Int+  , settingsOnFailed :: SomeException -> IO ()   }  defaultWorkerSettings :: WorkerSettings@@ -63,6 +64,7 @@     { settingsQueue = defaultQueue     , settingsId = Nothing     , settingsIdleDelay = 1+    , settingsOnFailed = \_ -> pure ()     }  envWorkerSettings :: IO WorkerSettings

@@ -15,6 +15,7 @@ import Control.Concurrent (killThread) import Data.Aeson import Data.Aeson.Casing+import Data.Aeson.Types (parseEither) import qualified Data.Text as T import Faktory.Client import Faktory.Job (Job, JobId, jobArg, jobJid, jobReserveForMicroseconds)@@ -89,7 +90,8 @@ processorLoop client settings workerSettings f = do   let     namespace = connectionInfoNamespace $ settingsConnection settings-    processAndAck job = do+    processAndAck job' = do+      job <- decodeJob job'       mResult <- timeout (jobReserveForMicroseconds job) $ f job       case mResult of         Nothing -> settingsLogError settings "Job reservation period expired."@@ -107,18 +109,21 @@     Right (Just job) ->       processAndAck job         `catches` [ Handler $ \(ex :: WorkerHalt) -> throw ex-                  , Handler $ \(ex :: SomeException) ->+                  , Handler $ \(ex :: SomeException) -> do+                      settingsOnFailed workerSettings ex                       failJob client job $ T.pack $ show ex                   ] +decodeJob :: (HasCallStack, FromJSON arg) => Job Value -> IO (Job arg)+decodeJob = either throwString pure . traverse (parseEither parseJSON)+ -- | <https://github.com/contribsys/faktory/wiki/Worker-Lifecycle#heartbeat> heartBeat :: Client -> WorkerId -> IO () heartBeat client workerId = do   threadDelaySeconds 25   command_ client "BEAT" [encode $ BeatPayload workerId] -fetchJob-  :: FromJSON args => Client -> Queue -> IO (Either String (Maybe (Job args)))+fetchJob :: Client -> Queue -> IO (Either String (Maybe (Job Value))) fetchJob client queue = commandJSON client "FETCH" [queueArg queue]  ackJob :: HasCallStack => Client -> Job args -> IO ()

@@ -15,7 +15,7 @@ spec = do   describe "jobBatchId" $ do     it "parses the batchId out of batch jobs" $ do-      (jobs, batchId) <- withWorker id $ withProducer $ \producer -> do+      (jobs, _, batchId) <- withWorker id $ withProducer $ \producer -> do         c <- buildJob @Text mempty producer "c"         d <- buildJob @Text mempty producer "d"         let options = description "foo" <> complete c <> success d@@ -100,7 +100,7 @@       fmap jobArg jobs `shouldMatchList` ["a", "b", "c", "d", "HALT"]      it "supports BATCH STATUS" $ do-      (_, batchId) <- withWorker id $ withProducer $ \producer -> do+      (_, _, batchId) <- withWorker id $ withProducer $ \producer -> do         c <- buildJob @Text mempty producer "c"         d <- buildJob @Text mempty producer "d"         let options = description "foo" <> complete c <> success d

@@ -1,5 +1,3 @@-{-# LANGUAGE TupleSections #-}- module Faktory.Test   ( module X   , workerTestCase@@ -34,10 +32,8 @@   => (WorkerSettings -> WorkerSettings)   -> (Producer -> IO ())   -> IO [Job Text]-workerTestCaseWith editSettings run = do-  a <- startWorker editSettings-  withProducer run-  haltWorker a+workerTestCaseWith editSettings =+  fmap (\(x, _, _) -> x) . withWorker editSettings . withProducer  withProducer :: (Producer -> IO a) -> IO a withProducer f = bracket newProducerEnv closeProducer f@@ -46,29 +42,41 @@   :: HasCallStack   => (WorkerSettings -> WorkerSettings)   -> IO a-  -> IO ([Job Text], a)+  -> IO ([Job Text], [SomeException], a) withWorker editSettings f = do   a <- startWorker editSettings   result <- f-  (,result) <$> haltWorker a+  (processed, failed) <- haltWorker a+  pure (processed, failed, result)  startWorker-  :: HasCallStack => (WorkerSettings -> WorkerSettings) -> IO (Async [Job Text])+  :: HasCallStack+  => (WorkerSettings -> WorkerSettings)+  -> IO (Async ([Job Text], [SomeException])) startWorker editSettings = do   withProducer $ void . flush   settings <- envSettings   workerSettings <- editSettings <$> envWorkerSettings   async $ do     processedJobs <- newMVar []+    failedJobs <- newMVar [] -    runWorker settings workerSettings $ \faktoryJob -> do+    let workerSettings' =+          workerSettings+            { settingsOnFailed = \ex ->+                modifyMVar_ failedJobs $ pure . (ex :)+            }++    runWorker settings workerSettings' $ \faktoryJob -> do       let job = jobArg faktoryJob       when (job == "WAIT") $ threadDelay 3000000       modifyMVar_ processedJobs $ pure . (faktoryJob :)       when (job == "BOOM") $ throw $ userError "BOOM"       when (job == "HALT") $ throw WorkerHalt -    readMVar processedJobs+    (,)+      <$> readMVar processedJobs+      <*> readMVar failedJobs  haltWorker :: Async a -> IO a haltWorker a = do

@@ -63,3 +63,15 @@       void $ perform @Text (reserveFor 4) producer "WAIT"      fmap jobArg jobs `shouldMatchList` ["WAIT", "HALT"]++  it "fails jobs that raise exceptions" $ do+    (_, failed, _) <- withWorker id $ withProducer $ \producer -> do+      void $ perform @Text mempty producer "BOOM"++    failed `shouldSatisfy` (== 1) . length++  it "fails jobs that didn't parse" $ do+    (_, failed, _) <- withWorker id $ withProducer $ \producer -> do+      void $ perform @Int mempty producer 42++    failed `shouldSatisfy` (== 1) . length