consumers 2.3.2.0 → 2.3.3.0
raw patch · 8 files changed
+37/−37 lines, 8 filesdep +aesondep −text-showPVP: major bump suggested
API removals or changes: PVP suggests a major version bump
Dependencies added: aeson
Dependencies removed: text-show
API changes (from Hackage documentation)
+ Database.PostgreSQL.Consumers.Config: [ccJobLogData] :: ConsumerConfig m idx job -> !job -> [Pair]
- Database.PostgreSQL.Consumers.Config: ConsumerConfig :: !RawSQL () -> !RawSQL () -> ![SQL] -> !row -> job -> !job -> idx -> !Maybe Channel -> !Int -> !Int -> !job -> m Result -> !SomeException -> job -> m Action -> ConsumerConfig m idx job
+ Database.PostgreSQL.Consumers.Config: ConsumerConfig :: !RawSQL () -> !RawSQL () -> ![SQL] -> !row -> job -> !job -> idx -> !Maybe Channel -> !Int -> !Int -> !job -> m Result -> !SomeException -> job -> m Action -> !job -> [Pair] -> ConsumerConfig m idx job
Files
- CHANGELOG.md +3/−0
- README.md +5/−1
- consumers.cabal +4/−6
- example/Example.hs +2/−8
- src/Database/PostgreSQL/Consumers/Components.hs +16/−11
- src/Database/PostgreSQL/Consumers/Config.hs +3/−0
- src/Database/PostgreSQL/Consumers/Utils.hs +2/−3
- test/Test.hs +2/−8
CHANGELOG.md view
@@ -1,3 +1,6 @@+# consumers-2.3.3.0 (2025-03-03)+* 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.
README.md view
@@ -1,9 +1,13 @@ # consumers [](https://hackage.haskell.org/package/consumers)-[](https://github.com/scrive/consumers/actions?query=branch%3Amaster)+[](https://github.com/scrive/consumers/actions/workflows/haskell-ci.yml) Library for setting up concurrent consumers of data stored inside PostgreSQL database in a simple, declarative manner. See the `examples/` directory for a usage example.++If you want to add metrics, see the+[`consumers-metrics-prometheus`](https://hackage.haskell.org/package/consumers-metrics-prometheus)+package to seamlessly instrument your consumer.
consumers.cabal view
@@ -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@@ -17,7 +17,7 @@ copyright: Scrive AB category: Concurrency, Database build-type: Simple-tested-with: GHC == { 8.10.7, 9.0.2, 9.2.8, 9.4.8, 9.6.6, 9.8.2, 9.10.1 }+tested-with: GHC == { 8.10.7, 9.0.2, 9.2.8, 9.4.8, 9.6.6, 9.8.4, 9.10.1, 9.12.1 } bug-reports: https://github.com/scrive/consumers/issues source-repository head@@ -54,6 +54,7 @@ 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 @@ hpqtypes, hpqtypes-extras, log-base,- text,- text-show,- transformers+ text hs-source-dirs: example @@ -107,7 +106,6 @@ mtl, stm, text,- text-show, time, transformers, transformers-base
example/Example.hs view
@@ -21,7 +21,6 @@ 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 @@ , ccMaxRunningJobs = 1 , ccProcessJob = processJob , ccOnException = handleException+ , ccJobLogData = \(i, _) -> ["job_id" .= i] } -- Add a job to the consumer's queue.@@ -141,13 +141,7 @@ -- 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'.
src/Database/PostgreSQL/Consumers/Components.hs view
@@ -32,6 +32,10 @@ -- | Run the consumer. The purpose of the returned monadic action is to wait for -- currently processed jobs and clean up. This function is best used in -- conjunction with 'finalize' to seamlessly handle the finalization.+--+-- If you want to add metrics, see the+-- [@consumers-metrics-prometheus@](https://hackage.haskell.org/package/consumers-metrics-prometheus)+-- package to seamlessly instrument your consumer. runConsumer :: ( MonadBaseControl IO m , MonadLog m@@ -119,9 +123,17 @@ 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: --@@ -136,8 +148,7 @@ 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@@ -387,7 +398,7 @@ (_, 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@@ -402,12 +413,6 @@ 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.
src/Database/PostgreSQL/Consumers/Config.hs view
@@ -5,6 +5,7 @@ ) 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 @@ -- ^ 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. }
src/Database/PostgreSQL/Consumers/Utils.hs view
@@ -16,7 +16,6 @@ import Control.Monad.Trans.Control import Data.Maybe import Data.Text qualified as T-import Data.Typeable import Database.PostgreSQL.PQTypes.Class import Database.PostgreSQL.PQTypes.SQL.Raw @@ -36,7 +35,7 @@ -- All exceptions other than 'StopExecution' thrown to threads spawned by -- 'forkP' and 'gforkP' are propagated back to the parent thread. data StopExecution = StopExecution- deriving (Show, Typeable)+ deriving (Show) instance Exception StopExecution where toException = E.asyncExceptionToException@@ -44,7 +43,7 @@ -- | Exception thrown from a child thread. data ThrownFrom = ThrownFrom String SomeException- deriving (Show, Typeable)+ deriving (Show) instance Exception ThrownFrom
test/Test.hs view
@@ -21,7 +21,6 @@ import System.Environment import System.Exit import Test.HUnit qualified as T-import TextShow data TestEnvSt = TestEnvSt { teCurrentTime :: UTCTime@@ -162,6 +161,7 @@ , ccMaxRunningJobs = 20 , ccProcessJob = processJob , ccOnException = handleException+ , ccJobLogData = \(i, _) -> ["job_id" .= i] } putJob :: Int32 -> TestEnv ()@@ -184,13 +184,7 @@ 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 =