diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,7 +1,12 @@ +# consumers-2.3.1.0 (2023-11-27) +* Drop support for advisory locks. +* Add support for GHC 9.6 and 9.8. +* Apply `ccOnException` when freeing jobs reserved by an inactive consumer. + # 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 +* 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. @@ -22,8 +27,8 @@ * Support hpqtypes-1.7.0.0 and hpqtypes-extras-1.9.0.0. # consumers-2.2.0.0 (2019-02-19) -* When jobs are relased from a dead consumer, reset their finished_at column -* Start the consumer only when ccMaxRunningJobs >= 1 +* When jobs are relased from a dead consumer, reset their finished_at column. +* Start the consumer only when ccMaxRunningJobs >= 1. # consumers-2.1.2.0 (2018-07-11) diff --git a/consumers.cabal b/consumers.cabal --- a/consumers.cabal +++ b/consumers.cabal @@ -1,5 +1,5 @@ name: consumers -version: 2.3.0.0 +version: 2.3.1.0 synopsis: Concurrent PostgreSQL data consumers description: Library for setting up concurrent consumers of data @@ -18,7 +18,8 @@ category: Concurrency, Database build-type: Simple cabal-version: >=1.10 -tested-with: GHC ==8.8.4 || ==8.10.7 || ==9.0.2 || ==9.2.3 +tested-with: GHC ==8.8.4 || ==8.10.7 || ==9.0.2 || ==9.2.8 || ==9.4.7 || ==9.6.3 + || ==9.8.1 Source-repository head Type: git @@ -32,19 +33,18 @@ Database.PostgreSQL.Consumers.Utils build-depends: base >= 4.13 && < 5 - , containers >= 0.5 && < 0.7 - , exceptions >= 0.10 && < 0.11 - , extra >= 1.6 && < 1.8 - , hpqtypes >= 1.7 && < 2.0 - , lifted-base >= 0.2 && < 0.3 - , lifted-threads >= 1.0 && < 1.1 - , log-base >= 0.11 && < 0.12 - , monad-control >= 1.0 && < 1.1 - , monad-time >= 0.4 && < 0.5 - , mtl >= 2.2 && < 2.3 - , stm >= 2.4 && < 2.6 - , time >= 1.6 && < 2.0 - , transformers-base >= 0.4 && < 0.5 + , containers >= 0.5 + , exceptions >= 0.10 + , hpqtypes >= 1.11 + , lifted-base >= 0.2 + , lifted-threads >= 1.0 + , log-base >= 0.11 + , monad-control >= 1.0 + , monad-time >= 0.4 + , mtl >= 2.2 + , stm >= 2.4 + , time >= 1.6 + , transformers-base >= 0.4 hs-source-dirs: src @@ -54,12 +54,14 @@ default-extensions: DeriveDataTypeable , FlexibleContexts , GeneralizedNewtypeDeriving + , LambdaCase , NoImplicitPrelude , OverloadedStrings , RankNTypes , RecordWildCards , ScopedTypeVariables , TupleSections + , TypeApplications , TypeFamilies , UndecidableInstances @@ -74,7 +76,7 @@ build-depends: base, consumers, hpqtypes, - hpqtypes-extras >= 1.9 && < 2.0, + hpqtypes-extras, log-base, text, text-show, diff --git a/example/Example.hs b/example/Example.hs --- a/example/Example.hs +++ b/example/Example.hs @@ -44,7 +44,7 @@ ConnectionSource connSource = simpleSource connSettings -- Monad stack initialisation. - withSimpleStdOutLogger $ \logger -> + withStdOutLogger $ \logger -> runLogT "consumers-example" logger defaultLogLevel $ runDBT connSource defaultTransactionSettings $ do diff --git a/src/Database/PostgreSQL/Consumers/Components.hs b/src/Database/PostgreSQL/Consumers/Components.hs --- a/src/Database/PostgreSQL/Consumers/Components.hs +++ b/src/Database/PostgreSQL/Consumers/Components.hs @@ -75,13 +75,14 @@ skipLockedTest :: Either DBException () <- try . runDBT cs defaultTransactionSettings $ runSQL_ "SELECT TRUE FOR UPDATE SKIP LOCKED" - let useSkipLocked = either (const False) (const True) skipLockedTest + -- If we can't lock rows using 'skip locked' throw an exception + either (const $ error "PostgreSQL version with support for SKIP LOCKED is required") pure skipLockedTest cid <- registerConsumer cc cs localData ["consumer_id" .= show cid] $ do listener <- spawnListener cc cs semaphore monitor <- localDomain "monitor" $ spawnMonitor cc cs cid - dispatcher <- localDomain "dispatcher" $ spawnDispatcher cc useSkipLocked + dispatcher <- localDomain "dispatcher" $ spawnDispatcher cc cs cid semaphore runningJobsInfo runningJobs mIdleSignal return . localDomain "finalizer" $ do stopExecution listener @@ -148,7 +149,7 @@ -- for activity and periodically updates its own. spawnMonitor :: forall m idx job. (MonadBaseControl IO m, MonadLog m, MonadMask m, MonadTime m, - Show idx, FromSQL idx) + Show idx, FromSQL idx, ToSQL idx) => ConsumerConfig m idx job -> ConnectionSourceM m -> ConsumerID @@ -168,35 +169,42 @@ else do logInfo_ $ "Consumer is not registered" throwM ThreadKilled - -- Freeing jobs locked by inactive consumers needs to happen - -- exactly once, otherwise it's possible to free it twice, after - -- it was already marked as reserved by other consumer, so let's - -- run it in serializable transaction. - (inactiveConsumers, freedJobs) <- runDBT cs tsSerializable $ do + (inactiveConsumers, freedJobs) <- runDBT cs ts $ do now <- currentTime - -- Delete all inactive (assumed dead) consumers and get their ids. - runSQL_ $ smconcat [ - "DELETE FROM" <+> raw ccConsumersTable - , " WHERE last_activity +" iminutes 1 <+> "<= " now - , " AND name =" unRawSQL ccJobsTable - , " RETURNING id::bigint" + -- Reserve all inactive (assumed dead) consumers and get their ids. We don't + -- delete them here, because if the coresponding reserved_by column in the + -- jobs table has an IMMEDIATE foreign key with the ON DELETE SET NULL + -- property, we will not be able to determine stuck jobs in the next step. + runSQL_ $ smconcat + [ "SELECT id::bigint" + , "FROM" <+> raw ccConsumersTable + , "WHERE last_activity +" iminutes 1 <+> "<= " now + , " AND name =" unRawSQL ccJobsTable + , "FOR UPDATE SKIP LOCKED" ] - inactive :: [Int64] <- fetchMany runIdentity - -- Reset reserved jobs manually, do not rely on the foreign key constraint - -- to do its job. We also reset finished_at to correctly bump number of - -- attempts on the next try. - freedJobs :: [idx] <- if null inactive - then return [] - else do + fetchMany (runIdentity @Int64) >>= \case + [] -> pure (0, []) + inactive -> do + -- Fetch all stuck jobs and run ccOnException on them to determine + -- actions. This is necessary e.g. to be able to apply exponential + -- backoff to them correctly. runSQL_ $ smconcat - [ "UPDATE" <+> raw ccJobsTable - , "SET reserved_by = NULL" - , " , finished_at = NULL" + [ "SELECT" <+> mintercalate ", " ccJobSelectors + , "FROM" <+> raw ccJobsTable , "WHERE reserved_by = ANY(" Array1 inactive <+> ")" - , "RETURNING id" + , "FOR UPDATE SKIP LOCKED" ] - fetchMany runIdentity - return (length inactive, freedJobs) + stuckJobs <- fetchMany ccJobFetcher + unless (null stuckJobs) $ do + results <- forM stuckJobs $ \job -> do + action <- lift $ ccOnException (toException ThreadKilled) job + pure (ccJobIndex job, Failed action) + runSQL_ $ updateJobsQuery ccJobsTable results now + runSQL_ $ smconcat + [ "DELETE FROM" <+> raw ccConsumersTable + , "WHERE id = ANY(" Array1 inactive <+> ")" + ] + pure (length inactive, map ccJobIndex stuckJobs) when (inactiveConsumers > 0) $ do logInfo "Unregistered inactive consumers" $ object [ "inactive_consumers" .= inactiveConsumers @@ -206,17 +214,12 @@ "freed_jobs" .= map show freedJobs ] liftBase . threadDelay $ 30 * 1000000 -- wait 30 seconds - where - tsSerializable = ts { - tsIsolationLevel = Serializable - } -- | Spawn a thread that reserves and processes jobs. spawnDispatcher :: forall m idx job. ( MonadBaseControl IO m, MonadLog m, MonadMask m, MonadTime m , Show idx, ToSQL idx ) => ConsumerConfig m idx job - -> Bool -> ConnectionSourceM m -> ConsumerID -> MVar () @@ -224,7 +227,7 @@ -> TVar Int -> Maybe (TMVar Bool) -> m ThreadId -spawnDispatcher ConsumerConfig{..} useSkipLocked cs cid semaphore +spawnDispatcher ConsumerConfig{..} cs cid semaphore runningJobsInfo runningJobs mIdleSignal = forkP "dispatcher" . forever $ do void $ takeMVar semaphore @@ -288,29 +291,13 @@ reservedJobs :: UTCTime -> SQL reservedJobs now = smconcat [ "SELECT id FROM" <+> raw ccJobsTable - -- Converting id to text and hashing it may seem silly, - -- especially when we're dealing with integers in the - -- first place, but even in such case the overhead is - -- small enough (converting 100k integers to text and - -- hashing them takes around 15 ms on i7) to be worth the - -- generality. Note that even if IDs of two pending jobs - -- produce the same hash, it just means that in the worst - -- case they will be processed by the same consumer. - , if useSkipLocked - then "WHERE TRUE" - else "WHERE pg_try_advisory_xact_lock(" - unRawSQL ccJobsTable - <> "::regclass::integer, hashtext(id::text))" - , " AND reserved_by IS NULL" + , "WHERE" + , " reserved_by IS NULL" , " AND run_at IS NOT NULL" , " AND run_at <= " now , " ORDER BY run_at" , "LIMIT" limit - -- Use SKIP LOCKED if available. Otherwise utilise - -- advisory locks. - , if useSkipLocked - then "FOR UPDATE SKIP LOCKED" - else "FOR UPDATE" + , "FOR UPDATE SKIP LOCKED" ] -- | Spawn each job in a separate thread. @@ -345,56 +332,65 @@ updateJobs :: [(idx, Result)] -> m () updateJobs results = runDBT cs ts $ do now <- currentTime - runSQL_ $ smconcat [ - "WITH removed AS (" - , " DELETE FROM" <+> raw ccJobsTable - , " WHERE id = ANY(" Array1 deletes <+> ")" - , ")" - , "UPDATE" <+> raw ccJobsTable <+> "SET" - , " reserved_by = NULL" - , ", run_at = CASE" - , " WHEN FALSE THEN run_at" - , smconcat $ M.foldrWithKey (retryToSQL now) [] retries - , " ELSE NULL" -- processed - , " END" - , ", finished_at = CASE" - , " WHEN id = ANY(" Array1 successes <+> ") THEN " now - , " ELSE NULL" - , " END" - , "WHERE id = ANY(" Array1 (map fst updates) <+> ")" - ] - where - retryToSQL now (Left int) ids = - ("WHEN id = ANY(" Array1 ids <+> ") THEN " now <> " +" int :) - retryToSQL _ (Right time) ids = - ("WHEN id = ANY(" Array1 ids <+> ") THEN" time :) + runSQL_ $ updateJobsQuery ccJobsTable results now - retries = foldr step M.empty $ map f updates - where - f (idx, result) = case result of - Ok action -> (idx, action) - Failed action -> (idx, action) +---------------------------------------- - step (idx, action) iretries = case action of - MarkProcessed -> iretries - RerunAfter int -> M.insertWith (++) (Left int) [idx] iretries - RerunAt time -> M.insertWith (++) (Right time) [idx] iretries - Remove -> error - "updateJobs: Remove should've been filtered out" +-- | Generate a single SQL query for updating all given jobs. +updateJobsQuery + :: (Show idx, ToSQL idx) + => RawSQL () + -> [(idx, Result)] + -> UTCTime + -> SQL +updateJobsQuery jobsTable results now = smconcat + [ "WITH removed AS (" + , " DELETE FROM" <+> raw jobsTable + , " WHERE id = ANY(" Array1 deletes <+> ")" + , ")" + , "UPDATE" <+> raw jobsTable <+> "SET" + , " reserved_by = NULL" + , ", run_at = CASE" + , " WHEN FALSE THEN run_at" + , smconcat $ M.foldrWithKey retryToSQL [] retries + , " ELSE NULL" -- processed + , " END" + , ", finished_at = CASE" + , " WHEN id = ANY(" Array1 successes <+> ") THEN " now + , " ELSE NULL" + , " END" + , "WHERE id = ANY(" Array1 (map fst updates) <+> ")" + ] + where + retryToSQL (Left int) ids = + ("WHEN id = ANY(" Array1 ids <+> ") THEN " now <> " +" int :) + retryToSQL (Right time) ids = + ("WHEN id = ANY(" Array1 ids <+> ") THEN" time :) - successes = foldr step [] updates - where - step (idx, Ok _) acc = idx : acc - step (_, Failed _) acc = acc + retries = foldr step M.empty $ map getAction updates + where + getAction (idx, result) = case result of + Ok action -> (idx, action) + Failed action -> (idx, action) - (deletes, updates) = foldr step ([], []) results - where - step job@(idx, result) (ideletes, iupdates) = case result of - Ok Remove -> (idx : ideletes, iupdates) - Failed Remove -> (idx : ideletes, iupdates) - _ -> (ideletes, job : iupdates) + step (idx, action) iretries = case action of + MarkProcessed -> iretries + RerunAfter int -> M.insertWith (++) (Left int) [idx] iretries + RerunAt time -> M.insertWith (++) (Right time) [idx] iretries + Remove -> error "updateJobs: Remove should've been filtered out" ----------------------------------------- + successes = foldr step [] updates + where + step (idx, Ok _) acc = idx : acc + step (_, Failed _) acc = acc + + (deletes, updates) = foldr step ([], []) results + where + step job@(idx, result) (ideletes, iupdates) = case result of + Ok Remove -> (idx : ideletes, iupdates) + Failed Remove -> (idx : ideletes, iupdates) + _ -> (ideletes, job : iupdates) + ts :: TransactionSettings ts = defaultTransactionSettings { diff --git a/test/Test.hs b/test/Test.hs --- a/test/Test.hs +++ b/test/Test.hs @@ -86,7 +86,7 @@ { csConnInfo = connString } ConnectionSource connSource = simpleSource connSettings - withSimpleStdOutLogger $ \logger -> + withStdOutLogger $ \logger -> runTestEnv connSource logger $ do createTables idleSignal <- liftIO $ atomically $ newEmptyTMVar