Add ccJobLogData to ConsumerConfig to improve logging (#30)

arybczak · web-flow · commit 20de8777d7c0 · 2025-03-03T13:59:05.000+01:00
diff --git a/consumers/CHANGELOG.md b/consumers/CHANGELOG.md
@@ -1,3 +1,6 @@
+# consumers-2.3.3.0 (2025-??-??)
+* Add `ccJobLogData` to `ConsumerConfig` to improve logging.
+
 # consumers-2.3.2.0 (2024-08-30)
 * Use prepared queries to improve parsing and planning time.
 * Prevent the consumer from crashlooping if `ccOnException` throws.
diff --git a/consumers/consumers.cabal b/consumers/consumers.cabal
@@ -1,6 +1,6 @@
 cabal-version:      3.0
 name:               consumers
-version:            2.3.2.0
+version:            2.3.3.0
 synopsis:           Concurrent PostgreSQL data consumers
 
 description:        Library for setting up concurrent consumers of data
@@ -54,6 +54,7 @@ library
                     Database.PostgreSQL.Consumers.Utils
 
   build-depends:    base              >= 4.14   && < 5
+                  , aeson             >= 2.0
                   , containers        >= 0.5
                   , exceptions        >= 0.10
                   , hpqtypes          >= 1.11
@@ -79,9 +80,7 @@ test-suite consumers-example
                       hpqtypes,
                       hpqtypes-extras,
                       log-base,
-                      text,
-                      text-show,
-                      transformers
+                      text
 
   hs-source-dirs:     example
 
@@ -107,7 +106,6 @@ test-suite consumers-test
                       mtl,
                       stm,
                       text,
-                      text-show,
                       time,
                       transformers,
                       transformers-base
diff --git a/consumers/example/Example.hs b/consumers/example/Example.hs
@@ -21,7 +21,6 @@ import Log
 import Log.Backend.StandardOutput
 import System.Environment
 import System.Exit
-import TextShow
 
 -- | Main application monad. See the 'log-base' and the 'hpqtypes'
 -- packages for documentation on 'DBT' and 'LogT'.
@@ -118,6 +117,7 @@ main = do
         , ccMaxRunningJobs = 1
         , ccProcessJob = processJob
         , ccOnException = handleException
+        , ccJobLogData = \(i, _) -> ["job_id" .= i]
         }
 
     -- Add a job to the consumer's queue.
@@ -141,13 +141,7 @@ main = do
     -- failure in different ways, such as: remove the job from the
     -- queue, mark it as processed, or schedule it for rerun.
     handleException :: SomeException -> (Int64, T.Text) -> AppM Action
-    handleException exc (idx, _msg) = do
-      logAttention "Job failed" $
-        object
-          [ "job_id" .= showt idx
-          , "exception" .= showt exc
-          ]
-      pure . RerunAfter $ imicroseconds 500000
+    handleException _ _ = pure . RerunAfter $ imicroseconds 500000
 
 -- | Table where jobs are stored. See
 -- 'Database.PostgreSQL.Consumers.Config.ConsumerConfig'.
diff --git a/consumers/src/Database/PostgreSQL/Consumers/Components.hs b/consumers/src/Database/PostgreSQL/Consumers/Components.hs
@@ -123,9 +123,17 @@ runConsumerWithMaybeIdleSignal cc0 cs mIdleSignal
   where
     cc =
       cc0
-        { ccOnException = \ex job -> do
+        { ccOnException = \ex job -> localData (ccJobLogData cc0 job) $ do
+            let doOnException = do
+                  action <- ccOnException cc0 ex job
+                  logInfo "Unexpected exception caught while processing job" $
+                    object
+                      [ "exception" .= show ex
+                      , "action" .= show action
+                      ]
+                  pure action
             -- Let asynchronous exceptions through (StopExecution in particular).
-            ccOnException cc0 ex job `ES.catchAny` \handlerEx -> do
+            doOnException `ES.catchAny` \handlerEx -> do
               -- Arbitrary delay, but better than letting exceptions from the
               -- handler through and potentially crashlooping the consumer:
               --
@@ -140,8 +148,7 @@ runConsumerWithMaybeIdleSignal cc0 cs mIdleSignal
               let action = RerunAfter $ idays 1
               logAttention "ccOnException threw an exception" $
                 object
-                  [ "job_id" .= show (ccJobIndex cc0 job)
-                  , "exception" .= show handlerEx
+                  [ "exception" .= show handlerEx
                   , "action" .= show action
                   ]
               pure action
@@ -391,7 +398,7 @@ spawnDispatcher ConsumerConfig {..} cs cid semaphore runningJobsInfo runningJobs
       (_, joinFork) <- mask $ \restore -> T.fork $ do
         tid <- myThreadId
         bracket_ (registerJob tid) (unregisterJob tid) . restore $ do
-          ccProcessJob job
+          localData (ccJobLogData job) $ ccProcessJob job
       pure (job, joinFork)
       where
         registerJob tid = atomically $ do
@@ -406,12 +413,6 @@ spawnDispatcher ConsumerConfig {..} cs cid semaphore runningJobsInfo runningJobs
         Right result -> pure (ccJobIndex job, result)
         Left ex -> do
           action <- ccOnException ex job
-          logAttention "Unexpected exception caught while processing job" $
-            object
-              [ "job_id" .= show (ccJobIndex job)
-              , "exception" .= show ex
-              , "action" .= show action
-              ]
           pure (ccJobIndex job, Failed action)
 
     -- Update status of the jobs.
diff --git a/consumers/src/Database/PostgreSQL/Consumers/Config.hs b/consumers/src/Database/PostgreSQL/Consumers/Config.hs
@@ -5,6 +5,7 @@ module Database.PostgreSQL.Consumers.Config
   ) where
 
 import Control.Exception (SomeException)
+import Data.Aeson.Types qualified as A
 import Data.Time
 import Database.PostgreSQL.PQTypes.FromRow
 import Database.PostgreSQL.PQTypes.Interval
@@ -113,4 +114,6 @@ data ConsumerConfig m idx job = forall row. FromRow row => ConsumerConfig
   -- ^ Action taken if a job processing function throws an exception. For
   -- robustness it's best to ensure that it doesn't throw. If it does, the
   -- exception will be logged and the job in question postponed by a day.
+  , ccJobLogData :: !(job -> [A.Pair])
+  -- ^ Data to attach to each log message while processing a job.
   }
diff --git a/consumers/test/Test.hs b/consumers/test/Test.hs
@@ -21,7 +21,6 @@ import Log.Backend.StandardOutput
 import System.Environment
 import System.Exit
 import Test.HUnit qualified as T
-import TextShow
 
 data TestEnvSt = TestEnvSt
   { teCurrentTime :: UTCTime
@@ -162,6 +161,7 @@ test = do
         , ccMaxRunningJobs = 20
         , ccProcessJob = processJob
         , ccOnException = handleException
+        , ccJobLogData = \(i, _) -> ["job_id" .= i]
         }
 
     putJob :: Int32 -> TestEnv ()
@@ -184,13 +184,7 @@ test = do
       pure (Ok Remove)
 
     handleException :: SomeException -> (Int64, Int32) -> TestEnv Action
-    handleException exc (idx, _countdown) = do
-      logAttention "Job failed" $
-        object
-          [ "job_id" .= showt idx
-          , "exception" .= showt exc
-          ]
-      pure . RerunAfter $ imicroseconds 500000
+    handleException _ _ = pure . RerunAfter $ imicroseconds 500000
 
 jobsTable :: Table
 jobsTable =
