consumers 2.2.0.6 → 2.3.0.0

raw patch · 5 files changed

+47/−28 lines, 5 filesdep ~basedep ~log-basedep ~monad-timePVP ok

version bump matches the API change (PVP)

Dependency ranges changed: base, log-base, monad-time

API changes (from Hackage documentation)

Files

• CHANGELOG.md +5/−0
• consumers.cabal +4/−5
• example/Example.hs +1/−0
• src/Database/PostgreSQL/Consumers/Config.hs +33/−22
• test/Test.hs +4/−1

@@ -1,3 +1,8 @@+# consumers-2.3.0.0 (2022-07-07)+* Update to monad-time 0.4.0.0+* Support for GHC 9.2+* Drop support for GHC < 8.8+ # consumers-2.2.0.6 (2021-10-22) * Improve efficiency of the SQL query that reserves jobs. 

@@ -1,5 +1,5 @@ name:               consumers-version:            2.2.0.6+version:            2.3.0.0 synopsis:           Concurrent PostgreSQL data consumers  description:        Library for setting up concurrent consumers of data@@ -18,8 +18,7 @@ category:           Concurrency, Database build-type:         Simple cabal-version:      >=1.10-tested-with:        GHC ==8.0.2 || ==8.2.2 || ==8.4.4 || ==8.6.5 || ==8.8.4 || ==8.10.4-                     || ==9.0.1+tested-with:        GHC ==8.8.4 || ==8.10.7 || ==9.0.2 || ==9.2.3  Source-repository head   Type:             git@@ -32,7 +31,7 @@                     Database.PostgreSQL.Consumers.Components,                     Database.PostgreSQL.Consumers.Utils -  build-depends:    base              >= 4.9    && < 5+  build-depends:    base              >= 4.13   && < 5                   , containers        >= 0.5    && < 0.7                   , exceptions        >= 0.10   && < 0.11                   , extra             >= 1.6    && < 1.8@@ -41,7 +40,7 @@                   , lifted-threads    >= 1.0    && < 1.1                   , log-base          >= 0.11   && < 0.12                   , monad-control     >= 1.0    && < 1.1-                  , monad-time        >= 0.3    && < 0.4+                  , monad-time        >= 0.4    && < 0.5                   , mtl               >= 2.2    && < 2.3                   , stm               >= 2.4    && < 2.6                   , time              >= 1.6    && < 2.0

@@ -111,6 +111,7 @@         runSQL_ $ "INSERT INTO consumers_example_jobs "           <> "(run_at, finished_at, reserved_by, attempts, message) "           <> "VALUES (NOW(), NULL, NULL, 0, 'hello')"+        notify "consumers_example_chan" ""         commit        -- Invoked when a job is ready to be processed.

