diff --git a/CHANGELOG.md b/CHANGELOG.md --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,19 @@ # Revision history for pgmq-config +## 0.2.0.0 -- 2026-04-23 + +Version unified with the rest of the pgmq-hs packages; no pgmq-config +0.1.4.0 was published to Hackage. + +### Bug Fixes + +* `ensureQueues` and `ensureQueuesEff` are now truly idempotent. Previously they + issued `pgmq.create`, `pgmq.enable_notify_insert`, `pgmq.create_fifo_index`, and + `pgmq.bind_topic` unconditionally on every call, which worked by accident for + standard/unlogged queues but caused trigger recreation on every boot and failed outright + for partitioned queues (because `pg_partman.create_parent` raises on re-registration). + Both entry points now route through the existing state-checking reconciliation path. + ## 0.1.3.0 -- 2026-03-12 * Initial release diff --git a/pgmq-config.cabal b/pgmq-config.cabal --- a/pgmq-config.cabal +++ b/pgmq-config.cabal @@ -1,6 +1,6 @@ cabal-version: 3.4 name: pgmq-config -version: 0.1.3.0 +version: 0.2.0.0 synopsis: Declarative queue configuration for PGMQ (PostgreSQL Message Queue) @@ -54,15 +54,15 @@ , hasql ^>=1.10 , hasql-pool ^>=1.4 , lens ^>=5.3 - , pgmq-core >=0.1.3 && <0.2 - , pgmq-hasql >=0.1.3 && <0.2 + , pgmq-core >=0.2 && <0.3 + , pgmq-hasql >=0.2 && <0.3 , text ^>=2.1 if flag(effectful) exposed-modules: Pgmq.Config.Effectful build-depends: - , effectful-core ^>=2.5 || ^>=2.6 - , pgmq-effectful >=0.1.3 && <0.2 + , effectful-core ^>=2.5 || ^>=2.6 + , pgmq-effectful >=0.2 && <0.3 hs-source-dirs: src default-language: GHC2024 @@ -102,3 +102,4 @@ , tasty ^>=1.5 , tasty-hunit ^>=0.10 , text + , time ^>=1.14 diff --git a/src/Pgmq/Config.hs b/src/Pgmq/Config.hs --- a/src/Pgmq/Config.hs +++ b/src/Pgmq/Config.hs @@ -26,7 +26,6 @@ where import Control.Lens ((^.)) -import Data.Foldable (for_) import Data.Generics.Labels () import Data.Set qualified as Set import Data.Text qualified as T @@ -43,11 +42,15 @@ ) -- | Ensure all declared queues exist with the desired settings. --- This is idempotent — safe to call on every application startup. +-- +-- Queries existing queues, topic bindings, and notification throttles first, +-- and only issues mutating calls for items that are missing. Safe to call on +-- every application startup: a second run on an unchanged config is a no-op +-- modulo the three list queries. +-- -- Operations are additive only: queues not in the config are left untouched. ensureQueues :: [QueueConfig] -> Session () -ensureQueues configs = - for_ configs applyQueueConfig +ensureQueues configs = () <$ ensureQueuesReport configs -- | Convenience wrapper that runs 'ensureQueues' against a connection pool. ensureQueuesWithPool :: Pool.Pool -> [QueueConfig] -> IO (Either Pool.UsageError ()) @@ -71,45 +74,6 @@ existingNotifySet = Set.fromList (map (\t -> t ^. #throttleQueueName) existingThrottles) concat <$> traverse (reconcileQueue existingQueueNames existingBindingSet existingNotifySet) configs - --- | Apply a single queue config without checking existing state. -applyQueueConfig :: QueueConfig -> Session () -applyQueueConfig cfg = do - let qn = cfg ^. #queueName - -- Create the queue - case cfg ^. #queueType of - StandardQueue -> - Sessions.createQueue qn - UnloggedQueue -> - Sessions.createUnloggedQueue qn - PartitionedQueue pc -> - Sessions.createPartitionedQueue - StmtTypes.CreatePartitionedQueue - { queueName = qn, - partitionInterval = pc ^. #partitionInterval, - retentionInterval = pc ^. #retentionInterval - } - - -- Enable notifications if configured - for_ (cfg ^. #notifyInsert) $ \nc -> - Sessions.enableNotifyInsert - StmtTypes.EnableNotifyInsert - { queueName = qn, - throttleIntervalMs = nc ^. #throttleMs - } - - -- Create FIFO index if requested - if cfg ^. #fifoIndex - then Sessions.createFifoIndex qn - else pure () - - -- Bind topic patterns - for_ (cfg ^. #topicBindings) $ \pat -> - Sessions.bindTopic - StmtTypes.BindTopic - { topicPattern = pat, - queueName = qn - } -- | Reconcile a single queue config against existing state, returning actions taken. reconcileQueue :: diff --git a/src/Pgmq/Config/Effectful.hs b/src/Pgmq/Config/Effectful.hs --- a/src/Pgmq/Config/Effectful.hs +++ b/src/Pgmq/Config/Effectful.hs @@ -23,9 +23,15 @@ ) -- | Ensure all declared queues exist using the Pgmq effect. +-- +-- Queries existing queues, topic bindings, and notification throttles first, +-- and only issues mutating calls for items that are missing. Safe to call on +-- every application startup: a second run on an unchanged config is a no-op +-- modulo the three list queries. +-- +-- Operations are additive only: queues not in the config are left untouched. ensureQueuesEff :: (Eff.Pgmq :> es) => [QueueConfig] -> Eff es () -ensureQueuesEff configs = - mapM_ applyQueueConfigEff configs +ensureQueuesEff configs = () <$ ensureQueuesReportEff configs -- | Like 'ensureQueuesEff', but returns a report of actions taken. ensureQueuesReportEff :: (Eff.Pgmq :> es) => [QueueConfig] -> Eff es [ReconcileAction] @@ -43,46 +49,6 @@ existingNotifySet = Set.fromList (map (\t -> t ^. #throttleQueueName) existingThrottles) concat <$> traverse (reconcileQueueEff existingQueueNames existingBindingSet existingNotifySet) configs - --- | Apply a single queue config without checking existing state. -applyQueueConfigEff :: (Eff.Pgmq :> es) => QueueConfig -> Eff es () -applyQueueConfigEff cfg = do - let qn = cfg ^. #queueName - case cfg ^. #queueType of - StandardQueue -> - Eff.createQueue qn - UnloggedQueue -> - Eff.createUnloggedQueue qn - PartitionedQueue pc -> - Eff.createPartitionedQueue - StmtTypes.CreatePartitionedQueue - { queueName = qn, - partitionInterval = pc ^. #partitionInterval, - retentionInterval = pc ^. #retentionInterval - } - - case cfg ^. #notifyInsert of - Nothing -> pure () - Just nc -> - Eff.enableNotifyInsert - StmtTypes.EnableNotifyInsert - { queueName = qn, - throttleIntervalMs = nc ^. #throttleMs - } - - if cfg ^. #fifoIndex - then Eff.createFifoIndex qn - else pure () - - mapM_ - ( \pat -> - Eff.bindTopic - StmtTypes.BindTopic - { topicPattern = pat, - queueName = qn - } - ) - (cfg ^. #topicBindings) -- | Reconcile a single queue config against existing state. reconcileQueueEff :: diff --git a/test/ConfigSpec.hs b/test/ConfigSpec.hs --- a/test/ConfigSpec.hs +++ b/test/ConfigSpec.hs @@ -8,9 +8,13 @@ import Control.Lens ((^.)) import Data.Generics.Labels () import Data.Text qualified as T +import Data.Time (UTCTime) import Data.Word (Word32) +import Hasql.Decoders qualified as D +import Hasql.Encoders qualified as E import Hasql.Pool qualified as Pool import Hasql.Session qualified +import Hasql.Statement (Statement, preparable) import Pgmq.Config import Pgmq.Hasql.Sessions qualified as Sessions import Pgmq.Types (QueueName, parseQueueName, parseTopicPattern, queueNameToText, topicPatternToText) @@ -28,7 +32,10 @@ testEnsureQueuesIncremental pool, testEnsureQueuesWithNotify pool, testEnsureQueuesWithFifo pool, - testEnsureQueuesWithTopicBinding pool + testEnsureQueuesWithTopicBinding pool, + testEnsureQueuesIsTrulyIdempotent pool, + testEnsureQueuesSilentIdempotent pool, + testEnsureQueuesSilentIncremental pool ] -- | Helper to run a session and fail on error @@ -202,3 +209,87 @@ actionForQueue qn (SkippedFifoIndex q) = q == qn actionForQueue qn (BoundTopic q _) = q == qn actionForQueue qn (SkippedTopicBinding q _) = q == qn + +-- | Silent-variant version of 'testEnsureQueuesIdempotent': call 'ensureQueues' +-- twice with the same standard-queue config and confirm the queue exists exactly +-- once afterwards. +testEnsureQueuesSilentIdempotent :: Pool.Pool -> TestTree +testEnsureQueuesSilentIdempotent pool = testCase "silent ensureQueues is idempotent" $ do + qn <- genQueueName + let configs = [standardQueue qn] + runSession pool (ensureQueues configs) + runSession pool (ensureQueues configs) + queues <- runSession pool Sessions.listQueues + let matching = filter (\q -> (q ^. #name) == qn) queues + length matching @?= 1 + cleanupQueue pool qn + +-- | Silent-variant version of 'testEnsureQueuesIncremental': after an initial +-- single-queue reconcile, a second reconcile that adds a new queue must leave +-- both queues in place. +testEnsureQueuesSilentIncremental :: Pool.Pool -> TestTree +testEnsureQueuesSilentIncremental pool = testCase "silent ensureQueues adds new queues incrementally" $ do + qn1 <- genQueueName + qn2 <- genQueueName + runSession pool (ensureQueues [standardQueue qn1]) + runSession pool (ensureQueues [standardQueue qn1, standardQueue qn2]) + queues <- runSession pool Sessions.listQueues + let names = map (^. #name) queues + assertBool "qn1 should exist" (qn1 `elem` names) + assertBool "qn2 should exist" (qn2 `elem` names) + cleanupQueue pool qn1 + cleanupQueue pool qn2 + +-- | Assert that calling the silent 'ensureQueues' a second time for a queue with +-- notify-insert configured does NOT reset 'last_notified_at' to the epoch. +-- +-- Under the pre-fix behaviour, 'ensureQueues' unconditionally calls +-- 'pgmq.enable_notify_insert', which first runs 'pgmq.disable_notify_insert' +-- (DELETE FROM pgmq.notify_insert_throttle) and then re-inserts the row with +-- the default 'last_notified_at = to_timestamp(0)'. After the fix, the second +-- call queries existing state and skips the notify mutation entirely, so a +-- timestamp bumped between the two calls is preserved. +testEnsureQueuesIsTrulyIdempotent :: Pool.Pool -> TestTree +testEnsureQueuesIsTrulyIdempotent pool = testCase "ensureQueues is truly idempotent for notify" $ do + qn <- genQueueName + let configs = [withNotifyInsert (Just 500) (standardQueue qn)] + -- First run creates the queue and enables notify. Fresh throttle row has + -- last_notified_at = to_timestamp(0) (the table default). + runSession pool (ensureQueues configs) + -- Bump last_notified_at to a recent timestamp so a reset would be observable. + bumped <- runSession pool (Hasql.Session.statement (queueNameToText qn) bumpLastNotifiedAtStmt) + assertBool "bumped timestamp should be post-epoch" (bumped > epochUtc) + -- Second run must be a no-op for notify (i.e., must not rewrite the row). + runSession pool (ensureQueues configs) + after <- runSession pool (Hasql.Session.statement (queueNameToText qn) pgNotifyLastAt) + assertBool + ( "last_notified_at must not be reset by second ensureQueues; was " + <> show after + ) + (after > epochUtc) + after @?= bumped + cleanupQueue pool qn + +-- | Read 'last_notified_at' for a queue. One-off statement for tests only. +pgNotifyLastAt :: Statement T.Text UTCTime +pgNotifyLastAt = preparable sql encoder decoder + where + sql = "SELECT last_notified_at FROM pgmq.notify_insert_throttle WHERE queue_name = $1" + encoder = E.param (E.nonNullable E.text) + decoder = D.singleRow (D.column (D.nonNullable D.timestamptz)) + +-- | Bump 'last_notified_at' to clock_timestamp() and return the new value. +-- Used to observe whether a subsequent ensureQueues resets it. +bumpLastNotifiedAtStmt :: Statement T.Text UTCTime +bumpLastNotifiedAtStmt = preparable sql encoder decoder + where + sql = + "UPDATE pgmq.notify_insert_throttle \ + \SET last_notified_at = clock_timestamp() \ + \WHERE queue_name = $1 RETURNING last_notified_at" + encoder = E.param (E.nonNullable E.text) + decoder = D.singleRow (D.column (D.nonNullable D.timestamptz)) + +-- | Postgres epoch as a UTCTime, i.e. 'to_timestamp(0)'. +epochUtc :: UTCTime +epochUtc = read "1970-01-01 00:00:00 UTC"