diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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. diff --git a/consumers.cabal b/consumers.cabal --- a/consumers.cabal +++ b/consumers.cabal @@ -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 diff --git a/example/Example.hs b/example/Example.hs --- a/example/Example.hs +++ b/example/Example.hs @@ -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. diff --git a/src/Database/PostgreSQL/Consumers/Config.hs b/src/Database/PostgreSQL/Consumers/Config.hs --- a/src/Database/PostgreSQL/Consumers/Config.hs +++ b/src/Database/PostgreSQL/Consumers/Config.hs @@ -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 diff --git a/test/Test.hs b/test/Test.hs --- a/test/Test.hs +++ b/test/Test.hs @@ -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