@@ -32,26 +32,29 @@ -- | Name of the database table where jobs are stored. The table needs to have -- the following columns in order to be suitable for acting as a job queue: ----- * id - represents ID of the job. Needs to be a primary key of a type+-- * __id__ - represents ID of the job. Needs to be a primary key of a type -- convertible to text, not nullable. ----- * run_at - represents the time at which the job will be--- processed. Needs to be nullable, of a type comparable with now()+-- * __run_at__ - represents the time at which the job will be+-- processed. Needs to be nullable, of a type comparable with @now()@ -- (TIMESTAMPTZ is recommended).+-- -- Note: a job with run_at set to NULL is never picked for processing. Useful -- for storing already processed/expired jobs for debugging purposes. ----- * finished_at - represents the time at which job processing was finished.--- Needs to be nullable, of a type you can assign now() to (TIMESTAMPTZ is+-- It's highly recommended to have an index on this column.+--+-- * __finished_at__ - represents the time at which job processing was finished.+-- Needs to be nullable, of a type you can assign @now()@ to (TIMESTAMPTZ is -- recommended). NULL means that the job was either never processed or that -- it was started and failed at least once. ----- * reserved_by - represents ID of the consumer that currently processes the+-- * __reserved_by__ - represents ID of the consumer that currently processes the -- job. Needs to be nullable, of the type corresponding to id in the table -- 'ccConsumersTable'. It's recommended (though not neccessary) to make it -- a foreign key referencing id in 'ccConsumersTable' with ON DELETE SET NULL. ----- * attempts - represents number of job processing attempts made so far. Needs+-- * __attempts__ - represents number of job processing attempts made so far. Needs -- to be not nullable, of type INTEGER. Initial value of a fresh job should be -- 0, therefore it makes sense to make the column default to 0. --@@ -59,14 +62,14 @@ -- | Name of a database table where registered consumers are stored. The table -- itself needs to have the following columns: ----- * id - represents ID of a consumer. Needs to be a primary key of the type+-- * __id__ - represents ID of a consumer. Needs to be a primary key of the type -- SERIAL or BIGSERIAL (recommended). ----- * name - represents jobs table of the consumer. Needs to be not nullable,+-- * __name__ - represents jobs table of the consumer. Needs to be not nullable, -- of type TEXT. Allows for tracking consumers of multiple queues with one -- table. Set to 'ccJobsTable'. ----- * last_activity - represents the last registered activity of the consumer.+-- * __last_activity__ - represents the last registered activity of the consumer. -- It's updated periodically by all currently running consumers every 30 -- seconds to prove that they are indeed running. They also check for the -- registered consumers that didn't update their status for a minute. If any@@ -83,19 +86,27 @@ , ccJobFetcher            :: !(row -> job) -- | Selector for taking out job ID from the job object. , ccJobIndex              :: !(job -> idx)--- | Notification channel used for listening for incoming jobs. If set--- to 'Nothing', no listening is performed and jobs are selected from--- the database every 'ccNotificationTimeout' microseconds.+-- | Notification channel used for listening for incoming jobs.+-- Whenever the consumer receives a notification, it checks the database+-- for any pending jobs (@'run_at <= NOW()'@) and runs them all.+-- If set to 'Nothing', no listening is performed and 'ccNotificationTimeout'+-- should be set to a positive number, otherwise no jobs would be ever run.+-- 'ccNotificationChannel' and 'ccNotificationTimeout' can be combined.+-- The consumer will check for pending jobs either when notification is received or+-- no notification is received for 'ccNotificationTimeout' microseconds since+-- the last check. , ccNotificationChannel   :: !(Maybe Channel)--- | Timeout of listening for incoming jobs, in microseconds. The consumer--- checks between the timeouts if there are any jobs in the database that--- needs to be processed, so even if 'ccNotificationChannel' is 'Just', you--- need to set it to a reasonable number if the jobs you process may fail--- and are retried later, as there is no way to signal with a notification--- that a job will need to be performed e.g. in 5 minutes. However, if--- 'ccNotificationChannel' is 'Just' and jobs are never retried, you can--- set it to -1, then listening will never timeout. Otherwise it needs to--- be a positive number.+-- | Timeout of checking for any pending jobs (@'run_at <= NOW()'@), in+-- microseconds. The consumer checks the database for any pending jobs after+-- 'ccNotificationTimeout' microseconds since the last check was performed,+-- runs them until all pending jobs are processed and after that, the cycle+-- repeats. Note that even if 'ccNotificationChannel' is 'Just', you need to set+-- 'ccNotificationTimeout' to a reasonable number if the jobs you process may fail+-- and are retried later, as there is no way to signal with a notification that+-- a job will need to be performed e.g. in 5 minutes. However, if+-- 'ccNotificationChannel' is 'Just' and jobs are never retried, you can set it+-- to -1, then listening will never timeout. Otherwise it needs to be a positive+-- number. , ccNotificationTimeout   :: !Int -- | Maximum amount of jobs that can be processed in parallel. , ccMaxRunningJobs        :: !Int

@@ -41,6 +41,7 @@  data TestEnvSt = TestEnvSt {     teCurrentTime :: UTCTime+  , teMonotonicTime :: Double   }  type InnerTestEnv = StateT TestEnvSt (DBT (LogT IO))@@ -57,6 +58,7 @@  instance MonadTime TestEnv where   currentTime = gets teCurrentTime+  monotonicTime = gets teMonotonicTime  modifyTestTime :: (MonadState TestEnvSt m) => (UTCTime -> UTCTime) -> m () modifyTestTime modtime = modify (\te -> te { teCurrentTime = modtime . teCurrentTime $ te })@@ -65,7 +67,7 @@ runTestEnv connSource logger m =     (runLogT "consumers-test" logger defaultLogLevel)   . (runDBT connSource defaultTransactionSettings)-  . (\m' -> fst <$> (runStateT m' $ TestEnvSt $ UTCTime (ModifiedJulianDay 0) 0))+  . (\m' -> fst <$> (runStateT m' $ TestEnvSt (UTCTime (ModifiedJulianDay 0) 0) 0))   . unTestEnv   $ m @@ -160,6 +162,7 @@         runSQL_ $ "INSERT INTO consumers_test_jobs "           <> "(run_at, finished_at, reserved_by, attempts, countdown) "           <> "VALUES (" <?> now <> " + interval '1 hour', NULL, NULL, 0, " <?> countdown <> ")"+        notify "consumers_test_chan" ""        processJob :: (Int64, Int32) -> TestEnv Result       processJob (_idx, countdown) = do