diff --git a/.github/workflows/haskell-ci.yml b/.github/workflows/haskell-ci.yml index 862fa02..85952c7 100644 --- a/.github/workflows/haskell-ci.yml +++ b/.github/workflows/haskell-ci.yml @@ -6,11 +6,11 @@ # # haskell-ci regenerate # -# For more information, see https://github.com/haskell-CI/haskell-ci +# For more information, see https://github.com/arybczak/haskell-ci # -# version: 0.19.20260104 +# version: 0.19.20260424 # -# REGENDATA ("0.19.20260104",["github","cabal.project","--config=cabal.haskell-ci"]) +# REGENDATA ("0.19.20260424",["github","cabal.project","--config=cabal.haskell-ci"]) # name: Haskell-CI on: @@ -23,6 +23,11 @@ on: merge_group: branches: - master + workflow_dispatch: + {} +concurrency: + group: ${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true jobs: linux: name: Haskell-CI - Linux - ${{ matrix.compiler }} @@ -30,10 +35,10 @@ jobs: timeout-minutes: 60 container: - image: buildpack-deps:jammy + image: buildpack-deps:resolute services: postgres: - image: postgres:14 + image: postgres:18 env: POSTGRES_PASSWORD: postgres options: --health-cmd pg_isready --health-interval 10s --health-timeout 5s --health-retries 5 @@ -46,9 +51,9 @@ jobs: compilerVersion: 9.14.1 setup-method: ghcup allow-failure: false - - compiler: ghc-9.12.2 + - compiler: ghc-9.12.4 compilerKind: ghc - compilerVersion: 9.12.2 + compilerVersion: 9.12.4 setup-method: ghcup allow-failure: false - compiler: ghc-9.10.3 @@ -66,22 +71,12 @@ jobs: compilerVersion: 9.6.7 setup-method: ghcup allow-failure: false - - compiler: ghc-9.4.8 - compilerKind: ghc - compilerVersion: 9.4.8 - setup-method: ghcup - allow-failure: false - - compiler: ghc-9.2.8 - compilerKind: ghc - compilerVersion: 9.2.8 - setup-method: ghcup - allow-failure: false fail-fast: false steps: - name: apt-get install run: | apt-get update - apt-get install -y --no-install-recommends gnupg ca-certificates dirmngr curl git software-properties-common libtinfo5 + apt-get install -y --no-install-recommends gnupg ca-certificates dirmngr curl git software-properties-common - name: Install GHCup run: | mkdir -p "$HOME/.ghcup/bin" @@ -166,7 +161,7 @@ jobs: chmod a+x $HOME/.cabal/bin/cabal-plan cabal-plan --version - name: checkout - uses: actions/checkout@v5 + uses: actions/checkout@v7 with: path: source - name: initial cabal.project for sdist @@ -198,23 +193,24 @@ jobs: echo " ghc-options: -Werror=missing-methods -Werror=missing-fields" >> cabal.project echo "package consumers-metrics-prometheus" >> cabal.project echo " ghc-options: -Werror=missing-methods -Werror=missing-fields" >> cabal.project - if [ $((HCNUMVER >= 90400)) -ne 0 ] ; then echo "package consumers" >> cabal.project ; fi - if [ $((HCNUMVER >= 90400)) -ne 0 ] ; then echo " ghc-options: -Werror=unused-packages" >> cabal.project ; fi - if [ $((HCNUMVER >= 90400)) -ne 0 ] ; then echo "package consumers-metrics-prometheus" >> cabal.project ; fi - if [ $((HCNUMVER >= 90400)) -ne 0 ] ; then echo " ghc-options: -Werror=unused-packages" >> cabal.project ; fi + echo "package consumers" >> cabal.project + echo " ghc-options: -Werror=unused-packages" >> cabal.project + echo "package consumers-metrics-prometheus" >> cabal.project + echo " ghc-options: -Werror=unused-packages" >> cabal.project echo "package consumers" >> cabal.project echo " ghc-options: -Werror=incomplete-patterns -Werror=incomplete-uni-patterns" >> cabal.project echo "package consumers-metrics-prometheus" >> cabal.project echo " ghc-options: -Werror=incomplete-patterns -Werror=incomplete-uni-patterns" >> cabal.project cat >> cabal.project <, copyright: Scrive AB category: Concurrency, Database build-type: Simple -tested-with: GHC == { 9.2.8, 9.4.8, 9.6.7, 9.8.4, 9.10.3, 9.12.2, 9.14.1 } +tested-with: GHC == { 9.6.7, 9.8.4, 9.10.3, 9.12.4, 9.14.1 } bug-reports: https://github.com/scrive/consumers/issues source-repository head @@ -42,10 +42,10 @@ library exposed-modules: Database.PostgreSQL.Consumers.Instrumented - build-depends: base >= 4.16 && < 5 - , consumers >= 2.3 + build-depends: base >= 4.18 && < 5 + , consumers >= 3.0 , exceptions >= 0.10 - , hpqtypes >= 1.13 + , hpqtypes >= 2.0 , lifted-base >= 0.2 , log-base >= 0.11 , monad-control >= 1.0 diff --git a/consumers-metrics-prometheus/src/Database/PostgreSQL/Consumers/Instrumented.hs b/consumers-metrics-prometheus/src/Database/PostgreSQL/Consumers/Instrumented.hs index 3a6a357..7ab0423 100644 --- a/consumers-metrics-prometheus/src/Database/PostgreSQL/Consumers/Instrumented.hs +++ b/consumers-metrics-prometheus/src/Database/PostgreSQL/Consumers/Instrumented.hs @@ -213,7 +213,7 @@ collectMetrics connSource ConsumerMetrics {..} ConsumerConfig {ccJobsTable, ccCo "SELECT count(id)::float8 FROM " <> raw ccConsumersTable <> " WHERE name =" unRawSQL ccJobsTable - fetchOne runIdentity + fetchOne fromSQL liftBase $ Prom.withLabel jobInfo jobName (`Prom.setGauge` info) overdue <- do @@ -221,7 +221,7 @@ collectMetrics connSource ConsumerMetrics {..} ConsumerConfig {ccJobsTable, ccCo "SELECT count(id)::float8 FROM " <> raw ccJobsTable <> " WHERE run_at <= now() AND reserved_by IS NULL" - fetchOne runIdentity + fetchOne fromSQL liftBase $ Prom.withLabel jobsOverdue jobName (`Prom.setGauge` overdue) -- | Alter a configuration to collect "job" metrics on 'ccProcessJob' diff --git a/consumers/consumers.cabal b/consumers/consumers.cabal index b9a3e96..d63ccb5 100644 --- a/consumers/consumers.cabal +++ b/consumers/consumers.cabal @@ -1,6 +1,6 @@ cabal-version: 3.0 name: consumers -version: 2.3.5.0 +version: 3.0.0.0 synopsis: Concurrent PostgreSQL data consumers description: Library for setting up concurrent consumers of data @@ -17,7 +17,7 @@ maintainer: Andrzej Rybczak , copyright: Scrive AB category: Concurrency, Database build-type: Simple -tested-with: GHC == { 9.2.8, 9.4.8, 9.6.7, 9.8.4, 9.10.3, 9.12.2, 9.14.1 } +tested-with: GHC == { 9.6.7, 9.8.4, 9.10.3, 9.12.4, 9.14.1 } bug-reports: https://github.com/scrive/consumers/issues source-repository head @@ -53,11 +53,11 @@ library Database.PostgreSQL.Consumers.Components, Database.PostgreSQL.Consumers.Utils - build-depends: base >= 4.16 && < 5 + build-depends: base >= 4.18 && < 5 , aeson >= 2.0 , containers >= 0.5 , exceptions >= 0.10 - , hpqtypes >= 1.13 + , hpqtypes >= 2.0 , lifted-base >= 0.2 , lifted-threads >= 1.0 , log-base >= 0.11 @@ -78,7 +78,7 @@ test-suite consumers-example build-depends: base, consumers, hpqtypes, - hpqtypes-extras >= 1.17.0.1, + hpqtypes-extras >= 2.0, log-base, text @@ -99,7 +99,7 @@ test-suite consumers-test exceptions, HUnit, hpqtypes, - hpqtypes-extras >= 1.19.0.0, + hpqtypes-extras >= 2.0, log-base, monad-control, monad-time, diff --git a/consumers/example/Example.hs b/consumers/example/Example.hs index 47f563a..99b6636 100644 --- a/consumers/example/Example.hs +++ b/consumers/example/Example.hs @@ -102,7 +102,7 @@ main = do { ccJobsTable = "consumers_example_jobs" , ccConsumersTable = "consumers_example_consumers" , ccJobSelectors = ["id", "message"] - , ccJobFetcher = id + , ccJobFetcher = (,) <$> fromSQL <*> fromSQL , ccJobIndex = \(i :: Int64, _msg :: T.Text) -> i , ccNotificationChannel = Just "consumers_example_chan" , ccNotificationTimeout = 10 * 1000000 -- 10 sec diff --git a/consumers/src/Database/PostgreSQL/Consumers/Components.hs b/consumers/src/Database/PostgreSQL/Consumers/Components.hs index 5302fad..2a23dcb 100644 --- a/consumers/src/Database/PostgreSQL/Consumers/Components.hs +++ b/consumers/src/Database/PostgreSQL/Consumers/Components.hs @@ -18,7 +18,6 @@ import Control.Monad.Catch import Control.Monad.Time import Control.Monad.Trans import Control.Monad.Trans.Control -import Data.Foldable qualified as F import Data.Function import Data.Int import Data.Map.Strict qualified as M @@ -255,7 +254,7 @@ spawnMonitor ConsumerConfig {..} cs cid = forkP "monitor" . forever $ do , " AND name =" unRawSQL ccJobsTable , "FOR UPDATE SKIP LOCKED" ] - fetchMany (runIdentity @Int64) >>= \case + fetchMany (fromSQL @Int64) >>= \case [] -> pure (0, []) inactive -> do -- Fetch all stuck jobs and run ccOnException on them to determine @@ -265,7 +264,7 @@ spawnMonitor ConsumerConfig {..} cs cid = forkP "monitor" . forever $ do smconcat [ "SELECT" <+> mintercalate ", " ccJobSelectors , "FROM" <+> raw ccJobsTable - , "WHERE reserved_by = ANY(" Array1 inactive <+> ")" + , "WHERE reserved_by = ANY(" inactive <+> ")" , "FOR UPDATE SKIP LOCKED" ] stuckJobs <- fetchMany ccJobFetcher @@ -277,7 +276,7 @@ spawnMonitor ConsumerConfig {..} cs cid = forkP "monitor" . forever $ do runPreparedSQL_ (preparedSqlName "removeInactive" ccConsumersTable) $ smconcat [ "DELETE FROM" <+> raw ccConsumersTable - , "WHERE id = ANY(" Array1 inactive <+> ")" + , "WHERE id = ANY(" inactive <+> ")" ] pure (length inactive, map ccJobIndex stuckJobs) when (inactiveConsumers > 0) $ do @@ -372,8 +371,7 @@ spawnDispatcher ConsumerConfig {..} cs cid semaphore runningJobsInfo runningJobs , "WHERE id IN (" <> reservedJobs now <> ")" , "RETURNING" <+> mintercalate ", " ccJobSelectors ] - -- Decode lazily as we want the transaction to be as short as possible. - (,n) . F.toList . fmap ccJobFetcher <$> queryResult + (,n) <$> fetchMany ccJobFetcher where reservedJobs :: UTCTime -> SQL reservedJobs now = @@ -433,7 +431,7 @@ updateJobsQuery jobsTable results now = smconcat [ "WITH removed AS (" , " DELETE FROM" <+> raw jobsTable - , " WHERE id = ANY(" Array1 deletes <+> ")" + , " WHERE id = ANY(" deletes <+> ")" , ")" , "UPDATE" <+> raw jobsTable <+> "SET" , " reserved_by = NULL" @@ -443,16 +441,16 @@ updateJobsQuery jobsTable results now = , " ELSE NULL" -- processed , " END" , ", finished_at = CASE" - , " WHEN id = ANY(" Array1 successes <+> ") THEN " now + , " WHEN id = ANY(" successes <+> ") THEN " now , " ELSE NULL" , " END" - , "WHERE id = ANY(" Array1 (map fst updates) <+> ")" + , "WHERE id = ANY(" map fst updates <+> ")" ] where retryToSQL (Left int) ids = - ("WHEN id = ANY(" Array1 ids <+> ") THEN " now <> " +" int :) + ("WHEN id = ANY(" ids <+> ") THEN " now <> " +" int :) retryToSQL (Right time) ids = - ("WHEN id = ANY(" Array1 ids <+> ") THEN" time :) + ("WHEN id = ANY(" ids <+> ") THEN" time :) retries = foldr (step . getAction) M.empty updates where diff --git a/consumers/src/Database/PostgreSQL/Consumers/Config.hs b/consumers/src/Database/PostgreSQL/Consumers/Config.hs index b759594..f364c3c 100644 --- a/consumers/src/Database/PostgreSQL/Consumers/Config.hs +++ b/consumers/src/Database/PostgreSQL/Consumers/Config.hs @@ -8,7 +8,7 @@ module Database.PostgreSQL.Consumers.Config import Control.Exception (SomeException) import Data.Aeson.Types qualified as A import Data.Time -import Database.PostgreSQL.PQTypes.FromRow +import Database.PostgreSQL.PQTypes.FromSQL import Database.PostgreSQL.PQTypes.Interval import Database.PostgreSQL.PQTypes.Notification import Database.PostgreSQL.PQTypes.SQL @@ -27,7 +27,7 @@ data Result = Ok Action | Failed Action deriving (Eq, Ord, Show) -- | Config of a consumer. -data ConsumerConfig m idx job = forall row. FromRow row => ConsumerConfig +data ConsumerConfig m idx job = ConsumerConfig { ccJobsTable :: !(RawSQL ()) -- ^ 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: @@ -79,8 +79,8 @@ data ConsumerConfig m idx job = forall row. FromRow row => ConsumerConfig , ccJobSelectors :: ![SQL] -- ^ Fields needed to be selected from the jobs table in order to assemble a -- job. - , ccJobFetcher :: !(row -> job) - -- ^ Function that transforms the list of fields into a job. + , ccJobFetcher :: !(RowDecoder job) + -- ^ Decoder that transforms the selected fields into a job. , ccJobIndex :: !(job -> idx) -- ^ Selector for taking out job ID from the job object. , ccNotificationChannel :: !(Maybe Channel) diff --git a/consumers/src/Database/PostgreSQL/Consumers/Consumer.hs b/consumers/src/Database/PostgreSQL/Consumers/Consumer.hs index a836db9..07abc75 100644 --- a/consumers/src/Database/PostgreSQL/Consumers/Consumer.hs +++ b/consumers/src/Database/PostgreSQL/Consumers/Consumer.hs @@ -15,16 +15,7 @@ import Database.PostgreSQL.PQTypes -- | ID of a consumer. newtype ConsumerID = ConsumerID Int64 - deriving (Eq, Ord) - -instance PQFormat ConsumerID where - pqFormat = pqFormat @Int64 -instance FromSQL ConsumerID where - type PQBase ConsumerID = PQBase Int64 - fromSQL mbase = ConsumerID <$> fromSQL mbase -instance ToSQL ConsumerID where - type PQDest ConsumerID = PQDest Int64 - toSQL (ConsumerID n) = toSQL n + deriving (Eq, Ord, PQFormat, FromSQL, ToSQL) instance Show ConsumerID where showsPrec p (ConsumerID n) = showsPrec p n @@ -44,7 +35,7 @@ registerConsumer ConsumerConfig {..} cs = runDBT cs defaultTransactionSettings $ , "(name, last_activity) VALUES (" unRawSQL ccJobsTable <> ", " now <> ")" , "RETURNING id" ] - fetchOne runIdentity + fetchOne fromSQL -- | Unregister consumer with a given ID. unregisterConsumer diff --git a/consumers/test/Test.hs b/consumers/test/Test.hs index fd2b61d..8d0fd07 100644 --- a/consumers/test/Test.hs +++ b/consumers/test/Test.hs @@ -89,7 +89,7 @@ test = do -- Each job creates 2 new jobs, so there should be 1024 jobs in table. runSQL_ "SELECT COUNT(*) from consumers_test_jobs" - rowcount0 :: Int64 <- fetchOne runIdentity + rowcount0 :: Int64 <- fetchOne fromSQL -- Move time 2 hours forward modifyTestTime $ addUTCTime (2 * 60 * 60) finalize @@ -100,7 +100,7 @@ test = do waitUntilTrue idleSignal -- Jobs are designed to double only 10 times, so there should be no jobs left now. runSQL_ "SELECT COUNT(*) from consumers_test_jobs" - rowcount1 :: Int64 <- fetchOne runIdentity + rowcount1 :: Int64 <- fetchOne fromSQL liftIO $ T.assertEqual "Number of jobs in table after 10 steps is 1024" 1024 rowcount0 liftIO $ T.assertEqual "Number of jobs in table after 11 steps is 0" 0 rowcount1 dropTables @@ -145,7 +145,7 @@ test = do { ccJobsTable = "consumers_test_jobs" , ccConsumersTable = "consumers_test_consumers" , ccJobSelectors = ["id", "countdown"] - , ccJobFetcher = id + , ccJobFetcher = (,) <$> fromSQL <*> fromSQL , ccJobIndex = \(i :: Int64, _ :: Int32) -> i , ccNotificationChannel = Just "consumers_test_chan" , -- select some small timeout