diff --git a/CHANGELOG.md b/CHANGELOG.md
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -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)
 
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.2.6
+version:         1.1.2.7
 license:         MIT
 license-file:    LICENSE
 copyright:       2018 Freckle Education
diff --git a/library/Faktory/Job.hs b/library/Faktory/Job.hs
--- a/library/Faktory/Job.hs
+++ b/library/Faktory/Job.hs
@@ -48,6 +48,7 @@
   , jobOptions :: JobOptions
   , jobFailure :: Maybe JobFailure
   }
+  deriving stock (Show, Functor, Foldable, Traversable)
 
 -- | Perform a Job with the given options
 --
diff --git a/library/Faktory/JobFailure.hs b/library/Faktory/JobFailure.hs
--- a/library/Faktory/JobFailure.hs
+++ b/library/Faktory/JobFailure.hs
@@ -15,6 +15,7 @@
   , jfErrorType :: Maybe Text
   , jfBacktrace :: Maybe [Text]
   }
+  deriving stock (Show)
 
 -- brittany-disable-next-binding
 
diff --git a/library/Faktory/Settings.hs b/library/Faktory/Settings.hs
--- a/library/Faktory/Settings.hs
+++ b/library/Faktory/Settings.hs
@@ -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
diff --git a/library/Faktory/Worker.hs b/library/Faktory/Worker.hs
--- a/library/Faktory/Worker.hs
+++ b/library/Faktory/Worker.hs
@@ -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 ()
diff --git a/tests/Faktory/Ent/BatchSpec.hs b/tests/Faktory/Ent/BatchSpec.hs
--- a/tests/Faktory/Ent/BatchSpec.hs
+++ b/tests/Faktory/Ent/BatchSpec.hs
@@ -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
diff --git a/tests/Faktory/Test.hs b/tests/Faktory/Test.hs
--- a/tests/Faktory/Test.hs
+++ b/tests/Faktory/Test.hs
@@ -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
diff --git a/tests/FaktorySpec.hs b/tests/FaktorySpec.hs
--- a/tests/FaktorySpec.hs
+++ b/tests/FaktorySpec.hs
@@ -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