packages feed

pgmq-config 0.1.3.0 → 0.2.0.0

raw patch · 5 files changed

+127/−91 lines, 5 filesdep +timedep ~pgmq-coredep ~pgmq-effectfuldep ~pgmq-hasqlPVP ok

version bump matches the API change (PVP)

Dependencies added: time

Dependency ranges changed: pgmq-core, pgmq-effectful, pgmq-hasql

API changes (from Hackage documentation)

Files

CHANGELOG.md view
@@ -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
pgmq-config.cabal view
@@ -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
src/Pgmq/Config.hs view
@@ -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 ::
src/Pgmq/Config/Effectful.hs view
@@ -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 ::
test/ConfigSpec.hs view
@@ -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"